switch (message.getDelivery()) {
                case Stateful:
                case AllHandlers:
                    if (endpoints.isEmpty()) {
                        return;
                    }
                    if (message.getDelivery() == InterHubMessage.Delivery.Stateful) {
                        initializer.onStatefulMessage(message);
                    }

        if (message.getDelivery() == InterHubMessage.Delivery.Stateful) {
            initializer.onStatefulMessage(message);
        }
        if (message instanceof Routable) {
            Routable routableMessage = (Routable) message;
            getChannel(routableMessage.getChannel()).dispatch(message);
        } else if (message.getDelivery() == InterHubMessage.Delivery.Stateful || message.getDelivery() == InterHubMessage.Delivery.AllHandlers) {

        InterHubMessage message = Stub() {
            getDelivery() >> InterHubMessage.Delivery.SingleHandler
        }
        return message
    }

    def stateful() {
        InterHubMessage message = Stub() {
            getDelivery() >> InterHubMessage.Delivery.Stateful
        }
        return message
    }

    def broadcast() {
        InterHubMessage message = Stub() {

import org.gradle.internal.serialize.Serializer;

/**
 * A stateful “backing” for a compilation operation.
 * <p>
 * The compilation may extract data from the source under compilation, made available after compilation by {@link #getExtractedData()}.
 * The exposed transformer typically gathers the data while transforming.
 * <p>
 * As these objects are stateful, they can only be used for a single compile operation.
 *

  // string description of gradient attribute.
  static StringRef GetGradientAttrName() { return "tf.gradient"; }

  // This attribute marks if a function is stateful.
  // Returns the string description of stateful attribute.
  static StringRef GetStatefulAttrName() { return "tf.signature.is_stateful"; }

  // Returns true if the op can be duplicated during transformations.

class QueueInitializerTest extends AbstractQueueTest {
    final QueueInitializer initializer = new QueueInitializer()
    final Dispatch<InterHubMessage> queue = Mock()

    def "does nothing when no stateful messages received"() {
        when:
        initializer.onQueueAdded(queue)

        then:
        0 * queue._
    }

    def "discards message on end of stream received"() {
        given:

import spock.lang.Issue
import spock.lang.Shared
import spock.lang.Timeout

import java.nio.channels.SocketChannel

@Timeout(60)
class TcpConnectorTest extends ConcurrentSpec {
    @Shared def serializer = Serializers.stateful(BaseSerializerFactory.STRING_SERIALIZER)
    final def idGenerator = new UUIDGenerator()
    final def addressFactory = new InetAddressFactory()
    final def outgoingConnector = new TcpOutgoingConnector()

    @Override
    public String toString() {
        return "[" + getClass().getSimpleName() + "]";
    }

    @Override
    public Delivery getDelivery() {
        return Delivery.Stateful;
    }

import java.net.InetAddress
import okhttp3.internal.interleave

/**
 * Implementation of HappyEyeballs Sorting Addresses.
 *
 * The current implementation does not address any of:
 *  - Async DNS split by IP class
 *  - Stateful handling of connectivity results
 *  - The prioritisation of addresses
 *
 * https://datatracker.ietf.org/doc/html/rfc8305#section-4
 */
fun reorderForHappyEyeballs(addresses: List<InetAddress>): List<InetAddress> {

        if (message instanceof EndOfStream) {
            endOfStream = (EndOfStream) message;
        } else {
            throw new UnsupportedOperationException(String.format("Received unexpected stateful message: %s", message));
        }
    }

    void onQueueAdded(Dispatch<InterHubMessage> queue) {
        if (endOfStream != null) {
            queue.dispatch(endOfStream);
        }
    }