/*
 * 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.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

import java.io.InputStream;
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.Future;
import java.util.concurrent.TimeUnit;

import org.apache.qpid.protonj2.client.AdvancedMessage;
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.Delivery;
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.Session;
import org.apache.qpid.protonj2.client.SessionOptions;
import org.apache.qpid.protonj2.client.exceptions.ClientConnectionRemotelyClosedException;
import org.apache.qpid.protonj2.client.exceptions.ClientException;
import org.apache.qpid.protonj2.client.exceptions.ClientIOException;
import org.apache.qpid.protonj2.client.exceptions.ClientIllegalStateException;
import org.apache.qpid.protonj2.client.exceptions.ClientLinkRemotelyClosedException;
import org.apache.qpid.protonj2.client.exceptions.ClientOperationTimedOutException;
import org.apache.qpid.protonj2.client.test.ImperativeClientTestCase;
import org.apache.qpid.protonj2.client.test.Wait;
import org.apache.qpid.protonj2.test.driver.ProtonTestServer;
import org.apache.qpid.protonj2.test.driver.codec.messaging.Modified;
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.codec.primitives.UnknownDescribedType;
import org.apache.qpid.protonj2.types.DescribedType;
import org.apache.qpid.protonj2.types.UnsignedLong;
import org.apache.qpid.protonj2.types.messaging.AmqpValue;
import org.apache.qpid.protonj2.types.messaging.Data;
import org.apache.qpid.protonj2.types.messaging.Section;
import org.apache.qpid.protonj2.types.transport.AmqpError;
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.Matcher;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@Timeout(20)
public class ReceiverTest extends ImperativeClientTestCase {

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

    @Test
    public void testCreateReceiverAndClose() throws Exception {
        doTestCreateReceiverAndCloseOrDetachLink(true);
    }

    @Test
    public void testCreateReceiverAndDetach() throws Exception {
        doTestCreateReceiverAndCloseOrDetachLink(false);
    }

    private void doTestCreateReceiverAndCloseOrDetachLink(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver()
                               .withSource().withAddress("test-queue").withDistributionMode(nullValue()).and()
                               .withTarget().withAddress("test-queue").and().
                               respond();
            peer.expectFlow().withLinkCredit(10);
            peer.expectDetach().withClosed(close).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());

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

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

            Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get(10, TimeUnit.SECONDS);

            assertSame(container, receiver.client());
            assertSame(connection, receiver.connection());
            assertSame(session, receiver.session());

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

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

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

    @Test
    public void testCreateReceiverAndCloseSync() throws Exception {
        doTestCreateReceiverAndCloseOrDetachLinkSync(true);
    }

    @Test
    public void testCreateReceiverAndDetachSync() throws Exception {
        doTestCreateReceiverAndCloseOrDetachLinkSync(false);
    }

    private void doTestCreateReceiverAndCloseOrDetachLinkSync(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.expectDetach().withClosed(close).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());

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

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

            Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get(10, TimeUnit.SECONDS);

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

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

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

    @Test
    public void testCreateReceiverAndCloseWithErrorSync() throws Exception {
        doTestCreateReceiverAndCloseOrDetachWithErrorSync(true);
    }

    @Test
    public void testCreateReceiverAndDetachWithErrorSync() throws Exception {
        doTestCreateReceiverAndCloseOrDetachWithErrorSync(false);
    }

    private void doTestCreateReceiverAndCloseOrDetachWithErrorSync(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            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);

            Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get(10, TimeUnit.SECONDS);

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

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

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

    @Test
    public void testReceiverOpenRejectedByRemote() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().respond().withNullSource();
            peer.expectFlow();
            peer.remoteDetach().withErrorCondition(AmqpError.UNAUTHORIZED_ACCESS.toString(), "Cannot read from this address").queue();
            peer.expectDetach();
            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());

            connection.openFuture().get();

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

            Receiver receiver = session.openReceiver("test-queue");
            try {
                receiver.openFuture().get();
                fail("Open of receiver 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.
            receiver.closeAsync().get();

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

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

    @Test
    public void testOpenReceiverTimesOutWhenNoAttachResponseReceivedTimeout() throws Exception {
        doTestOpenReceiverTimesOutWhenNoAttachResponseReceived(true);
    }

    @Test
    public void testOpenReceiverTimesOutWhenNoAttachResponseReceivedNoTimeout() throws Exception {
        doTestOpenReceiverTimesOutWhenNoAttachResponseReceived(false);
    }

    private void doTestOpenReceiverTimesOutWhenNoAttachResponseReceived(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue());
            peer.expectFlow();
            peer.expectDetach();
            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);
            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().openTimeout(10));

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

                fail("Should not complete the open 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 testOpenReceiverWaitWithTimeoutFailsWhenConnectionDrops() throws Exception {
        doTestOpenReceiverWaitFailsWhenConnectionDrops(true);
    }

    @Test
    public void testOpenReceiverWaitWithNoTimeoutFailsWhenConnectionDrops() throws Exception {
        doTestOpenReceiverWaitFailsWhenConnectionDrops(false);
    }

    private void doTestOpenReceiverWaitFailsWhenConnectionDrops(boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver();
            peer.expectFlow();
            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();
            Receiver receiver = session.openReceiver("test-queue");

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

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

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

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

    @Test
    public void testCloseReceiverTimesOutWhenNoCloseResponseReceivedTimeout() throws Exception {
        doTestCloseOrDetachReceiverTimesOutWhenNoCloseResponseReceived(true, true);
    }

    @Test
    public void testCloseReceiverTimesOutWhenNoCloseResponseReceivedNoTimeout() throws Exception {
        doTestCloseOrDetachReceiverTimesOutWhenNoCloseResponseReceived(true, false);
    }

    @Test
    public void testDetachReceiverTimesOutWhenNoCloseResponseReceivedTimeout() throws Exception {
        doTestCloseOrDetachReceiverTimesOutWhenNoCloseResponseReceived(false, true);
    }

    @Test
    public void testDetachReceiverTimesOutWhenNoCloseResponseReceivedNoTimeout() throws Exception {
        doTestCloseOrDetachReceiverTimesOutWhenNoCloseResponseReceived(false, false);
    }

    private void doTestCloseOrDetachReceiverTimesOutWhenNoCloseResponseReceived(boolean close, boolean timeout) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow();
            peer.expectDetach();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

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

            Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get(10, TimeUnit.SECONDS);

            try {
                if (close) {
                    if (timeout) {
                        receiver.closeAsync().get(10, TimeUnit.SECONDS);
                    } else {
                        receiver.closeAsync().get();
                    }
                } else {
                    if (timeout) {
                        receiver.detachAsync().get(10, TimeUnit.SECONDS);
                    } else {
                        receiver.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 testReceiverDrainAllOutstanding() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).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());

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

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

            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().creditWindow(0));
            receiver.openFuture().get(5, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            // Add some credit, verify not draining
            int credit = 7;
            Matcher<Boolean> drainMatcher = anyOf(equalTo(false), nullValue());
            peer.expectFlow().withDrain(drainMatcher).withLinkCredit(credit).withDeliveryCount(0);

            receiver.addCredit(credit);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            // Drain all the credit
            peer.expectFlow().withDrain(true).withLinkCredit(credit).withDeliveryCount(0)
                             .respond()
                             .withDrain(true).withLinkCredit(0).withDeliveryCount(credit);

            Future<? extends Receiver> draining = receiver.drain();
            draining.get(5, TimeUnit.SECONDS);

            // Close things down
            peer.expectClose().respond();
            connection.closeAsync().get(5, TimeUnit.SECONDS);

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

    @Test
    public void testDrainCompletesWhenReceiverHasNoCredit() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).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();
            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().creditWindow(0));
            receiver.openFuture().get(5, TimeUnit.SECONDS);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            Future<? extends Receiver> draining = receiver.drain();
            draining.get(5, TimeUnit.SECONDS);

            // Close things down
            peer.expectClose().respond();
            connection.closeAsync().get(5, TimeUnit.SECONDS);

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

    @Test
    public void testDrainAdditionalDrainCallThrowsWhenReceiverStillDraining() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.expectFlow().withDrain(true);
            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();
            Receiver receiver = session.openReceiver("test-queue").openFuture().get();

            receiver.drain();

            try {
                receiver.drain().get();
                fail("Drain call should fail timeout exceeded.");
            } catch (ExecutionException cliEx) {
                LOG.debug("Receiver threw error on drain call", cliEx);
                assertTrue(cliEx.getCause() instanceof ClientIllegalStateException);
            }

            connection.closeAsync().get();

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

    @Test
    public void testAddCreditFailsWhileDrainPending() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond().withInitialDeliveryCount(20);
            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();
            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().creditWindow(0));

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            // Add some credit, verify not draining
            int credit = 7;
            Matcher<Boolean> drainMatcher = anyOf(equalTo(false), nullValue());
            peer.expectFlow().withDrain(drainMatcher).withLinkCredit(credit);

            // Ensure we get the attach response with the initial delivery count.
            receiver.openFuture().get().addCredit(credit);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            // Drain all the credit
            peer.expectFlow().withDrain(true).withLinkCredit(credit).withDeliveryCount(20);
            peer.expectClose().respond();

            Future<? extends Receiver> draining = receiver.drain();
            assertFalse(draining.isDone());

            try {
                receiver.addCredit(1);
                fail("Should not allow add credit when drain is pending");
            } catch (ClientIllegalStateException ise) {
                // Expected
            }

            connection.closeAsync().get();

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

    @Test
    public void testAddCreditFailsWhenCreditWindowEnabled() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withLinkCredit(10);
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Receiver receiver = session.openReceiver("test-queue", new ReceiverOptions().creditWindow(10));

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

            try {
                receiver.addCredit(1);
                fail("Should not allow add credit when credit window configured");
            } catch (ClientIllegalStateException ise) {
                // Expected
            }

            connection.close();

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

    @Test
    public void testCreateDynamicReceiver() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue())
                               .withSource().withDynamic(true).withAddress((String) null)
                               .and().respond()
                               .withSource().withDynamic(true).withAddress("test-dynamic-node");
            peer.expectFlow();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

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

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

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

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

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

    @Test
    public void testCreateDynamicReceiverWthNodeProperties() throws Exception {
        final Map<String, Object> nodeProperties = new HashMap<>();
        nodeProperties.put("test-property-1", "one");
        nodeProperties.put("test-property-2", "two");
        nodeProperties.put("test-property-3", "three");

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue())
                               .withSource()
                                   .withDynamic(true)
                                   .withAddress((String) null)
                                   .withDynamicNodeProperties(nodeProperties)
                               .and().respond()
                               .withSource()
                                   .withDynamic(true)
                                   .withAddress("test-dynamic-node")
                                   .withDynamicNodeProperties(nodeProperties);
            peer.expectFlow();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

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

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

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

    @Test
    public void testCreateDynamicReceiverWithNoCreditWindow() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue())
                               .withSource().withDynamic(true).withAddress((String) null)
                               .and().respond()
                               .withSource().withDynamic(true).withAddress("test-dynamic-node");
            peer.expectAttach().ofSender().respond();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            ReceiverOptions receiverOptions = new ReceiverOptions().creditWindow(0);
            Receiver receiver = session.openDynamicReceiver(receiverOptions).openFuture().get();

            // Perform another round trip operation to ensure we see that no flow frame was
            // sent by the receiver
            session.openSender("test");

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

            receiver.close();
            connection.close();

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

    @Test
    public void testDynamicReceiverAddressWaitsForRemoteAttach() throws Exception {
        tryReadDynamicReceiverAddress(true);
    }

    @Test
    public void testDynamicReceiverAddressFailsAfterOpenTimeout() throws Exception {
        tryReadDynamicReceiverAddress(false);
    }

    private void tryReadDynamicReceiverAddress(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue())
                               .withSource().withDynamic(true).withAddress((String) null);
            peer.expectFlow();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().openTimeout(250);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            connection.openFuture().get();

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

            Receiver receiver = session.openDynamicReceiver();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            if (attachResponse) {
                peer.expectDetach().respond();
                peer.respondToLastAttach().withSource().withAddress("test-dynamic-node").and().later(10);
            } else {
                peer.expectDetach();
            }

            if (attachResponse) {
                assertNotNull(receiver.address(), "Remote should have assigned the address for the dynamic receiver");
                assertEquals("test-dynamic-node", receiver.address());
            } else {
                try {
                    receiver.address();
                    fail("Should failed to get address due to no attach response");
                } catch (ClientException ex) {
                    LOG.debug("Caught expected exception from address call", ex);
                }
            }

            try {
                receiver.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 testCreateReceiverWithQoSOfAtMostOnce() throws Exception {
        doTestCreateReceiverWithConfiguredQoS(DeliveryMode.AT_MOST_ONCE);
    }

    @Test
    public void testCreateReceiverWithQoSOfAtLeastOnce() throws Exception {
        doTestCreateReceiverWithConfiguredQoS(DeliveryMode.AT_LEAST_ONCE);
    }

    private void doTestCreateReceiverWithConfiguredQoS(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().withRole(Role.RECEIVER.getValue())
                               .withSndSettleMode(sndMode)
                               .withRcvSettleMode(ReceiverSettleMode.FIRST.byteValue())
                               .respond()
                               .withSndSettleMode(sndMode)
                               .withRcvSettleMode(ReceiverSettleMode.FIRST.byteValue());
            peer.expectFlow();
            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

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

            ReceiverOptions options = new ReceiverOptions().deliveryMode(qos);
            Receiver receiver = session.openReceiver("test-qos", options);
            receiver.openFuture().get(10, TimeUnit.SECONDS);

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

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

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

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

    @Test
    public void testReceiverGetSourceWaitsForRemoteAttach() throws Exception {
        tryReadReceiverSource(true);
    }

    @Test
    public void testReceiverGetSourceFailsAfterOpenTimeout() throws Exception {
        tryReadReceiverSource(false);
    }

    private void tryReadReceiverSource(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue());
            peer.expectFlow();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().openTimeout(250);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            connection.openFuture().get();

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

            Receiver receiver = session.openReceiver("test-receiver");

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

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

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

            try {
                receiver.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 testReceiverGetTargetWaitsForRemoteAttach() throws Exception {
        tryReadReceiverTarget(true);
    }

    @Test
    public void testReceiverGetTargetFailsAfterOpenTimeout() throws Exception {
        tryReadReceiverTarget(false);
    }

    private void tryReadReceiverTarget(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue());
            peer.expectFlow();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().openTimeout(250);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            connection.openFuture().get();

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

            Receiver receiver = session.openReceiver("test-receiver");

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

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

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

            try {
                receiver.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 testReceiverGetRemotePropertiesWaitsForRemoteAttach() throws Exception {
        tryReadReceiverRemoteProperties(true);
    }

    @Test
    public void testReceiverGetRemotePropertiesFailsAfterOpenTimeout() throws Exception {
        tryReadReceiverRemoteProperties(false);
    }

    private void tryReadReceiverRemoteProperties(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue());
            peer.expectFlow();
            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();
            session.openFuture().get();

            ReceiverOptions options = new ReceiverOptions().openTimeout(250);
            Receiver receiver = session.openReceiver("test-receiver", 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(receiver.properties(), "Remote should have responded with a remote properties value");
                assertEquals(expectedProperties, receiver.properties());
            } else {
                try {
                    receiver.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 {
                receiver.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 testReceiverGetRemoteOfferedCapabilitiesWaitsForRemoteAttach() throws Exception {
        tryReadReceiverRemoteOfferedCapabilities(true);
    }

    @Test
    public void testReceiverGetRemoteOfferedCapabilitiesFailsAfterOpenTimeout() throws Exception {
        tryReadReceiverRemoteOfferedCapabilities(false);
    }

    private void tryReadReceiverRemoteOfferedCapabilities(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue());
            peer.expectFlow();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().openTimeout(250);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            connection.openFuture().get();

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

            Receiver receiver = session.openReceiver("test-receiver");

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

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

            if (attachResponse) {
                assertNotNull(receiver.offeredCapabilities(), "Remote should have responded with a remote offered Capabilities value");
                assertEquals(1, receiver.offeredCapabilities().length);
                assertEquals("QUEUE", receiver.offeredCapabilities()[0]);
            } else {
                try {
                    receiver.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 {
                receiver.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 testReceiverGetRemoteDesiredCapabilitiesWaitsForRemoteAttach() throws Exception {
        tryReadReceiverRemoteDesiredCapabilities(true);
    }

    @Test
    public void testReceiverGetRemoteDesiredCapabilitiesFailsAfterOpenTimeout() throws Exception {
        tryReadReceiverRemoteDesiredCapabilities(false);
    }

    private void tryReadReceiverRemoteDesiredCapabilities(boolean attachResponse) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue());
            peer.expectFlow();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().openTimeout(250);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            connection.openFuture().get();

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

            Receiver receiver = session.openReceiver("test-receiver");

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

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

            if (attachResponse) {
                assertNotNull(receiver.desiredCapabilities(), "Remote should have responded with a remote desired Capabilities value");
                assertEquals(1, receiver.desiredCapabilities().length);
                assertEquals("Error-Free", receiver.desiredCapabilities()[0]);
            } else {
                try {
                    receiver.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 {
                receiver.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 testBlockingReceiveCancelledWhenReceiverClosed() throws Exception {
        doTestBlockingReceiveCancelledWhenReceiverClosedOrDetached(true);
    }

    @Test
    public void testBlockingReceiveCancelledWhenReceiverDetached() throws Exception {
        doTestBlockingReceiveCancelledWhenReceiverClosedOrDetached(false);
    }

    public void doTestBlockingReceiveCancelledWhenReceiverClosedOrDetached(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).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();
            ReceiverOptions options = new ReceiverOptions().creditWindow(0);
            Receiver receiver = session.openReceiver("test-queue", options);
            receiver.openFuture().get();

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectFlow().withLinkCredit(10);
            peer.execute(() -> {
                if (close) {
                    receiver.closeAsync();
                } else {
                    receiver.detachAsync();
                }
            }).queue();
            peer.expectDetach().withClosed(close).respond();
            peer.expectClose().respond();

            receiver.addCredit(10);

            try {
                receiver.receive();
                fail("Should throw to indicate that receiver was closed");
            } catch (ClientException ise) {
            }

            connection.closeAsync().get();

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

    @Test
    public void testBlockingReceiveCancelledWhenReceiverRemotelyClosed() throws Exception {
        doTestBlockingReceiveCancelledWhenReceiverRemotelyClosedOrDetached(true);
    }

    @Test
    public void testBlockingReceiveCancelledWhenReceiverRemotelyDetached() throws Exception {
        doTestBlockingReceiveCancelledWhenReceiverRemotelyClosedOrDetached(false);
    }

    public void doTestBlockingReceiveCancelledWhenReceiverRemotelyClosedOrDetached(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteDetach().withClosed(close)
                               .withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Address was manually deleted")
                               .afterDelay(10).queue();
            peer.expectDetach().withClosed(close);
            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();
            Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get();

            try {
                receiver.receive();
                fail("Client should throw to indicate remote closed the receiver forcibly.");
            } catch (ClientIllegalStateException ise) {
            }

            connection.closeAsync().get();

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

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

    @Test
    public void testDetachReceiverWithErrorCondition() 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().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow();
            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();
            final Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get();

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

            connection.closeAsync().get();

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

    @Test
    public void testOpenReceiverWithLinCapabilities() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue())
                               .withSource().withCapabilities("queue").and()
                               .respond();
            peer.expectFlow();
            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);
            ReceiverOptions receiverOptions = new ReceiverOptions();
            receiverOptions.sourceOptions().capabilities("queue");
            Receiver receiver = session.openReceiver("test-queue", receiverOptions);

            receiver.openFuture().get();
            receiver.close();

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

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

    @Test
    public void testReceiveMessageInSplitTransferFrames() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow();
            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();
            final Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get();

            final byte[] payload = createEncodedMessage(new AmqpValue<>("Hello World"));

            final byte[] slice1 = Arrays.copyOfRange(payload, 0, 2);
            final byte[] slice2 = Arrays.copyOfRange(payload, 2, 4);
            final byte[] slice3 = Arrays.copyOfRange(payload, 4, payload.length);

            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(true)
                                 .withMessageFormat(0)
                                 .withPayload(slice1).now();

            assertNull(receiver.tryReceive());

            peer.remoteTransfer().withHandle(0)
                                 .withMore(true)
                                 .withMessageFormat(0)
                                 .withPayload(slice2).now();

            assertNull(receiver.tryReceive());

            peer.remoteTransfer().withHandle(0)
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(slice3).now();

            peer.expectDisposition().withSettled(true).withState().accepted();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            Delivery delivery = receiver.receive();
            assertNotNull(delivery);
            Message<?> received = delivery.message();
            assertNotNull(received);
            assertTrue(received.body() instanceof String);
            String value = (String) received.body();
            assertEquals("Hello World", value);

            delivery.accept();
            receiver.closeAsync();
            connection.closeAsync().get();

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

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

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow();
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            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();
            final Receiver receiver = session.openReceiver("test-queue");

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            assertNull(receiver.receive(15, TimeUnit.MILLISECONDS));

            peer.expectDetach().respond();
            peer.expectClose().respond();
            peer.remoteTransfer().withHandle(0)
                                 .withMore(false)
                                 .withAborted(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).now();

            assertNull(receiver.receive(15, TimeUnit.MILLISECONDS));

            receiver.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testReceiverAddCreditOnAbortedTransferWhenNeeded() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow();
            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();
            ReceiverOptions options = new ReceiverOptions();
            options.creditWindow(1);
            final Receiver receiver = session.openReceiver("test-queue", options);
            receiver.openFuture().get();

            final byte[] payload = createEncodedMessage(new AmqpValue<>("Hello World"));

            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).now();

            assertNull(receiver.tryReceive());

            peer.expectFlow();
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(1)
                                 .withMessageFormat(0)
                                 .withMore(false)
                                 .withPayload(payload).queue();
            peer.expectDisposition().withSettled(true).withState().accepted();
            peer.expectFlow();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            // Send final aborted transfer to complete first transfer and allow next to commence.
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withMore(false)
                                 .withAborted(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).now();

            Delivery delivery = receiver.receive();
            assertNotNull(delivery);
            Message<?> received = delivery.message();
            assertNotNull(received);
            assertTrue(received.body() instanceof String);
            String value = (String) received.body();
            assertEquals("Hello World", value);

            receiver.closeAsync();
            connection.closeAsync().get();

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

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

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withLinkCredit(1);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            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();
            ReceiverOptions options = new ReceiverOptions();
            options.creditWindow(1);
            final Receiver receiver = session.openReceiver("test-queue", options);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            assertNull(receiver.receive(15, TimeUnit.MILLISECONDS));

            // Credit window is one and next transfer signals aborted so receiver should
            // top-up the credit window to allow more transfers to arrive.
            peer.expectFlow().withLinkCredit(1);
            peer.expectDetach().respond();
            peer.expectClose().respond();

            // Abort the delivery which should result in a credit top-up.
            peer.remoteTransfer().withHandle(0)
                                 .withMore(false)
                                 .withAborted(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).now();

            assertNull(receiver.receive(15, TimeUnit.MILLISECONDS));

            receiver.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testReceiveCallFailsWhenReceiverPreviouslyClosed() throws Exception {
        doTestReceiveCallFailsWhenReceiverDetachedOrClosed(true);
    }

    @Test
    public void testReceiveCallFailsWhenReceiverPreviouslyDetached() throws Exception {
        doTestReceiveCallFailsWhenReceiverDetachedOrClosed(false);
    }

    private void doTestReceiveCallFailsWhenReceiverDetachedOrClosed(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.expectDetach().withClosed(close).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();
            Receiver receiver = session.openReceiver("test-queue").openFuture().get();

            if (close) {
                receiver.closeAsync();
            } else {
                receiver.detachAsync();
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            try {
                receiver.receive();
                fail("Receive call should fail when link closed or detached.");
            } catch (ClientIllegalStateException cliEx) {
                LOG.debug("Receiver threw error on receive call", cliEx);
            }

            peer.expectClose().respond();

            connection.closeAsync().get();

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

    @Test
    public void testReceiveBlockedForMessageFailsWhenConnectionRemotelyClosed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteClose().withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Connection was deleted").afterDelay(25).queue();
            peer.expectClose();
            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();
            Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get();

            try {
                receiver.receive();
                fail("Receive should have failed when Connection remotely closed.");
            } catch (ClientConnectionRemotelyClosedException cliEx) {
                // Expected
            }

            connection.closeAsync().get();

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

    @Test
    public void testTimedReceiveBlockedForMessageFailsWhenConnectionRemotelyClosed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteClose().withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Connection was deleted").afterDelay(25).queue();
            peer.expectClose();
            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();
            Receiver receiver = session.openReceiver("test-queue");
            receiver.openFuture().get();

            try {
                receiver.receive(10, TimeUnit.SECONDS);
                fail("Receive 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 testReceiveTimedCallFailsWhenReceiverClosed() throws Exception {
        doTestReceiveTimedCallFailsWhenReceiverDetachedOrClosed(true);
    }

    @Test
    public void testReceiveTimedCallFailsWhenReceiverDetached() throws Exception {
        doTestReceiveTimedCallFailsWhenReceiverDetachedOrClosed(false);
    }

    private void doTestReceiveTimedCallFailsWhenReceiverDetachedOrClosed(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.expectDetach().withClosed(close).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();
            Receiver receiver = session.openReceiver("test-queue").openFuture().get();

            if (close) {
                receiver.closeAsync();
            } else {
                receiver.detachAsync();
            }

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);

            try {
                receiver.receive(60, TimeUnit.SECONDS);
                fail("Receive call should fail when link closed or detached.");
            } catch (ClientIllegalStateException cliEx) {
                LOG.debug("Receiver threw error on receive call", cliEx);
            }

            peer.expectClose().respond();

            connection.closeAsync().get();

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

    @Test
    public void testDrainFutureSignalsFailureWhenReceiverClosed() throws Exception {
        doTestDrainFutureSignalsFailureWhenReceiverClosedOrDetached(true);
    }

    @Test
    public void testDrainFutureSignalsFailureWhenReceiverDetached() throws Exception {
        doTestDrainFutureSignalsFailureWhenReceiverClosedOrDetached(false);
    }

    private void doTestDrainFutureSignalsFailureWhenReceiverClosedOrDetached(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectFlow().withDrain(true).withLinkCredit(10);
            peer.execute(() -> {
                if (close) {
                    receiver.closeAsync();
                } else {
                    receiver.detachAsync();
                }
            }).queue();
            peer.expectDetach().withClosed(close).respond();
            peer.expectClose().respond();

            try {
                receiver.drain().get(10, TimeUnit.SECONDS);
                fail("Drain call should fail when link closed or detached.");
            } catch (ExecutionException cliEx) {
                LOG.debug("Receiver threw error on drain call", cliEx);
                assertTrue(cliEx.getCause() instanceof ClientException);
            }

            connection.closeAsync().get();

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

    @Test
    public void testDrainFutureSignalsFailureWhenReceiverRemotelyClosed() throws Exception {
        doTestDrainFutureSignalsFailureWhenReceiverRemotelyClosedOrDetached(true);
    }

    @Test
    public void testDrainFutureSignalsFailureWhenReceiverRemotelyDetached() throws Exception {
        doTestDrainFutureSignalsFailureWhenReceiverRemotelyClosedOrDetached(false);
    }

    private void doTestDrainFutureSignalsFailureWhenReceiverRemotelyClosedOrDetached(boolean close) throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow().withLinkCredit(10);
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectFlow().withDrain(true).withLinkCredit(10);
            peer.remoteDetach().withClosed(close)
                               .withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Address was manually deleted").queue();
            peer.expectDetach().withClosed(close);
            peer.expectClose().respond();

            try {
                receiver.drain().get(10, TimeUnit.SECONDS);
                fail("Drain call should fail when link closed or detached.");
            } catch (ExecutionException cliEx) {
                LOG.debug("Receiver threw error on drain call", cliEx);
                assertTrue(cliEx.getCause() instanceof ClientException);
            }

            connection.closeAsync().get();

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

    @Test
    public void testDrainFutureSignalsFailureWhenSessionRemotelyClosed() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.expectFlow().withDrain(true);
            peer.remoteEnd().withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Session was closed").afterDelay(5).queue();
            peer.expectEnd();
            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();
            Receiver receiver = session.openReceiver("test-queue").openFuture().get();

            try {
                receiver.drain().get(10, TimeUnit.SECONDS);
                fail("Drain call should fail when session closed by remote.");
            } catch (ExecutionException cliEx) {
                LOG.debug("Receiver threw error on drain call", cliEx);
                assertTrue(cliEx.getCause() instanceof ClientException);
            }

            connection.closeAsync().get();

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

    @Test
    public void testDrainFutureSignalsFailureWhenConnectionDrops() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.expectFlow().withDrain(true);
            peer.dropAfterLastHandler();
            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();
            Receiver receiver = session.openReceiver("test-queue").openFuture().get();

            try {
                receiver.drain().get(10, TimeUnit.SECONDS);
                fail("Drain call should fail when the connection drops.");
            } catch (ExecutionException cliEx) {
                LOG.debug("Receiver threw error on drain call", cliEx);
                assertTrue(cliEx.getCause() instanceof ClientException);
            }

            connection.closeAsync().get();

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

    @Test
    public void testDrainFutureSignalsFailureWhenDrainTimeoutExceeded() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.expectFlow().withDrain(true);
            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();
            ReceiverOptions receiverOptions = new ReceiverOptions().drainTimeout(15);
            Receiver receiver = session.openReceiver("test-queue", receiverOptions).openFuture().get();

            try {
                receiver.drain().get();
                fail("Drain call should fail timeout exceeded.");
            } catch (ExecutionException cliEx) {
                LOG.debug("Receiver threw error on drain call", cliEx);
                assertTrue(cliEx.getCause() instanceof ClientOperationTimedOutException);
            }

            connection.closeAsync().get();

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

    @Test
    public void testDrainFutureSignalsFailureWhenSessionDrainTimeoutExceeded() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.expectFlow().withDrain(true);
            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());
            SessionOptions sessionOptions = new SessionOptions().drainTimeout(20);
            Session session = connection.openSession(sessionOptions);
            Receiver receiver = session.openReceiver("test-queue").openFuture().get();

            try {
                receiver.drain().get();
                fail("Drain call should fail timeout exceeded.");
            } catch (ExecutionException cliEx) {
                LOG.debug("Receiver threw error on drain call", cliEx);
                assertTrue(cliEx.getCause() instanceof ClientOperationTimedOutException);
            }

            connection.closeAsync().get();

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

    @Test
    public void testDrainFutureSignalsFailureWhenConnectionDrainTimeoutExceeded() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.expectFlow().withDrain(true);
            peer.expectClose().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions connectionOptions = new ConnectionOptions().drainTimeout(20);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), connectionOptions);
            Session session = connection.openSession();
            Receiver receiver = session.openReceiver("test-queue").openFuture().get();

            try {
                receiver.drain().get();
                fail("Drain call should fail timeout exceeded.");
            } catch (ExecutionException cliEx) {
                LOG.debug("Receiver threw error on drain call", cliEx);
                assertTrue(cliEx.getCause() instanceof ClientOperationTimedOutException);
            }

            connection.closeAsync().get();

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

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

            URI remoteURI = peer.getServerURI();

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Receiver receiver = session.openReceiver("test");

            try {
                receiver.receive();
                fail("Receive should fail with remotely closed error after remote drops");
            } catch (ClientConnectionRemotelyClosedException cliEx) {
            }

            connection.closeAsync().get();

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

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

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withSettled(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            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();
            Receiver receiver = session.openReceiver("test-queue");

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

            Delivery delivery = receiver.receive(5, TimeUnit.SECONDS);
            assertNotNull(delivery);

            Message<String> message = delivery.message();
            assertNotNull(message);

            try {
                delivery.rawInputStream();
                fail("Should not be able to use the inputstream once message is requested");
            } catch (ClientIllegalStateException cliEx) {
                // Expected
            }

            receiver.closeAsync();
            connection.closeAsync().get();

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

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

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withSettled(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            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();
            Receiver receiver = session.openReceiver("test-queue");

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

            Delivery delivery = receiver.receive(5, TimeUnit.SECONDS);
            assertNotNull(delivery);
            assertNull(delivery.annotations());

            try {
                delivery.rawInputStream();
                fail("Should not be able to use the inputstream once message is requested");
            } catch (ClientIllegalStateException cliEx) {
                // Expected
            }

            receiver.closeAsync();
            connection.closeAsync().get();

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

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

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withSettled(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            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();
            Receiver receiver = session.openReceiver("test-queue");

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

            Delivery delivery = receiver.receive(5, TimeUnit.SECONDS);
            assertNotNull(delivery);
            InputStream stream = delivery.rawInputStream();
            assertNotNull(stream);

            assertEquals(payload.length, stream.available());
            byte[] bytesRead = new byte[payload.length];
            assertEquals(payload.length, stream.read(bytesRead));
            assertArrayEquals(payload, bytesRead);

            try {
                delivery.message();
                fail("Should not be able to use the message API once raw stream is requested");
            } catch (ClientIllegalStateException cliEx) {
                // Expected
            }

            try {
                delivery.annotations();
                fail("Should not be able to use the annotations API once raw stream is requested");
            } catch (ClientIllegalStateException cliEx) {
                // Expected
            }

            receiver.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testReceiveDeliveryWithMultipleDataSections() throws Exception {
        final Data section1 = new Data(new byte[] { 0, 1, 2, 3 });
        final Data section2 = new Data(new byte[] { 0, 1, 2, 3 });
        final Data section3 = new Data(new byte[] { 0, 1, 2, 3 });

        final byte[] payload = createEncodedMessage(section1, section2, section3);

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withSettled(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            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();
            Receiver receiver = session.openReceiver("test-queue");

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

            Delivery delivery = receiver.receive(5, TimeUnit.SECONDS);
            assertNotNull(delivery);

            AdvancedMessage<?> message = delivery.message().toAdvancedMessage();
            assertNotNull(message);

            assertEquals(3, message.bodySections().size());
            List<Section<?>> section = new ArrayList<>(message.bodySections());
            assertEquals(section1, section.get(0));
            assertEquals(section2, section.get(1));
            assertEquals(section3, section.get(2));

            receiver.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testSessionWindowExpandedAsIncomingFramesArrive() throws Exception {
        final byte[] payload1 = new byte[255];
        final byte[] payload2 = new byte[255];

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().withMaxFrameSize(1024).respond();
            peer.expectBegin().withIncomingWindow(1).respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withIncomingWindow(1).withLinkCredit(10);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(true)
                                 .withSettled(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload1).queue();
            peer.expectFlow().withIncomingWindow(1).withLinkCredit(10);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withSettled(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload2).queue();
            peer.expectFlow().withIncomingWindow(1).withLinkCredit(9);
            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);
            SessionOptions sessionOpts = new SessionOptions().incomingCapacity(1024);
            Session session = connection.openSession(sessionOpts);
            Receiver receiver = session.openReceiver("test-queue");

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

            Delivery delivery = receiver.receive(5, TimeUnit.SECONDS);
            assertNotNull(delivery);

            InputStream stream = delivery.rawInputStream();
            assertNotNull(stream);

            assertEquals(payload1.length + payload2.length, stream.available());
            byte[] bytesRead = new byte[payload1.length + payload2.length];
            assertEquals(payload1.length + payload2.length, stream.read(bytesRead));

            receiver.closeAsync();
            connection.closeAsync().get();

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

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

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDisposition().optional();
            peer.dropAfterLastHandler(1);
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            final CountDownLatch disconnected = new CountDownLatch(1);

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions();
            options.disconnectedHandler((c, e) -> disconnected.countDown());
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            final Receiver receiver = connection.openReceiver("test-queue");
            final Delivery delivery = receiver.receive();

            peer.waitForScriptToComplete();

            assertNotNull(delivery);

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

            // Data already read so it will be already available for read but should be unable to
            // since the connection has dropped.
            assertNotEquals(-1, delivery.rawInputStream().read());

            connection.close();

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

    @Test
    public void testCreateReceiverWithDefaultSourceAndTargetOptions() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver()
                               .withSource().withAddress("test-queue")
                                            .withDistributionMode(nullValue())
                                            .withDefaultTimeout()
                                            .withDurable(TerminusDurability.NONE)
                                            .withExpiryPolicy(TerminusExpiryPolicy.LINK_DETACH)
                                            .withDefaultOutcome(new Modified().setDeliveryFailed(true))
                                            .withCapabilities(nullValue())
                                            .withFilter(nullValue())
                                            .withOutcomes("amqp:accepted:list", "amqp:rejected:list", "amqp:released:list", "amqp:modified:list")
                                            .also()
                               .withTarget().withAddress(notNullValue())
                                            .withCapabilities(nullValue())
                                            .withDurable(nullValue())
                                            .withExpiryPolicy(nullValue())
                                            .withDefaultTimeout()
                                            .withDynamic(anyOf(nullValue(), equalTo(false)))
                                            .withDynamicNodeProperties(nullValue())
                               .and().respond();
            peer.expectFlow().withLinkCredit(10);
            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();
            Receiver receiver = session.openReceiver("test-queue").openFuture().get();

            receiver.close();
            session.close();
            connection.close();

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

    @Test
    public void testCreateReceiverWithUserConfiguredSourceAndTargetOptions() 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().ofReceiver()
                               .withSource().withAddress("test-queue")
                                            .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(notNullValue())
                                            .withCapabilities("QUEUE")
                                            .withDurable(TerminusDurability.CONFIGURATION)
                                            .withExpiryPolicy(TerminusExpiryPolicy.SESSION_END)
                                            .withTimeout(42)
                                            .withDynamic(anyOf(nullValue(), equalTo(false)))
                                            .withDynamicNodeProperties(nullValue())
                               .and().respond();
            peer.expectFlow().withLinkCredit(10);
            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();
            ReceiverOptions receiverOptions = new ReceiverOptions();

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

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

            Receiver receiver = session.openReceiver("test-queue", receiverOptions).openFuture().get();

            receiver.close();
            session.close();
            connection.close();

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

    private class AmqpJmsSelectorType implements DescribedType {

        private final String selector;

        public AmqpJmsSelectorType(String selector) {
            this.selector = selector;
        }

        @Override
        public Object getDescriptor() {
            return UnsignedLong.valueOf(0x0000468C00000004L);
        }

        @Override
        public Object getDescribed() {
            return this.selector;
        }

        @Override
        public String toString() {
            return "AmqpJmsSelectorType{" + selector + "}";
        }
    }

    public class AmqpJmsNoLocalType implements DescribedType {

        private final String noLocal;

        public AmqpJmsNoLocalType() {
            this.noLocal = "NoLocalFilter{}";
        }

        @Override
        public Object getDescriptor() {
            return UnsignedLong.valueOf(0x0000468C00000003L);
        }

        @Override
        public Object getDescribed() {
            return this.noLocal;
        }
    }

    private class PeerJmsSelectorType extends UnknownDescribedType {

        public PeerJmsSelectorType(String selector) {
            super(org.apache.qpid.protonj2.test.driver.codec.primitives.UnsignedLong.valueOf(0x0000468C00000004L), selector);
        }
    }

    private class PeerNoLocalFilterType extends UnknownDescribedType {

        public PeerNoLocalFilterType() {
            super(org.apache.qpid.protonj2.test.driver.codec.primitives.UnsignedLong.valueOf(0x0000468C00000003L), "NoLocalFilter{}");
        }
    }

    @Test
    public void testCreateReceiverWithUserConfiguredSourceWithJMSStyleSelector() throws Exception {
        final DescribedType clientJmsSelector = new AmqpJmsSelectorType("myProperty=42");
        final Map<String, Object> filters = new HashMap<>();
        filters.put("jms-selector", clientJmsSelector);

        final PeerJmsSelectorType peerJmsSelector = new PeerJmsSelectorType("myProperty=42");
        final Map<String, Object> filtersAtPeer = new HashMap<>();
        filtersAtPeer.put("jms-selector", peerJmsSelector);

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver()
                               .withSource().withAddress("test-queue")
                                            .withDistributionMode("copy")
                                            .withTimeout(128)
                                            .withDurable(TerminusDurability.UNSETTLED_STATE)
                                            .withExpiryPolicy(TerminusExpiryPolicy.CONNECTION_CLOSE)
                                            .withDefaultOutcome(new Released())
                                            .withCapabilities("QUEUE")
                                            .withFilter(filtersAtPeer)
                                            .withOutcomes("amqp:accepted:list", "amqp:rejected:list")
                                            .also()
                               .withTarget().withAddress(notNullValue())
                                            .withCapabilities("QUEUE")
                                            .withDurable(TerminusDurability.CONFIGURATION)
                                            .withExpiryPolicy(TerminusExpiryPolicy.SESSION_END)
                                            .withTimeout(42)
                                            .withDynamic(anyOf(nullValue(), equalTo(false)))
                                            .withDynamicNodeProperties(nullValue())
                               .and().respond();
            peer.expectFlow().withLinkCredit(10);
            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();
            ReceiverOptions receiverOptions = new ReceiverOptions();

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

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

            Receiver receiver = session.openReceiver("test-queue", receiverOptions).openFuture().get();

            receiver.close();
            session.close();
            connection.close();

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

    @Test
    public void testCreateReceiverWithUserConfiguredSourceWithJMSStyleSelectorAndNoLocalFilter() throws Exception {
        final DescribedType clientJmsSelector = new AmqpJmsSelectorType("myProperty=42");
        final DescribedType clientNoLocalFilter = new AmqpJmsNoLocalType();

        final Map<String, Object> filters = new HashMap<>();
        filters.put("no-local", clientNoLocalFilter);
        filters.put("jms-selector", clientJmsSelector);

        final PeerJmsSelectorType peerJmsSelector = new PeerJmsSelectorType("myProperty=42");
        final PeerNoLocalFilterType peerNoLocalFilter = new PeerNoLocalFilterType();
        final Map<String, Object> filtersAtPeer = new HashMap<>();
        filtersAtPeer.put("jms-selector", peerJmsSelector);
        filtersAtPeer.put("no-local", peerNoLocalFilter);

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver()
                               .withSource().withAddress("test-queue")
                                            .withDistributionMode("copy")
                                            .withTimeout(128)
                                            .withDurable(TerminusDurability.UNSETTLED_STATE)
                                            .withExpiryPolicy(TerminusExpiryPolicy.CONNECTION_CLOSE)
                                            .withDefaultOutcome(new Released())
                                            .withCapabilities("QUEUE")
                                            .withFilter(filtersAtPeer)
                                            .withOutcomes("amqp:accepted:list", "amqp:rejected:list")
                                            .also()
                               .withTarget().withAddress(notNullValue())
                                            .withCapabilities("QUEUE")
                                            .withDurable(TerminusDurability.CONFIGURATION)
                                            .withExpiryPolicy(TerminusExpiryPolicy.SESSION_END)
                                            .withTimeout(42)
                                            .withDynamic(anyOf(nullValue(), equalTo(false)))
                                            .withDynamicNodeProperties(nullValue())
                               .and().respond();
            peer.expectFlow().withLinkCredit(10);
            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();
            ReceiverOptions receiverOptions = new ReceiverOptions();

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

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

            Receiver receiver = session.openReceiver("test-queue", receiverOptions).openFuture().get();

            receiver.close();
            session.close();
            connection.close();

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

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

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

            URI remoteURI = peer.getServerURI();

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

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

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

            connection.closeAsync().get();

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

    @Test
    public void testReceiverCreditReplenishedAfterSyncReceiveAutoAccept() throws Exception {
       doTestReceiverCreditReplenishedAfterSyncReceive(true);
    }

    @Test
    public void testReceiverCreditReplenishedAfterSyncReceiveManualAccept() throws Exception {
       doTestReceiverCreditReplenishedAfterSyncReceive(false);
    }

    public void doTestReceiverCreditReplenishedAfterSyncReceive(boolean autoAccept) throws Exception {
       byte[] payload = createEncodedMessage(new AmqpValue<String>("Hello World"));

       try (ProtonTestServer peer = new ProtonTestServer()) {
           peer.expectSASLAnonymousConnect();
           peer.expectOpen().respond();
           peer.expectBegin().respond();
           peer.expectAttach().ofReceiver().respond();
           peer.expectFlow().withLinkCredit(10);
           for (int i = 0; i < 10; ++i) {
               peer.remoteTransfer().withDeliveryId(i)
                                    .withMore(false)
                                    .withMessageFormat(0)
                                    .withPayload(payload).queue();
           }
           peer.start();

           URI remoteURI = peer.getServerURI();

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

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

           ReceiverOptions options = new ReceiverOptions();
           options.autoAccept(autoAccept);
           options.creditWindow(10);

           Receiver receiver = connection.openReceiver("test-receiver", options);

           Wait.waitFor(() -> receiver.queuedDeliveries() == 10);

           peer.waitForScriptToComplete();
           if (autoAccept)
           {
               peer.expectDisposition().withFirst(0);
               peer.expectDisposition().withFirst(1);
           }

           // Consume messages 1 and 2 which should not provoke credit replenishment
           // as there are still 8 outstanding which is above the 70% mark
           assertNotNull(receiver.receive()); // #1
           assertNotNull(receiver.receive()); // #2

           peer.waitForScriptToComplete();
           peer.expectAttach().ofSender().respond();
           peer.expectDetach().respond();
           if (autoAccept)
           {
               peer.expectDisposition().withFirst(2);
           }
           peer.expectFlow().withLinkCredit(3);

           // Ensure that no additional frames from last receive overlap with this one
           connection.openSender("test").openFuture().get().close();

           // Now consume message 3 which will trip the replenish barrier and the
           // credit should be updated to reflect that we still have 7 queued
           assertNotNull(receiver.receive());  // #3

           peer.waitForScriptToComplete();
           if (autoAccept) {
               peer.expectDisposition().withFirst(3);
               peer.expectDisposition().withFirst(4);
           }

           // Consume messages 4 and 5 which should not provoke credit replenishment
           // as there are still 5 outstanding plus the credit we sent last time
           // which is above the 70% mark
           assertNotNull(receiver.receive()); // #4
           assertNotNull(receiver.receive()); // #5

           peer.waitForScriptToComplete();
           peer.expectAttach().ofSender().respond();
           peer.expectDetach().respond();
           if (autoAccept) {
               peer.expectDisposition().withFirst(5);
           }
           peer.expectFlow().withLinkCredit(6);

           // Ensure that no additional frames from last receive overlap with this one
           connection.openSender("test").openFuture().get().close();

           // Consume number 6 which means we only have 4 outstanding plus the three
           // that we sent last time we flowed which is 70% of possible prefetch so
           // we should flow to top off credit which would be 6 since we have four
           // still pending
           assertNotNull(receiver.receive()); // #6

           peer.waitForScriptToComplete();
           if (autoAccept) {
               peer.expectDisposition().withFirst(6);
               peer.expectDisposition().withFirst(7);
           }

           // Consume deliveries 7 and 8 which should not flow as we should be
           // above the threshold of 70% since we would now have 2 outstanding
           // and 6 credits on the link
           assertNotNull(receiver.receive()); // #7
           assertNotNull(receiver.receive()); // #8

           peer.waitForScriptToComplete();
           if (autoAccept) {
               peer.expectDisposition().withFirst(8);
               peer.expectDisposition().withFirst(9);
           }

           // Now consume 9 and 10 but we still shouldn't flow more credit because
           // the link credit is above the 50% mark for overall credit windowing.
           assertNotNull(receiver.receive()); // #9
           assertNotNull(receiver.receive()); // #10

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

           connection.close();

           peer.waitForScriptToComplete();
        }
    }

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

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().withRole(Role.RECEIVER.getValue()).respond();
            peer.expectFlow().withLinkCredit(10);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withSettled(true)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue().splitWrite(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());
            Session session = connection.openSession();
            Receiver receiver = session.openReceiver("test-queue");

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

            Delivery delivery = receiver.receive(5, TimeUnit.SECONDS);
            assertNotNull(delivery);

            Message<?> message = delivery.message();
            assertEquals("Hello World!", message.body());

            receiver.closeAsync();
            connection.closeAsync().get();

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

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

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond().withNextOutgoingId(0);
            peer.expectAttach().respond();
            peer.expectFlow().withLinkCredit(3);
            peer.remoteTransfer().withDeliveryId(0)
                                 .withDeliveryTag(new byte[] {1})
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDisposition().withFirst(0)
                                    .withSettled(true)
                                    .withState().accepted();
            peer.remoteTransfer().withDeliveryId(1)
                                 .withDeliveryTag(new byte[] {2})
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDisposition().withFirst(1)
                                    .withSettled(true)
                                    .withState().accepted();
            peer.remoteTransfer().withDeliveryId(2)
                                 .withDeliveryTag(new byte[] {3})
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDisposition().withFirst(2)
                                    .withSettled(true)
                                    .withState().accepted();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            ReceiverOptions options = new ReceiverOptions();
            options.creditWindow(0);
            options.autoSettle(true);

            Client container = Client.create();
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort());
            Session session = connection.openSession();
            Receiver receiver = session.openReceiver("test-queue", options);

            receiver.addCredit(3);

            Delivery delivery1 = receiver.receive();
            Wait.assertTrue(() -> delivery1.settled());
            Delivery delivery2 = receiver.receive();
            Wait.assertTrue(() -> delivery2.settled());
            Delivery delivery3 = receiver.receive();
            Wait.assertTrue(() -> delivery3.settled());

            assertNotNull(delivery1);
            assertNotNull(delivery2);
            assertNotNull(delivery3);

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

            receiver.close();
            session.close();
            connection.close();

            // Check post conditions and done.
            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
        }
    }

    @Test
    public void testReceiveAcceptsTimeoutAndWaitsForDelivery() throws Exception {
        doTestReceiveAcceptsNegtiveValuesAsInfiniteTimeouts(1, TimeUnit.MINUTES);
    }

    @Test
    public void testReceiveAcceptsNegtiveValuesAsInfiniteTimeoutsMinusOneMillisends() throws Exception {
        doTestReceiveAcceptsNegtiveValuesAsInfiniteTimeouts(-1, TimeUnit.MILLISECONDS);
    }

    @Test
    public void testReceiveAcceptsNegtiveValuesAsInfiniteTimeoutsMinusOneSeconds() throws Exception {
        doTestReceiveAcceptsNegtiveValuesAsInfiniteTimeouts(-1, TimeUnit.SECONDS);
    }

    public void doTestReceiveAcceptsNegtiveValuesAsInfiniteTimeouts(long timeout, TimeUnit units) throws Exception {
        final byte[] payload = createEncodedMessage(new AmqpValue<>("Hello World"));

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).afterDelay(25).queue();
            peer.expectDisposition().withFirst(0)
                                    .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());
            final Receiver receiver = connection.openReceiver("test-queue");
            final Delivery delivery = receiver.receive(timeout, units);

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

            assertNotNull(delivery);

            receiver.close();
            connection.close();

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

    @Test
    public void testReceiveMessageWithNullWrappedInAmqpValue() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withMessage().withBody().withValue((String) null)
                                 .also()
                                 .splitWrite(true)
                                 .afterDelay(25)
                                 .queue();
            peer.expectDisposition().withFirst(0)
                                    .withSettled(true)
                                    .withState().accepted();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            final Client container = Client.create();
            final ConnectionOptions options = new ConnectionOptions();
            final Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            final Receiver receiver = connection.openReceiver("test-queue");
            final Delivery delivery = receiver.receive();
            final Message<String> message = delivery.message();

            assertNull(message.body());

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

            assertNotNull(delivery);

            receiver.close();
            connection.close();

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

    @Test
    public void testReceiveMessageWithNullWrappedInDataSection() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withMessage().withBody().withData((byte[]) null)
                                 .also()
                                 .splitWrite(true)
                                 .afterDelay(25)
                                 .queue();
            peer.expectDisposition().withFirst(0)
                                    .withSettled(true)
                                    .withState().accepted();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            final Client container = Client.create();
            final ConnectionOptions options = new ConnectionOptions();
            final Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            final Receiver receiver = connection.openReceiver("test-queue");
            final Delivery delivery = receiver.receive();
            final Message<byte[]> message = delivery.message();

            assertNull(message.body());

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

            assertNotNull(delivery);

            receiver.close();
            connection.close();

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