/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.qpid.protonj2.client.impl;

import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.qpid.protonj2.client.Client;
import org.apache.qpid.protonj2.client.ClientOptions;
import org.apache.qpid.protonj2.client.Connection;
import org.apache.qpid.protonj2.client.ConnectionOptions;
import org.apache.qpid.protonj2.client.ErrorCondition;
import org.apache.qpid.protonj2.client.Message;
import org.apache.qpid.protonj2.client.NextReceiverPolicy;
import org.apache.qpid.protonj2.client.Receiver;
import org.apache.qpid.protonj2.client.ReceiverOptions;
import org.apache.qpid.protonj2.client.Sender;
import org.apache.qpid.protonj2.client.Session;
import org.apache.qpid.protonj2.client.Tracker;
import org.apache.qpid.protonj2.client.exceptions.ClientConnectionRedirectedException;
import org.apache.qpid.protonj2.client.exceptions.ClientConnectionRemotelyClosedException;
import org.apache.qpid.protonj2.client.exceptions.ClientException;
import org.apache.qpid.protonj2.client.exceptions.ClientIOException;
import org.apache.qpid.protonj2.client.exceptions.ClientUnsupportedOperationException;
import org.apache.qpid.protonj2.client.test.ImperativeClientTestCase;
import org.apache.qpid.protonj2.client.test.Wait;
import org.apache.qpid.protonj2.test.driver.ProtonTestServer;
import org.apache.qpid.protonj2.test.driver.ProtonTestServerOptions;
import org.apache.qpid.protonj2.test.driver.codec.messaging.TerminusDurability;
import org.apache.qpid.protonj2.test.driver.codec.messaging.TerminusExpiryPolicy;
import org.apache.qpid.protonj2.test.driver.codec.security.SaslCode;
import org.apache.qpid.protonj2.test.driver.matchers.messaging.SourceMatcher;
import org.apache.qpid.protonj2.types.UnsignedInteger;
import org.apache.qpid.protonj2.types.messaging.AmqpValue;
import org.apache.qpid.protonj2.types.transport.AMQPHeader;
import org.apache.qpid.protonj2.types.transport.AmqpError;
import org.apache.qpid.protonj2.types.transport.ConnectionError;
import org.apache.qpid.protonj2.types.transport.Role;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Test for the Connection class
 */
@Timeout(20)
public class ConnectionTest extends ImperativeClientTestCase {

    private static final Logger LOG = LoggerFactory.getLogger(ConnectionTest.class);

    @Test
    public void testConnectFailsDueToServerStopped() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            peer.close();

            Client container = Client.create();

            try {
                Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
                connection.openFuture().get();
                fail("Should fail to connect");
            } catch (ExecutionException ex) {
                LOG.info("Connection create failed due to: ", ex);
                assertTrue(ex.getCause() instanceof ClientException);
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateTwoDistinctConnectionsFromSingleClientInstance() throws Exception {
        try (ProtonTestServer firstPeer = new ProtonTestServer(testServerOptions());
             ProtonTestServer secondPeer = new ProtonTestServer(testServerOptions())) {

            firstPeer.expectSASLAnonymousConnect();
            firstPeer.expectOpen().withContainerId(Matchers.any(String.class)).respond();
            firstPeer.expectClose().respond();
            firstPeer.start();

            secondPeer.expectSASLAnonymousConnect();
            secondPeer.expectOpen().withContainerId(Matchers.any(String.class)).respond();
            secondPeer.expectClose().respond();
            secondPeer.start();

            final URI firstURI = firstPeer.getServerURI();
            final URI secondURI = secondPeer.getServerURI();

            Client container = Client.create();
            Connection connection1 = container.connect(firstURI.getHost(), firstURI.getPort(), connectionOptions());
            Connection connection2 = container.connect(secondURI.getHost(), secondURI.getPort(), connectionOptions());

            connection1.openFuture().get();
            connection2.openFuture().get();

            connection1.closeAsync().get();
            connection2.closeAsync().get();

            firstPeer.waitForScriptToComplete();
            secondPeer.waitForScriptToComplete();
        }
    }

    @Test
    public void testCreateConnectionToNonSaslPeer() throws Exception {
        doConnectionWithUnexpectedHeaderTestImpl(AMQPHeader.getAMQPHeader().toArray());
    }

    @Test
    public void testCreateConnectionToNonAmqpPeer() throws Exception {
        doConnectionWithUnexpectedHeaderTestImpl(new byte[] { 'N', 'O', 'T', '-', 'A', 'M', 'Q', 'P' });
    }

    private void doConnectionWithUnexpectedHeaderTestImpl(byte[] responseHeader) throws Exception, IOException {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLHeader().respondWithBytes(responseHeader);
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            ConnectionOptions options = connectionOptions("guest", "guest");
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            try {
                connection.openFuture().get(10, TimeUnit.SECONDS);
            } catch (ExecutionException ex) {}

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionString() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionSignalsEvent() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().respond();
            peer.start();

            final URI remoteURI = peer.getServerURI();
            final CountDownLatch connected = new CountDownLatch(1);

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(
                remoteURI.getHost(), remoteURI.getPort(), connectionOptions().connectedHandler((conn, event) -> connected.countDown()));

            connection.openFuture().get(10, TimeUnit.SECONDS);

            assertTrue(connected.await(5, TimeUnit.SECONDS));

            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionWithConfiguredContainerId() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().withContainerId("container-id-test").respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            ClientOptions options = new ClientOptions().id("container-id-test");
            Client container = Client.create(options);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionWithUnconfiguredContainerId() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().withContainerId(Matchers.any(String.class)).respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            ClientOptions options = new ClientOptions();
            Client container = Client.create(options);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionStringWithDefaultTcpPort() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            ConnectionOptions options = connectionOptions();
            options.transportOptions().defaultTcpPort(remoteURI.getPort());
            Connection connection = container.connect(remoteURI.getHost(), options);

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionEstablishedHandlerGetsCalled() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            final CountDownLatch established = new CountDownLatch(1);
            ConnectionOptions options = connectionOptions();

            options.connectedHandler((connection, location) -> {
                LOG.info("Connection signaled that it was established");
                established.countDown();
            });

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            assertTrue(established.await(10, TimeUnit.SECONDS));

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionFailedHandlerGetsCalled() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin();
            peer.dropAfterLastHandler(10);
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            final CountDownLatch failed = new CountDownLatch(1);
            ConnectionOptions options = connectionOptions();

            options.disconnectedHandler((connection, location) -> {
                LOG.info("Connection signaled that it has failed");
                failed.countDown();
            });

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.openSession();

            assertTrue(failed.await(10, TimeUnit.SECONDS));

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionWithCredentialsChoosesSASLPlainIfOffered() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLPlainConnect("user", "pass", "PLAIN", "ANONYMOUS");
            peer.expectOpen().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            final CountDownLatch established = new CountDownLatch(1);
            ConnectionOptions options = connectionOptions("user", "pass");

            options.connectedHandler((connection, location) -> {
                LOG.info("Connection signaled that it was established");
                established.countDown();
            });

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            assertTrue(established.await(10, TimeUnit.SECONDS));

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionWithSASLDisabledToSASLEnabledHost() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectAMQPHeader().respondWithSASLHeader();
            peer.dropAfterLastHandler(); // Prevent any additional data to be parsed
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            final ConnectionOptions options = connectionOptions();
            options.saslOptions().saslEnabled(false);

            final Client container = Client.create();
            final Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            try {
                connection.openFuture().get(10, TimeUnit.SECONDS);
                fail("Should not successfully connect to remote");
            } catch(ExecutionException ex) {
                assertTrue(ex.getCause() instanceof ClientConnectionRemotelyClosedException);
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionCloseGetsResponseWithErrorDoesNotThrowTimedGet() throws Exception {
        doTestConnectionCloseGetsResponseWithErrorDoesNotThrow(true);
    }

    @Test
    public void testConnectionCloseGetsResponseWithErrorDoesNotThrowUntimedGet() throws Exception {
        doTestConnectionCloseGetsResponseWithErrorDoesNotThrow(false);
    }

    protected void doTestConnectionCloseGetsResponseWithErrorDoesNotThrow(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().respond().withErrorCondition(ConnectionError.CONNECTION_FORCED.toString(), "Not accepting connections");
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            if (timeout) {
                connection.openFuture().get(10, TimeUnit.SECONDS);
                // Should close normally and not throw error as we initiated the close.
                connection.closeAsync().get(10, TimeUnit.SECONDS);
            } else {
                connection.openFuture().get();
                // Should close normally and not throw error as we initiated the close.
                connection.closeAsync().get();
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testRemotelyCloseConnectionWithRedirect() throws Exception {
        final String redirectVhost = "vhost";
        final String redirectNetworkHost = "localhost";
        final int redirectPort = 5677;
        final String redirectScheme = "wss";
        final String redirectPath = "/websockets";

        // Tell the test peer to close the connection when executing its last handler
        final Map<String, Object> errorInfo = new HashMap<>();
        errorInfo.put(ClientConstants.OPEN_HOSTNAME.toString(), redirectVhost);
        errorInfo.put(ClientConstants.NETWORK_HOST.toString(), redirectNetworkHost);
        errorInfo.put(ClientConstants.PORT.toString(), redirectPort);
        errorInfo.put(ClientConstants.SCHEME.toString(), redirectScheme);
        errorInfo.put(ClientConstants.PATH.toString(), redirectPath);

        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().reject(ConnectionError.REDIRECT.toString(), "Not accepting connections", errorInfo);
            peer.expectBegin().optional();
            peer.expectClose();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            try {
                connection.defaultSession().openFuture().get();
                fail("Should not be able to connect since the connection is redirected.");
            } catch (Exception ex) {
                LOG.debug("Received expected exception from session open: {}", ex.getMessage());
                Throwable cause = ex.getCause();
                assertTrue(cause instanceof ClientConnectionRedirectedException);

                ClientConnectionRedirectedException connectionRedirect = (ClientConnectionRedirectedException) ex.getCause();

                assertEquals(redirectVhost, connectionRedirect.getHostname());
                assertEquals(redirectNetworkHost, connectionRedirect.getNetworkHost());
                assertEquals(redirectPort, connectionRedirect.getPort());
                assertEquals(redirectScheme, connectionRedirect.getScheme());
                assertEquals(redirectPath, connectionRedirect.getPath());

                URI redirect = connectionRedirect.getRedirectionURI();

                assertEquals(redirectNetworkHost, redirect.getHost());
                assertEquals(redirectPort, redirect.getPort());
                assertEquals(redirectScheme, redirect.getScheme());
                assertEquals(redirectPath, redirect.getPath());
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionBlockingCloseGetsResponseWithErrorDoesNotThrow() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().respond().withErrorCondition(ConnectionError.CONNECTION_FORCED.toString(), "Not accepting connections");
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get();
            // Should close normally and not throw error as we initiated the close.
            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionClosedWithErrorToRemoteSync() throws Exception {
        doTestConnectionClosedWithErrorToRemote(false);
    }

    @Test
    public void testConnectionClosedWithErrorToRemoteAsync() throws Exception {
        doTestConnectionClosedWithErrorToRemote(true);
    }

    private void doTestConnectionClosedWithErrorToRemote(boolean async) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().withError(ConnectionError.CONNECTION_FORCED.toString(), "Closed").respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get();
            if (async) {
                connection.closeAsync(ErrorCondition.create(ConnectionError.CONNECTION_FORCED.toString(), "Closed")).get();
            } else {
                connection.close(ErrorCondition.create(ConnectionError.CONNECTION_FORCED.toString(), "Closed"));
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionRemoteClosedAfterOpened() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().reject(ConnectionError.CONNECTION_FORCED.toString(), "Not accepting connections");
            peer.expectClose();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete();
        }
    }

    @Test
    public void testConnectionRemoteClosedAfterOpenedWithEmptyErrorConditionDescription() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().reject(ConnectionError.CONNECTION_FORCED.toString(), (String) null);
            peer.expectClose();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete();
        }
    }

    @Test
    public void testConnectionRemoteClosedAfterOpenedWithNoRemoteErrorCondition() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().reject();
            peer.expectClose();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete();
        }
    }

    @Test
    public void testConnectionOpenFutureWaitCancelledOnConnectionDropWithTimeout() throws Exception {
        doTestConnectionOpenFutureWaitCancelledOnConnectionDrop(true);
    }

    @Test
    public void testConnectionOpenFutureWaitCancelledOnConnectionDropNoTimeout() throws Exception {
        doTestConnectionOpenFutureWaitCancelledOnConnectionDrop(false);
    }

    protected void doTestConnectionOpenFutureWaitCancelledOnConnectionDrop(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.close();

            try {
                if (timeout) {
                    connection.openFuture().get(10, TimeUnit.SECONDS);
                } else {
                    connection.openFuture().get();
                }
                fail("Should have thrown an execution error due to connection drop");
            } catch (ExecutionException error) {
                LOG.info("connection open failed with error: ", error);
            }

            try {
                if (timeout) {
                    connection.closeAsync().get(10, TimeUnit.SECONDS);
                } else {
                    connection.closeAsync().get();
                }
            } catch (Throwable error) {
                LOG.info("connection close failed with error: ", error);
                fail("Close should ignore connect error and complete without error.");
            }

            peer.waitForScriptToComplete();
        }
    }

    @Test
    public void testRemotelyCloseConnectionDuringSessionCreation() throws Exception {
        final String BREAD_CRUMB = "ErrorMessageBreadCrumb";

        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin();
            peer.remoteClose().withErrorCondition(AmqpError.NOT_ALLOWED.toString(), BREAD_CRUMB).queue();
            peer.expectClose();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            connection.openFuture().get();

            Session session = connection.openSession();

            try {
                session.openFuture().get();
                fail("Open should throw error when waiting for remote open and connection remotely closed.");
            } catch (ExecutionException error) {
                LOG.info("Session open failed with error: ", error);
                assertNotNull(error.getMessage(), "Expected exception to have a message");
                assertTrue(error.getMessage().contains(BREAD_CRUMB), "Expected breadcrumb to be present in message");
                assertNotNull(error.getCause(), "Execution error should convey the cause");
                assertTrue(error.getCause() instanceof ClientConnectionRemotelyClosedException);
            }

            session.closeAsync().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            connection.closeAsync().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionOpenTimeoutWhenNoRemoteOpenArrivesTimeout() throws Exception {
        doTestConnectionOpenTimeoutWhenNoRemoteOpenArrives(true);
    }

    @Test
    public void testConnectionOpenTimeoutWhenNoRemoteOpenArrivesNoTimeout() throws Exception {
        doTestConnectionOpenTimeoutWhenNoRemoteOpenArrives(false);
    }

    private void doTestConnectionOpenTimeoutWhenNoRemoteOpenArrives(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen();
            peer.expectClose();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            final ConnectionOptions options = connectionOptions().openTimeout(100);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            try {
                if (timeout) {
                    connection.openFuture().get(10, TimeUnit.SECONDS);
                } else {
                    connection.openFuture().get();
                }

                fail("Open should timeout when no open response and complete future with error.");
            } catch (Throwable error) {
                LOG.info("connection open failed with error: ", error);
            }

            peer.waitForScriptToComplete(6, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionOpenWaitWithTimeoutCanceledWhenConnectionDrops() throws Exception {
        doTestConnectionOpenWaitCanceledWhenConnectionDrops(true);
    }

    @Test
    public void testConnectionOpenWaitWithNoTimeoutCanceledWhenConnectionDrops() throws Exception {
        doTestConnectionOpenWaitCanceledWhenConnectionDrops(false);
    }

    private void doTestConnectionOpenWaitCanceledWhenConnectionDrops(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen();
            peer.dropAfterLastHandler(10);
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);
            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            try {
                if (timeout) {
                    connection.openFuture().get(10, TimeUnit.SECONDS);
                } else {
                    connection.openFuture().get();
                }

                fail("Open should timeout when no open response and complete future with error.");
            } catch (ExecutionException error) {
                LOG.info("connection open failed with error: ", error);
                assertTrue(error.getCause() instanceof ClientIOException);
            }

            connection.client();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionCloseTimeoutWhenNoRemoteCloseArrivesTimeout() throws Exception {
        doTestConnectionCloseTimeoutWhenNoRemoteCloseArrives(true);
    }

    @Test
    public void testConnectionCloseTimeoutWhenNoRemoteCloseArrivesNoTimeout() throws Exception {
        doTestConnectionCloseTimeoutWhenNoRemoteCloseArrives(false);
    }

    private void doTestConnectionCloseTimeoutWhenNoRemoteCloseArrives(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            final ConnectionOptions options = connectionOptions().closeTimeout(75);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            connection.openFuture().get(10, TimeUnit.SECONDS);

            // Shouldn't throw from close, nothing to be done anyway.
            try {
                if (timeout) {
                    connection.closeAsync().get(10, TimeUnit.SECONDS);
                } else {
                    connection.closeAsync().get();
                }
            } catch (Throwable error) {
                LOG.info("connection close failed with error: ", error);
                fail("Close should ignore lack of close response and complete without error.");
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionCloseWaitWithTimeoutCompletesAfterRemoteConnectionDrops() throws Exception {
        doTestConnectionCloseWaitCompletesAfterRemoteConnectionDrops(true);
    }

    @Test
    public void testConnectionCloseWaitWithNoTimeoutCompletesAfterRemoteConnectionDrops() throws Exception {
        doTestConnectionCloseWaitCompletesAfterRemoteConnectionDrops(false);
    }

    private void doTestConnectionCloseWaitCompletesAfterRemoteConnectionDrops(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose();
            peer.dropAfterLastHandler(10);
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get(10, TimeUnit.SECONDS);

            // Shouldn't throw from close, nothing to be done anyway.
            try {
                if (timeout) {
                    connection.closeAsync().get(10, TimeUnit.SECONDS);
                } else {
                    connection.closeAsync().get();
                }
            } catch (Throwable error) {
                LOG.info("connection close failed with error: ", error);
                fail("Close should treat Connection drop as success and complete without error.");
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateDefaultSenderFailsOnConnectionWithoutSupportForAnonymousRelay() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(
                remoteURI.getHost(), remoteURI.getPort(), connectionOptions()).openFuture().get();

            try {
                connection.defaultSender();
                fail("Should not be able to get the default sender when remote does not offer anonymous relay");
            } catch (ClientUnsupportedOperationException unsupported) {
                LOG.info("Caught expected error: ", unsupported);
            }

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateDefaultSenderOnConnectionWithSupportForAnonymousRelay() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().withDesiredCapabilities(ClientConstants.ANONYMOUS_RELAY.toString())
                             .respond()
                             .withOfferedCapabilities(ClientConstants.ANONYMOUS_RELAY.toString());
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.SENDER.getValue()).respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            connection.openFuture().get(10, TimeUnit.SECONDS);

            Sender defaultSender = connection.defaultSender().openFuture().get(5, TimeUnit.SECONDS);
            assertNotNull(defaultSender);

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionRecreatesAnonymousRelaySenderAfterRemoteCloseOfSender() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().withDesiredCapabilities(ClientConstants.ANONYMOUS_RELAY.toString())
                             .respond()
                             .withOfferedCapabilities(ClientConstants.ANONYMOUS_RELAY.toString());
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.SENDER.getValue()).respond();
            peer.remoteDetach().queue();
            peer.expectDetach();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            connection.openFuture().get(10, TimeUnit.SECONDS);

            Sender defaultSender = connection.defaultSender().openFuture().get(5, TimeUnit.SECONDS);
            assertNotNull(defaultSender);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectAttach().withRole(Role.SENDER.getValue()).respond();
            peer.expectClose().respond();

            defaultSender = connection.defaultSender().openFuture().get(5, TimeUnit.SECONDS);
            assertNotNull(defaultSender);

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateDynamicReceiver() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue())
                               .withSource(new SourceMatcher().withDynamic(true).withAddress(nullValue()))
                               .respond();
            peer.expectFlow();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            connection.openFuture().get(10, TimeUnit.SECONDS);

            Receiver receiver = connection.openDynamicReceiver();
            receiver.openFuture().get(10, TimeUnit.SECONDS);

            assertNotNull(receiver.address(), "Remote should have assigned the address for the dynamic receiver");

            receiver.closeAsync().get(10, TimeUnit.SECONDS);

            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateDynamicReceiverWithNodeProperties() throws Exception {
        Map<String, Object> dynamicNodeProperties = new HashMap<>();
        dynamicNodeProperties.put("test", "vale");

        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue())
                               .withSource()
                               .withDynamic(true)
                               .withAddress(nullValue())
                               .withDynamicNodeProperties(dynamicNodeProperties)
                               .also()
                               .respond();
            peer.expectFlow();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            connection.openFuture().get(10, TimeUnit.SECONDS);

            Receiver receiver = connection.openDynamicReceiver(dynamicNodeProperties);
            receiver.openFuture().get(10, TimeUnit.SECONDS);

            assertNotNull(receiver.address(), "Remote should have assigned the address for the dynamic receiver");

            receiver.closeAsync().get(10, TimeUnit.SECONDS);

            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateDynamicReceiverWithReceiverOptions() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue())
                               .withDesiredCapabilities("queue")
                               .respond();
            peer.expectFlow();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            connection.openFuture().get(10, TimeUnit.SECONDS);

            ReceiverOptions options = new ReceiverOptions();
            options.desiredCapabilities("queue");
            Receiver receiver = connection.openDynamicReceiver(options);
            receiver.openFuture().get(10, TimeUnit.SECONDS);

            assertNotNull(receiver.address(), "Remote should have assigned the address for the dynamic receiver");

            receiver.closeAsync().get(10, TimeUnit.SECONDS);

            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionSenderOpenHeldUntilConnectionOpenedAndRelaySupportConfirmed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen();
            peer.expectBegin();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            Sender sender = connection.defaultSender();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            // This should happen after we inject the held open and attach
            peer.expectAttach().withRole(Role.SENDER.getValue()).withTarget().withAddress(Matchers.nullValue()).and().respond();
            peer.expectClose().respond();

            // Inject held responses to get the ball rolling again
            peer.remoteOpen().withOfferedCapabilities("ANONYMOUS-RELAY").now();
            peer.respondToLastBegin().now();

            try {
                sender.openFuture().get();
            } catch (ExecutionException ex) {
                fail("Open of Sender failed waiting for response: " + ex.getCause());
            }

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionSenderIsSingleton() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond().withOfferedCapabilities("ANONYMOUS-RELAY");
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.SENDER.getValue()).withTarget().withAddress(Matchers.nullValue()).and().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            Sender sender1 = connection.defaultSender();
            Sender sender2 = connection.defaultSender();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectClose().respond();

            try {
                sender1.openFuture().get();
            } catch (ExecutionException ex) {
                fail("Open of Sender failed waiting for response: " + ex.getCause());
            }

            try {
                sender2.openFuture().get();
            } catch (ExecutionException ex) {
                fail("Open of Sender failed waiting for response: " + ex.getCause());
            }

            assertSame(sender1, sender2);

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionSenderOpenFailsWhenAnonymousRelayNotSupported() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());
            Sender sender = connection.defaultSender();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectClose().respond();

            try {
                sender.openFuture().get();
                fail("Open of Sender should have failed waiting for response when anonymous relay not supported");
            } catch (ExecutionException ex) {
            }

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionGetRemotePropertiesWaitsForRemoteBegin() throws Exception {
        tryReadConnectionRemoteProperties(true);
    }

    @Test
    public void testConnectionGetRemotePropertiesFailsAfterOpenTimeout() throws Exception {
        tryReadConnectionRemoteProperties(false);
    }

    private void tryReadConnectionRemoteProperties(boolean openResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();

            ConnectionOptions options = connectionOptions().openTimeout(100);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            Map<String, Object> expectedProperties = new HashMap<>();
            expectedProperties.put("TEST", "test-property");

            if (openResponse) {
                peer.expectClose().respond();
                peer.remoteOpen().withProperties(expectedProperties).later(10);
            } else {
                peer.expectClose();
            }

            if (openResponse) {
                assertNotNull(connection.properties(), "Remote should have responded with a remote properties value");
                assertEquals(expectedProperties, connection.properties());
            } else {
                try {
                    connection.properties();
                    fail("Should failed to get remote state due to no open response");
                } catch (ClientException ex) {
                    LOG.debug("Caught expected exception from blocking call", ex);
                }
            }

            connection.closeAsync().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionGetRemoteOfferedCapabilitiesWaitsForRemoteBegin() throws Exception {
        tryReadConnectionRemoteOfferedCapabilities(true);
    }

    @Test
    public void testConnectionGetRemoteOfferedCapabilitiesFailsAfterOpenTimeout() throws Exception {
        tryReadConnectionRemoteOfferedCapabilities(false);
    }

    private void tryReadConnectionRemoteOfferedCapabilities(boolean openResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            ConnectionOptions options = connectionOptions().openTimeout(100);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            if (openResponse) {
                peer.expectClose().respond();
                peer.remoteOpen().withOfferedCapabilities("transactions").later(10);
            } else {
                peer.expectClose();
            }

            if (openResponse) {
                assertNotNull(connection.offeredCapabilities(), "Remote should have responded with a remote offered Capabilities value");
                assertEquals(1, connection.offeredCapabilities().length);
                assertEquals("transactions", connection.offeredCapabilities()[0]);
            } else {
                try {
                    connection.offeredCapabilities();
                    fail("Should failed to get remote state due to no open response");
                } catch (ClientException ex) {
                    LOG.debug("Caught expected exception from blocking call", ex);
                }
            }

            connection.closeAsync().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionGetRemoteDesiredCapabilitiesWaitsForRemoteBegin() throws Exception {
        tryReadConnectionRemoteDesiredCapabilities(true);
    }

    @Test
    public void testConnectionGetRemoteDesiredCapabilitiesFailsAfterOpenTimeout() throws Exception {
        tryReadConnectionRemoteDesiredCapabilities(false);
    }

    private void tryReadConnectionRemoteDesiredCapabilities(boolean openResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            ConnectionOptions options = connectionOptions().openTimeout(100);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            if (openResponse) {
                peer.expectClose().respond();
                peer.remoteOpen().withDesiredCapabilities("Error-Free").later(10);
            } else {
                peer.expectClose();
            }

            if (openResponse) {
                assertNotNull(connection.desiredCapabilities(), "Remote should have responded with a remote desired Capabilities value");
                assertEquals(1, connection.desiredCapabilities().length);
                assertEquals("Error-Free", connection.desiredCapabilities()[0]);
            } else {
                try {
                    connection.desiredCapabilities();
                    fail("Should failed to get remote state due to no open response");
                } catch (ClientException ex) {
                    LOG.debug("Caught expected exception from blocking call", ex);
                }
            }

            connection.closeAsync().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCloseWithErrorCondition() throws Exception {
        final String condition = "amqp:precondition-failed";
        final String description = "something bad happened.";

        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().withError(condition, description).respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(
                remoteURI.getHost(), remoteURI.getPort(), connectionOptions()).openFuture().get();

            connection.close(ErrorCondition.create(condition, description, null));

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testAnonymousSenderOpenHeldUntilConnectionOpenedAndSupportConfirmed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen();
            peer.expectBegin();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Sender sender = connection.openAnonymousSender();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            // This should happen after we inject the held open and attach
            peer.expectAttach().ofSender().withTarget().withAddress(Matchers.nullValue()).and().respond();
            peer.expectClose().respond();

            // Inject held responses to get the ball rolling again
            peer.remoteOpen().withOfferedCapabilities("ANONYMOUS-RELAY").now();
            peer.respondToLastBegin().now();

            try {
                sender.openFuture().get();
            } catch (ExecutionException ex) {
                fail("Open of Sender failed waiting for response: " + ex.getCause());
            }

            connection.closeAsync();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testSendHeldUntilConnectionOpenedAndSupportConfirmed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond().withOfferedCapabilities("ANONYMOUS-RELAY");
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().withTarget().withAddress(nullValue()).and().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {0}).respond().withSettled(true).withState().accepted();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());

            try {
                Tracker tracker = connection.send(Message.create("Hello World"));
                assertNotNull(tracker);
                tracker.awaitAccepted();
            } catch (ClientException ex) {
                fail("Open of Sender failed waiting for response: " + ex.getCause());
            }

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testConnectionLevelSendFailsWhenAnonymousRelayNotAdvertisedByRemote() throws Exception {
        doTestConnectionLevelSendFailsWhenAnonymousRelayNotAdvertisedByRemote(false);
    }

    @Test
    public void testConnectionLevelSendFailsWhenAnonymousRelayNotAdvertisedByRemoteAfterAlreadyOpened() throws Exception {
        doTestConnectionLevelSendFailsWhenAnonymousRelayNotAdvertisedByRemote(true);
    }

    private void doTestConnectionLevelSendFailsWhenAnonymousRelayNotAdvertisedByRemote(boolean openWait) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            // Ensures that the Begin arrives regard of a race on open without anonymous relay support
            connection.defaultSession();

            if (openWait) {
                connection.openFuture().get();
            }

            try {
                connection.send(Message.create("Hello World"));
                fail("Open of Sender should fail as remote did not advertise anonymous relay support: ");
            } catch (ClientUnsupportedOperationException ex) {
            }

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testOpenAnonymousSenderFailsWhenAnonymousRelayNotAdvertisedByRemote() throws Exception {
        doTestOpenAnonymousSenderFailsWhenAnonymousRelayNotAdvertisedByRemote(false);
    }

    @Test
    public void testOpenAnonymousSenderFailsWhenAnonymousRelayNotAdvertisedByRemoteAfterAlreadyOpened() throws Exception {
        doTestOpenAnonymousSenderFailsWhenAnonymousRelayNotAdvertisedByRemote(true);
    }

    private void doTestOpenAnonymousSenderFailsWhenAnonymousRelayNotAdvertisedByRemote(boolean openWait) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            // Ensures that the Begin arrives regard of a race on open without anonymous relay support
            connection.defaultSession();

            if (openWait) {
                connection.openFuture().get();
            }

            try {
                connection.openAnonymousSender().openFuture().get();
                fail("Open of Sender should fail as remote did not advertise anonymous relay support: ");
            } catch (ClientUnsupportedOperationException ex) {
            } catch (ExecutionException ex) {
                assertTrue(ex.getCause() instanceof ClientUnsupportedOperationException);
            }

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testOpenDurableReceiverFromConnection() throws Exception {
        final String address = "test-topic";
        final String subscriptionName = "mySubscriptionName";

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver()
                               .withName(subscriptionName)
                               .withSource()
                                   .withAddress(address)
                                   .withDurable(TerminusDurability.UNSETTLED_STATE)
                                   .withExpiryPolicy(TerminusExpiryPolicy.NEVER)
                                   .withDistributionMode("copy")
                               .and().respond();
            peer.expectFlow();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Receiver receiver = connection.openDurableReceiver(address, subscriptionName);

            receiver.openFuture().get();
            receiver.closeAsync().get();

            connection.closeAsync().get();

            peer.waitForScriptToComplete(6, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testUserSpeicifedNextReceiverPolicyOverridesConfiguration() throws Exception {
        byte[] payload = createEncodedMessage(new AmqpValue<String>("Hello World"));

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.remoteTransfer().withHandle(1)
                                 .withDeliveryId(1)
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.remoteTransfer().withHandle(1)
                                 .withDeliveryId(2)
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.remoteTransfer().withHandle(2)
                                 .withDeliveryId(3)
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().defaultNextReceiverPolicy(NextReceiverPolicy.SMALLEST_BACKLOG);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            ReceiverOptions receiverOptions = new ReceiverOptions().creditWindow(10).autoAccept(false);
            Receiver receiver1 = connection.openReceiver("test-receiver1", receiverOptions).openFuture().get();
            Receiver receiver2 = connection.openReceiver("test-receiver2", receiverOptions).openFuture().get();
            Receiver receiver3 = connection.openReceiver("test-receiver3", receiverOptions).openFuture().get();

            peer.waitForScriptToComplete();

            Wait.waitFor(() -> receiver1.queuedDeliveries() == 1);
            Wait.waitFor(() -> receiver2.queuedDeliveries() == 2);
            Wait.waitFor(() -> receiver3.queuedDeliveries() == 1);

            Receiver next = connection.nextReceiver(NextReceiverPolicy.LARGEST_BACKLOG);
            assertSame(next, receiver2);

            peer.waitForScriptToComplete();

            peer.waitForScriptToComplete();
            peer.expectClose().respond();

            connection.close();

            peer.waitForScriptToComplete();
        }
    }

    @Test
    public void testNextReceiverThrowsAfterConnectionClosedRandom() throws Exception {
        doTestNextReceiverThrowsAfterSessionClosed(NextReceiverPolicy.RANDOM);
    }

    @Test
    public void testNextReceiverThrowsAfterConnectionClosedLargestBacklog() throws Exception {
        doTestNextReceiverThrowsAfterSessionClosed(NextReceiverPolicy.LARGEST_BACKLOG);
    }

    @Test
    public void testNextReceiverThrowsAfterConnectionClosedSmallestBacklog() throws Exception {
        doTestNextReceiverThrowsAfterSessionClosed(NextReceiverPolicy.SMALLEST_BACKLOG);
    }

    @Test
    public void testNextReceiverThrowsAfterConnectionClosedFirstAvailable() throws Exception {
        doTestNextReceiverThrowsAfterSessionClosed(NextReceiverPolicy.FIRST_AVAILABLE);
    }

    public void doTestNextReceiverThrowsAfterSessionClosed(NextReceiverPolicy policy) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            final CountDownLatch started = new CountDownLatch(1);
            final CountDownLatch done = new CountDownLatch(1);
            final AtomicReference<Exception> error = new AtomicReference<>();

            final Client container = Client.create();
            final ConnectionOptions options = new ConnectionOptions().defaultNextReceiverPolicy(policy);
            final Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            ForkJoinPool.commonPool().execute(() -> {
                try {
                    started.countDown();
                    connection.nextReceiver();
                } catch (ClientException e) {
                    error.set(e);
                } finally {
                    done.countDown();
                }
            });

            peer.waitForScriptToComplete();
            peer.expectClose().respond();

            assertTrue(started.await(10, TimeUnit.SECONDS));

            connection.closeAsync();

            assertTrue(done.await(10, TimeUnit.SECONDS));
            assertTrue(error.get() instanceof ClientConnectionRemotelyClosedException);

            peer.waitForScriptToComplete();
        }
    }

    @Test
    public void testCreateConnectionWithNoVHostCarriesRemoteHost() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.start();

            final URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            peer.expectSASLHeader().respondWithSASLHeader();
            peer.remoteSaslMechanisms().withMechanisms("ANONYMOUS").queue();
            peer.expectSaslInit().withMechanism("ANONYMOUS").withHostname(remoteURI.getHost());
            peer.remoteSaslOutcome().withCode(SaslCode.OK).queue();
            peer.expectAMQPHeader().respondWithAMQPHeader();
            peer.expectOpen().withHostname(remoteURI.getHost()).respond();
            peer.expectClose().respond();

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions());

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionWithSpecifiedVHost() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLHeader().respondWithSASLHeader();
            peer.remoteSaslMechanisms().withMechanisms("ANONYMOUS").queue();
            peer.expectSaslInit().withMechanism("ANONYMOUS").withHostname("test");
            peer.remoteSaslOutcome().withCode(SaslCode.OK).queue();
            peer.expectAMQPHeader().respondWithAMQPHeader();
            peer.expectOpen().withHostname("test").respond();
            peer.expectClose().respond();
            peer.start();

            final URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            ConnectionOptions options = connectionOptions().virtualHost("test");
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionWithEmptyVHostSendsNullValueInOpen() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLHeader().respondWithSASLHeader();
            peer.remoteSaslMechanisms().withMechanisms("ANONYMOUS").queue();
            peer.expectSaslInit().withMechanism("ANONYMOUS").withHostname(nullValue());
            peer.remoteSaslOutcome().withCode(SaslCode.OK).queue();
            peer.expectAMQPHeader().respondWithAMQPHeader();
            peer.expectOpen().withHostname(nullValue()).respond();
            peer.expectClose().respond();
            peer.start();

            final URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            ConnectionOptions options = connectionOptions().virtualHost("");
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testCreateConnectionWithNoIdleTimeout() throws Exception {
        doTestCreateConnectionWithIdleTimeoutSendsExpectedValue(0, 0);
    }

    @Test
    public void testCreateConnectionWithHalfIdleTimeout() throws Exception {
        doTestCreateConnectionWithIdleTimeoutSendsExpectedValue(10_000, 5_000);
    }

    @Test
    public void testCreateConnectionWithHalfLargeIdleTimeout() throws Exception {
        doTestCreateConnectionWithIdleTimeoutSendsExpectedValue(UnsignedInteger.MAX_VALUE.longValue(), UnsignedInteger.MAX_VALUE.longValue() / 2);
    }

    private void doTestCreateConnectionWithIdleTimeoutSendsExpectedValue(long setValue, long expectedValue) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer(testServerOptions())) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().withIdleTimeOut(expectedValue).respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Connect test started, peer listening on: {}", remoteURI);

            Client container = Client.create();
            ConnectionOptions options = connectionOptions().idleTimeout(setValue);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            connection.openFuture().get(10, TimeUnit.SECONDS);
            connection.closeAsync().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Disabled("Disabled due to requirement of hard coded port")
    @Test
    public void testLocalPortOption() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

            LOG.info("Test started, peer listening on: {}", remoteURI);

            final int localPort = 5671;

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions();
            options.transportOptions().localPort(localPort);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);

            connection.openFuture().get();

            assertEquals(localPort, peer.getConnectionRemotePort());

            connection.close();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    protected ProtonTestServerOptions testServerOptions() {
        return new ProtonTestServerOptions();
    }

    protected ConnectionOptions connectionOptions() {
        return new ConnectionOptions();
    }

    protected ConnectionOptions connectionOptions(String user, String password) {
        return new ConnectionOptions().user(user).password(password);
    }
}
