thread.blockUntil.received
        server.stop()
        client.stop()

        then:
        1 * service.doStuff("1")
        1 * service.doStuff("2")
        1 * service.doStuff("3")
        1 * service.doStuff("4") >> { instant.received }

        cleanup:
        client?.stop()
        server?.stop()
    }

    def "client can send messages to server"() {

		t.Error("close stop, method exit timeout.")
	case <-signal:
		t.Log("queue return.")
	}
}

func TestClosed(t *testing.T) {
	t.Run("immediate close", func(t *testing.T) {
		stop := make(chan struct{})
		q := NewQueue(0)
		go q.Run(stop)
		close(stop)
		if err := WaitForClose(q, 10*time.Second); err != nil {
			t.Error(err)
		}
	})
	t.Run("no tasks after close", func(t *testing.T) {

        1 * asyncProcessor1.startProcessing(asyncResultProcessor)
        1 * asyncProcessor1.processTestClass(test)

        when:
        processor.stop()

        then:
        1 * asyncProcessor1.stop()
        1 * actor1.stop()
        1 * resultProcessorActor.stop()
    }

    def startsMultipleProcessorsOnDemandAndStopsAtEnd() {
        TestClassRunInfo test = Mock()
        TestClassProcessor processor1 = Mock()

        def result1 = worker.run("value")
        def result2 = worker.run("value")
        def result3 = worker.run("value")
        worker.stop()

        then:
        result1 == "value:1"
        result2 == "value:2"
        result3 == "value:3"

        cleanup:
        worker?.stop()
    }

    def "receives memory status from worker process"() {
        when:

// next and stop.
//
// Next returns the next value in the sequence
// and a boolean indicating whether the value is valid.
// When the sequence is over, next returns the zero V and false.
// It is valid to call next after reaching the end of the sequence
// or after calling stop. These calls will continue
// to return the zero V and false.
//
// Stop ends the iteration. It must be called when the caller is

// Run starts the controller. This must be called.
func (c *crdWatcher) Run(stop <-chan struct{}) {
	c.mutex.Lock()
	if c.stop != nil {
		// Run already called. Because we call this from client.RunAndWait this isn't uncommon
		c.mutex.Unlock()
		return
	}
	c.stop = stop
	c.mutex.Unlock()
	kube.WaitForCacheSync("crd watcher", stop, c.crds.HasSynced)
	c.queue.Run(stop)
	c.crds.ShutdownHandlers()
}

	// Regression test to ensure no race conditions during monitor shutdown
	store := memory.Make(collections.Mocks)
	m := memory.NewMonitor(store)
	stop := make(chan struct{})
	go m.Run(stop)
	m.ScheduleProcessEvent(memory.ConfigEvent{})
	close(stop)
	m.ScheduleProcessEvent(memory.ConfigEvent{})
}

func TestEventConsistency(t *testing.T) {
	store := memory.Make(collections.Mocks)

	}
}

func (b *B) startit() chan<- struct{} {
	stop := make(chan struct{})
	b.wg.Add(1)
	go func() {
		defer b.wg.Done()
		var v uint64
		for {
			select {
			case <-stop:
				b.v = v
				return
			case <-time.After(1 * time.Millisecond):
				r, err := b.f()
				if err != nil {
					panic("bad")
				}
				v = r
			}
		}
	}()
	return stop
}

var S, G int

//go:noinline

        when:
        handler.reservePort()
        handler.start(10, {})
        handler.stop()

        then:
        1 * factory.create(_ as String) >> lockRequestListener
        1 * factory.create(_ as String) >> releaseLockActionExecutor
        1 * lockRequestListener.stop()
        1 * releaseLockActionExecutor.stop()
    }

    def "stopping is safe even if the handler was not initialized"() {
        when:

	// The channel should not be waited on after Stop() is invoked. It is allowed
	// to cache the returned value of C() for the lifetime of the Timer.
	C() <-chan time.Time
	// Next is invoked by wait functions to signal timers that the next interval
	// should begin. You may only use Next() if you have drained the channel C().
	// You should not call Next() after Stop() is invoked.
	Next()