/*
 * 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.CoreMatchers.notNullValue;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
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.AtomicBoolean;

import org.apache.qpid.protonj2.client.Client;
import org.apache.qpid.protonj2.client.Connection;
import org.apache.qpid.protonj2.client.ConnectionOptions;
import org.apache.qpid.protonj2.client.DeliveryMode;
import org.apache.qpid.protonj2.client.DeliveryState;
import org.apache.qpid.protonj2.client.DistributionMode;
import org.apache.qpid.protonj2.client.DurabilityMode;
import org.apache.qpid.protonj2.client.ErrorCondition;
import org.apache.qpid.protonj2.client.ExpiryPolicy;
import org.apache.qpid.protonj2.client.Message;
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.SenderOptions;
import org.apache.qpid.protonj2.client.Session;
import org.apache.qpid.protonj2.client.Source;
import org.apache.qpid.protonj2.client.Target;
import org.apache.qpid.protonj2.client.Tracker;
import org.apache.qpid.protonj2.client.exceptions.ClientConnectionRemotelyClosedException;
import org.apache.qpid.protonj2.client.exceptions.ClientDeliveryStateException;
import org.apache.qpid.protonj2.client.exceptions.ClientException;
import org.apache.qpid.protonj2.client.exceptions.ClientLinkRedirectedException;
import org.apache.qpid.protonj2.client.exceptions.ClientLinkRemotelyClosedException;
import org.apache.qpid.protonj2.client.exceptions.ClientOperationTimedOutException;
import org.apache.qpid.protonj2.client.exceptions.ClientResourceRemotelyClosedException;
import org.apache.qpid.protonj2.client.exceptions.ClientSendTimedOutException;
import org.apache.qpid.protonj2.client.exceptions.ClientUnsupportedOperationException;
import org.apache.qpid.protonj2.client.impl.ClientDeliveryState.ClientRejected;
import org.apache.qpid.protonj2.client.test.ImperativeClientTestCase;
import org.apache.qpid.protonj2.engine.DeliveryTagGenerator;
import org.apache.qpid.protonj2.test.driver.ProtonTestServer;
import org.apache.qpid.protonj2.test.driver.codec.messaging.Released;
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.matchers.messaging.DeliveryAnnotationsMatcher;
import org.apache.qpid.protonj2.test.driver.matchers.transport.TransferPayloadCompositeMatcher;
import org.apache.qpid.protonj2.test.driver.matchers.types.EncodedAmqpValueMatcher;
import org.apache.qpid.protonj2.types.DeliveryTag;
import org.apache.qpid.protonj2.types.messaging.Rejected;
import org.apache.qpid.protonj2.types.transport.AmqpError;
import org.apache.qpid.protonj2.types.transport.LinkError;
import org.apache.qpid.protonj2.types.transport.ReceiverSettleMode;
import org.apache.qpid.protonj2.types.transport.Role;
import org.apache.qpid.protonj2.types.transport.SenderSettleMode;
import org.hamcrest.Matchers;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Timeout(20)
public class SenderTest extends ImperativeClientTestCase {

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

    @Test
    public void testCreateSenderAndClose() throws Exception {
        doTestCreateSenderAndCloseOrDetach(true);
    }

    @Test
    public void testCreateSenderAndDetach() throws Exception {
        doTestCreateSenderAndCloseOrDetach(false);
    }

    private void doTestCreateSenderAndCloseOrDetach(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender()
                               .withSource().withAddress("test-queue").and()
                               .withTarget().withAddress("test-queue").and()
                               .respond();
            peer.expectDetach().withClosed(close).respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

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

            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            if (close) {
                sender.closeAsync().get(10, TimeUnit.SECONDS);
            } else {
                sender.detachAsync().get(10, TimeUnit.SECONDS);
            }

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

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

    @Test
    public void testCreateSenderAndCloseSync() throws Exception {
        doTestCreateSenderAndCloseOrDetachSync(true);
    }

    @Test
    public void testCreateSenderAndDetachSync() throws Exception {
        doTestCreateSenderAndCloseOrDetachSync(false);
    }

    private void doTestCreateSenderAndCloseOrDetachSync(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.expectDetach().withClosed(close).respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

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

            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            if (close) {
                sender.close();
            } else {
                sender.detach();
            }

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

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

    @Test
    public void testCreateSenderAndCloseWithErrorSync() throws Exception {
        doTestCreateSenderAndCloseOrDetachWithErrorSync(true);
    }

    @Test
    public void testCreateSenderAndDetachWithErrorSync() throws Exception {
        doTestCreateSenderAndCloseOrDetachWithErrorSync(false);
    }

    private void doTestCreateSenderAndCloseOrDetachWithErrorSync(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.expectDetach().withError("amqp-resource-deleted", "an error message").withClosed(close).respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

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

            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            if (close) {
                sender.close(ErrorCondition.create("amqp-resource-deleted", "an error message", null));
            } else {
                sender.detach(ErrorCondition.create("amqp-resource-deleted", "an error message", null));
            }

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

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

    @Test
    public void testSenderOpenRejectedByRemote() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().respond().withNullTarget();
            peer.remoteDetach().withErrorCondition(AmqpError.UNAUTHORIZED_ACCESS.toString(), "Cannot read from this address").queue();
            peer.expectDetach();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

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

            Sender sender = session.openSender("test-queue");
            try {
                sender.openFuture().get(10, TimeUnit.SECONDS);
                fail("Open of sender should fail due to remote indicating pending close.");
            } catch (ExecutionException exe) {
                assertNotNull(exe.getCause());
                assertTrue(exe.getCause() instanceof ClientLinkRemotelyClosedException);
                ClientLinkRemotelyClosedException linkClosed = (ClientLinkRemotelyClosedException) exe.getCause();
                assertNotNull(linkClosed.getErrorCondition());
                assertEquals(AmqpError.UNAUTHORIZED_ACCESS.toString(), linkClosed.getErrorCondition().condition());
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            // Should not result in any close being sent now, already closed.
            sender.closeAsync().get(10, TimeUnit.SECONDS);

            peer.expectClose().respond();
            connection.closeAsync().get(10, TimeUnit.SECONDS);

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

    @Test
    public void testRemotelyCloseSenderLinkWithRedirect() throws Exception {
        final String redirectVhost = "vhost";
        final String redirectNetworkHost = "localhost";
        final String redirectAddress = "redirect-queue";
        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);
        errorInfo.put(ClientConstants.ADDRESS.toString(), redirectAddress);

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond().withNullTarget();
            peer.remoteDetach().withClosed(true)
                               .withErrorCondition(LinkError.REDIRECT.toString(), "Not accepting links here", errorInfo).queue();
            peer.expectDetach();
            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());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");

            try {
                sender.openFuture().get();
                fail("Should not be able to create sender since the remote is redirecting.");
            } catch (Exception ex) {
                LOG.debug("Received expected exception from sender open: {}", ex.getMessage());
                Throwable cause = ex.getCause();
                assertTrue(cause instanceof ClientLinkRedirectedException);

                ClientLinkRedirectedException linkRedirect = (ClientLinkRedirectedException) ex.getCause();

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

                URI redirect = linkRedirect.getRedirectionURI();

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

            connection.close();

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

    @Test
    public void testOpenSenderTimesOutWhenNoAttachResponseReceivedTimeout() throws Exception {
        doTestOpenSenderTimesOutWhenNoAttachResponseReceived(true);
    }

    @Test
    public void testOpenSenderTimesOutWhenNoAttachResponseReceivedNoTimeout() throws Exception {
        doTestOpenSenderTimesOutWhenNoAttachResponseReceived(false);
    }

    private void doTestOpenSenderTimesOutWhenNoAttachResponseReceived(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender();
            peer.expectDetach();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession().openFuture().get();
            Sender sender = session.openSender("test-queue", new SenderOptions().openTimeout(10));

            try {
                if (timeout) {
                    sender.openFuture().get(20, TimeUnit.SECONDS);
                } else {
                    sender.openFuture().get();
                }

                fail("Should not complete the open future without an error");
            } catch (ExecutionException exe) {
                Throwable cause = exe.getCause();
                assertTrue(cause instanceof ClientOperationTimedOutException);
            }

            LOG.info("Closing connection after waiting for sender open");

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

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

    @Test
    public void testOpenSenderWaitWithTimeoutFailsWhenConnectionDrops() throws Exception {
        doTestOpenSenderWaitFailsWhenConnectionDrops(true);
    }

    @Test
    public void testOpenSenderWaitWithNoTimeoutFailsWhenConnectionDrops() throws Exception {
        doTestOpenSenderWaitFailsWhenConnectionDrops(false);
    }

    private void doTestOpenSenderWaitFailsWhenConnectionDrops(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender();
            peer.dropAfterLastHandler(10);
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");

            Thread.sleep(10);

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

                fail("Should not complete the open future without an error");
            } catch (ExecutionException exe) {
                Throwable cause = exe.getCause();
                LOG.trace("Caught exception caused by: {}", exe);
                assertTrue(cause instanceof ClientConnectionRemotelyClosedException);
            }

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

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

    @Test
    public void testCloseSenderTimesOutWhenNoCloseResponseReceivedTimeout() throws Exception {
        doTestCloseOrDetachSenderTimesOutWhenNoCloseResponseReceived(true, true);
    }

    @Test
    public void testCloseSenderTimesOutWhenNoCloseResponseReceivedNoTimeout() throws Exception {
        doTestCloseOrDetachSenderTimesOutWhenNoCloseResponseReceived(true, false);
    }

    @Test
    public void testDetachSenderTimesOutWhenNoCloseResponseReceivedTimeout() throws Exception {
        doTestCloseOrDetachSenderTimesOutWhenNoCloseResponseReceived(false, true);
    }

    @Test
    public void testDetachSenderTimesOutWhenNoCloseResponseReceivedNoTimeout() throws Exception {
        doTestCloseOrDetachSenderTimesOutWhenNoCloseResponseReceived(false, false);
    }

    private void doTestCloseOrDetachSenderTimesOutWhenNoCloseResponseReceived(boolean close, boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.expectDetach();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions();
            options.closeTimeout(10);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            try {
                if (close) {
                    if (timeout) {
                        sender.closeAsync().get(10, TimeUnit.SECONDS);
                    } else {
                        sender.closeAsync().get();
                    }
                } else {
                    if (timeout) {
                        sender.detachAsync().get(10, TimeUnit.SECONDS);
                    } else {
                        sender.detachAsync().get();
                    }
                }

                fail("Should not complete the close or detach future without an error");
            } catch (ExecutionException exe) {
                Throwable cause = exe.getCause();
                assertTrue(cause instanceof ClientOperationTimedOutException);
            }

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

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

    @Test
    public void testSendTimesOutWhenNoCreditIssued() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions();
            options.sendTimeout(1);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            Message<String> message = Message.create("Hello World");
            try {
                sender.send(message);
                fail("Should throw a send timed out exception");
            } catch (ClientSendTimedOutException ex) {
                // Expected error, ignore
            }

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

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

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

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

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions();
            options.sendTimeout(10);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            Message<String> message = Message.create("Hello World");
            try {
                sender.send(message);
                fail("Should throw a send timed out exception");
            } catch (ClientSendTimedOutException ex) {
                // Expected error, ignore
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.remoteFlow().withLinkCredit(1).now();
            peer.expectAttach().ofSender().respond();
            peer.expectTransfer().withMessage().withValue("Hello World 2");
            peer.expectDetach().respond();
            peer.expectClose().respond();

            // Ensure the send happens after the remote has sent a flow with credit
            session.openSender("test-queue-2").openFuture().get();

            try {
                sender.send(Message.create("Hello World 2"));
            } catch (ClientException ex) {
                LOG.trace("Error on second send", ex);
                fail("Should not throw an exception");
            }

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

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

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

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

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions();
            options.sendTimeout(200);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            // Expect a transfer but only after the flow which is delayed to allow the
            // client time to block on credit.
            peer.expectTransfer().withNonNullPayload();
            peer.remoteFlow().withDeliveryCount(0)
                             .withLinkCredit(1)
                             .withIncomingWindow(1024)
                             .withOutgoingWindow(10)
                             .withNextIncomingId(0)
                             .withNextOutgoingId(1).later(30);
            peer.expectDetach().respond();
            peer.expectClose().respond();

            Message<String> message = Message.create("Hello World");
            try {
                LOG.debug("Attempting send with sender: {}", sender);
                sender.send(message);
            } catch (ClientSendTimedOutException ex) {
                fail("Should not throw a send timed out exception");
            }

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

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

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

    @Test
    public void testSendWhenCreditIsAvailable() throws Exception {
        doTestSendWhenCreditIsAvailable(false, false);
    }

    @Test
    public void testTrySendWhenCreditIsAvailable() throws Exception {
        doTestSendWhenCreditIsAvailable(true, false);
    }

    @Test
    public void testSendWhenCreditIsAvailableWithDeliveryAnnotations() throws Exception {
        doTestSendWhenCreditIsAvailable(false, true);
    }

    @Test
    public void testTrySendWhenCreditIsAvailableWithDeliveryAnnotations() throws Exception {
        doTestSendWhenCreditIsAvailable(true, true);
    }

    private void doTestSendWhenCreditIsAvailable(boolean trySend, boolean addDeliveryAnnotations) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withDeliveryCount(0)
                             .withLinkCredit(10)
                             .withIncomingWindow(1024)
                             .withOutgoingWindow(10)
                             .withNextIncomingId(0)
                             .withNextOutgoingId(1).queue();
            peer.expectAttach().ofReceiver().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            // This ensures that the flow to sender is processed before we try-send
            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().creditWindow(0));
            receiver.openFuture().get(10, TimeUnit.SECONDS);

            Map<String, Object> deliveryAnnotations = new HashMap<>();
            deliveryAnnotations.put("da1", 1);
            deliveryAnnotations.put("da2", 2);
            deliveryAnnotations.put("da3", 3);
            DeliveryAnnotationsMatcher daMatcher = new DeliveryAnnotationsMatcher(true);
            daMatcher.withEntry("da1", Matchers.equalTo(1));
            daMatcher.withEntry("da2", Matchers.equalTo(2));
            daMatcher.withEntry("da3", Matchers.equalTo(3));
            EncodedAmqpValueMatcher bodyMatcher = new EncodedAmqpValueMatcher("Hello World");
            TransferPayloadCompositeMatcher payloadMatcher = new TransferPayloadCompositeMatcher();
            if (addDeliveryAnnotations) {
                payloadMatcher.setDeliveryAnnotationsMatcher(daMatcher);
            }
            payloadMatcher.setMessageContentMatcher(bodyMatcher);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withPayload(payloadMatcher);
            peer.expectDetach().respond();
            peer.expectClose().respond();

            Message<String> message = Message.create("Hello World");

            final Tracker tracker;
            if (trySend) {
                if (addDeliveryAnnotations) {
                    tracker = sender.trySend(message, deliveryAnnotations);
                } else {
                    tracker = sender.trySend(message);
                }
            } else {
                if (addDeliveryAnnotations) {
                    tracker = sender.send(message, deliveryAnnotations);
                } else {
                    tracker = sender.send(message);
                }
            }

            assertNotNull(tracker);

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

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

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

    @Test
    public void testTrySendWhenNoCreditAvailable() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions();
            options.sendTimeout(1);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            Message<String> message = Message.create("Hello World");
            assertNull(sender.trySend(message));

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

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

    @Test
    public void testCreateSenderWithQoSOfAtMostOnce() throws Exception {
        doTestCreateSenderWithConfiguredQoS(DeliveryMode.AT_MOST_ONCE);
    }

    @Test
    public void testCreateSenderWithQoSOfAtLeastOnce() throws Exception {
        doTestCreateSenderWithConfiguredQoS(DeliveryMode.AT_LEAST_ONCE);
    }

    private void doTestCreateSenderWithConfiguredQoS(DeliveryMode qos) throws Exception {
        byte sndMode = qos == DeliveryMode.AT_MOST_ONCE ? SenderSettleMode.SETTLED.byteValue() : SenderSettleMode.UNSETTLED.byteValue();

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender()
                               .withSndSettleMode(sndMode)
                               .withRcvSettleMode(ReceiverSettleMode.FIRST.byteValue())
                               .respond()
                               .withSndSettleMode(sndMode)
                               .withRcvSettleMode(ReceiverSettleMode.FIRST.byteValue());
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            SenderOptions options = new SenderOptions().deliveryMode(qos);
            Sender sender = session.openSender("test-qos", options);
            sender.openFuture().get();

            assertEquals("test-qos", sender.address());

            sender.closeAsync();

            connection.closeAsync().get();

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

    @Test
    public void testSendAutoSettlesOnceRemoteSettles() throws Exception {
        doTestSentMessageGetsAutoSettledAfterRemoteSettles(false);
    }

    @Test
    public void testTrySendAutoSettlesOnceRemoteSettles() throws Exception {
        doTestSentMessageGetsAutoSettledAfterRemoteSettles(true);
    }

    private void doTestSentMessageGetsAutoSettledAfterRemoteSettles(boolean trySend) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withDeliveryCount(0)
                             .withLinkCredit(10)
                             .withIncomingWindow(1024)
                             .withOutgoingWindow(10)
                             .withNextIncomingId(0)
                             .withNextOutgoingId(1).queue();
            peer.expectAttach().ofReceiver().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get(10, TimeUnit.SECONDS);

            // This ensures that the flow to sender is processed before we try-send
            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().creditWindow(0));
            receiver.openFuture().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload()
                                 .respond()
                                 .withSettled(true).withState().accepted();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            Message<String> message = Message.create("Hello World");

            final Tracker tracker;
            if (trySend) {
                tracker = sender.trySend(message);
            } else {
                tracker = sender.send(message);
            }

            assertNotNull(tracker);
            assertNotNull(tracker.settlementFuture().get(5, TimeUnit.SECONDS));
            assertEquals(tracker.remoteState().getType(), DeliveryState.Type.ACCEPTED);

            sender.closeAsync();

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

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

    @Test
    public void testSendDoesNotAutoSettlesOnceRemoteSettlesIfAutoSettleOff() throws Exception {
        doTestSentMessageNotAutoSettledAfterRemoteSettles(false);
    }

    @Test
    public void testTrySendDoesNotAutoSettlesOnceRemoteSettlesIfAutoSettleOff() throws Exception {
        doTestSentMessageNotAutoSettledAfterRemoteSettles(true);
    }

    private void doTestSentMessageNotAutoSettledAfterRemoteSettles(boolean trySend) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withDeliveryCount(0)
                             .withLinkCredit(10)
                             .withIncomingWindow(1024)
                             .withOutgoingWindow(10)
                             .withNextIncomingId(0)
                             .withNextOutgoingId(1).queue();
            peer.expectAttach().ofReceiver().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue", new SenderOptions().autoSettle(false));
            sender.openFuture().get(10, TimeUnit.SECONDS);

            // This ensures that the flow to sender is processed before we try-send
            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().creditWindow(0));
            receiver.openFuture().get(10, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload()
                                 .respond()
                                 .withSettled(true).withState().accepted();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            Message<String> message = Message.create("Hello World");

            final Tracker tracker;
            if (trySend) {
                tracker = sender.trySend(message);
            } else {
                tracker = sender.send(message);
            }

            assertNotNull(tracker);
            assertTrue(tracker.settlementFuture().get(5, TimeUnit.SECONDS).remoteSettled());
            assertEquals(tracker.remoteState().getType(), DeliveryState.Type.ACCEPTED);
            assertNull(tracker.state());
            assertFalse(tracker.settled());

            sender.closeAsync();

            connection.closeAsync().get();

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

    @Test
    public void testSenderSendingSettledCompletesTrackerAcknowledgeFuture() throws Exception {
        doTestSenderSendingSettledCompletesTrackerAcknowledgeFuture(false);
    }

    @Test
    public void testSenderTrySendingSettledCompletesTrackerAcknowledgeFuture() throws Exception {
        doTestSenderSendingSettledCompletesTrackerAcknowledgeFuture(true);
    }

    private void doTestSenderSendingSettledCompletesTrackerAcknowledgeFuture(boolean trySend) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender()
                               .withSenderSettleModeSettled()
                               .withReceiverSettlesFirst()
                               .respond()
                               .withSenderSettleModeSettled()
                               .withReceivervSettlesFirst();
            peer.remoteFlow().withLinkCredit(10).queue();
            peer.expectAttach().respond();  // Open a receiver to ensure sender link has processed
            peer.expectFlow();              // the inbound flow frame we sent previously before send.
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            Session session = connection.openSession().openFuture().get();

            SenderOptions options = new SenderOptions().deliveryMode(DeliveryMode.AT_MOST_ONCE);
            Sender sender = session.openSender("test-qos", options);
            assertEquals("test-qos", sender.address());
            session.openReceiver("dummy").openFuture().get();

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

            final Message<String> message = Message.create("Hello World");
            final Tracker tracker;
            if (trySend) {
                tracker = sender.trySend(message);
            } else {
                tracker = sender.send(message);
            }

            assertNotNull(tracker);
            assertTrue(tracker.settlementFuture().isDone());
            assertTrue(tracker.settlementFuture().get().settled());
            assertFalse(tracker.settlementFuture().get().remoteSettled());

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

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

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

    @Test
    public void testSenderIncrementsTransferTagOnEachSend() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(10).queue();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            Session session = connection.openSession().openFuture().get();
            SenderOptions options = new SenderOptions().deliveryMode(DeliveryMode.AT_LEAST_ONCE).autoSettle(false);
            Sender sender = session.openSender("test-tags", options).openFuture().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {0}).respond().withSettled(true).withState().accepted();
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {1}).respond().withSettled(true).withState().accepted();
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {2}).respond().withSettled(true).withState().accepted();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            final Message<String> message = Message.create("Hello World");
            final Tracker tracker1 = sender.send(message);
            final Tracker tracker2 = sender.send(message);
            final Tracker tracker3 = sender.send(message);

            assertNotNull(tracker1);
            assertNotNull(tracker1.settlementFuture().get().settled());
            assertNotNull(tracker2);
            assertNotNull(tracker2.settlementFuture().get().settled());
            assertNotNull(tracker3);
            assertNotNull(tracker3.settlementFuture().get().settled());

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

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

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

    @Test
    public void testSenderSendsSettledInAtLeastOnceMode() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(10).queue();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            Session session = connection.openSession().openFuture().get();
            SenderOptions options = new SenderOptions().deliveryMode(DeliveryMode.AT_MOST_ONCE).autoSettle(false);
            Sender sender = session.openSender("test-tags", options).openFuture().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {}).withSettled(true);
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {}).withSettled(true);
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {}).withSettled(true);
            peer.expectDetach().respond();
            peer.expectClose().respond();

            final Message<String> message = Message.create("Hello World");
            final Tracker tracker1 = sender.send(message);
            final Tracker tracker2 = sender.send(message);
            final Tracker tracker3 = sender.send(message);

            assertNotNull(tracker1);
            assertNotNull(tracker1.settlementFuture().get().settled());
            assertNotNull(tracker2);
            assertNotNull(tracker2.settlementFuture().get().settled());
            assertNotNull(tracker3);
            assertNotNull(tracker3.settlementFuture().get().settled());

            sender.closeAsync().get();

            connection.closeAsync().get();

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

    @Test
    public void testCreateAnonymousSenderWhenRemoteDoesNotOfferSupportForIt() 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());
            Session session = connection.openSession().openFuture().get();

            try {
                session.openAnonymousSender();
                fail("Should not be able to open an anonymous sender when remote does not offer anonymous relay");
            } catch (ClientUnsupportedOperationException unsupported) {
                LOG.info("Caught expected error: ", unsupported);
            }

            connection.closeAsync();

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

    @Test
    public void testCreateAnonymousSenderBeforeKnowingRemoteDoesNotOfferSupportForIt() 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());
            Session session = connection.openSession();
            Sender anonymousSender = session.openAnonymousSender();
            Message<String> message = Message.create("Hello World").to("my-queue");

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

            try {
                anonymousSender.send(message);
                fail("Should not be able to open an anonymous sender when remote does not offer anonymous relay");
            } catch (ClientUnsupportedOperationException unsupported) {
                LOG.info("Caught expected error: ", unsupported);
            }

            connection.closeAsync();

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

    @Test
    public void testCreateAnonymousSenderAppliesOptions() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond().withOfferedCapabilities("ANONYMOUS-RELAY");
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().withSenderSettleModeSettled()
                                          .withReceiverSettlesFirst()
                                          .withTarget().withAddress(Matchers.nullValue()).and().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());
            Session session = connection.openSession();
            SenderOptions senderOptions = new SenderOptions().deliveryMode(DeliveryMode.AT_MOST_ONCE);
            Sender anonymousSender = session.openAnonymousSender(senderOptions);

            anonymousSender.openFuture().get();

            connection.closeAsync();

            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());
            Session session = connection.openSession();
            Sender sender = session.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 testSenderGetRemotePropertiesWaitsForRemoteAttach() throws Exception {
        tryReadSenderRemoteProperties(true);
    }

    @Test
    public void testSenderGetRemotePropertiesFailsAfterOpenTimeout() throws Exception {
        tryReadSenderRemoteProperties(false);
    }

    private void tryReadSenderRemoteProperties(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender();
            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());
            connection.openFuture().get();

            Session session = connection.openSession();
            session.openFuture().get();

            SenderOptions options = new SenderOptions().openTimeout(100);
            Sender sender = session.openSender("test-sender", options);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

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

            if (attachResponse) {
                peer.expectDetach().respond();
                peer.respondToLastAttach().withPropertiesMap(expectedProperties).later(10);
            } else {
                peer.expectDetach();
            }

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

            try {
                sender.closeAsync().get();
            } catch (ExecutionException ex) {
                LOG.debug("Caught unexpected exception from close call", ex);
                fail("Should not fail to close when connection not closed and detach sent.");
            }

            LOG.debug("*** Test read remote properties ***");

            peer.expectClose().respond();
            connection.closeAsync().get();

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

    @Test
    public void testGetRemoteOfferedCapabilitiesWaitsForRemoteAttach() throws Exception {
        tryReadRemoteOfferedCapabilities(true);
    }

    @Test
    public void testGetRemoteOfferedCapabilitiesFailsAfterOpenTimeout() throws Exception {
        tryReadRemoteOfferedCapabilities(false);
    }

    private void tryReadRemoteOfferedCapabilities(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().openTimeout(100);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession();
            Sender sender = session.openSender("test-sender");

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            if (attachResponse) {
                peer.expectDetach().respond();
                peer.respondToLastAttach().withOfferedCapabilities("QUEUE").later(10);
            } else {
                peer.expectDetach();
            }

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

            try {
                sender.closeAsync().get();
            } catch (ExecutionException ex) {
                LOG.debug("Caught unexpected exception from close call", ex);
                fail("Should not fail to close when connection not closed and detach sent.");
            }

            peer.expectClose().respond();
            connection.closeAsync().get();

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

    @Test
    public void testGetRemoteDesiredCapabilitiesWaitsForRemoteAttach() throws Exception {
        tryReadRemoteDesiredCapabilities(true);
    }

    @Test
    public void testGetRemoteDesiredCapabilitiesFailsAfterOpenTimeout() throws Exception {
        tryReadRemoteDesiredCapabilities(false);
    }

    private void tryReadRemoteDesiredCapabilities(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().openTimeout(100);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession();
            Sender sender = session.openSender("test-sender");

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            if (attachResponse) {
                peer.expectDetach().respond();
                peer.respondToLastAttach().withDesiredCapabilities("Error-Free").later(10);
            } else {
                peer.expectDetach();
            }

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

            try {
                sender.closeAsync().get();
            } catch (ExecutionException ex) {
                LOG.debug("Caught unexpected exception from close call", ex);
                fail("Should not fail to close when connection not closed and detach sent.");
            }

            peer.expectClose().respond();
            connection.closeAsync().get();

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

    @Test
    public void testOpenSenderWithLinCapabilities() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.SENDER.getValue())
                               .withTarget().withCapabilities("queue").and()
                               .respond();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession().openFuture().get(10, TimeUnit.SECONDS);
            SenderOptions senderOptions = new SenderOptions();
            senderOptions.targetOptions().capabilities("queue");
            Sender sender = session.openSender("test-queue", senderOptions);

            sender.openFuture().get();
            sender.close();

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

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

    @Test
    public void testCloseSenderWithErrorCondition() throws Exception {
        doTestCloseOrDetachWithErrorCondition(true);
    }

    @Test
    public void testDetachSenderWithErrorCondition() throws Exception {
        doTestCloseOrDetachWithErrorCondition(false);
    }

    public void doTestCloseOrDetachWithErrorCondition(boolean close) throws Exception {
        final String condition = "amqp:link:detach-forced";
        final String description = "something bad happened.";

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.expectDetach().withClosed(close).withError(condition, description).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());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-sender");
            sender.openFuture().get();

            if (close) {
                sender.closeAsync(ErrorCondition.create(condition, description, null));
            } else {
                sender.detachAsync(ErrorCondition.create(condition, description, null));
            }

            connection.closeAsync().get();

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

    @Test
    public void testSendMultipleMessages() throws Exception {
        final int CREDIT = 20;

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withDeliveryCount(0).withLinkCredit(CREDIT).queue();
            peer.expectAttach().ofReceiver().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get();

            // This ensures that the flow to sender is processed before we try-send
            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().creditWindow(0));
            receiver.openFuture().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            final List<Tracker> sentMessages = new ArrayList<>();

            for (int i = 0; i < CREDIT; ++i) {
                peer.expectTransfer().withDeliveryId(i)
                                     .withNonNullPayload()
                                     .withSettled(false)
                                     .respond()
                                     .withSettled(true)
                                     .withState().accepted();
            }
            peer.expectDetach().respond();
            peer.expectClose().respond();

            Message<String> message = Message.create("Hello World");

            for (int i = 0; i < CREDIT; ++i) {
                final Tracker tracker = sender.send(message);
                sentMessages.add(tracker);
                tracker.settlementFuture().get();
            }
            assertEquals(CREDIT, sentMessages.size());

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testSendBlockedForCreditFailsWhenLinkRemotelyClosed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteDetach().withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Link was deleted").afterDelay(25).queue();
            peer.expectDetach();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get();

            Message<String> message = Message.create("Hello World");

            try {
                sender.send(message);
                fail("Send should have timed out.");
            } catch (ClientResourceRemotelyClosedException cliEx) {
                // Expected send to throw indicating that the remote closed the link
            }

            connection.closeAsync().get();

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

    @Test
    public void testSendBlockedForCreditFailsWhenSessionRemotelyClosed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteEnd().withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Session was deleted").afterDelay(25).queue();
            peer.expectEnd();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get();

            Message<String> message = Message.create("Hello World");

            try {
                sender.send(message);
                fail("Send should have timed out.");
            } catch (ClientResourceRemotelyClosedException cliEx) {
                // Expected send to throw indicating that the remote closed the session
            }

            connection.closeAsync().get();

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

    @Test
    public void testSendBlockedForCreditFailsWhenConnectionRemotelyClosed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteClose().withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Connection was deleted").afterDelay(25).queue();
            peer.expectClose();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get();

            Message<String> message = Message.create("Hello World");

            try {
                sender.send(message);
                fail("Send should have failed when Connection remotely closed.");
            } catch (ClientConnectionRemotelyClosedException cliEx) {
                // Expected send to throw indicating that the remote closed the connection
            }

            connection.closeAsync().get();

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

    @Test
    public void testSendBlockedForCreditFailsWhenConnectionDrops() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.dropAfterLastHandler(25);
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");
            sender.openFuture().get();

            Message<String> message = Message.create("Hello World");

            try {
                sender.send(message);
                fail("Send should have timed out.");
            } catch (ClientConnectionRemotelyClosedException cliEx) {
                // Expected send to throw indicating that the remote closed unexpectedly
            }

            connection.closeAsync().get();

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

    @Test
    public void testSendAfterConnectionDropsThrowsConnectionRemotelyClosedError() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().withTarget().withAddress("test").and().respond();
            peer.dropAfterLastHandler(25);
            peer.start();

            final CountDownLatch dropped = new CountDownLatch(1);

            ConnectionOptions options = new ConnectionOptions();
            options.disconnectedHandler((connection, event) -> {
                dropped.countDown();
            });

            URI remoteURI = peer.getServerURI();

            Message<String> message = Message.create("test-message");
            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession();
            Sender sender = session.openSender("test");

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

            try {
                sender.send(message);
                fail("Send should fail with remotely closed error after remote drops");
            } catch (ClientConnectionRemotelyClosedException cliEx) {
                // Expected
            }

            try {
                sender.trySend(message);
                fail("trySend should fail with remotely closed error after remote drops");
            } catch (ClientConnectionRemotelyClosedException cliEx) {
                // Expected
            }

            connection.closeAsync().get();

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

    @Test
    public void testAwaitSettlementFutureFailedAfterConnectionDropped() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().withTarget().withAddress("test").and().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectTransfer();
            peer.dropAfterLastHandler();
            peer.start();

            URI remoteURI = peer.getServerURI();

            Message<String> message = Message.create("test-message");
            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test");

            Tracker tracker = null;
            try {
                tracker = sender.send(message);
            } catch (ClientConnectionRemotelyClosedException cliEx) {
                fail("Send not should fail with remotely closed error after remote drops");
            }

            // Connection should be dropped at this point and next call should test that after
            // the drop the future has been completed
            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            try {
                tracker.settlementFuture().get();
                fail("Wait for settlement should fail with remotely closed error after remote drops");
            } catch (ExecutionException exe) {
                assertTrue(exe.getCause() instanceof ClientConnectionRemotelyClosedException);
            }

            connection.closeAsync().get();

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

    @Test
    public void testAwaitSettlementFailedOnConnectionDropped() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().withTarget().withAddress("test").and().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectTransfer();
            peer.dropAfterLastHandler(30);
            peer.start();

            URI remoteURI = peer.getServerURI();

            Message<String> message = Message.create("test-message");
            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test");

            Tracker tracker = null;
            try {
                tracker = sender.send(message);
            } catch (ClientConnectionRemotelyClosedException cliEx) {
                fail("Send should not fail with remotely closed error after remote drops");
            }

            // Most of the time this should await before connection drops testing that
            // the drop completes waiting callers.
            try {
                tracker.awaitSettlement();
                fail("Wait for settlement should fail with remotely closed error after remote drops");
            } catch (ClientConnectionRemotelyClosedException cliRCEx) {
            }

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

    @Test
    public void testBlockedSendThrowsConnectionRemotelyClosedError() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().withTarget().withAddress("test").and().respond();
            peer.dropAfterLastHandler(25);
            peer.start();

            URI remoteURI = peer.getServerURI();

            Message<String> message = Message.create("test-message");
            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test");

            try {
                sender.send(message);
                fail("Send should fail with remotely closed error after remote drops");
            } catch (ClientConnectionRemotelyClosedException cliEx) {
                // Expected
            }

            connection.closeAsync().get();

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

    @Test
    void testAutoFlushDuringWriteThatExceedConfiguredBufferLimitSessionCreditLimitOnTransfer() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().withNextOutgoingId(0).respond();
            peer.expectAttach().ofSender().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().maxFrameSize(1024);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Sender sender = connection.openSender("test-queue");

            final byte[] payload = new byte[4800];
            Arrays.fill(payload, (byte) 1);

            final AtomicBoolean sendFailed = new AtomicBoolean();
            ForkJoinPool.commonPool().execute(() -> {
                try {
                    sender.send(Message.create(payload));
                } catch (Exception e) {
                    LOG.info("send failed with error: ", e);
                    sendFailed.set(true);
                }
            });

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload().withMore(true).withFrameSize(1024);
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(1).withLinkCredit(10).queue();
            peer.expectTransfer().withNonNullPayload().withMore(true).withFrameSize(1024);
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(2).withLinkCredit(10).queue();
            peer.expectTransfer().withNonNullPayload().withMore(true).withFrameSize(1024);
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(3).withLinkCredit(10).queue();
            peer.expectTransfer().withNonNullPayload().withMore(true).withFrameSize(1024);
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(4).withLinkCredit(10).queue();
            peer.expectTransfer().withNonNullPayload().withMore(false).accept();

            // Grant the credit to start meeting the above expectations
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(0).withLinkCredit(10).now();

            peer.waitForScriptToComplete(500, TimeUnit.SECONDS);
            peer.expectDetach().respond();
            peer.expectClose().respond();

            assertFalse(sendFailed.get());

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    void testAutoFlushDuringWriteWithRollingIncomingWindowUpdates() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().withNextOutgoingId(0).respond();
            peer.expectAttach().ofSender().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().maxFrameSize(1024);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Sender sender = connection.openSender("test-queue");

            final byte[] payload = new byte[4800];
            Arrays.fill(payload, (byte) 1);

            final AtomicBoolean sendFailed = new AtomicBoolean();
            ForkJoinPool.commonPool().execute(() -> {
                try {
                    sender.send(Message.create(payload));
                } catch (Exception e) {
                    LOG.info("send failed with error: ", e);
                    sendFailed.set(true);
                }
            });

            // Credit should will be refilling as transfers arrive vs being exhausted on each
            // incoming transfer and the send awaiting more credit.
            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload().withMore(true);
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(2).withLinkCredit(10).queue();
            peer.expectTransfer().withNonNullPayload().withMore(true);
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(3).withLinkCredit(10).queue();
            peer.expectTransfer().withNonNullPayload().withMore(true);
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(4).withLinkCredit(10).queue();
            peer.expectTransfer().withNonNullPayload().withMore(true);
            peer.expectTransfer().withNonNullPayload().withMore(false).accept();

            // Grant the credit to start meeting the above expectations
            peer.remoteFlow().withIncomingWindow(2).withNextIncomingId(0).withLinkCredit(10).now();

            peer.waitForScriptToComplete(500, TimeUnit.SECONDS);
            peer.expectDetach().respond();
            peer.expectClose().respond();

            assertFalse(sendFailed.get());

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testSendTimesOutIfNotAllMessageFramesCanBeSent() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().withNextOutgoingId(0).respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withIncomingWindow(2).withNextIncomingId(0).withLinkCredit(1).queue();
            peer.expectTransfer().withDeliveryId(0).withNonNullPayload().withMore(true);
            peer.expectTransfer().withNonNullPayload().withMore(true);
            peer.expectTransfer().withNullPayload().withAborted(true);
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().maxFrameSize(1024);
            options.sendTimeout(25);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Sender sender = connection.openSender("test-queue").openFuture().get();

            final byte[] payload = new byte[4800];
            Arrays.fill(payload, (byte) 1);

            try {
                sender.send(Message.create(payload));
            } catch (ClientSendTimedOutException e) {
                LOG.trace("send failed with expected error: ", e);
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.remoteFlow().withIncomingWindow(1).withNextIncomingId(4).withLinkCredit(1).now();
            peer.expectAttach().ofSender().respond();
            peer.expectTransfer().withDeliveryId(1).withMessage().withValue("Hello World 2");
            peer.expectDetach().respond();
            peer.expectClose().respond();

            // Ensure the send happens after the remote has sent a flow with credit
            connection.openSender("test-queue-2").openFuture().get();

            try {
                sender.send(Message.create("Hello World 2"));
            } catch (ClientException ex) {
                LOG.trace("Error on second send", ex);
                fail("Should not throw an exception");
            }

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    void testConcurrentSendOnlyBlocksForInitialSendInProgress() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().accepted();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().accepted();
            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.openSender("test-queue").openFuture().get();
            // Ensure that sender gets its flow before the sends are triggered.
            connection.openReceiver("test-queue").openFuture().get();

            final byte[] payload = new byte[1024];
            Arrays.fill(payload, (byte) 1);

            // One should block on the send waiting for the others send to finish
            // otherwise they should not care about concurrency of sends.

            final AtomicBoolean sendFailed = new AtomicBoolean();
            ForkJoinPool.commonPool().execute(() -> {
                try {
                    LOG.info("Test send 1 is preparing to fire:");
                    Tracker tracker = sender.send(Message.create(payload));
                    tracker.awaitSettlement(10, TimeUnit.SECONDS);
                } catch (Exception e) {
                    LOG.info("Test send 1 failed with error: ", e);
                    sendFailed.set(true);
                }
            });

            ForkJoinPool.commonPool().execute(() -> {
                try {
                    LOG.info("Test send 2 is preparing to fire:");
                    Tracker tracker = sender.send(Message.create(payload));
                    tracker.awaitSettlement(10, TimeUnit.SECONDS);
                } catch (Exception e) {
                    LOG.info("Test send 2 failed with error: ", e);
                    sendFailed.set(true);
                }
            });

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

            assertFalse(sendFailed.get());

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testConcurrentSendBlocksBehindSendWaitingForCredit() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().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());
            Sender sender = connection.openSender("test-queue").openFuture().get();

            final byte[] payload = new byte[1024];
            Arrays.fill(payload, (byte) 1);

            final CountDownLatch send1Started = new CountDownLatch(1);
            final CountDownLatch send2Completed = new CountDownLatch(1);

            final AtomicBoolean sendFailed = new AtomicBoolean();
            ForkJoinPool.commonPool().execute(() -> {
                try {
                    LOG.info("Test send 1 is preparing to fire:");
                    ForkJoinPool.commonPool().execute(() -> send1Started.countDown());
                    sender.send(Message.create(payload));
                } catch (Exception e) {
                    LOG.info("Test send 1 failed with error: ", e);
                    sendFailed.set(true);
                }
            });

            ForkJoinPool.commonPool().execute(() -> {
                try {
                    assertTrue(send1Started.await(10, TimeUnit.SECONDS));
                    LOG.info("Test send 2 is preparing to fire:");
                    Tracker tracker = sender.send(Message.create(payload));
                    tracker.awaitSettlement(10, TimeUnit.SECONDS);
                    send2Completed.countDown();
                } catch (Exception e) {
                    LOG.info("Test send 2 failed with error: ", e);
                    sendFailed.set(true);
                }
            });

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().accepted();
            peer.remoteFlow().withIncomingWindow(1).withDeliveryCount(1).withNextIncomingId(2).withLinkCredit(1).queue();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().accepted();

            // Now Trigger transfer / flow event chain
            peer.remoteFlow().withIncomingWindow(1).withDeliveryCount(0).withNextIncomingId(1).withLinkCredit(1).later(10);

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

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

            assertFalse(sendFailed.get());

            sender.closeAsync().get();
            connection.closeAsync().get();

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

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

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().maxFrameSize(1024);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Sender sender = connection.openSender("test-queue");

            final byte[] payload = new byte[1536];
            Arrays.fill(payload, (byte) 1);

            final CountDownLatch send1Started = new CountDownLatch(1);
            final CountDownLatch send2Completed = new CountDownLatch(1);

            final AtomicBoolean sendFailed = new AtomicBoolean();
            ForkJoinPool.commonPool().execute(() -> {
                try {
                    LOG.info("Test send 1 is preparing to fire:");
                    ForkJoinPool.commonPool().execute(() -> send1Started.countDown());
                    sender.send(Message.create(payload));
                } catch (Exception e) {
                    LOG.info("Test send 1 failed with error: ", e);
                    sendFailed.set(true);
                }
            });

            ForkJoinPool.commonPool().execute(() -> {
                try {
                    assertTrue(send1Started.await(10, TimeUnit.SECONDS));
                    LOG.info("Test send 2 is preparing to fire:");
                    Tracker tracker = sender.send(Message.create(payload));
                    tracker.awaitSettlement(10, TimeUnit.SECONDS);
                    send2Completed.countDown();
                } catch (Exception e) {
                    LOG.info("Test send 2 failed with error: ", e);
                    sendFailed.set(true);
                }
            });

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload().withMore(true);
            peer.remoteFlow().withIncomingWindow(1).withDeliveryCount(0).withNextIncomingId(2).withLinkCredit(1).queue();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().accepted();
            peer.remoteFlow().withIncomingWindow(1).withDeliveryCount(1).withNextIncomingId(3).withLinkCredit(1).queue();
            peer.expectTransfer().withNonNullPayload().withMore(true);
            peer.remoteFlow().withIncomingWindow(1).withDeliveryCount(1).withNextIncomingId(4).withLinkCredit(1).queue();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().accepted();

            // Trigger chain of events after script is configured.
            peer.remoteFlow().withIncomingWindow(1).withDeliveryCount(0).withNextIncomingId(1).withLinkCredit(1).now();

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

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

            assertFalse(sendFailed.get());

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testCreateSenderWithDefaultSourceAndTargetOptions() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender()
                               .withSource().withAddress(notNullValue())
                                            .withDistributionMode(nullValue())
                                            .withDefaultTimeout()
                                            .withDurable(TerminusDurability.NONE)
                                            .withExpiryPolicy(TerminusExpiryPolicy.LINK_DETACH)
                                            .withDefaultOutcome(nullValue())
                                            .withCapabilities(nullValue())
                                            .withFilter(nullValue())
                                            .withOutcomes("amqp:accepted:list", "amqp:rejected:list", "amqp:released:list", "amqp:modified:list")
                                            .also()
                               .withTarget().withAddress("test-queue")
                                            .withCapabilities(nullValue())
                                            .withDurable(nullValue())
                                            .withExpiryPolicy(nullValue())
                                            .withDefaultTimeout()
                                            .withDynamic(anyOf(nullValue(), equalTo(false)))
                                            .withDynamicNodeProperties(nullValue())
                               .and().respond();
            peer.expectDetach().respond();
            peer.expectEnd().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());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue").openFuture().get();

            sender.close();
            session.close();
            connection.close();

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

    @Test
    public void testCreateSenderWithUserConfiguredSourceAndTargetOptions() throws Exception {
        final Map<String, Object> filters = new HashMap<>();
        filters.put("x-opt-filter", "a = b");

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender()
                               .withSource().withAddress(notNullValue())
                                            .withDistributionMode("copy")
                                            .withTimeout(128)
                                            .withDurable(TerminusDurability.UNSETTLED_STATE)
                                            .withExpiryPolicy(TerminusExpiryPolicy.CONNECTION_CLOSE)
                                            .withDefaultOutcome(new Released())
                                            .withCapabilities("QUEUE")
                                            .withFilter(filters)
                                            .withOutcomes("amqp:accepted:list", "amqp:rejected:list")
                                            .also()
                               .withTarget().withAddress("test-queue")
                                            .withCapabilities("QUEUE")
                                            .withDurable(TerminusDurability.CONFIGURATION)
                                            .withExpiryPolicy(TerminusExpiryPolicy.SESSION_END)
                                            .withTimeout(42)
                                            .withDynamic(anyOf(nullValue(), equalTo(false)))
                                            .withDynamicNodeProperties(nullValue())
                               .and().respond();
            peer.expectDetach().respond();
            peer.expectEnd().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());
            Session session = connection.openSession();
            SenderOptions senderOptions = new SenderOptions();

            senderOptions.sourceOptions().capabilities("QUEUE");
            senderOptions.sourceOptions().distributionMode(DistributionMode.COPY);
            senderOptions.sourceOptions().timeout(128);
            senderOptions.sourceOptions().durabilityMode(DurabilityMode.UNSETTLED_STATE);
            senderOptions.sourceOptions().expiryPolicy(ExpiryPolicy.CONNECTION_CLOSE);
            senderOptions.sourceOptions().defaultOutcome(DeliveryState.released());
            senderOptions.sourceOptions().filters(filters);
            senderOptions.sourceOptions().outcomes(DeliveryState.Type.ACCEPTED, DeliveryState.Type.REJECTED);

            senderOptions.targetOptions().capabilities("QUEUE");
            senderOptions.targetOptions().durabilityMode(DurabilityMode.CONFIGURATION);
            senderOptions.targetOptions().expiryPolicy(ExpiryPolicy.SESSION_CLOSE);
            senderOptions.targetOptions().timeout(42);

            Sender sender = session.openSender("test-queue", senderOptions).openFuture().get();

            sender.close();
            session.close();
            connection.close();

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

    @Test
    public void testWaitForAcceptedReturnsOnRemoteAcceptance() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().accepted();
            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.openSender("test-queue").openFuture().get();
            Tracker tracker = sender.send(Message.create("Hello World"));
            tracker.awaitAccepted();

            assertTrue(tracker.remoteSettled());
            assertTrue(tracker.remoteState().isAccepted());

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

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testWaitForAcceptanceFailsIfRemoteSendsRejected() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectTransfer().withNonNullPayload()
                                 .withMore(false)
                                 .respond()
                                 .withSettled(true)
                                 .withState().rejected("RejectedA", "RejectedB");
            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.openSender("test-queue").openFuture().get();
            Tracker tracker = sender.send(Message.create("Hello World"));

            try {
                tracker.awaitAccepted(10, TimeUnit.SECONDS);
                fail("Should not succeed since remote sent something other than Accepted");
            } catch (ClientDeliveryStateException dlvEx) {
                // Expected
            }

            assertTrue(tracker.remoteSettled());
            assertFalse(tracker.remoteState().isAccepted());
            assertTrue(tracker.remoteState().getType().equals(DeliveryState.Type.REJECTED));

            ClientRejected remoteState = (ClientRejected) tracker.remoteState();
            Rejected protonState = (Rejected) remoteState.getProtonDeliveryState();

            assertNotNull(protonState);
            assertNotNull(protonState.getError());
            assertEquals("RejectedA", protonState.getError().getCondition().toString());
            assertEquals("RejectedB", protonState.getError().getDescription());

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

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testWaitForAcceptanceFailsIfRemoteSendsReleased() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().released();
            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.openSender("test-queue").openFuture().get();
            Tracker tracker = sender.send(Message.create("Hello World"));

            try {
                tracker.awaitAccepted(10, TimeUnit.SECONDS);
                fail("Should not succeed since remote sent something other than Accepted");
            } catch (ClientDeliveryStateException dlvEx) {
                // Expected
            }

            assertTrue(tracker.remoteSettled());
            assertFalse(tracker.remoteState().isAccepted());
            assertTrue(tracker.remoteState().getType().equals(DeliveryState.Type.RELEASED));

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

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testWaitForAcceptanceFailsIfRemoteSendsModified() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true).withState().modified();
            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.openSender("test-queue").openFuture().get();
            Tracker tracker = sender.send(Message.create("Hello World"));

            try {
                tracker.awaitAccepted(10, TimeUnit.SECONDS);
                fail("Should not succeed since remote sent something other than Accepted");
            } catch (ClientDeliveryStateException dlvEx) {
                // Expected
            }

            assertTrue(tracker.remoteSettled());
            assertFalse(tracker.remoteState().isAccepted());
            assertTrue(tracker.remoteState().getType().equals(DeliveryState.Type.MODIFIED));

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

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testWaitForAcceptanceFailsIfRemoteSendsNoDisposition() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectTransfer().withNonNullPayload().withMore(false).respond().withSettled(true);
            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.openSender("test-queue").openFuture().get();
            Tracker tracker = sender.send(Message.create("Hello World"));

            try {
                tracker.awaitAccepted(10, TimeUnit.SECONDS);
                fail("Should not succeed since remote sent something other than Accepted");
            } catch (ClientDeliveryStateException dlvEx) {
                // Expected
            }

            assertTrue(tracker.remoteSettled());
            assertNull(tracker.remoteState());

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

            sender.closeAsync().get();
            connection.closeAsync().get();

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

    @Test
    public void testSenderLinkNameOptionAppliedWhenSet() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().withName("custom-link-name").respond();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            SenderOptions senderOptions = new SenderOptions().linkName("custom-link-name");
            Sender sender = session.openSender("test-queue", senderOptions);

            sender.openFuture().get();
            sender.close();

            connection.close();

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

    @Test
    public void testInspectRemoteSourceMatchesValuesSent() throws Exception {
        Map<String, Object> remoteFilters = new HashMap<>();
        remoteFilters.put("filter-1", "value1");

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond().withSource().withOutcomes("Accepted", "Released")
                                                                 .withCapabilities("Queue")
                                                                 .withDistributionMode("COPY")
                                                                 .withDynamic(false)
                                                                 .withExpiryPolicy(TerminusExpiryPolicy.SESSION_END)
                                                                 .withDurability(TerminusDurability.UNSETTLED_STATE)
                                                                 .withDefaultOutcome(Released.getInstance())
                                                                 .withTimeout(Integer.MAX_VALUE)
                                                                 .withFilterMap(remoteFilters)
                                                                 .withAddress("test-queue");
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");

            Source remoteSource = sender.source();

            assertTrue(remoteSource.outcomes().contains(DeliveryState.Type.ACCEPTED));
            assertTrue(remoteSource.capabilities().contains("Queue"));
            assertEquals("test-queue", remoteSource.address());
            assertFalse(remoteSource.dynamic());
            assertNull(remoteSource.dynamicNodeProperties());
            assertEquals(DistributionMode.COPY, remoteSource.distributionMode());
            assertEquals(DeliveryState.released(), remoteSource.defaultOutcome());
            assertEquals(Integer.MAX_VALUE, remoteSource.timeout());
            assertEquals(DurabilityMode.UNSETTLED_STATE, remoteSource.durabilityMode());
            assertEquals(ExpiryPolicy.SESSION_CLOSE, remoteSource.expiryPolicy());
            assertEquals(remoteFilters, remoteSource.filters());

            sender.close();
            connection.close();

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

    @Test
    public void testInspectRemoteTargetMatchesValuesSent() throws Exception {
        Map<String, Object> remoteFilters = new HashMap<>();
        remoteFilters.put("filter-1", "value1");

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond().withTarget().withCapabilities("Queue")
                                                                 .withDynamic(false)
                                                                 .withExpiryPolicy(TerminusExpiryPolicy.SESSION_END)
                                                                 .withDurability(TerminusDurability.UNSETTLED_STATE)
                                                                 .withTimeout(Integer.MAX_VALUE)
                                                                 .withAddress("test-queue");
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test-queue");

            Target remoteTarget = sender.target();

            assertTrue(remoteTarget.capabilities().contains("Queue"));
            assertEquals("test-queue", remoteTarget.address());
            assertFalse(remoteTarget.dynamic());
            assertEquals(Integer.MAX_VALUE, remoteTarget.timeout());
            assertEquals(DurabilityMode.UNSETTLED_STATE, remoteTarget.durabilityMode());
            assertEquals(ExpiryPolicy.SESSION_CLOSE, remoteTarget.expiryPolicy());

            sender.close();
            connection.close();

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

    private static DeliveryTagGenerator customTagGenerator() {
        return new DeliveryTagGenerator() {

            private int count = 1;

            @Override
            public DeliveryTag nextTag() {
                switch (count++) {
                    case 1:
                        return new DeliveryTag.ProtonDeliveryTag(new byte[] { 1, 1, 1 });
                    case 2:
                        return new DeliveryTag.ProtonDeliveryTag(new byte[] { 2, 2, 2 });
                    case 3:
                        return new DeliveryTag.ProtonDeliveryTag(new byte[] { 3, 3, 3 });
                    default:
                        throw new UnsupportedOperationException("Only supports creating three tags");
                }
            }
        };
    }

    @Test
    public void testSenderUsesCustomDeliveryTagGeneratorConfiguration() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(10).queue();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            Session session = connection.openSession().openFuture().get();
            SenderOptions options = new SenderOptions().deliveryMode(DeliveryMode.AT_LEAST_ONCE)
                                                       .autoSettle(false)
                                                       .deliveryTagGeneratorSupplier(SenderTest::customTagGenerator);
            Sender sender = session.openSender("test-tags", options).openFuture().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {1, 1, 1}).respond().withSettled(true).withState().accepted();
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {2, 2, 2}).respond().withSettled(true).withState().accepted();
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {3, 3, 3}).respond().withSettled(true).withState().accepted();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            final Message<String> message = Message.create("Hello World");
            final Tracker tracker1 = sender.send(message);
            final Tracker tracker2 = sender.send(message);
            final Tracker tracker3 = sender.send(message);

            assertNotNull(tracker1);
            assertNotNull(tracker1.settlementFuture().get().settled());
            assertNotNull(tracker2);
            assertNotNull(tracker2.settlementFuture().get().settled());
            assertNotNull(tracker3);
            assertNotNull(tracker3.settlementFuture().get().settled());

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

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

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

    @Test
    public void testCannotCreateSenderWhenTagGeneratorReturnsNull() 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("Sender test started, peer listening on: {}", remoteURI);

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

            Session session = connection.openSession().openFuture().get();
            SenderOptions options = new SenderOptions().deliveryMode(DeliveryMode.AT_LEAST_ONCE)
                                                       .autoSettle(false)
                                                       .deliveryTagGeneratorSupplier(() -> null);
            try {
                session.openSender("test-tags", options).openFuture().get();
                fail("Should not create a sender if the tag generator is not supplied");
            } catch (ClientException cliEx) {
                // Expected
            }

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

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

    @RepeatedTest(2)
    public void testAwaitSettlementAndCheckRemoteState() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().withTarget().withAddress("test").and().respond();
            peer.remoteFlow().withLinkCredit(5).queue();
            peer.expectTransfer().accept();
            peer.expectTransfer().accept();
            peer.expectTransfer().accept();
            peer.expectTransfer().accept();
            peer.expectTransfer().accept();
            peer.start();

            URI remoteURI = peer.getServerURI();

            Message<String> message = Message.create("test-message");
            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Sender sender = session.openSender("test");

            final Tracker tracker1 = sender.send(message);
            final Tracker tracker2 = sender.send(message);
            final Tracker tracker3 = sender.send(message);
            final Tracker tracker4 = sender.send(message);
            final Tracker tracker5 = sender.send(message);

            assertEquals(ClientDeliveryState.ClientAccepted.getInstance() ,tracker1.awaitSettlement().remoteState());
            assertEquals(ClientDeliveryState.ClientAccepted.getInstance() ,tracker2.awaitSettlement().remoteState());
            assertEquals(ClientDeliveryState.ClientAccepted.getInstance() ,tracker3.awaitSettlement().remoteState());
            assertEquals(ClientDeliveryState.ClientAccepted.getInstance() ,tracker4.awaitSettlement().remoteState());
            assertEquals(ClientDeliveryState.ClientAccepted.getInstance() ,tracker5.awaitSettlement().remoteState());
            assertTrue(tracker1.remoteSettled());
            assertTrue(tracker2.remoteSettled());
            assertTrue(tracker3.remoteSettled());
            assertTrue(tracker4.remoteSettled());
            assertTrue(tracker5.remoteSettled());

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

    @Test
    public void testSendIsRejectedAndTrackerReflectsState() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            Session session = connection.openSession().openFuture().get();
            SenderOptions options = new SenderOptions().deliveryMode(DeliveryMode.AT_LEAST_ONCE).autoSettle(false);
            Sender sender = session.openSender("test-tags", options).openFuture().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectTransfer().withNonNullPayload()
                                 .withDeliveryTag(new byte[] {0})
                                 .respond()
                                 .withSettled(true)
                                 .withState().rejected("no-space", "disk full", Map.of("test", "test"));
            peer.expectDetach().respond();
            peer.expectClose().respond();

            final Message<String> message = Message.create("Hello World");
            final Tracker tracker = sender.send(message);

            assertNotNull(tracker);
            assertNotNull(tracker.settlementFuture().get().settled());
            assertEquals(DeliveryState.Type.REJECTED, tracker.remoteState().getType());

            final org.apache.qpid.protonj2.client.Rejected rejected =
                (org.apache.qpid.protonj2.client.Rejected) tracker.remoteState();

            assertNotNull(rejected);
            assertEquals("no-space", rejected.getCondition());
            assertEquals("disk full", rejected.getDescription());
            assertEquals(Map.of("test", "test"), rejected.getInfo());

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

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

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