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

import java.io.OutputStream;
import java.net.URI;
import java.util.concurrent.TimeUnit;

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.DeliveryState;
import org.apache.qpid.protonj2.client.Message;
import org.apache.qpid.protonj2.client.OutputStreamOptions;
import org.apache.qpid.protonj2.client.Receiver;
import org.apache.qpid.protonj2.client.ReceiverOptions;
import org.apache.qpid.protonj2.client.Sender;
import org.apache.qpid.protonj2.client.Session;
import org.apache.qpid.protonj2.client.StreamSender;
import org.apache.qpid.protonj2.client.StreamSenderMessage;
import org.apache.qpid.protonj2.client.Tracker;
import org.apache.qpid.protonj2.client.exceptions.ClientConnectionRemotelyClosedException;
import org.apache.qpid.protonj2.client.exceptions.ClientException;
import org.apache.qpid.protonj2.client.exceptions.ClientIllegalStateException;
import org.apache.qpid.protonj2.client.exceptions.ClientTransactionDeclarationException;
import org.apache.qpid.protonj2.client.exceptions.ClientTransactionNotActiveException;
import org.apache.qpid.protonj2.client.exceptions.ClientTransactionRolledBackException;
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.matchers.messaging.HeaderMatcher;
import org.apache.qpid.protonj2.test.driver.matchers.transport.TransferPayloadCompositeMatcher;
import org.apache.qpid.protonj2.test.driver.matchers.types.EncodedDataMatcher;
import org.apache.qpid.protonj2.types.messaging.Accepted;
import org.apache.qpid.protonj2.types.messaging.AmqpValue;
import org.apache.qpid.protonj2.types.messaging.Header;
import org.apache.qpid.protonj2.types.messaging.Modified;
import org.apache.qpid.protonj2.types.messaging.Rejected;
import org.apache.qpid.protonj2.types.messaging.Released;
import org.apache.qpid.protonj2.types.transactions.TransactionErrors;
import org.apache.qpid.protonj2.types.transport.AmqpError;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

//@Timeout(30)
public class TransactionsTest extends ImperativeClientTestCase {

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

    @Test
    public void testCoordinatorLinkSupportedOutcomes() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().withSource().withOutcomes(Accepted.DESCRIPTOR_SYMBOL.toString(),
                                                                     Rejected.DESCRIPTOR_SYMBOL.toString(),
                                                                     Released.DESCRIPTOR_SYMBOL.toString(),
                                                                     Modified.DESCRIPTOR_SYMBOL.toString()).and().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            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().openFuture().get();

            session.beginTransaction();
            session.commitTransaction();

            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testTimedOutExceptionOnBeginWithNoResponse() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare();
            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();
            ConnectionOptions options = new ConnectionOptions().requestTimeout(50);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession().openFuture().get();

            try {
                session.beginTransaction();
                fail("Begin should have timed out after no response.");
            } catch (ClientTransactionDeclarationException expected) {
                // Expect this to time out.
            }

            try {
                session.commitTransaction();
                fail("Commit should have failed due to no active transaction.");
            } catch (ClientIllegalStateException expected) {
                // Expect this to fail since transaction not declared
            }

            try {
                session.rollbackTransaction();
                fail("Rollback should have failed due to no active transaction.");
            } catch (ClientIllegalStateException expected) {
                // Expect this to fail since transaction not declared
            }

            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testTimedOutExceptionOnBeginWithNoResponseThenRecoverWithNextBegin() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare();
            peer.expectDetach().respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().requestTimeout(150);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession().openFuture().get();

            try {
                session.beginTransaction();
                fail("Begin should have timed out after no response.");
            } catch (ClientTransactionDeclarationException expected) {
                // Expect this to time out.
            }

            try {
                session.commitTransaction();
                fail("Commit should have failed due to no active transaction.");
            } catch (ClientIllegalStateException expected) {
                // Expect this to fail since transaction not declared
            }

            try {
                session.rollbackTransaction();
                fail("Rollback should have failed due to no active transaction.");
            } catch (ClientIllegalStateException expected) {
                // Expect this to fail since transaction not declared
            }

            peer.waitForScriptToComplete(500, TimeUnit.SECONDS);
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept();
            peer.expectDischarge().accept();
            peer.expectEnd().respond();
            peer.expectClose().respond();

            session.beginTransaction();
            session.commitTransaction();
            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testTimedOutExceptionOnBeginWithNoResponseThenRecoverWithNextBeginAndDelayedDetachResponse() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare();
            peer.expectDetach().respond().afterDelay(20);
            peer.start();

            URI remoteURI = peer.getServerURI();

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

            Client container = Client.create();
            ConnectionOptions options = new ConnectionOptions().requestTimeout(150);
            Connection connection = container.connect(remoteURI.getHost(), remoteURI.getPort(), options);
            Session session = connection.openSession().openFuture().get();

            try {
                session.beginTransaction();
                fail("Begin should have timed out after no response.");
            } catch (ClientTransactionDeclarationException expected) {
                // Expect this to time out.
            }

            try {
                session.commitTransaction();
                fail("Commit should have failed due to no active transaction.");
            } catch (ClientIllegalStateException expected) {
                // Expect this to fail since transaction not declared
            }

            try {
                session.rollbackTransaction();
                fail("Rollback should have failed due to no active transaction.");
            } catch (ClientIllegalStateException expected) {
                // Expect this to fail since transaction not declared
            }

            peer.waitForScriptToComplete(500, TimeUnit.SECONDS);
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept();
            peer.expectDischarge().accept();
            peer.expectEnd().respond();
            peer.expectClose().respond();

            session.beginTransaction();
            session.commitTransaction();
            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testExceptionOnBeginWhenCoordinatorLinkRefused() throws Exception {
        final String errorMessage = "CoordinatorLinkRefusal-breadcrumb";

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().reject(true, AmqpError.NOT_IMPLEMENTED.toString(), errorMessage);
            peer.expectDetach();
            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().openFuture().get();

            try {
                session.beginTransaction();
                fail("Begin should have failed after link closed.");
            } catch (ClientTransactionDeclarationException expected) {
                // Expect this to time out.
                String message = expected.getMessage();
                assertTrue(message.contains(errorMessage));
            }

            try {
                session.commitTransaction();
                fail("Commit should have failed due to no active transaction.");
            } catch (ClientTransactionNotActiveException expected) {
                // Expect this as the begin failed on coordinator rejected
            }

            try {
                session.rollbackTransaction();
                fail("Rollback should have failed due to no active transaction.");
            } catch (ClientTransactionNotActiveException expected) {
                // Expect this as the begin failed on coordinator rejected
            }

            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testExceptionOnBeginWhenCoordinatorLinkClosedAfterDeclare() throws Exception {
        final String errorMessage = "CoordinatorLinkClosed-breadcrumb";

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare();
            peer.remoteDetach().withClosed(true)
                               .withErrorCondition(AmqpError.NOT_IMPLEMENTED.toString(), errorMessage).queue();
            peer.expectDetach();
            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().openFuture().get();

            try {
                session.beginTransaction();
                fail("Begin should have failed after link closed.");
            } catch (ClientException expected) {
                // Expect this to time out.
                String message = expected.getMessage();
                assertTrue(message.contains(errorMessage));
            }

            try {
                session.commitTransaction();
                fail("Commit should have failed due to no active transaction.");
            } catch (ClientTransactionNotActiveException expected) {
                // Expect this as the begin failed on coordinator close
            }

            try {
                session.rollbackTransaction();
                fail("Rollback should have failed due to no active transaction.");
            } catch (ClientTransactionNotActiveException expected) {
                // Expect this as the begin failed on coordinator close
            }

            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testExceptionOnBeginWhenCoordinatorLinkClosedAfterDeclareAllowsNewTransactionDeclaration() throws Exception {
        final String errorMessage = "CoordinatorLinkClosed-breadcrumb";

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare();
            peer.remoteDetach().withClosed(true)
                               .withErrorCondition(AmqpError.NOT_IMPLEMENTED.toString(), errorMessage).queue();
            peer.expectDetach();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept();
            peer.expectDischarge().accept();
            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().openFuture().get();

            try {
                session.beginTransaction();
                fail("Begin should have failed after link closed.");
            } catch (ClientException expected) {
                // Expect this to time out.
                String message = expected.getMessage();
                assertTrue(message.contains(errorMessage));
            }

            // Try again and expect to return to normal state now.
            session.beginTransaction();
            session.commitTransaction();

            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testExceptionOnCommitWhenCoordinatorLinkClosedAfterDischargeSent() throws Exception {
        final String errorMessage = "CoordinatorLinkClosed-breadcrumb";

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept();
            peer.expectDischarge();
            peer.remoteDetach().withClosed(true)
                               .withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), errorMessage).queue();
            peer.expectDetach();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept();
            peer.expectDischarge().accept();
            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().openFuture().get();

            session.beginTransaction();

            try {
                session.commitTransaction();
                fail("Commit should have failed after link closed.");
            } catch (ClientTransactionRolledBackException expected) {
                // Expect this to time out.
                String message = expected.getMessage();
                assertTrue(message.contains(errorMessage));
            }

            session.beginTransaction();
            session.rollbackTransaction();

            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testExceptionOnCommitWhenCoordinatorLinkClosedAfterTxnDeclaration() throws Exception {
        doTestExceptionOnDischargeWhenCoordinatorLinkClosedAfterTxnDeclaration(true);
    }

    @Test
    public void testExceptionOnRollbackWhenCoordinatorLinkClosedAfterTxnDeclaration() throws Exception {
        doTestExceptionOnDischargeWhenCoordinatorLinkClosedAfterTxnDeclaration(false);
    }

    private void doTestExceptionOnDischargeWhenCoordinatorLinkClosedAfterTxnDeclaration(boolean commit) throws Exception {
        final String errorMessage = "CoordinatorLinkClosed-breadcrumb";

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept();
            peer.remoteDetachLastCoordinatorLink().withClosed(true)
                                                  .withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), errorMessage).queue();
            peer.expectDischarge().optional();  // No discharge if close processed before commit or rollback triggered
            peer.expectDetach();
            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().openFuture().get();

            session.beginTransaction();

            if (commit) {
                try {
                    session.commitTransaction();
                    fail("Commit should have failed after link closed.");
                } catch (ClientTransactionRolledBackException expected) {
                    // Expect this to time out.
                    String message = expected.getMessage();
                    assertTrue(message.contains(errorMessage));
                }
            } else {
                try {
                    session.rollbackTransaction();
                } catch (Exception ex) {
                    LOG.debug("Caught unexpected exception from rollback", ex);
                    fail("Rollback should not have failed after link closed.");
                }
            }

            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testExceptionOnCommitWhenCoordinatorRejectsDischarge() throws Exception {
        final String errorMessage = "Transaction aborted due to timeout";
        final byte[] txnId1 = new byte[] { 0, 1, 2, 3 };
        final byte[] txnId2 = new byte[] { 1, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(4).queue();
            peer.expectDeclare().accept(txnId1);
            peer.expectDischarge().withFail(false)
                                  .withTxnId(txnId1)
                                  .reject(TransactionErrors.TRANSACTION_TIMEOUT.toString(), "Transaction aborted due to timeout");
            peer.expectDeclare().accept(txnId2);
            peer.expectDischarge().withFail(true).withTxnId(txnId2).accept();
            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().openFuture().get();

            session.beginTransaction();

            try {
                session.commitTransaction();
                fail("Commit should have failed after link closed.");
            } catch (ClientTransactionRolledBackException expected) {
                // Expect this to time out.
                String message = expected.getMessage();
                assertTrue(message.contains(errorMessage));
            }

            session.beginTransaction();
            session.rollbackTransaction();

            session.closeAsync();
            connection.closeAsync().get();

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

    @Test
    public void testExceptionOnRollbackWhenCoordinatorRejectsDischarge() throws Exception {
        final String errorMessage = "Transaction aborted due to timeout";
        final byte[] txnId1 = new byte[] { 0, 1, 2, 3 };
        final byte[] txnId2 = new byte[] { 1, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(4).queue();
            peer.expectDeclare().accept(txnId1);
            peer.expectDischarge().withFail(true)
                                  .withTxnId(txnId1)
                                  .reject(TransactionErrors.TRANSACTION_TIMEOUT.toString(), "Transaction aborted due to timeout");
            peer.expectDeclare().accept(txnId2);
            peer.expectDischarge().withFail(false).withTxnId(txnId2).accept();
            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().openFuture().get();

            session.beginTransaction();

            try {
                session.rollbackTransaction();
                fail("Commit should have failed after link closed.");
            } catch (ClientTransactionRolledBackException expected) {
                // Expect this to time out.
                String message = expected.getMessage();
                assertTrue(message.contains(errorMessage));
            }

            session.beginTransaction();
            session.commitTransaction();

            session.closeAsync();
            connection.closeAsync().get();

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

    /**
     * Create a transaction and then close the Session which result in the remote rolling back
     * the transaction by default so the client doesn't manually roll it back itself.
     *
     * @throws Exception
     */
    @Test
    public void testBeginTransactionAndClose() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept();
            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().openFuture().get();

            session.beginTransaction();

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

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

    @Test
    public void testBeginAndCommitTransaction() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            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().openFuture().get();

            session.beginTransaction();
            session.commitTransaction();

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

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

    @Test
    public void testBeginAndRollbackTransaction() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.expectDischarge().withFail(true).withTxnId(txnId).accept();
            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().openFuture().get();

            session.beginTransaction();
            session.rollbackTransaction();

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

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

    @Test
    public void testTransactionDeclaredDispositionWithoutTxnId() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectDeclare().accept(null);
            peer.expectClose().withError(AmqpError.DECODE_ERROR.toString(), "The txn-id field cannot be omitted").respond();
            peer.start();

            URI remoteURI = peer.getServerURI();

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

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

            try {
                session.beginTransaction();
                fail("Should not complete transaction begin due to client connection failure on decode issue.");
            } catch (ClientException ex) {
                // expected to fail
            }

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

            peer.waitForScriptToCompleteIgnoreErrors();
        }
    }

    @Test
    public void testBeginAndCommitTransactions() throws Exception {
        final byte[] txnId1 = new byte[] { 0, 1, 2, 3 };
        final byte[] txnId2 = new byte[] { 1, 1, 2, 3 };
        final byte[] txnId3 = new byte[] { 2, 1, 2, 3 };
        final byte[] txnId4 = new byte[] { 3, 1, 2, 3 };
        final byte[] txnId5 = new byte[] { 4, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(10).queue();
            peer.expectDeclare().accept(txnId1);
            peer.expectDischarge().withFail(false).withTxnId(txnId1).accept();
            peer.expectDeclare().accept(txnId2);
            peer.expectDischarge().withFail(false).withTxnId(txnId2).accept();
            peer.expectDeclare().accept(txnId3);
            peer.expectDischarge().withFail(false).withTxnId(txnId3).accept();
            peer.expectDeclare().accept(txnId4);
            peer.expectDischarge().withFail(false).withTxnId(txnId4).accept();
            peer.expectDeclare().accept(txnId5);
            peer.expectDischarge().withFail(false).withTxnId(txnId5).accept();
            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().openFuture().get();

            for (int i = 0; i < 5; ++i) {
                LOG.info("Transaction declare and discharge cycle: {}", i);
                session.beginTransaction();
                session.commitTransaction();
            }

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

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

    @Test
    public void testCannotBeginSecondTransactionWhileFirstIsActive() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept();
            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().openFuture().get();

            session.beginTransaction();

            try {
                session.beginTransaction();
                fail("Should not be allowed to begin another transaction");
            } catch (ClientIllegalStateException cliEx) {
                // Expected
            }

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

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

    @Test
    public void testSendMessageInsideOfTransaction() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.expectTransfer().withHandle(0)
                                 .withNonNullPayload()
                                 .withState().transactional().withTxnId(txnId).and()
                                 .respond()
                                 .withState().transactional().withTxnId(txnId).withAccepted().and()
                                 .withSettled(true);
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            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().openFuture().get();
            Sender sender = session.openSender("address").openFuture().get();

            session.beginTransaction();

            final Tracker tracker = sender.send(Message.create("test-message"));

            assertNotNull(tracker);
            assertNotNull(tracker.settlementFuture().get());
            assertEquals(tracker.remoteState().getType(), DeliveryState.Type.TRANSACTIONAL,
                         "Delivery inside transaction should have Transactional state");
            assertNotNull(tracker.state());
            assertEquals(tracker.state().getType(), DeliveryState.Type.TRANSACTIONAL,
                         "Delivery inside transaction should have Transactional state: " + tracker.state().getType());
            Wait.assertTrue("Delivery in transaction should be locally settled after response", () -> tracker.settled());

            session.commitTransaction();

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

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

    @Test
    public void testAwaitSettlementWorksForMessageSentInTransaction() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.expectTransfer().withHandle(0)
                                 .withNonNullPayload()
                                 .withState().transactional().withTxnId(txnId).and()
                                 .respond()
                                 .withState().transactional().withTxnId(txnId).withAccepted().and()
                                 .withSettled(true);
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            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().openFuture().get();
            Sender sender = session.openSender("address").openFuture().get();

            session.beginTransaction();

            final Tracker tracker = sender.send(Message.create("test-message"));

            assertNotNull(tracker);
            assertNotNull(tracker.awaitAccepted());
            assertTrue(tracker.remoteState().isAccepted());
            assertEquals(tracker.remoteState().getType(), DeliveryState.Type.TRANSACTIONAL,
                         "Delivery inside transaction should have Transactional state");
            assertEquals(tracker.state().getType(), DeliveryState.Type.TRANSACTIONAL,
                         "Delivery inside transaction should have Transactional state: " + tracker.state().getType());
            Wait.assertTrue("Delivery in transaction should be locally settled after response", () -> tracker.settled());

            session.commitTransaction();

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

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

    @Test
    public void testSendMessagesInsideOfUniqueTransactions() throws Exception {
        final byte[] txnId1 = new byte[] { 0, 1, 2, 3 };
        final byte[] txnId2 = new byte[] { 1, 1, 2, 3 };
        final byte[] txnId3 = new byte[] { 2, 1, 2, 3 };
        final byte[] txnId4 = new byte[] { 3, 1, 2, 3 };

        final byte[][] txns = new byte[][] { txnId1, txnId2, txnId3, txnId4 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(txns.length).queue();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(txns.length * 2).queue();
            for (int i = 0; i < txns.length; ++i) {
                peer.expectDeclare().accept(txns[i]);
                peer.expectTransfer().withHandle(0)
                                     .withNonNullPayload()
                                     .withState().transactional().withTxnId(txns[i]).and()
                                     .respond()
                                     .withState().transactional().withTxnId(txns[i]).withAccepted().and()
                                     .withSettled(true);
                peer.expectDischarge().withFail(false).withTxnId(txns[i]).accept();
            }
            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().openFuture().get();
            Sender sender = session.openSender("address").openFuture().get();

            for (int i = 0; i < txns.length; ++i) {
                session.beginTransaction();

                final Tracker tracker = sender.send(Message.create("test-message-" + i));

                assertNotNull(tracker);
                assertNotNull(tracker.settlementFuture().get());
                assertEquals(tracker.remoteState().getType(), DeliveryState.Type.TRANSACTIONAL);
                assertNotNull(tracker.state());
                assertEquals(tracker.state().getType(), DeliveryState.Type.TRANSACTIONAL,
                    "Delivery inside transaction should have Transactional state: " + tracker.state().getType());
                Wait.assertTrue("Delivery in transaction should be locally settled after response", () -> tracker.settled());

                session.commitTransaction();
            }

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

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

    @Test
    public void testReceiveMessageInsideOfTransaction() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.start();

            final URI remoteURI = peer.getServerURI();
            final byte[] payload = createEncodedMessage(new AmqpValue<>("Hello World"));

            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.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDisposition().withSettled(true)
                                    .withState().transactional().withTxnId(txnId).withAccepted();
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            session.beginTransaction();

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

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

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

    @Test
    public void testReceiveMessageInsideOfTransactionNoAutoSettleSenderSettles() throws Exception {
        doTestReceiveMessageInsideOfTransactionNoAutoSettle(true);
    }

    @Test
    public void testReceiveMessageInsideOfTransactionNoAutoSettleSenderDoesNotSettle() throws Exception {
        doTestReceiveMessageInsideOfTransactionNoAutoSettle(false);
    }

    private void doTestReceiveMessageInsideOfTransactionNoAutoSettle(boolean settle) throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.start();

            final URI remoteURI = peer.getServerURI();
            final byte[] payload = createEncodedMessage(new AmqpValue<>("Hello World"));

            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().autoAccept(false).autoSettle(false);
            Receiver receiver = session.openReceiver("test-queue", options).openFuture().get();

            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDisposition().withSettled(true)
                                    .withState().transactional().withTxnId(txnId).withAccepted();
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            session.beginTransaction();

            Delivery delivery = receiver.receive(100, TimeUnit.MILLISECONDS);
            assertNotNull(delivery);
            assertFalse(delivery.settled());
            assertNull(delivery.state());

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

            // Manual Accept within the transaction, settlement is ignored.
            delivery.disposition(DeliveryState.accepted(), settle);

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

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

    @Test
    public void testReceiveMessageInsideOfTransactionButAcceptAndSettleOutside() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.start();

            final URI remoteURI = peer.getServerURI();
            final byte[] payload = createEncodedMessage(new AmqpValue<>("Hello World"));

            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().autoAccept(false).autoSettle(false);
            Receiver receiver = session.openReceiver("test-queue", options).openFuture().get();

            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            peer.expectDisposition().withSettled(true).withState().accepted();

            session.beginTransaction();

            Delivery delivery = receiver.receive(100, TimeUnit.MILLISECONDS);
            assertNotNull(delivery);
            assertFalse(delivery.settled());
            assertNull(delivery.state());

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

            session.commitTransaction();

            // Manual Accept outside the transaction and no auto settle or accept
            // so no transactional enlistment.
            delivery.disposition(DeliveryState.accepted(), true);

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

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

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

    @Test
    public void testTransactionCommitFailWithEmptyRejectedDisposition() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.expectTransfer().withHandle(0)
                                 .withNonNullPayload()
                                 .withState().transactional().withTxnId(txnId).and()
                                 .respond()
                                 .withState().transactional().withTxnId(txnId).withAccepted().and()
                                 .withSettled(true);
            peer.expectDischarge().withFail(false).withTxnId(txnId).reject();
            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().openFuture().get();
            Sender sender = session.openSender("address").openFuture().get();

            session.beginTransaction();

            final Tracker tracker = sender.send(Message.create("test-message"));
            assertNotNull(tracker.settlementFuture().get());
            assertEquals(tracker.remoteState().getType(), DeliveryState.Type.TRANSACTIONAL);

            try {
                session.commitTransaction();
                fail("Commit should fail with Rollback exception");
            } catch (ClientTransactionRolledBackException cliRbEx) {
                // Expected roll back due to discharge rejection
            }

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

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

    @Test
    public void testDeclareTransactionAfterConnectionDrops() throws Exception {
        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            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().openFuture().get();

            peer.waitForScriptToComplete();

            try {
                session.beginTransaction();
                fail("Should have failed to discharge transaction");
            } catch (ClientException cliEx) {
                // Expected error as connection was dropped
                LOG.debug("Client threw error on begin after connection drop", cliEx);
            }

            connection.closeAsync().get();

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

    @Test
    public void testCommitTransactionAfterConnectionDropsFollowingTxnDeclared() throws Exception {
        dischargeTransactionAfterConnectionDropsFollowingTxnDeclared(true);
    }

    @Test
    public void testRollbackTransactionAfterConnectionDropsFollowingTxnDeclared() throws Exception {
        dischargeTransactionAfterConnectionDropsFollowingTxnDeclared(false);
    }

    public void dischargeTransactionAfterConnectionDropsFollowingTxnDeclared(boolean commit) throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            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().openFuture().get();

            session.beginTransaction();

            peer.waitForScriptToComplete();

            if (commit) {
                try {
                    session.commitTransaction();
                    fail("Should have failed to commit transaction");
                } catch (ClientException cliEx) {
                    // Expected error as connection was dropped
                }
            } else {
                try {
                    session.rollbackTransaction();
                } catch (ClientConnectionRemotelyClosedException cliEx) {
                    // Can get an error if the session processes the close before the
                    // roll back is called.  Mitigating that is tricky and still leaves
                    // the user needing to handle error when session is actually closed
                    // via Session.close()
                } catch (Exception ex) {
                    LOG.info("Caught unexpected error: {}", ex);
                    fail("Connection drops will implicitly roll back TXN on remote");
                }
            }

            connection.closeAsync().get();

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

    @Test
    public void testSendMessagesNoOpWhenTransactionInDoubt() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectDeclare().accept(txnId);
            peer.remoteDetach().withClosed(true)
                               .withErrorCondition(AmqpError.RESOURCE_DELETED.toString(), "Coordinator").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());
            Session session = connection.openSession().openFuture().get();

            session.beginTransaction();

            // After the wait TXN should be in doubt and send should no-op
            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(1).queue();
            peer.expectEnd().respond();
            peer.expectClose().respond();

            final Sender sender = session.openSender("address").openFuture().get();

            for (int i = 0; i < 10; ++i) {
                final Tracker tracker = sender.send(Message.create("test-message-"));

                assertNotNull(tracker);
                assertNotNull(tracker.settlementFuture().get());
                assertEquals(ClientDeliveryState.ClientAccepted.getInstance(), tracker.remoteState());
                assertTrue(tracker.remoteSettled());
                assertNull(tracker.state());
                assertFalse(tracker.settled());
                assertFalse(tracker.awaitAccepted().settled());
                assertFalse(tracker.awaitSettlement().settled());
                assertFalse(tracker.awaitAccepted(1, TimeUnit.SECONDS).settled());
                assertFalse(tracker.awaitSettlement(1, TimeUnit.SECONDS).settled());
                assertSame(sender, tracker.sender());

                // These should no-op since message was never sent.
                tracker.settle();
                tracker.disposition(ClientDeliveryState.ClientAccepted.getInstance(), true);
            }

            try {
                session.commitTransaction();
                fail("Should not be able to commit as remote closed coordinator");
            } catch (ClientTransactionRolledBackException cliTxRbEx) {
                // Expected
            }

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

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

    @Test
    void testStreamSenderMessageCanOperatesWithinTransaction() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofSender().respond();
            peer.remoteFlow().withLinkCredit(2).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());
            StreamSender sender = connection.openStreamSender("test-queue");
            StreamSenderMessage message = sender.beginMessage();

            // Populate all Header values
            Header header = new Header();
            header.setDurable(true);
            header.setPriority((byte) 1);
            header.setTimeToLive(65535);
            header.setFirstAcquirer(true);
            header.setDeliveryCount(2);

            message.header(header);

            OutputStreamOptions options = new OutputStreamOptions();
            OutputStream stream = message.body(options);

            HeaderMatcher headerMatcher = new HeaderMatcher(true);
            headerMatcher.withDurable(true);
            headerMatcher.withPriority((byte) 1);
            headerMatcher.withTtl(65535);
            headerMatcher.withFirstAcquirer(true);
            headerMatcher.withDeliveryCount(2);
            EncodedDataMatcher dataMatcher = new EncodedDataMatcher(new byte[] { 0, 1, 2, 3 });
            TransferPayloadCompositeMatcher payloadMatcher = new TransferPayloadCompositeMatcher();
            payloadMatcher.setHeadersMatcher(headerMatcher);
            payloadMatcher.setMessageContentMatcher(dataMatcher);

            peer.waitForScriptToComplete(5, TimeUnit.SECONDS);
            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(5).queue();
            peer.expectDeclare().accept(txnId);
            peer.expectTransfer().withHandle(0)
                                 .withMore(true)
                                 .withPayload(payloadMatcher)
                                 .withState().transactional().withTxnId(txnId).and()
                                 .respond()
                                 .withState().transactional().withTxnId(txnId).withAccepted().and()
                                 .withSettled(true);
            peer.expectTransfer().withMore(false).withNullPayload();
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            peer.expectDetach().respond();
            peer.expectEnd().respond();
            peer.expectClose().respond();

            sender.session().beginTransaction();

            // Stream won't output until some body bytes are written since the buffer was not
            // filled by the header write.  Then the close will complete the stream message.
            stream.write(new byte[] { 0, 1, 2, 3 });
            stream.flush();
            stream.close();

            sender.session().commitTransaction();
            sender.closeAsync().get();

            connection.closeAsync().get();

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

    @Test
    public void testAcceptAndReleaseInSameTransaction() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.start();

            final URI remoteURI = peer.getServerURI();
            final byte[] payload = createEncodedMessage(new AmqpValue<>("Hello World"));

            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().autoAccept(false).autoSettle(false);
            Receiver receiver = session.openReceiver("test-queue", options).openFuture().get();

            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(1)
                                 .withDeliveryTag(new byte[] { 2 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDisposition().withSettled(true)
                                    .withState().transactional().withTxnId(txnId).withAccepted();
            peer.expectDisposition().withSettled(true)
                                    .withState().transactional().withTxnId(txnId).withReleased();
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            session.beginTransaction();

            final Delivery delivery1 = receiver.receive(100, TimeUnit.MILLISECONDS);
            final Delivery delivery2 = receiver.receive(100, TimeUnit.MILLISECONDS);

            assertNotNull(delivery1);
            assertFalse(delivery1.settled());
            assertNull(delivery1.state());
            assertNotNull(delivery2);
            assertFalse(delivery2.settled());
            assertNull(delivery2.state());

            delivery1.accept();
            delivery2.release();

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

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

    @Test
    public void testModifiedDispositionInTransaction() throws Exception {
        final byte[] txnId = new byte[] { 0, 1, 2, 3 };

        try (ProtonTestServer peer = new ProtonTestServer()) {
            peer.expectSASLAnonymousConnect();
            peer.expectOpen().respond();
            peer.expectBegin().respond();
            peer.expectAttach().ofReceiver().respond();
            peer.expectFlow();
            peer.start();

            final URI remoteURI = peer.getServerURI();
            final byte[] payload = createEncodedMessage(new AmqpValue<>("Hello World"));

            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().autoAccept(false).autoSettle(false);
            Receiver receiver = session.openReceiver("test-queue", options).openFuture().get();

            peer.expectCoordinatorAttach().respond();
            peer.remoteFlow().withLinkCredit(2).queue();
            peer.expectDeclare().accept(txnId);
            peer.remoteTransfer().withHandle(0)
                                 .withDeliveryId(0)
                                 .withDeliveryTag(new byte[] { 1 })
                                 .withMore(false)
                                 .withMessageFormat(0)
                                 .withPayload(payload).queue();
            peer.expectDisposition().withSettled(true)
                                    .withState()
                                    .transactional()
                                    .withTxnId(txnId)
                                    .withModified(true, true);
            peer.expectDischarge().withFail(false).withTxnId(txnId).accept();
            peer.expectDetach().respond();
            peer.expectClose().respond();

            session.beginTransaction();

            final Delivery delivery = receiver.receive(100, TimeUnit.MILLISECONDS);

            assertNotNull(delivery);
            assertFalse(delivery.settled());
            assertNull(delivery.state());

            delivery.modified(true, true);

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

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