queue.queue(message2)
        queue.queue(message3)

        when:
        def messages1 = []
        def messages2 = []
        queue.getChannel(id1).newEndpoint().take(messages1)
        queue.getChannel(id2).newEndpoint().take(messages2)

        then:
        messages1 == [message1, message3]
        messages2 == [message2]
    }

    def "forwards channel message to channel queue"() {

        given:
        def message1 = unicast()
        def message2 = broadcast()
        def message3 = unicast()
        def endpoint1 = queue.newEndpoint()
        def endpoint2 = queue.newEndpoint()

        and:
        queue.dispatch(message1)
        queue.dispatch(message2)
        queue.dispatch(message3)

        when:
        queue.empty(endpoint1)

	"k8s.io/apimachinery/pkg/util/uuid"
)

type testSocket struct {
	path     string
	endpoint string
}

// newEndpoint constructs a unique name for a Linux Abstract Socket to be used in a test.
// This package uses Linux Domain Sockets to remove the need for clean-up of socket files.
func newEndpoint() *testSocket {
	p := fmt.Sprintf("@%s.sock", uuid.NewUUID())

	return &testSocket{
		path:     p,

)

// Tests lock rpc client.
func TestLockRESTlient(t *testing.T) {
	// These should not be connectable.
	endpoint, err := NewEndpoint("http://localhost:9876")
	if err != nil {
		t.Fatalf("unexpected error %v", err)
	}
	endpointLocal, err := NewEndpoint("http://localhost:9012")
	if err != nil {
		t.Fatalf("unexpected error %v", err)
	}
	endpointLocal.IsLocal = true

const (
	testProviderName = "providerName"
)

type testSocket struct {
	path     string
	endpoint string
}

// newEndpoint constructs a unique name for a Linux Abstract Socket to be used in a test.
// This package uses Linux Domain Sockets to remove the need for clean-up of socket files.
func newEndpoint() *testSocket {
	p := fmt.Sprintf("@%s.sock", uuid.NewUUID())

	return &testSocket{
		path:     p,

        def channel = new ChannelIdentifier("channel")
        def outgoingMessage = new ChannelMessage(channel, "payload")

        given:
        def incoming = incomingQueue.getChannel(channel).newEndpoint()
        outgoingQueue.dispatch(outgoingMessage)

        when:
        connections.noFurtherConnections()
        def messages = []
        incoming.take(messages)

        then:

		if err != nil {
			return err
		}
		for _, ip := range getBindAddresses(ip) {
			ep, err := s.newEndpoint(p, ip, "")
			if err != nil {
				return err
			}
			s.endpoints = append(s.endpoints, ep)
		}
	}

	if len(s.UDSServer) > 0 {
		ep, err := s.newEndpoint(nil, "", s.UDSServer)
		if err != nil {
			return err
		}
		s.endpoints = append(s.endpoints, ep)
	}

    }

    /**
     * Adds a new incoming connection.
     */
    public ConnectionState add(RemoteConnection<InterHubMessage> connection) {
        EndPointQueue queue = outgoingQueue.newEndpoint();
        ConnectionState state = new ConnectionState(this, connection, queue);
        connections.add(state);
        return state;
    }

    /**

                default:
                    throw new IllegalArgumentException("Unknown delivery type: " + message.getDelivery());
            }
        }
    }

    public EndPointQueue newEndpoint() {
        EndPointQueue endPointQueue = new EndPointQueue(this, lock.newCondition());
        endpoints.add(endPointQueue);
        initializer.onQueueAdded(endPointQueue);
        return endPointQueue;
    }

	if err != nil {
		t.Fatalf("unexpected error %v", err)
	}
	url.Path = t.TempDir()

	globalMinioHost, globalMinioPort = mustSplitHostPort(url.Host)

	endpoint, err := NewEndpoint(url.String())
	if err != nil {
		t.Fatalf("NewEndpoint failed %v", endpoint)
	}

	if err = endpoint.UpdateIsLocal(); err != nil {
		t.Fatalf("UpdateIsLocal failed %v", err)
	}

	endpoint.PoolIdx = 0
	endpoint.SetIdx = 0