[logdog] Replace Tumble with push queues.

logdog/appengine/coordinator/endpoints/services/terminateStream_test.go

 // Copyright 2015 The LUCI Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
 // You may obtain a copy of the License at
 //
 //      http://www.apache.org/licenses/LICENSE-2.0
 //
 // Unless required by applicable law or agreed to in writing, software
 // distributed under the License is distributed on an "AS IS" BASIS,
@@ skipping 91 matching lines @@
                                         })
                                         So(tls.State.TerminalIndex, ShouldEqual, 1337)
                                         So(tls.State.Terminated(), ShouldBeTrue)
                                         So(tls.State.ArchivalState(), ShouldEqual, coordinator.NotArchived)
 
                                         Convey(`Replaces the pessimistic archival mutation with an optimistic one.`, func() {
                                                 // We have replaced the pessimistic archival mutation with an
                                                 // optimistic one. Assert that this happened by advancing time by
                                                 // the optimistic period and confirming the published archival
                                                 // request.
-                                                env.IterateTumbleAll(c)
+                                                env.RunTaskQueues(c, tls)
                                                 So(env.ArchivalPublisher.Hashes(), ShouldResemble, []string{})
 
                                                 // Add our settle delay, confirm that archival is scheduled.
                                                 env.Clock.Add(10 * time.Second)
-                                                env.IterateTumbleAll(c)
+                                                env.RunTaskQueues(c, tls)
                                                 So(env.ArchivalPublisher.Hashes(), ShouldResemble, []string{string(tls.Stream.ID)})
 
                                                 // Add our pessimistic delay, confirm that no additional tasks
                                                 // are scheduled (because pessimistic was replaced).
                                                 env.ArchivalPublisher.Clear()
                                                 env.Clock.Add(time.Hour)
-                                                env.IterateTumbleAll(c)
+                                                env.RunTaskQueues(c, tls)
                                                 So(env.ArchivalPublisher.Hashes(), ShouldResemble, []string{})
                                         })
 
                                         Convey(`Can be marked terminal again (idempotent).`, func() {
                                                 _, err := svr.TerminateStream(c, &req)
                                                 So(err, ShouldBeRPCOK)
 
                                                 // Reload state and confirm.
                                                 So(tls.Get(c), ShouldBeNil)
 
@@ skipping 98 matching lines @@
                                 })
 
                                 So(getParams(), ShouldResemble, &coordinator.ArchivalParams{
                                         SettleDelay:    10 * time.Second,
                                         CompletePeriod: 0,
                                 })
                         })
                 })
         })
 }