thread.blockUntil.connected
        operation.stop {
            acceptor.stop()
        }

        then:
        operation.stop.end > instant.actionFinished

        cleanup:
        acceptor?.stop()
        connection?.stop()
    }

    def "can receive message from peer after peer has closed connection"() {

	}
	cleanupStop := false
	stop := opts.Stop
	if stop == nil {
		// If we created the stop, clean it up. Otherwise, caller is responsible
		cleanupStop = true
		stop = make(chan struct{})
	}
	f := namespace.NewDiscoveryNamespacesFilter(
		kclient.New[*corev1.Namespace](opts.Client),
		opts.MeshWatcher,
		stop,
	)
	kubelib.SetObjectFilter(opts.Client, f)

        when:
        stoppable.stop()

        then:
        1 * a.stop()
        1 * b.stop()
    }

    def stopsAllElementsWhenOneFailsToStop() {
        Stoppable a = Mock()
        Stoppable b = Mock()
        RuntimeException failure = new RuntimeException()
        stoppable.add(a)
        stoppable.add(b)

        when:
        stoppable.stop()

        then:

        manager.stop()

        then:
        noExceptionThrown()
        1 * client1.isFailed() >> true
        0 * client1.stop()
        1 * client2.stop()
    }

    def "can stop session-scoped clients"() {
        listenerManager = new DefaultListenerManager(Scope.BuildSession)

		enabled:        true,
	}
	l.AddRunFunction(func(stop <-chan struct{}) {
		<-stop
	})
	for _, fn := range fns {
		l.AddRunFunction(fn)
	}
	stop := make(chan struct{})
	go l.Run(stop)

	retry.UntilOrFail(t, func() bool {
		return l.isLeader() == expectLeader
	}, retry.Converge(5), retry.Delay(time.Millisecond*100), retry.Timeout(time.Second*10))
	return l, stop
}

type fakeDefaultWatcher struct {

		callCount++
		return configs, err
	}
	mon := NewMonitor("", store, someConfigFunc, "")
	stop := make(chan struct{})
	defer func() { close(stop) }()
	mon.Start(stop)

	go func() {
		updateTicker := time.NewTicker(100 * time.Millisecond)
		numUpdates := 10
		for {
			select {
			case <-stop:
				updateTicker.Stop()
				return
			case <-updateTicker.C:
				mon.updateCh <- struct{}{}
				numUpdates--

        operation.stop {
            executor.stop(timeoutMs, TimeUnit.MILLISECONDS)
        }

        then:
        IllegalStateException e = thrown()
        e.message == 'Timeout waiting for concurrent jobs to complete.'

        and:
        operation.stop.duration in approx(timeoutMs)
    }

    def stopRethrowsFirstExecutionException() {
        given:

        daemonConnection.stop()

        then:
        0 * handler._
    }

    def "does not notify disconnect handler on stop"() {
        Runnable handler = Mock()

        when:
        daemonConnection.onDisconnect(handler)
        daemonConnection.stop()

        then:
        0 * handler._
    }

    def "can stop after disconnect handler fails"() {

            }
        }

        then:
        instant.action1Done > instant.action2Done
    }

    def "can stop multiple times"() {
        when:
        lifecycle.stop()
        lifecycle.stop()
        lifecycle.requestStop()
        lifecycle.requestStop()
        lifecycle.stop()

        then:
        noExceptionThrown()
    }

        then:
        1 * delegate.processTestClass(test2)
        then:
        1 * delegate.stop()
        0 * _._
    }

    def "stopNow does nothing after stop completed"() {
        when:
        processor.startProcessing(resultProcessor)
        processor.processTestClass(test1)
        processor.stop()

        then:
        1 * factory.create() >> delegate