diff --git a/src/NMS.AMQP/Apache-NMS-AMQP.csproj b/src/NMS.AMQP/Apache-NMS-AMQP.csproj
index 5f67b68..dfb16a4 100644
--- a/src/NMS.AMQP/Apache-NMS-AMQP.csproj
+++ b/src/NMS.AMQP/Apache-NMS-AMQP.csproj
@@ -49,8 +49,8 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <None Include="..\..\LICENSE.txt" Pack="true" PackagePath="LICENSE.txt" />
-        <None Include="..\..\NOTICE.txt" Pack="true" PackagePath="NOTICE.txt" />
+        <None Include="..\..\LICENSE.txt" Pack="true" PackagePath="LICENSE.txt"/>
+        <None Include="..\..\NOTICE.txt" Pack="true" PackagePath="NOTICE.txt"/>
     </ItemGroup>
 
     <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs b/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
index 0ca5b55..38f16d4 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
@@ -44,9 +44,10 @@
         public AmqpProvider Provider { get; }
         private readonly ITransportContext transport;
         private readonly Uri remoteUri;
-        private global::Amqp.Connection underlyingConnection;
+        private Connection underlyingConnection;
         private readonly AmqpMessageFactory messageFactory;
         private AmqpConnectionSession connectionSession;
+        private TaskCompletionSource<bool> tsc;
 
         public AmqpConnection(AmqpProvider provider, ITransportContext transport, ConnectionInfo info)
         {
@@ -68,39 +69,39 @@
         internal async Task Start()
         {
             Address address = UriUtil.ToAddress(remoteUri, Info.username, Info.password);
-            underlyingConnection = await transport.CreateAsync(address, CreateOpenFrame(Info), OnOpened);
+            this.tsc = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            underlyingConnection = await transport.CreateAsync(address, new AmqpHandler(this)).ConfigureAwait(false);
             underlyingConnection.AddClosedCallback(Provider.OnInternalClosed);
+            
+            // Wait for connection to be opened
+            await tsc.Task;
 
             // Create a Session for this connection that is used for Temporary Destinations
             // and perhaps later on management and advisory monitoring.
-
             // TODO: change the way how connection session id is obtained
             SessionInfo sessionInfo = new SessionInfo(Info.Id);
             sessionInfo.AcknowledgementMode = AcknowledgementMode.AutoAcknowledge;
 
             connectionSession = new AmqpConnectionSession(this, sessionInfo);
-            await connectionSession.Start();
+            await connectionSession.Start().ConfigureAwait(false);
         }
 
-        private Open CreateOpenFrame(ConnectionInfo connInfo)
+        internal void OnLocalOpen(Open open)
         {
-            return new Open
+            open.ContainerId = Info.ClientId;
+            open.ChannelMax = Info.channelMax;
+            open.MaxFrameSize = Convert.ToUInt32(Info.maxFrameSize);
+            open.HostName = remoteUri.Host;
+            open.IdleTimeOut = Convert.ToUInt32(Info.idleTimout);
+            open.DesiredCapabilities = new[]
             {
-                ContainerId = connInfo.ClientId,
-                ChannelMax = connInfo.channelMax,
-                MaxFrameSize = Convert.ToUInt32(connInfo.maxFrameSize),
-                HostName = remoteUri.Host,
-                IdleTimeOut = Convert.ToUInt32(connInfo.idleTimout),
-                DesiredCapabilities = new[]
-                {
-                    SymbolUtil.OPEN_CAPABILITY_SOLE_CONNECTION_FOR_CONTAINER,
-                    SymbolUtil.OPEN_CAPABILITY_DELAYED_DELIVERY,
-                    SymbolUtil.OPEN_CAPABILITY_ANONYMOUS_RELAY
-                }
+                SymbolUtil.OPEN_CAPABILITY_SOLE_CONNECTION_FOR_CONTAINER,
+                SymbolUtil.OPEN_CAPABILITY_DELAYED_DELIVERY,
+                SymbolUtil.OPEN_CAPABILITY_ANONYMOUS_RELAY
             };
         }
 
-        private void OnOpened(global::Amqp.IConnection connection, Open open)
+        internal void OnRemoteOpened(Open open)
         {
             if (SymbolUtil.CheckAndCompareFields(open.Properties, SymbolUtil.CONNECTION_ESTABLISH_FAILED, SymbolUtil.BOOLEAN_TRUE))
             {
@@ -121,6 +122,7 @@
                     Info.QueuePrefix = queuePrefix;
                 }
 
+                this.tsc.SetResult(true);
                 Provider.FireConnectionEstablished();
             }
         }
@@ -172,7 +174,7 @@
         {
             return temporaryDestinations.TryGetValue(destination.Id, out AmqpTemporaryDestination amqpTemporaryDestination) ? amqpTemporaryDestination : null;
         }
-        
+
         public void RemoveTemporaryDestination(Id destinationId)
         {
             temporaryDestinations.TryRemove(destinationId, out _);
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpHandler.cs b/src/NMS.AMQP/Provider/Amqp/AmqpHandler.cs
new file mode 100644
index 0000000..6841d78
--- /dev/null
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpHandler.cs
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+
+using Amqp.Framing;
+using Amqp.Handler;
+
+namespace Apache.NMS.AMQP.Provider.Amqp
+{
+    internal class AmqpHandler : IHandler
+    {
+        private readonly AmqpConnection connection;
+
+        public AmqpHandler(AmqpConnection connection)
+        {
+            this.connection = connection;
+        }
+
+        public bool CanHandle(EventId id)
+        {
+            switch (id)
+            {
+                case EventId.SendDelivery:
+                case EventId.ConnectionRemoteOpen:
+                case EventId.ConnectionLocalOpen:
+                    return true;
+                default:
+                    return false;
+            }
+        }
+
+        public void Handle(Event protocolEvent)
+        {
+            switch (protocolEvent.Id)
+            {
+                case EventId.SendDelivery when protocolEvent.Context is IDelivery delivery:
+                    delivery.Batchable = false;
+                    break;
+                case EventId.ConnectionRemoteOpen when protocolEvent.Context is Open open:
+                    this.connection.OnRemoteOpened(open);
+                    break;
+                case EventId.ConnectionLocalOpen when protocolEvent.Context is Open open:
+                    this.connection.OnLocalOpen(open);
+                    break;
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs b/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs
index 9fcaa6e..d767577 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpProducer.cs
@@ -121,13 +121,7 @@
                     if (envelope.SendAsync)
                         SendAsync(message, transactionalState);
                     else
-                    {
-                        // TODO: Should be unified after https://github.com/Azure/amqpnetlite/pull/374 is sorted out
-                        if (transactionalState != null)
-                            SendSync(message, transactionalState);
-                        else
-                            senderLink.Send(message, TimeSpan.FromMilliseconds(session.Connection.Provider.SendTimeout));
-                    }
+                        SendSync(message, transactionalState);
                 }
                 catch (TimeoutException tex)
                 {
diff --git a/src/NMS.AMQP/Transport/ITransportContext.cs b/src/NMS.AMQP/Transport/ITransportContext.cs
index d3132a9..ad225be 100644
--- a/src/NMS.AMQP/Transport/ITransportContext.cs
+++ b/src/NMS.AMQP/Transport/ITransportContext.cs
@@ -18,6 +18,7 @@
 using System.Threading.Tasks;
 using Amqp;
 using Amqp.Framing;
+using Amqp.Handler;
 
 namespace Apache.NMS.AMQP.Transport
 {
@@ -37,6 +38,6 @@
 
         ITransportContext Copy();
 
-        Task<Amqp.Connection> CreateAsync(Address address, Open open = null, OnOpened onOpened = null);
+        Task<Connection> CreateAsync(Address address, IHandler handler);
     }
 }
diff --git a/src/NMS.AMQP/Transport/SecureTransportContext.cs b/src/NMS.AMQP/Transport/SecureTransportContext.cs
index a4a3eb8..cc3913c 100644
--- a/src/NMS.AMQP/Transport/SecureTransportContext.cs
+++ b/src/NMS.AMQP/Transport/SecureTransportContext.cs
@@ -24,6 +24,7 @@
 using System.Threading.Tasks;
 using Amqp;
 using Amqp.Framing;
+using Amqp.Handler;
 using Apache.NMS.AMQP.Util;
 
 namespace Apache.NMS.AMQP.Transport
@@ -228,7 +229,7 @@
 
         #region IProviderSecureTransportContext Methods
 
-        public override Task<Amqp.Connection> CreateAsync(Address address, Open open = null, OnOpened onOpened = null)
+        public override Task<Amqp.Connection> CreateAsync(Address address, IHandler handler)
         {
             // Load local certificates
             this.connectionBuilder.SSL.ClientCertificates.AddRange(LoadClientCertificates());
@@ -242,7 +243,7 @@
                 throw new NMSSecurityException(string.Format("Invalid SSL Protocol {0} selected from system supported protocols {1}", this.SSLProtocol, PropertyUtil.ToString(SupportedProtocols)));
             }
             
-            return base.CreateAsync(address, open, onOpened);
+            return base.CreateAsync(address, handler);
         }
 
         #endregion
diff --git a/src/NMS.AMQP/Transport/TransportContext.cs b/src/NMS.AMQP/Transport/TransportContext.cs
index 5e1666f..229d8e7 100644
--- a/src/NMS.AMQP/Transport/TransportContext.cs
+++ b/src/NMS.AMQP/Transport/TransportContext.cs
@@ -20,6 +20,8 @@
 using System.Threading.Tasks;
 using Amqp;
 using Amqp.Framing;
+using Amqp.Handler;
+using Apache.NMS.AMQP.Provider.Amqp;
 using Apache.NMS.AMQP.Util;
 
 namespace Apache.NMS.AMQP.Transport
@@ -38,7 +40,7 @@
         internal TransportContext()
         {
             connectionBuilder = new Amqp.ConnectionFactory();
-            connectionBuilder.SASL.Profile = Amqp.Sasl.SaslProfile.Anonymous;
+            connectionBuilder.SASL.Profile = Amqp.Sasl.SaslProfile.Anonymous;            
         }
 
         static TransportContext()
@@ -156,9 +158,9 @@
             return copy;
         }
 
-        public virtual Task<Amqp.Connection> CreateAsync(Address address, Open open = null, OnOpened onOpened = null)
+        public virtual Task<Connection> CreateAsync(Address address, IHandler handler)
         {
-            return connectionBuilder.CreateAsync(address, open, onOpened);
+            return connectionBuilder.CreateAsync(address, handler);    
         }
 
         protected virtual void CopyInto(TransportContext copy)
diff --git a/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationTest.cs
index 110f65d..d34d529 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationTest.cs
@@ -419,9 +419,9 @@
                 producer.Send(message, MsgDeliveryMode.Persistent, (MsgPriority) priority, NMSConstants.defaultTimeToLive);
 
                 Assert.AreEqual((MsgPriority) priority, message.NMSPriority);
-                
+
                 connection.Close();
-                
+
                 testPeer.WaitForAllMatchersToComplete(1000);
             }
         }
@@ -438,7 +438,7 @@
                 ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
                 IQueue destination = session.GetQueue("myQueue");
                 IMessageProducer producer = session.CreateProducer(destination);
-                
+
                 string text = "myMessage";
                 string actualMessageId = null;
                 testPeer.ExpectTransfer(m =>
@@ -448,18 +448,18 @@
                     actualMessageId = m.Properties.MessageId;
                 });
                 testPeer.ExpectClose();
-                
+
                 ITextMessage message = session.CreateTextMessage(text);
                 Assert.IsNull(message.NMSMessageId, "NMSMessageId should not yet be set");
-                
+
                 producer.Send(message);
-                
+
                 Assert.IsNotNull(message.NMSMessageId);
                 Assert.IsNotEmpty(message.NMSMessageId, "NMSMessageId should be set");
                 Assert.IsTrue(message.NMSMessageId.StartsWith("ID:"), "MMS 'ID:' prefix not found");
-                
+
                 connection.Close();
-                
+
                 testPeer.WaitForAllMatchersToComplete(1000);
                 // Get the value that was actually transmitted/received, verify it is a string, compare to what we have locally
                 Assert.AreEqual(message.NMSMessageId, actualMessageId, "Expected NMSMessageId value to be present in AMQP message");
@@ -489,7 +489,7 @@
                 ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
                 IQueue destination = session.GetQueue("myQueue");
                 IMessageProducer producer = session.CreateProducer(destination);
-                
+
                 string text = "myMessage";
                 testPeer.ExpectTransfer(m =>
                 {
@@ -498,11 +498,11 @@
                     Assert.AreEqual(text, (m.BodySection as AmqpValue).Value);
                 });
                 testPeer.ExpectClose();
-                
+
                 ITextMessage message = session.CreateTextMessage(text);
-                
+
                 Assert.IsNull(message.NMSMessageId, "NMSMessageId should not yet be set");
-                
+
                 if (existingId)
                 {
                     string existingMessageId = "ID:this-should-be-overwritten-in-send";
@@ -511,13 +511,13 @@
                 }
 
                 producer.DisableMessageID = true;
-                
+
                 producer.Send(message);
 
                 Assert.IsNull(message.NMSMessageId, "NMSMessageID should be null");
-                
+
                 connection.Close();
-                
+
                 testPeer.WaitForAllMatchersToComplete(2000);
             }
         }
@@ -526,7 +526,7 @@
         public void TestRemotelyCloseProducer()
         {
             string breadCrumb = "ErrorMessageBreadCrumb";
-            
+
             ManualResetEvent producerClosed = new ManualResetEvent(false);
             Mock<INmsConnectionListener> mockConnectionListener = new Mock<INmsConnectionListener>();
             mockConnectionListener
@@ -537,23 +537,23 @@
             {
                 NmsConnection connection = (NmsConnection) EstablishConnection(testPeer);
                 connection.AddConnectionListener(mockConnectionListener.Object);
-                
+
                 testPeer.ExpectBegin();
                 ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
-                
+
                 // Create a producer, then remotely end it afterwards.
                 testPeer.ExpectSenderAttach();
                 testPeer.RemotelyDetachLastOpenedLinkOnLastOpenedSession(expectDetachResponse: true, closed: true, errorType: AmqpError.RESOURCE_DELETED, breadCrumb);
-                
+
                 IQueue destination = session.GetQueue("myQueue");
                 IMessageProducer producer = session.CreateProducer(destination);
-                
+
                 // Verify the producer gets marked closed
                 testPeer.WaitForAllMatchersToComplete(1000);
-                
+
                 Assert.True(producerClosed.WaitOne(TimeSpan.FromMilliseconds(1000)), "Producer closed callback didn't trigger");
                 Assert.That(() => producer.DisableMessageID, Throws.Exception.InstanceOf<IllegalStateException>(), "Producer never closed");
-                
+
                 // Try closing it explicitly, should effectively no-op in client.
                 // The test peer will throw during close if it sends anything.
                 producer.Close();
@@ -567,23 +567,23 @@
             {
                 IConnection connection = EstablishConnection(testPeer, optionsString: "nms.sendTimeout=500");
                 testPeer.ExpectBegin();
-                
+
                 ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
                 IQueue queue = session.GetQueue("myQueue");
-                
+
                 ITextMessage message = session.CreateTextMessage("text");
-                
+
                 // Expect the producer to attach. Don't send any credit so that the client will
                 // block on a send and we can test our timeouts.
                 testPeer.ExpectSenderAttachWithoutGrantingCredit();
                 testPeer.ExpectClose();
-                
+
                 IMessageProducer producer = session.CreateProducer(queue);
-                
+
                 Assert.Catch<Exception>(() => producer.Send(message), "Send should time out.");
-                
+
                 connection.Close();
-                
+
                 testPeer.WaitForAllMatchersToComplete(1000);
             }
         }
@@ -600,23 +600,20 @@
                 IQueue queue = session.GetQueue("myQueue");
 
                 ITextMessage message = session.CreateTextMessage("text");
-                
+
                 // Expect the producer to attach and grant it some credit, it should send
                 // a transfer which we will not send any response for which should cause the
                 // send operation to time out.
                 testPeer.ExpectSenderAttach();
                 testPeer.ExpectTransferButDoNotRespond(messageMatcher: Assert.NotNull);
-                
-                // When send operation timed out, released and settled disposition is issued by the provider
-                testPeer.ExpectDispositionThatIsReleasedAndSettled();
                 testPeer.ExpectClose();
-                
+
                 IMessageProducer producer = session.CreateProducer(queue);
-                
+
                 Assert.Catch<Exception>(() => producer.Send(message), "Send should time out.");
-                
+
                 connection.Close();
-                
+
                 testPeer.WaitForAllMatchersToComplete(1000);
             }
         }
@@ -627,22 +624,22 @@
             using (TestAmqpPeer testPeer = new TestAmqpPeer())
             {
                 IConnection connection = EstablishConnection(testPeer);
-                
+
                 testPeer.ExpectBegin();
                 testPeer.ExpectSenderAttach();
-                
+
                 ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
                 IQueue destination = session.GetQueue("myQueue");
                 IMessageProducer producer = session.CreateProducer(destination);
-                
+
                 testPeer.ExpectTransfer(Assert.IsNotNull);
-                
-                
+
+
                 producer.Send(session.CreateMessage());
-                
+
                 testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
                 producer.Close();
-                
+
                 testPeer.WaitForAllMatchersToComplete(1000);
             }
         }
@@ -657,24 +654,92 @@
 
                 testPeer.ExpectBegin();
                 testPeer.ExpectSenderAttach();
-                
+
                 ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
                 IQueue destination = session.GetQueue("myQueue");
                 IMessageProducer producer = session.CreateProducer(destination);
-                
+
                 testPeer.ExpectTransfer(Assert.IsNotNull);
-                
+
                 connection.Stop();
-                
+
                 producer.Send(session.CreateMessage());
-                
+
                 testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
                 testPeer.ExpectClose();
-                
+
                 producer.Close();
                 connection.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+  
+        [Test, Timeout(20_000)]
+        public void TestSendingMessagePersistentSetsBatchableFalse()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                IConnection connection = EstablishConnection(testPeer);
+                connection.Start();
+
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                IQueue destination = session.GetQueue("myQueue");
+                IMessageProducer producer = session.CreateProducer(destination);
+                testPeer.ExpectTransfer(messageMatcher: Assert.IsNotNull,
+                    stateMatcher: Assert.IsNull,
+                    settled: false,
+                    sendResponseDisposition: true,
+                    responseState: new Accepted(),
+                    responseSettled: true,
+                    batchable: false);
+
+                IMessage message = session.CreateMessage();
+                producer.Send(message: message, deliveryMode: MsgDeliveryMode.Persistent, MsgPriority.Normal, NMSConstants.defaultTimeToLive);
                 
                 testPeer.WaitForAllMatchersToComplete(1000);
+
+                testPeer.ExpectClose();
+                connection.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageNonPersistentSetsBatchableFalse()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                IConnection connection = EstablishConnection(testPeer);
+                connection.Start();
+
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                IQueue destination = session.GetQueue("myQueue");
+                IMessageProducer producer = session.CreateProducer(destination);
+                testPeer.ExpectTransfer(messageMatcher: Assert.IsNotNull,
+                    stateMatcher: Assert.IsNull,
+                    settled: true,
+                    sendResponseDisposition: true,
+                    responseState: new Accepted(),
+                    responseSettled: true,
+                    batchable: false);
+
+                IMessage message = session.CreateMessage();
+                producer.Send(message: message, deliveryMode: MsgDeliveryMode.NonPersistent, MsgPriority.Normal, NMSConstants.defaultTimeToLive);
+                
+                testPeer.WaitForAllMatchersToComplete(1000);
+
+                testPeer.ExpectClose();
+                connection.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
             }
         }
     }
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
index 0b67c9c..9a519f2 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
@@ -675,12 +675,14 @@
             bool sendResponseDisposition,
             DeliveryState responseState,
             bool responseSettled,
-            int dispositionDelay = 0
+            int dispositionDelay = 0,
+            bool batchable = false
         )
         {
             var transferMatcher = new FrameMatcher<Transfer>()
                 .WithAssertion(transfer => Assert.AreEqual(settled, transfer.Settled))
                 .WithAssertion(transfer => stateMatcher(transfer.State))
+                .WithAssertion(transfer => Assert.AreEqual(batchable, transfer.Batchable))
                 .WithAssertion(messageMatcher);
 
             if (sendResponseDisposition)