when:
        parallel.run {
            dispatch.dispatchTo(target1)
            dispatch.dispatch('message1')
            parallel.syncAt(1)

            dispatch.dispatchTo(target2)
            dispatch.dispatch('message2')
            parallel.syncAt(2)

            dispatch.dispatch('message3')
            parallel.syncAt(3)
        }

        dispatch.stop()

        then:

        state = State.Init;
        if (dispatch != null) {
            dispatchTo(dispatch);
        }
    }

    /**
     * Starts dispatching messages to the given handler. The handler does not need to be thread-safe.
     */
    public void dispatchTo(final Dispatch<? super T> dispatch) {
        final InterruptibleRunnable dispatcher = new InterruptibleRunnable(new Runnable() {
            @Override

     * The instance is not injected directly into this service as the instance will be constructed in some other scope.
     * </p>
     */
    void dispatchTo(UserInputReceiver userInput);

    /**
     * Discards the current {@link UserInputReceiver} instance used by this service.
     */
    void stopDispatching();

        if (userInput == null) {
            throw new IllegalStateException("User input has not been initialized.");
        }
        return userInput;
    }

    @Override
    public void dispatchTo(UserInputReceiver userInput) {
        if (!delegate.compareAndSet(null, userInput)) {
            throw new IllegalStateException("User input has already been initialized.");
        }
    }

    @Override

        userInput.dispatchTo(forwarder);
    }

    @Override
    public void stop() {
        userInput.stopDispatching();
        forwarder.stop();
        executor.shutdown();
    }

    private static class ForwardingUserInput implements UserInputReceiver {
        private final Dispatch<? super InputMessage> dispatch;
        private final BufferedReader reader;

  static Dispatcher legacyAsync() {
    return new LegacyAsyncDispatcher();
  }

  /**
   * Returns a dispatcher that dispatches events to subscribers immediately as they're posted
   * without using an intermediate queue to change the dispatch order. This is effectively a
   * depth-first dispatch order, vs. breadth-first when using a queue.
   */
  static Dispatcher immediate() {
    return ImmediateDispatcher.INSTANCE;
  }

  static Dispatcher legacyAsync() {
    return new LegacyAsyncDispatcher();
  }

  /**
   * Returns a dispatcher that dispatches events to subscribers immediately as they're posted
   * without using an intermediate queue to change the dispatch order. This is effectively a
   * depth-first dispatch order, vs. breadth-first when using a queue.
   */
  static Dispatcher immediate() {
    return ImmediateDispatcher.INSTANCE;
  }

   * request bodies.
   */
  open fun peek(): MockResponse {
    return MockResponse(socketPolicy = KeepOpen)
  }

  /**
   * Release any resources held by this dispatcher. Any requests that are currently being dispatched
   * should return immediately. Responses returned after shutdown will not be transmitted: their
   * socket connections have already been closed.
   */
  open fun shutdown() {}

        expect:
        def dispatch1 = BroadcastDispatch.empty(TestListener)
        dispatch1.removeAll([]).is(dispatch1)
        dispatch1.removeAll([listener2, listener3]).is(dispatch1)

        def dispatch2 = dispatch1.add(listener1)
        dispatch2.removeAll([]).is(dispatch2)
        dispatch2.removeAll([listener2, listener3]).is(dispatch2)

        def dispatch3 = dispatch2.add(listener2)

      Queues.newConcurrentLinkedQueue();

  private Dispatcher dispatcher;

  public void testPerThreadQueuedDispatcher() {
    dispatcher = Dispatcher.perThreadDispatchQueue();
    dispatcher.dispatch(1, integerSubscribers.iterator());

    assertThat(dispatchedSubscribers)
        .containsExactly(
            i1,
            i2,
            i3, // Integer subscribers are dispatched to first.
            s1,