diff --git a/src/NMS.AMQP/NmsMessageProducer.cs b/src/NMS.AMQP/NmsMessageProducer.cs
index 6257177..3f7e3db 100644
--- a/src/NMS.AMQP/NmsMessageProducer.cs
+++ b/src/NMS.AMQP/NmsMessageProducer.cs
@@ -263,6 +263,11 @@
             }
             set
             {
+                if (!session.Connection.ConnectionInfo.DelayedDeliverySupported)
+                {
+                    throw new NotSupportedException("Delayed Delivery is not supported");
+                }
+                
                 CheckClosed();
                 deliveryDelay = value;
             }
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs b/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
index 1e86ced..6b40430 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
@@ -147,9 +147,9 @@
                     }
 
                     if (Array.Exists(capabilities,
-                        symbol => Equals(symbol, SymbolUtil.OPEN_CAPABILITY_DELAYED_DELIVERY)))
+                        symbol => Equals(symbol, SymbolUtil.OPEN_CAPABILITY_SHARED_SUBS)))
                     {
-                        Info.DelayedDeliverySupported = true;
+                        Info.SharedSubsSupported = true;
                     }
                 }
 
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpConsumer.cs b/src/NMS.AMQP/Provider/Amqp/AmqpConsumer.cs
index 7670d2f..d4ceeac 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpConsumer.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpConsumer.cs
@@ -130,7 +130,7 @@
                 bool supported = false;
                 if (remoteOfferedCapabilities != null)
                 {
-                    if (Array.Exists(remoteOfferedCapabilities, symbol => symbol == SymbolUtil.OPEN_CAPABILITY_SHARED_SUBS))
+                    if (Array.Exists(remoteOfferedCapabilities, symbol => SymbolUtil.OPEN_CAPABILITY_SHARED_SUBS.Equals(symbol)))
                     {
                         supported = true;
                     }
diff --git a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMessageFacade.cs b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMessageFacade.cs
index d5332f1..4799ada 100644
--- a/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMessageFacade.cs
+++ b/src/NMS.AMQP/Provider/Amqp/Message/AmqpNmsMessageFacade.cs
@@ -269,15 +269,16 @@
                     case ulong _:
                     case int _:
                     case uint _:
-                        return new DateTime(621355968000000000L + (long) deliveryTime * 10000L, DateTimeKind.Utc);
+                        return new DateTime(621355968000000000L + Convert.ToInt64(deliveryTime) * 10000L, DateTimeKind.Utc);
                     default:
                         return syntheticDeliveryTime;
                 }
             }
             set
             {
+                // Assumption that if it is being set through property, then it is with purpose of send out this value 
                 syntheticDeliveryTime = value;
-                RemoveMessageAnnotation(SymbolUtil.NMS_DELIVERY_TIME);
+                SetMessageAnnotation(SymbolUtil.NMS_DELIVERY_TIME, new DateTimeOffset(value).ToUnixTimeMilliseconds());
             }
         }
 
@@ -488,6 +489,7 @@
             target.connection = connection;
             target.consumerDestination = consumerDestination;
             target.syntheticExpiration = syntheticExpiration;
+            target.syntheticDeliveryTime = syntheticDeliveryTime;
             target.amqpTimeToLiveOverride = amqpTimeToLiveOverride;
             target.destination = destination;
             target.replyTo = replyTo;
@@ -508,7 +510,7 @@
             return MessageAnnotations != null && MessageAnnotations.Map.ContainsKey(annotationName);
         }
 
-        public void SetMessageAnnotation(Symbol symbolKeyName, string value)
+        public void SetMessageAnnotation(Symbol symbolKeyName, object value)
         {
             LazyCreateMessageAnnotations();
             MessageAnnotations.Map.Add(symbolKeyName, value);
diff --git a/src/NMS.AMQP/Util/SymbolUtil.cs b/src/NMS.AMQP/Util/SymbolUtil.cs
index bb64387..e3fe2c3 100644
--- a/src/NMS.AMQP/Util/SymbolUtil.cs
+++ b/src/NMS.AMQP/Util/SymbolUtil.cs
@@ -39,7 +39,7 @@
         public static readonly Symbol OPEN_CAPABILITY_SOLE_CONNECTION_FOR_CONTAINER = new Symbol("sole-connection-for-container");
         public static readonly Symbol OPEN_CAPABILITY_ANONYMOUS_RELAY = new Symbol("ANONYMOUS-RELAY");
         public static readonly Symbol OPEN_CAPABILITY_DELAYED_DELIVERY = new Symbol("DELAYED_DELIVERY");
-        public static readonly Symbol OPEN_CAPABILITY_SHARED_SUBS = new Symbol("SHARED_SUBS");
+        public static readonly Symbol OPEN_CAPABILITY_SHARED_SUBS = new Symbol("SHARED-SUBS");
 
         // Attach Frame 
         public readonly static Symbol ATTACH_EXPIRY_POLICY_LINK_DETACH = new Symbol("link-detach");
diff --git a/test/Apache-NMS-AMQP-Interop-Test/AmqpTestSupport.cs b/test/Apache-NMS-AMQP-Interop-Test/AmqpTestSupport.cs
index 2155d86..bf309c4 100644
--- a/test/Apache-NMS-AMQP-Interop-Test/AmqpTestSupport.cs
+++ b/test/Apache-NMS-AMQP-Interop-Test/AmqpTestSupport.cs
@@ -18,6 +18,7 @@
 using System;
 using Apache.NMS;
 using Apache.NMS.AMQP;
+using NMS.AMQP.Test.TestAmqp;
 using NUnit.Framework;
 
 namespace NMS.AMQP.Test
@@ -30,19 +31,32 @@
 
         protected string TestName => TestContext.CurrentContext.Test.Name;
 
+        static AmqpTestSupport()
+        {
+            Tracer.Trace = new NLogAdapter();
+        }
+        
         [TearDown]
         public void TearDown()
         {
             Connection?.Close();
         }
 
-        protected IConnection CreateAmqpConnection()
+        protected IConnection CreateAmqpConnectionStarted(string clientId = null)
+        {
+            var connection = CreateAmqpConnection(clientId);
+            connection.Start();
+            return connection;
+        }
+        
+        protected IConnection CreateAmqpConnection(string clientId = null)
         {
             string brokerUri = Environment.GetEnvironmentVariable("NMS_AMQP_TEST_URI") ?? "amqp://127.0.0.1:5672";
             string userName = Environment.GetEnvironmentVariable("NMS_AMQP_TEST_CU") ?? "admin";
             string password = Environment.GetEnvironmentVariable("NMS_AMQP_TEST_CPWD") ?? "admin";
 
             NmsConnectionFactory factory = new NmsConnectionFactory(brokerUri);
+            factory.ClientId = clientId;
             return factory.CreateConnection(userName, password);
         }
 
@@ -108,11 +122,7 @@
             IQueue queue = session.GetQueue(TestName);
             IMessageConsumer consumer = session.CreateConsumer(queue);
 
-            IMessage message;
-            do
-            {
-                message = consumer.Receive(timeout);
-            } while (message != null);
+            PurgeConsumer(consumer, timeout);
 
             amqpConnection.Close();
         }
@@ -125,13 +135,18 @@
             ITopic queue = session.GetTopic(TestName);
             IMessageConsumer consumer = session.CreateConsumer(queue);
 
+            PurgeConsumer(consumer, timeout);
+
+            amqpConnection.Close();
+        }
+
+        protected void PurgeConsumer(IMessageConsumer consumer, TimeSpan timeout)
+        {
             IMessage message;
             do
             {
                 message = consumer.Receive(timeout);
             } while (message != null);
-
-            amqpConnection.Close();
         }
     }
 }
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Interop-Test/Apache-NMS-AMQP-Interop-Test.csproj b/test/Apache-NMS-AMQP-Interop-Test/Apache-NMS-AMQP-Interop-Test.csproj
index f7ce890..e167532 100644
--- a/test/Apache-NMS-AMQP-Interop-Test/Apache-NMS-AMQP-Interop-Test.csproj
+++ b/test/Apache-NMS-AMQP-Interop-Test/Apache-NMS-AMQP-Interop-Test.csproj
@@ -16,11 +16,11 @@
 -->
 <Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <TargetFrameworks>net462;netcoreapp2.2</TargetFrameworks>
+    <TargetFrameworks>net462;netcoreapp3.1</TargetFrameworks>
     <TargetFramework Condition="'$(AppTargetFramework)' != ''">$(AppTargetFramework)</TargetFramework>
     <RootNamespace>NMS.AMQP.Test</RootNamespace>
     <AssemblyName>NMS.AMQP.Interop.Test</AssemblyName>
-    <LangVersion>7.3</LangVersion>
+    <LangVersion>8</LangVersion>
   </PropertyGroup>
   
   <ItemGroup>
@@ -33,5 +33,12 @@
   
   <ItemGroup>
     <ProjectReference Include="..\..\src\NMS.AMQP\Apache-NMS-AMQP.csproj" />
+    <ProjectReference Include="..\Apache-NMS-AMQP-Test\Apache-NMS-AMQP-Test.csproj" />
+  </ItemGroup>
+  
+  <ItemGroup>
+    <None Update="NLog.config">
+      <CopyToOutputDirectory>PreserveNewest</CopyToOutputDirectory>
+    </None>
   </ItemGroup>
 </Project>
diff --git a/test/Apache-NMS-AMQP-Interop-Test/NLog.config b/test/Apache-NMS-AMQP-Interop-Test/NLog.config
new file mode 100644
index 0000000..c0f1581
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Interop-Test/NLog.config
@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+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.
+  -->
+<nlog xmlns="http://www.nlog-project.org/schemas/NLog.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+
+    <targets>
+        <target name="logconsole" xsi:type="Console" layout="${time} | ${level} | ${callsite:includeNamespace=false:methodName=false} | ${message}" />
+    </targets>
+
+    <rules>
+        <logger name="*" minlevel="Debug" writeTo="logconsole" />
+    </rules>
+</nlog>
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Interop-Test/NmsMessageConsumerTest.cs b/test/Apache-NMS-AMQP-Interop-Test/NmsMessageConsumerTest.cs
index 4529783..9f174d4 100644
--- a/test/Apache-NMS-AMQP-Interop-Test/NmsMessageConsumerTest.cs
+++ b/test/Apache-NMS-AMQP-Interop-Test/NmsMessageConsumerTest.cs
@@ -16,6 +16,11 @@
  */
 
 using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
 using Apache.NMS;
 using NUnit.Framework;
 
@@ -84,6 +89,223 @@
             Assert.IsNull(messageConsumer.Receive(TimeSpan.FromSeconds(1)));
         }
 
+
+        [Test, Timeout(60_000)]
+        public void TestDurableSubscription()
+        {
+            Connection = CreateAmqpConnection();
+            Connection.Start();
+
+            int counter = 0;
+
+
+            using ISession sessionProducer = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+
+            string subscriptionName = "mySubscriptionName";
+            ITopic topicProducer = sessionProducer.GetTopic(TestName);
+            using IMessageProducer producer = sessionProducer.CreateProducer(topicProducer);
+
+            // First durable consumer, reads message but does not unsubscribe
+            using (var connectionSubscriber = CreateAmqpConnectionStarted("CLIENT1"))
+            using (ISession session = connectionSubscriber.CreateSession(AcknowledgementMode.AutoAcknowledge))
+            {
+                using ITopic topic = session.GetTopic(TestName);
+                using (IMessageConsumer messageConsumer = session.CreateDurableConsumer(topic, subscriptionName, null))
+                {
+                    // Purge topic
+                    PurgeConsumer(messageConsumer, TimeSpan.FromSeconds(0.5));
+
+                    ITextMessage producerMessage = sessionProducer.CreateTextMessage("text" + (counter++));
+                    producer.Send(producerMessage, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.Zero);
+
+                    var message = messageConsumer.Receive();
+                    Assert.AreEqual("text0", message.Body<string>());
+                }
+            }
+
+            // Write some more messages while subscription is closed
+            for (int t = 0; t < 3; t++)
+            {
+                ITextMessage producerMessage = sessionProducer.CreateTextMessage("text" + (counter++));
+                producer.Send(producerMessage, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.Zero);
+            }
+
+            // Second durable consumer, reads message that were send during no-subscription and unsubscribe
+            using (var connectionSubscriber = CreateAmqpConnectionStarted("CLIENT1"))
+            using (ISession session = connectionSubscriber.CreateSession(AcknowledgementMode.AutoAcknowledge))
+            {
+                using ITopic topic = session.GetTopic(TestName);
+                using (IMessageConsumer messageConsumer = session.CreateDurableConsumer(topic, subscriptionName, null))
+                {
+                    for (int t = 1; t <= 3; t++)
+                    {
+                        var message = messageConsumer.Receive();
+                        Assert.AreEqual("text" + t, message.Body<string>());
+                    }
+
+                    // Assert topic is empty after those msgs
+                    var msgAtTheEnd = messageConsumer.Receive(TimeSpan.FromSeconds(1));
+                    Assert.IsNull(msgAtTheEnd);
+
+                    Assert.Throws<IllegalStateException>(() => session.Unsubscribe(subscriptionName)); // Error unsubscribing while consumer is on
+                }
+
+                session.Unsubscribe(subscriptionName);
+            }
+
+
+            // Send some messages again to verify we will not get them when create durable subscription
+            for (int t = 0; t < 3; t++)
+            {
+                ITextMessage producerMessage = sessionProducer.CreateTextMessage("text" + (counter++));
+                producer.Send(producerMessage, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.Zero);
+            }
+
+            // Third durable subscriber, expect NOT to read messages during no-subscription period
+            using (var connectionSubscriber = CreateAmqpConnectionStarted("CLIENT1"))
+            using (ISession session = connectionSubscriber.CreateSession(AcknowledgementMode.AutoAcknowledge))
+            {
+                using ITopic topic = session.GetTopic(TestName);
+                using (IMessageConsumer messageConsumer = session.CreateDurableConsumer(topic, subscriptionName, null))
+                {
+                    // Assert topic is empty 
+                    var msgAtTheEnd = messageConsumer.Receive(TimeSpan.FromSeconds(1));
+                    Assert.IsNull(msgAtTheEnd);
+                }
+
+                // And unsubscribe again
+                session.Unsubscribe(subscriptionName);
+            }
+        }
+
+
+        [Test, Timeout(60_000)]
+        public void TestSharedSubscription()
+        {
+            IMessageConsumer GetConsumer(string subscriptionName, String clientId)
+            {
+                var connection = CreateAmqpConnectionStarted(clientId);
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                var topic = session.GetTopic(TestName);
+                var messageConsumer = session.CreateSharedConsumer(topic, subscriptionName);
+                return messageConsumer;
+            }
+
+            Connection = CreateAmqpConnection();
+            Connection.Start();
+
+            string subscriptionName = "mySubscriptionName";
+            
+
+            var receivedMessages = new List<int>();
+
+            var messageConsumer1 = GetConsumer(subscriptionName, null);
+            var messageConsumer2 = GetConsumer(subscriptionName, null);
+            messageConsumer1.Listener += (msg) =>
+            {
+                receivedMessages.Add(1);
+                msg.Acknowledge();
+            };
+            messageConsumer2.Listener += (msg) =>
+            {
+                receivedMessages.Add(2);
+                msg.Acknowledge();
+            };
+            
+            // Now send some messages
+            using ISession sessionProducer = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+            ITopic topicProducer = sessionProducer.GetTopic(TestName);
+            using IMessageProducer producer = sessionProducer.CreateProducer(topicProducer);
+            for (int t = 0; t < 10; t++)
+            {
+                ITextMessage producerMessage = sessionProducer.CreateTextMessage("text" + t);
+                producer.Send(producerMessage, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.Zero);
+            }
+
+            // Give it some time to process
+            Thread.Sleep(TimeSpan.FromSeconds(2));
+            
+            // Assert message was routed to multiple consumers
+            Assert.AreEqual(2, receivedMessages.Distinct().Count());
+            Assert.AreEqual(10, receivedMessages.Count);
+        }
+        
+        [Test, Timeout(60_000)]
+        public void TestSharedDurableSubscription()
+        {
+            (IMessageConsumer,ISession,IConnection) GetConsumer(string subscriptionName, String clientId)
+            {
+                var connection = CreateAmqpConnection(clientId);
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                var topic = session.GetTopic(TestName);
+                var messageConsumer = session.CreateSharedDurableConsumer(topic, subscriptionName);
+                return (messageConsumer, session, connection);
+            }
+
+            Connection = CreateAmqpConnection();
+            Connection.Start();
+
+            string subscriptionName = "mySubscriptionName";
+            int messageSendCount = 1099;   
+
+            var receivedMessages = new ConcurrentBag<int>();
+
+            
+            IConnection connectionConsumer1, connectionConsumer2;
+            IMessageConsumer messageConsumer1, messageConsumer2;
+            
+            (messageConsumer1, _, connectionConsumer1) = GetConsumer(subscriptionName, null);
+            (messageConsumer2, _, connectionConsumer2) = GetConsumer(subscriptionName, null);
+            connectionConsumer1.Start();
+            connectionConsumer2.Start();
+            
+            messageConsumer1.Close();
+            messageConsumer2.Close();
+            connectionConsumer1.Close();
+            connectionConsumer2.Close();
+            
+            // Now send some messages
+            using ISession sessionProducer = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+            ITopic topicProducer = sessionProducer.GetTopic(TestName);
+            using IMessageProducer producer = sessionProducer.CreateProducer(topicProducer);
+            for (int t = 0; t < messageSendCount; t++)
+            {
+                ITextMessage producerMessage = sessionProducer.CreateTextMessage("text" + t);
+                producer.Send(producerMessage, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.Zero);
+            }
+
+            // Create consumers again and expect messages to be delivered to them
+            ISession sessionConsumer1, sessionConsumer2;
+            (messageConsumer1, sessionConsumer1, connectionConsumer1) = GetConsumer(subscriptionName, null);
+            (messageConsumer2, sessionConsumer2, connectionConsumer2) = GetConsumer(subscriptionName, null);
+            messageConsumer1.Listener += (msg) =>
+            {
+                receivedMessages.Add(1);
+                msg.Acknowledge();
+            };
+            messageConsumer2.Listener += (msg) =>
+            {
+                receivedMessages.Add(2);
+                msg.Acknowledge();
+            };
+            Task.Run(() => connectionConsumer1.Start()); // parallel to give both consumers chance to start at the same time
+            Task.Run(() => connectionConsumer2.Start());
+            
+            // Give it some time to process
+            Thread.Sleep(TimeSpan.FromSeconds(5));
+            
+            // Assert message was routed to multiple consumers
+            Assert.AreEqual(2, receivedMessages.Distinct().Count());
+            Assert.AreEqual(messageSendCount, receivedMessages.Count);
+            
+            messageConsumer1.Close();
+            messageConsumer2.Close();
+            sessionConsumer1.Unsubscribe(subscriptionName);
+            sessionConsumer2.Unsubscribe(subscriptionName);
+            
+        }
+
+
         [Test, Timeout(60_000)]
         public void TestSelectNoLocal()
         {
diff --git a/test/Apache-NMS-AMQP-Interop-Test/NmsMessageProducerTest.cs b/test/Apache-NMS-AMQP-Interop-Test/NmsMessageProducerTest.cs
new file mode 100644
index 0000000..5bd182d
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Interop-Test/NmsMessageProducerTest.cs
@@ -0,0 +1,77 @@
+/*
+ * 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 System;
+using System.Threading;
+using System.Threading.Tasks;
+using Apache.NMS;
+using NUnit.Framework;
+
+namespace NMS.AMQP.Test
+{
+    [TestFixture]
+    public class NmsMessageProducerTest : AmqpTestSupport
+    {
+        [Test, Timeout(60_000)]
+        public void TestDeliveryDelay()
+        {
+            PurgeQueue(TimeSpan.FromMilliseconds(500));
+
+            var deliveryDelay = TimeSpan.FromSeconds(7);
+            
+            Connection = CreateAmqpConnection();
+            Connection.Start();
+
+            ISession session = Connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+            IQueue queue = session.GetQueue(TestName);
+            IMessageProducer producer = session.CreateProducer(queue);
+            producer.DeliveryDelay = deliveryDelay;
+
+            DateTime? receivingTime = null;
+            IMessageConsumer consumer = session.CreateConsumer(queue);
+            var receivingTask = Task.Run(() =>
+            {
+                while (true)
+                {
+                    var message = consumer.Receive(TimeSpan.FromMilliseconds(100));
+                    if (message != null && message.Body<string>() == "Hello")
+                    {
+                        receivingTime = DateTime.Now;
+                        return;
+                    }
+                }
+            });
+            
+            
+            DateTime sendTime = DateTime.Now;
+            ITextMessage message = session.CreateTextMessage("Hello");
+            producer.Send(message, MsgDeliveryMode.Persistent, MsgPriority.Normal, TimeSpan.Zero);
+
+            // Wait that delivery delay
+            Thread.Sleep(deliveryDelay);
+
+            receivingTask.Wait(TimeSpan.FromSeconds(20)); // make sure its done
+
+            var measuredDelay = (receivingTime.Value - sendTime);
+            
+            Assert.Greater(measuredDelay.TotalMilliseconds, deliveryDelay.TotalMilliseconds* 0.5);
+            Assert.Less(measuredDelay.TotalMilliseconds, deliveryDelay.TotalMilliseconds*1.5);
+        }
+
+      
+    }
+}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj b/test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj
index 68c6937..2de2f86 100644
--- a/test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj
+++ b/test/Apache-NMS-AMQP-Test/Apache-NMS-AMQP-Test.csproj
@@ -16,7 +16,7 @@
 -->
 <Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <TargetFrameworks>net462;netcoreapp2.2</TargetFrameworks>
+    <TargetFrameworks>net462;netcoreapp3.1</TargetFrameworks>
     <TargetFramework Condition="'$(AppTargetFramework)' != ''">$(AppTargetFramework)</TargetFramework>
     <RootNamespace>NMS.AMQP.Test</RootNamespace>
     <AssemblyName>NMS.AMQP.Test</AssemblyName>
diff --git a/test/Apache-NMS-AMQP-Test/Integration/IntegrationTestFixture.cs b/test/Apache-NMS-AMQP-Test/Integration/IntegrationTestFixture.cs
index 9a13133..fac9499 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/IntegrationTestFixture.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/IntegrationTestFixture.cs
@@ -50,6 +50,26 @@
             return connection;
         }
 
+        protected INMSContext EstablishNMSContext(TestAmqpPeer testPeer, string optionsString = null, Symbol[] serverCapabilities = null, Fields serverProperties = null, bool setClientId = true, AcknowledgementMode acknowledgementMode = AcknowledgementMode.AutoAcknowledge)
+        {
+            testPeer.ExpectSaslPlain("guest", "guest");
+            testPeer.ExpectOpen(serverCapabilities: serverCapabilities, serverProperties: serverProperties);
+
+            // Each connection creates a session for managing temporary destinations etc.
+            testPeer.ExpectBegin();
+
+            var remoteUri = BuildUri(testPeer, optionsString);
+            var connectionFactory = new NmsConnectionFactory(remoteUri);
+            var context = connectionFactory.CreateContext("guest", "guest", acknowledgementMode);
+            if (setClientId)
+            {
+                // Set a clientId to provoke the actual AMQP connection process to occur.
+                context.ClientId = "ClientName";
+            }
+            
+            return context;
+        }
+        
         private static string BuildUri(TestAmqpPeer testPeer, string optionsString)
         {
             string baseUri = "amqp://127.0.0.1:" + testPeer.ServerPort.ToString();
diff --git a/test/Apache-NMS-AMQP-Test/Integration/MessageDeliveryTimeTest.cs b/test/Apache-NMS-AMQP-Test/Integration/MessageDeliveryTimeTest.cs
new file mode 100644
index 0000000..dbe936a
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Test/Integration/MessageDeliveryTimeTest.cs
@@ -0,0 +1,181 @@
+﻿/*
+ * 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 System;
+using Amqp.Framing;
+using Amqp.Types;
+using Apache.NMS;
+using Apache.NMS.AMQP.Util;
+using NMS.AMQP.Test.TestAmqp;
+using NUnit.Framework;
+
+namespace NMS.AMQP.Test.Integration
+{
+    [TestFixture]
+    public class MessageDeliveryTimeTest : IntegrationTestFixture
+    {
+        [Test, Timeout(20000)]
+        public void TestReceiveMessageWithoutDeliveryTimeSet()
+        {
+            DoReceiveMessageDeliveryTime(null, null);
+        }
+
+        [Test, Timeout(20000)]
+        public void TestDeliveryTimeIsDateTime()
+        {
+            DateTime deliveryTime = DateTimeOffset.FromUnixTimeMilliseconds(CurrentTimeInMillis() + 12345).DateTime.ToUniversalTime();
+            DoReceiveMessageDeliveryTime(deliveryTime, deliveryTime);
+        }
+
+        [Test, Timeout(20000)]
+        public void TestDeliveryTimeIsULong()
+        {
+            ulong deliveryTime = (ulong) (CurrentTimeInMillis() + 12345);
+            DoReceiveMessageDeliveryTime(deliveryTime, DateTimeOffset.FromUnixTimeMilliseconds((long) deliveryTime).DateTime);
+        }
+
+        [Test, Timeout(20000)]
+        public void TestDeliveryTimeIsLong()
+        {
+            long deliveryTime = (CurrentTimeInMillis() + 12345);
+            DoReceiveMessageDeliveryTime(deliveryTime, DateTimeOffset.FromUnixTimeMilliseconds(deliveryTime).DateTime);
+        }
+
+        [Test, Timeout(20000)]
+        public void TestDeliveryTimeIsInt()
+        {
+            int deliveryTime = (int) (CurrentTimeInMillis() + 12345);
+            DoReceiveMessageDeliveryTime(deliveryTime, DateTimeOffset.FromUnixTimeMilliseconds(deliveryTime).DateTime);
+        }
+
+        [Test, Timeout(20000)]
+        public void TestDeliveryTimeIsUInt()
+        {
+            uint deliveryTime = (uint) (CurrentTimeInMillis() + 12345);
+            DoReceiveMessageDeliveryTime(deliveryTime, DateTimeOffset.FromUnixTimeMilliseconds(deliveryTime).DateTime);
+        }
+
+        private long CurrentTimeInMillis()
+        {
+            return new DateTimeOffset(DateTime.UtcNow).ToUnixTimeMilliseconds();
+        }
+
+        private void DoReceiveMessageDeliveryTime(object setDeliveryTimeAnnotation, DateTime? expectedDeliveryTime)
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var connection = EstablishConnection(testPeer, "amqp.traceFrames=true");
+                connection.Start();
+                testPeer.ExpectBegin();
+                var session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                var queue = session.GetQueue("myQueue");
+
+                var message = CreateMessageWithNullContent();
+                if (setDeliveryTimeAnnotation != null)
+                {
+                    message.MessageAnnotations = message.MessageAnnotations ?? new MessageAnnotations();
+                    message.MessageAnnotations[SymbolUtil.NMS_DELIVERY_TIME] = setDeliveryTimeAnnotation;
+                }
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message);
+                testPeer.ExpectDisposition(true, (deliveryState) => { });
+
+                DateTime startingTimeFrom = DateTime.UtcNow;
+                var messageConsumer = session.CreateConsumer(queue);
+                var receivedMessage = messageConsumer.Receive(TimeSpan.FromMilliseconds(3000));
+                DateTime receivingTime = DateTime.UtcNow;
+
+                testPeer.WaitForAllMatchersToComplete(3000);
+
+                testPeer.ExpectClose();
+                connection.Close();
+
+                testPeer.WaitForAllMatchersToComplete(3000);
+
+                Assert.IsNotNull(receivedMessage);
+                if (expectedDeliveryTime != null)
+                {
+                    Assert.AreEqual(receivedMessage.NMSDeliveryTime, expectedDeliveryTime.Value);
+                }
+                else
+                {
+                    Assert.LessOrEqual(receivedMessage.NMSDeliveryTime, receivingTime);
+                    Assert.GreaterOrEqual(receivedMessage.NMSDeliveryTime, startingTimeFrom);
+                }
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestDeliveryDelayNotSupportedThrowsException()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                IConnection connection = base.EstablishConnection(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                IQueue queue = session.GetQueue("myQueue");
+                IMessageProducer producer = session.CreateProducer(queue);
+                Assert.Throws<NotSupportedException>(() => producer.DeliveryDelay = TimeSpan.FromMinutes(17));
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestDeliveryDelayHasItsReflectionInAmqpAnnotations()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                // Determine current time
+                TimeSpan deliveryDelay = TimeSpan.FromMinutes(17);
+                long currentUnixEpochTime = new DateTimeOffset(DateTime.UtcNow + deliveryDelay).ToUnixTimeMilliseconds();
+                long currentUnixEpochTime2 = new DateTimeOffset(DateTime.UtcNow + deliveryDelay + deliveryDelay).ToUnixTimeMilliseconds();
+
+                IConnection connection = base.EstablishConnection(testPeer,
+                    serverCapabilities: new Symbol[] {SymbolUtil.OPEN_CAPABILITY_DELAYED_DELIVERY, SymbolUtil.OPEN_CAPABILITY_SOLE_CONNECTION_FOR_CONTAINER});
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                IQueue queue = session.GetQueue("myQueue");
+                IMessageProducer producer = session.CreateProducer(queue);
+                producer.DeliveryDelay = deliveryDelay;
+
+                // Create and transfer a new message
+                testPeer.ExpectTransfer(message =>
+                {
+                    Assert.GreaterOrEqual((long) message.MessageAnnotations[SymbolUtil.NMS_DELIVERY_TIME], currentUnixEpochTime);
+                    Assert.Less((long) message.MessageAnnotations[SymbolUtil.NMS_DELIVERY_TIME], currentUnixEpochTime2);
+
+                    Assert.IsTrue(message.Header.Durable);
+                });
+                testPeer.ExpectClose();
+
+                ITextMessage textMessage = session.CreateTextMessage();
+
+                producer.Send(textMessage);
+                Assert.AreEqual(MsgDeliveryMode.Persistent, textMessage.NMSDeliveryMode);
+
+                connection.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/Integration/NMSConsumerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/NMSConsumerIntegrationTest.cs
new file mode 100644
index 0000000..c1bb33b
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Test/Integration/NMSConsumerIntegrationTest.cs
@@ -0,0 +1,975 @@
+﻿/*
+ * 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 System;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Amqp.Framing;
+using Apache.NMS;
+using Apache.NMS.AMQP.Message;
+using Apache.NMS.AMQP.Util;
+using NMS.AMQP.Test.TestAmqp;
+using NUnit.Framework;
+
+namespace NMS.AMQP.Test.Integration
+{
+    // Adapted from ConsumerIntegrationTest to use NMSContext
+    [TestFixture]
+    public class NMSConsumerIntegrationTest : IntegrationTestFixture
+    {
+        [Test, Timeout(20_000)]
+        public void TestCloseConsumer()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlow();
+
+                IQueue queue = context.GetQueue("myQueue");
+                var consumer = context.CreateConsumer(queue);
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+                consumer.Close();
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        // TODO No connection Listener in context
+        // [Test, Timeout(20_000)]
+        // public void TestRemotelyCloseConsumer()
+        // {
+        //     Mock<INmsConnectionListener> mockConnectionListener = new Mock<INmsConnectionListener>();
+        //     string errorMessage = "buba";
+        //
+        //     using (TestAmqpPeer testPeer = new TestAmqpPeer())
+        //     {
+        //         ManualResetEvent consumerClosed = new ManualResetEvent(false);
+        //         ManualResetEvent exceptionFired = new ManualResetEvent(false);
+        //
+        //         mockConnectionListener
+        //             .Setup(listener => listener.OnConsumerClosed(It.IsAny<IMessageConsumer>(), It.IsAny<Exception>()))
+        //             .Callback(() => consumerClosed.Set());
+        //
+        //         var context = (NmsContext) EstablishNMSContext(testPeer, "amqp.traceFrames=true");
+        //         context.ConnectionInterruptedListener += () => { consumerClosed.Set(); };// AddConnectionListener(mockConnectionListener.Object);}
+        //         // context.list ConnectionInterruptedListener += () => { consumerClosed.Set(); };// AddConnectionListener(mockConnectionListener.Object);}
+        //         context.ExceptionListener += exception => { exceptionFired.Set(); };
+        //
+        //         testPeer.ExpectBegin();
+        //         // ISession session = context.CreateSession(AcknowledgementMode.AutoAcknowledge);
+        //
+        //         // Create a consumer, then remotely end it afterwards.
+        //         testPeer.ExpectReceiverAttach();
+        //         testPeer.ExpectLinkFlow();
+        //         testPeer.RemotelyDetachLastOpenedLinkOnLastOpenedSession(expectDetachResponse: true, closed: true, errorType: AmqpError.RESOURCE_DELETED, errorMessage: errorMessage, delayBeforeSend: 400);
+        //
+        //         IQueue queue = context.GetQueue("myQueue");
+        //         var consumer = context.CreateConsumer(queue);
+        //         
+        //         
+        //         // Verify the consumer gets marked closed
+        //         testPeer.WaitForAllMatchersToComplete(1000);
+        //
+        //         Assert.True(consumerClosed.WaitOne(2000), "Consumer closed callback didn't trigger");
+        //         Assert.False(exceptionFired.WaitOne(20), "Exception listener shouldn't fire with no MessageListener");
+        //
+        //         // Try closing it explicitly, should effectively no-op in client.
+        //         // The test peer will throw during close if it sends anything.
+        //         consumer.Close();
+        //     }
+        // }
+
+        // [Test, Timeout(20_000)]
+        // public void TestRemotelyCloseConsumerWithMessageListenerFiresExceptionListener()
+        // {
+        //     Mock<INmsConnectionListener> mockConnectionListener = new Mock<INmsConnectionListener>();
+        //     string errorMessage = "buba";
+        //
+        //     using (TestAmqpPeer testPeer = new TestAmqpPeer())
+        //     {
+        //         ManualResetEvent consumerClosed = new ManualResetEvent(false);
+        //         ManualResetEvent exceptionFired = new ManualResetEvent(false);
+        //
+        //         mockConnectionListener
+        //             .Setup(listener => listener.OnConsumerClosed(It.IsAny<IMessageConsumer>(), It.IsAny<Exception>()))
+        //             .Callback(() => consumerClosed.Set());
+        //
+        //         NmsConnection connection = (NmsConnection) EstablishConnection(testPeer);
+        //         connection.AddConnectionListener(mockConnectionListener.Object);
+        //         connection.ExceptionListener += exception => { exceptionFired.Set(); };
+        //
+        //         testPeer.ExpectBegin();
+        //         ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+        //
+        //         // Create a consumer, then remotely end it afterwards.
+        //         testPeer.ExpectReceiverAttach();
+        //         testPeer.ExpectLinkFlow();
+        //         testPeer.RemotelyDetachLastOpenedLinkOnLastOpenedSession(expectDetachResponse: true, closed: true, errorType: AmqpError.RESOURCE_DELETED, errorMessage: errorMessage, 10);
+        //
+        //         IQueue queue = session.GetQueue("myQueue");
+        //         IMessageConsumer consumer = session.CreateConsumer(queue);
+        //
+        //         consumer.Listener += message => { };
+        //
+        //         // Verify the consumer gets marked closed
+        //         testPeer.WaitForAllMatchersToComplete(1000);
+        //
+        //         Assert.True(consumerClosed.WaitOne(2000), "Consumer closed callback didn't trigger");
+        //         Assert.True(exceptionFired.WaitOne(2000), "Exception listener should have fired with a MessageListener");
+        //
+        //         // Try closing it explicitly, should effectively no-op in client.
+        //         // The test peer will throw during close if it sends anything.
+        //         consumer.Close();
+        //     }
+        // }
+
+        [Test, Timeout(20_000)]
+        public void TestReceiveMessageWithReceiveZeroTimeout()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue queue = context.GetQueue("myQueue");
+
+                testPeer.ExpectReceiverAttach();
+
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: new Amqp.Message() { BodySection = new AmqpValue() { Value = null } }, count: 1);
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+
+                var consumer = context.CreateConsumer(queue);
+                IMessage message = consumer.Receive();
+                Assert.NotNull(message, "A message should have been received");
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(10000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestExceptionInOnMessageReleasesInAutoAckMode()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue queue = context.GetQueue("myQueue");
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: new Amqp.Message() { BodySection = new AmqpValue() { Value = null } }, count: 1);
+                testPeer.ExpectDispositionThatIsReleasedAndSettled();
+
+                var consumer = context.CreateConsumer(queue);
+                consumer.Listener += message => throw new Exception();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(10000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCloseDurableTopicSubscriberDetachesWithCloseFalse()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+                
+                string topicName = "myTopic";
+                string subscriptionName = "mySubscription";
+                ITopic topic = context.GetTopic(topicName);
+
+                testPeer.ExpectDurableSubscriberAttach(topicName, subscriptionName);
+                testPeer.ExpectLinkFlow();
+
+                var durableConsumer = context.CreateDurableConsumer(topic, subscriptionName, null, false);
+
+                testPeer.ExpectDetach(expectClosed: false, sendResponse: true, replyClosed: false);
+                durableConsumer.Close();
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestConsumerReceiveThrowsIfConnectionLost()
+        {
+            DoTestConsumerReceiveThrowsIfConnectionLost(false);
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestConsumerTimedReceiveThrowsIfConnectionLost()
+        {
+            DoTestConsumerReceiveThrowsIfConnectionLost(true);
+        }
+
+        private void DoTestConsumerReceiveThrowsIfConnectionLost(bool useTimeout)
+        {
+            ManualResetEvent consumerReady = new ManualResetEvent(false);
+
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+
+                testPeer.ExpectBegin();
+
+                IQueue queue = context.GetQueue("queue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlow();
+                testPeer.RunAfterLastHandler(() => { consumerReady.WaitOne(2000); });
+                testPeer.DropAfterLastMatcher(delay: 10);
+
+                var consumer = context.CreateConsumer(queue);
+                consumerReady.Set();
+
+                try
+                {
+                    if (useTimeout)
+                    {
+                        consumer.Receive(TimeSpan.FromMilliseconds(10_0000));
+                    }
+                    else
+                    {
+                        consumer.Receive();
+                    }
+
+
+                    Assert.Fail("An exception should have been thrown");
+                }
+                catch (NMSException)
+                {
+                    // Expected
+                }
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        //  TODO No connection Listener in context
+        // [Test, Timeout(20_000)]
+        // public void TestConsumerReceiveNoWaitThrowsIfConnectionLost()
+        // {
+        //     ManualResetEvent disconnected = new ManualResetEvent(false);
+        //
+        //     using (TestAmqpPeer testPeer = new TestAmqpPeer())
+        //     {
+        //         NmsContext context = (NmsContext) EstablishNMSContext(testPeer);
+        //         Mock<INmsConnectionListener> connectionListener = new Mock<INmsConnectionListener>();
+        //
+        //         connectionListener
+        //             .Setup(listener => listener.OnConnectionFailure(It.IsAny<NMSException>()))
+        //             .Callback(() => { disconnected.Set(); });
+        //
+        //         context.AddConnectionListener(connectionListener.Object);
+        //
+        //         context.Start();
+        //
+        //         testPeer.ExpectBegin();
+        //
+        //         IQueue queue = context.GetQueue("queue");
+        //
+        //         testPeer.ExpectReceiverAttach();
+        //         testPeer.ExpectLinkFlow();
+        //         testPeer.RemotelyCloseConnection(expectCloseResponse: true, errorCondition: ConnectionError.CONNECTION_FORCED, errorMessage: "buba");
+        //
+        //         var consumer = context.CreateConsumer(queue);
+        //
+        //         Assert.True(disconnected.WaitOne(), "Connection should be disconnected");
+        //
+        //         try
+        //         {
+        //             consumer.ReceiveNoWait();
+        //             Assert.Fail("An exception should have been thrown");
+        //         }
+        //         catch (NMSException)
+        //         {
+        //             // Expected
+        //         }
+        //     }
+        // }
+
+        [Test, Timeout(20_000)]
+        public void TestSetMessageListenerAfterStartAndSend()
+        {
+            int messageCount = 4;
+            CountdownEvent latch = new CountdownEvent(messageCount);
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), messageCount);
+
+                var consumer = context.CreateConsumer(destination);
+
+                for (int i = 0; i < messageCount; i++)
+                {
+                    testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+                }
+
+                consumer.Listener += message => latch.Signal();
+
+                Assert.True(latch.Wait(4000), "Messages not received within given timeout. Count remaining: " + latch.CurrentCount);
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+
+                consumer.Close();
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        // TODO Connection is started anyway when creating consumer
+        // [Test, Timeout(20_000)]
+        // public void TestNoReceivedMessagesWhenConnectionNotStarted()
+        // {
+        //     using (TestAmqpPeer testPeer = new TestAmqpPeer())
+        //     {
+        //         var context = EstablishNMSContext(testPeer);
+        //         
+        //         testPeer.ExpectBegin();
+        //
+        //         IQueue destination = context.GetQueue("myQueue");
+        //
+        //         testPeer.ExpectReceiverAttach();
+        //         testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), count: 3);
+        //
+        //         // cREATING CONSUMER STARTS CONNECTION
+        //         var consumer = context.CreateConsumer(destination);
+        //
+        //         // Wait for a message to arrive then try and receive it, which should not happen
+        //         // since the connection is not started.
+        //         Assert.Null(consumer.Receive(TimeSpan.FromMilliseconds(100)));
+        //
+        //         testPeer.ExpectEnd();
+        //         testPeer.ExpectClose();
+        //         context.Close();
+        //
+        //         testPeer.WaitForAllMatchersToComplete(2000);
+        //     }
+        // }
+
+        // TODO Connection is started anyway when creating consumer
+        // [Test, Timeout(20_000)]
+        // public void TestNoReceivedNoWaitMessagesWhenConnectionNotStarted()
+        // {
+        //     using (TestAmqpPeer testPeer = new TestAmqpPeer())
+        //     {
+        //         var context = EstablishNMSContext(testPeer);
+        //
+        //         testPeer.ExpectBegin();
+        //
+        //         IQueue destination = context.GetQueue("myQueue");
+        //
+        //         testPeer.ExpectReceiverAttach();
+        //         testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), count: 3);
+        //
+        //         var consumer = context.CreateConsumer(destination);
+        //
+        //         // Wait for a message to arrive then try and receive it, which should not happen
+        //         // since the connection is not started.
+        //         Assert.Null(consumer.ReceiveNoWait());
+        //
+        //         testPeer.ExpectEnd();
+        //         testPeer.ExpectClose();
+        //         context.Close();
+        //
+        //         testPeer.WaitForAllMatchersToComplete(2000);
+        //     }
+        // }
+
+        [Test, Timeout(20_000)]
+        public void TestSyncReceiveFailsWhenListenerSet()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlow();
+
+                var consumer = context.CreateConsumer(destination);
+
+                consumer.Listener += message => { };
+
+                Assert.Catch<NMSException>(() => consumer.Receive(), "Should have thrown an exception.");
+                Assert.Catch<NMSException>(() => consumer.Receive(TimeSpan.FromMilliseconds(1000)), "Should have thrown an exception.");
+                Assert.Catch<NMSException>(() => consumer.ReceiveNoWait(), "Should have thrown an exception.");
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateProducerInOnMessage()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                IQueue outbound = context.GetQueue("ForwardDest");
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), 1);
+
+                testPeer.ExpectSenderAttach();
+                testPeer.ExpectTransfer(messageMatcher: Assert.NotNull);
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+
+                var consumer = context.CreateConsumer(destination);
+
+                consumer.Listener += message =>
+                {
+                    var producer = context.CreateProducer();
+                    producer.Send(outbound, message);
+                    producer.Close();
+                };
+
+                testPeer.WaitForAllMatchersToComplete(10_000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestMessageListenerCallsConnectionCloseThrowsIllegalStateException()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), 1);
+
+                var consumer = context.CreateConsumer(destination);
+
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+
+                ManualResetEvent latch = new ManualResetEvent(false);
+                Exception exception = null;
+                consumer.Listener += message =>
+                {
+                    try
+                    {
+                        context.Close();
+                    }
+                    catch (Exception e)
+                    {
+                        exception = e;
+                    }
+
+                    latch.Set();
+                };
+
+                Assert.True(latch.WaitOne(4000), "Messages not received within given timeout.");
+                Assert.IsNotNull(exception);
+                Assert.IsInstanceOf<IllegalStateException>(exception);
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+                consumer.Close();
+
+                testPeer.ExpectEnd();
+                // testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestMessageListenerCallsConnectionStopThrowsIllegalStateException()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), 1);
+
+                var consumer = context.CreateConsumer(destination);
+
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+
+                ManualResetEvent latch = new ManualResetEvent(false);
+                Exception exception = null;
+                consumer.Listener += message =>
+                {
+                    try
+                    {
+                        context.Stop();
+                    }
+                    catch (Exception e)
+                    {
+                        exception = e;
+                    }
+
+                    latch.Set();
+                };
+
+                Assert.True(latch.WaitOne(3000), "Messages not received within given timeout.");
+                Assert.IsNotNull(exception);
+                Assert.IsInstanceOf<IllegalStateException>(exception);
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+                consumer.Close();
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestMessageListenerCallsSessionCloseThrowsIllegalStateException()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), 1);
+
+                var consumer = context.CreateConsumer(destination);
+
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+
+                ManualResetEvent latch = new ManualResetEvent(false);
+                Exception exception = null;
+                consumer.Listener += message =>
+                {
+                    try
+                    {
+                        context.Close();
+                    }
+                    catch (Exception e)
+                    {
+                        exception = e;
+                    }
+
+                    latch.Set();
+                };
+
+                Assert.True(latch.WaitOne(3000), "Messages not received within given timeout.");
+                Assert.IsNotNull(exception);
+                Assert.IsInstanceOf<IllegalStateException>(exception);
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+                consumer.Close();
+
+                testPeer.ExpectEnd();
+                // testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        // TODO: To be fixed
+        [Test, Timeout(20_000), Ignore("Ignore")]
+        public void TestMessageListenerClosesItsConsumer()
+        {
+            var latch = new ManualResetEvent(false);
+            var exceptionListenerFired = new ManualResetEvent(false);
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                context.ExceptionListener += _ => exceptionListenerFired.Set();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), 1);
+
+                var consumer = context.CreateConsumer(destination);
+
+                testPeer.ExpectLinkFlow(drain: true, sendDrainFlowResponse: true, creditMatcher: credit => Assert.AreEqual(99, credit)); // Not sure if expected credit is right
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+
+                Exception exception = null;
+                consumer.Listener += message =>
+                {
+                    try
+                    {
+                        consumer.Close();
+                        latch.Set();
+                    }
+                    catch (Exception e)
+                    {
+                        exception = e;
+                    }
+                };
+
+                Assert.True(latch.WaitOne(TimeSpan.FromMilliseconds(1000)), "Process not completed within given timeout");
+                Assert.IsNull(exception, "No error expected during close");
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                Assert.False(exceptionListenerFired.WaitOne(20), "Exception listener shouldn't have fired");
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestRecoverOrderingWithAsyncConsumer()
+        {
+            ManualResetEvent latch = new ManualResetEvent(false);
+            Exception asyncError = null;
+
+            int recoverCount = 5;
+            int messageCount = 8;
+            int testPayloadLength = 255;
+            string payload = Encoding.UTF8.GetString(new byte[testPayloadLength]);
+
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer, acknowledgementMode:AcknowledgementMode.ClientAcknowledge);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(
+                    message: new Amqp.Message() { BodySection = new AmqpValue() { Value = payload } },
+                    count: messageCount,
+                    drain: false,
+                    nextIncomingId: 1,
+                    addMessageNumberProperty: true,
+                    sendDrainFlowResponse: false,
+                    sendSettled: false,
+                    creditMatcher: credit => Assert.Greater(credit, messageCount)
+                );
+
+                var consumer = context.CreateConsumer(destination);
+                
+                bool complete = false;
+                int messageSeen = 0;
+                int expectedIndex = 0;
+                consumer.Listener += message =>
+                {
+                    if (complete)
+                    {
+                        return;
+                    }
+
+                    try
+                    {
+                        int actualIndex = message.Properties.GetInt(TestAmqpPeer.MESSAGE_NUMBER);
+                        Assert.AreEqual(expectedIndex, actualIndex, "Received Message Out Of Order");
+
+                        // don't ack the message until we receive it X times
+                        if (messageSeen < recoverCount)
+                        {
+                            context.Recover();
+                            messageSeen++;
+                        }
+                        else
+                        {
+                            messageSeen = 0;
+                            expectedIndex++;
+
+                            // Have the peer expect the accept the disposition (1-based, hence pre-incremented).
+                            testPeer.ExpectDisposition(settled: true,
+                                stateMatcher: state => Assert.AreEqual(state.Descriptor.Code, MessageSupport.ACCEPTED_INSTANCE.Descriptor.Code
+                                ));
+
+                            message.Acknowledge();
+
+                            if (expectedIndex == messageCount)
+                            {
+                                complete = true;
+                                latch.Set();
+                            }
+                        }
+                    }
+                    catch (Exception e)
+                    {
+                        complete = true;
+                        asyncError = e;
+                        latch.Set();
+                    }
+                };
+
+                Assert.True(latch.WaitOne(TimeSpan.FromSeconds(15)), "Messages not received within given timeout.");
+                Assert.IsNull(asyncError, "Unexpected exception");
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestConsumerCloseWaitsForAsyncDeliveryToComplete()
+        {
+            ManualResetEvent latch = new ManualResetEvent(false);
+
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), count: 1);
+
+                var consumer = context.CreateConsumer(destination);
+
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+
+                consumer.Listener += _ =>
+                {
+                    latch.Set();
+                    Task.Delay(TimeSpan.FromMilliseconds(100)).GetAwaiter().GetResult();
+                };
+
+                Assert.True(latch.WaitOne(TimeSpan.FromMilliseconds(3000)), "Messages not received within given timeout.");
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+                consumer.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSessionCloseWaitsForAsyncDeliveryToComplete()
+        {
+            ManualResetEvent latch = new ManualResetEvent(false);
+
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), count: 1);
+
+                var consumer = context.CreateConsumer(destination);
+
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+
+                consumer.Listener += _ =>
+                {
+                    latch.Set();
+                    Task.Delay(TimeSpan.FromMilliseconds(100)).GetAwaiter().GetResult();
+                };
+
+                Assert.True(latch.WaitOne(TimeSpan.FromMilliseconds(3000)), "Messages not received within given timeout.");
+
+                
+                testPeer.WaitForAllMatchersToComplete(2000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestConnectionCloseWaitsForAsyncDeliveryToComplete()
+        {
+            ManualResetEvent latch = new ManualResetEvent(false);
+
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                IQueue destination = context.GetQueue("myQueue");
+                context.Start();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: CreateMessageWithContent(), count: 1);
+
+                var consumer = context.CreateConsumer(destination);
+
+                testPeer.ExpectDispositionThatIsAcceptedAndSettled();
+
+                consumer.Listener += _ =>
+                {
+                    latch.Set();
+                    Task.Delay(TimeSpan.FromMilliseconds(100)).GetAwaiter().GetResult();
+                };
+
+                Assert.True(latch.WaitOne(TimeSpan.FromMilliseconds(3000)), "Messages not received within given timeout.");
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestRecoveredMessageShouldNotBeMutated()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer, acknowledgementMode:AcknowledgementMode.ClientAcknowledge);
+                context.Start();
+
+                testPeer.ExpectBegin();
+                IQueue destination = context.GetQueue("myQueue");
+                string originalPayload = "testMessage";
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlowRespondWithTransfer(message: new Amqp.Message { BodySection = new AmqpValue() { Value = originalPayload } }, count: 1);
+
+                var consumer = context.CreateConsumer(destination);
+                NmsTextMessage message = consumer.Receive() as NmsTextMessage;
+                Assert.NotNull(message);
+                message.IsReadOnlyBody = false;
+                message.Text = message.Text + "Received";
+                context.Recover();
+
+                ITextMessage recoveredMessage = consumer.Receive() as ITextMessage;
+                Assert.IsNotNull(recoveredMessage);
+                Assert.AreNotEqual(message.Text, recoveredMessage.Text);
+                Assert.AreEqual(originalPayload, recoveredMessage.Text);
+                Assert.AreNotSame(message, recoveredMessage);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/Integration/NMSContextIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/NMSContextIntegrationTest.cs
new file mode 100644
index 0000000..a1b43e8
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Test/Integration/NMSContextIntegrationTest.cs
@@ -0,0 +1,278 @@
+﻿/*
+ * 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 Apache.NMS;
+using NMS.AMQP.Test.TestAmqp;
+using NUnit.Framework;
+
+namespace NMS.AMQP.Test.Integration
+{
+    // Adapted from SessionIntegrationTest to use NMSContext
+    [TestFixture]
+    public class NMSContextIntegrationTest : IntegrationTestFixture
+    {
+        [Test, Timeout(20_000)]
+        public void TestClose()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                INMSContext context = EstablishNMSContext(testPeer);
+
+                testPeer.ExpectClose();
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateProducer()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+
+                testPeer.ExpectSenderAttach();
+
+                var producer = context.CreateProducer();
+
+                testPeer.ExpectDetach(true, true, true);
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                producer.Close();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateConsumer()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlow();
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                var consumer = context.CreateConsumer(context.GetQueue("myQueue"));
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateConsumerWithEmptySelector()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlow();
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlow();
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                IQueue queue = context.GetQueue("myQueue");
+                context.CreateConsumer(queue, "");
+                context.CreateConsumer(queue, "", noLocal: false);
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateConsumerWithNullSelector()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlow();
+                testPeer.ExpectReceiverAttach();
+                testPeer.ExpectLinkFlow();
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                IQueue queue = context.GetQueue("myQueue");
+                context.CreateConsumer(queue, null);
+                context.CreateConsumer(queue, null, noLocal: false);
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateDurableConsumer()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                string topicName = "myTopic";
+                ITopic topic = context.GetTopic(topicName);
+                string subscriptionName = "mySubscription";
+
+                testPeer.ExpectDurableSubscriberAttach(topicName, subscriptionName);
+                testPeer.ExpectLinkFlow();
+
+                var durableConsumer = context.CreateDurableConsumer(topic, subscriptionName, null, false);
+                Assert.NotNull(durableConsumer, "MessageConsumer object was null");
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+
+        [Test, Timeout(20_000)]
+        public void TestCreateTemporaryQueue()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+
+                testPeer.ExpectBegin();
+
+                string dynamicAddress = "myTempQueueAddress";
+                testPeer.ExpectTempQueueCreationAttach(dynamicAddress);
+
+                ITemporaryQueue temporaryQueue = context.CreateTemporaryQueue();
+                Assert.NotNull(temporaryQueue, "TemporaryQueue object was null");
+                Assert.NotNull(temporaryQueue.QueueName, "TemporaryQueue queue name was null");
+                Assert.AreEqual(dynamicAddress, temporaryQueue.QueueName, "TemporaryQueue name not as expected");
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateTemporaryTopic()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+
+                testPeer.ExpectBegin();
+
+                string dynamicAddress = "myTempTopicAddress";
+                testPeer.ExpectTempTopicCreationAttach(dynamicAddress);
+
+                ITemporaryTopic temporaryTopic = context.CreateTemporaryTopic();
+                Assert.NotNull(temporaryTopic, "TemporaryTopic object was null");
+                Assert.NotNull(temporaryTopic.TopicName, "TemporaryTopic name was null");
+                Assert.AreEqual(dynamicAddress, temporaryTopic.TopicName, "TemporaryTopic name not as expected");
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateSharedConsumer()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                string topicName = "myTopic";
+                ITopic topic = context.GetTopic(topicName);
+                string subscriptionName = "mySubscription";
+
+                testPeer.ExpectSharedSubscriberAttach(topicName, subscriptionName);
+                testPeer.ExpectLinkFlow();
+
+                var durableConsumer = context.CreateSharedConsumer(topic, subscriptionName, null); //, false);
+                Assert.NotNull(durableConsumer, "MessageConsumer object was null");
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(20000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestCreateSharedDurableConsumer()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+
+                string topicName = "myTopic";
+                ITopic topic = context.GetTopic(topicName);
+                string subscriptionName = "mySubscription";
+
+                testPeer.ExpectSharedDurableSubscriberAttach(topicName, subscriptionName);
+                testPeer.ExpectLinkFlow();
+
+                var durableConsumer = context.CreateSharedDurableConsumer(topic, subscriptionName, null); //, false);
+                Assert.NotNull(durableConsumer, "MessageConsumer object was null");
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/Integration/NMSProducerIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/NMSProducerIntegrationTest.cs
new file mode 100644
index 0000000..22563a0
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Test/Integration/NMSProducerIntegrationTest.cs
@@ -0,0 +1,722 @@
+/*
+ * 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 System;
+using Amqp.Framing;
+using Amqp.Types;
+using Apache.NMS;
+using Apache.NMS.AMQP.Util;
+using NMS.AMQP.Test.TestAmqp;
+using NUnit.Framework;
+
+namespace NMS.AMQP.Test.Integration
+{
+    // Adapted from ProducerIntegrationTest to use NMSContext
+    [TestFixture]
+    public class NMSProducerIntegrationTest : IntegrationTestFixture
+    {
+        private const long TICKS_PER_MILLISECOND = 10000;
+
+        [Test, Timeout(20_000)]
+        public void TestCloseSender()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = base.EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue queue = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                producer.Close();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSentTextMessageCanBeModified()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = base.EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue queue = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                // Create and transfer a new message
+                String text = "myMessage";
+                testPeer.ExpectTransfer(x => Assert.AreEqual(text, (x.BodySection as AmqpValue).Value));
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                ITextMessage message = context.CreateTextMessage(text);
+                producer.Send(queue, message);
+
+                Assert.AreEqual(text, message.Text);
+                message.Text = text + text;
+                Assert.AreEqual(text + text, message.Text);
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestDefaultDeliveryModeProducesDurableMessages()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = base.EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue queue = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                // Create and transfer a new message
+                testPeer.ExpectTransfer(message => Assert.IsTrue(message.Header.Durable));
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                ITextMessage textMessage = context.CreateTextMessage();
+
+                producer.Send(queue, textMessage);
+                Assert.AreEqual(MsgDeliveryMode.Persistent, textMessage.NMSDeliveryMode);
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestProducerOverridesMessageDeliveryMode()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = base.EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue queue = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                // Create and transfer a new message, explicitly setting the deliveryMode on the
+                // message (which applications shouldn't) to NON_PERSISTENT and sending it to check
+                // that the producer ignores this value and sends the message as PERSISTENT(/durable)
+                testPeer.ExpectTransfer(message => Assert.IsTrue(message.Header.Durable));
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                ITextMessage textMessage = context.CreateTextMessage();
+                textMessage.NMSDeliveryMode = MsgDeliveryMode.NonPersistent;
+                Assert.AreEqual(MsgDeliveryMode.NonPersistent, textMessage.NMSDeliveryMode);
+
+                producer.Send(queue, textMessage);
+
+                Assert.AreEqual(MsgDeliveryMode.Persistent, textMessage.NMSDeliveryMode);
+
+                context.Close();
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageNonPersistentProducerSetDurableFalse()
+        {
+            DoSendingMessageNonPersistentTestImpl(true);
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageNonPersistentProducerOmitsHeader()
+        {
+            DoSendingMessageNonPersistentTestImpl(false);
+        }
+
+        private void DoSendingMessageNonPersistentTestImpl(bool setPriority)
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                //Add capability to indicate support for ANONYMOUS-RELAY
+                Symbol[] serverCapabilities = {SymbolUtil.OPEN_CAPABILITY_ANONYMOUS_RELAY};
+                var context = EstablishNMSContext(testPeer, serverCapabilities: serverCapabilities);
+                testPeer.ExpectBegin();
+
+                string queueName = "myQueue";
+                Action<object> targetMatcher = t =>
+                {
+                    var target = t as Target;
+                    Assert.IsNotNull(target);
+                };
+
+
+                testPeer.ExpectSenderAttach(targetMatcher: targetMatcher, sourceMatcher: Assert.NotNull, senderSettled: false);
+
+                IQueue queue = context.GetQueue(queueName);
+                INMSProducer producer = context.CreateProducer();
+
+                byte priority = 5;
+                String text = "myMessage";
+                testPeer.ExpectTransfer(messageMatcher: message =>
+                    {
+                        if (setPriority)
+                        {
+                            Assert.IsFalse(message.Header.Durable);
+                            Assert.AreEqual(priority, message.Header.Priority);
+                        }
+
+                        Assert.AreEqual(text, (message.BodySection as AmqpValue).Value);
+                    }, stateMatcher: Assert.IsNull,
+                    settled: false,
+                    sendResponseDisposition: true,
+                    responseState: new Accepted(),
+                    responseSettled: true);
+
+                ITextMessage textMessage = context.CreateTextMessage(text);
+
+                producer.DeliveryMode = MsgDeliveryMode.NonPersistent;
+                if (setPriority)
+                    producer.Priority = (MsgPriority) priority;
+
+                producer.Send(queue, textMessage);
+
+                Assert.AreEqual(MsgDeliveryMode.NonPersistent, textMessage.NMSDeliveryMode, "Should have NonPersistent delivery mode set");
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageSetsNMSDestination()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                string text = "myMessage";
+                ITextMessage message = context.CreateTextMessage(text);
+
+                testPeer.ExpectTransfer(m => Assert.AreEqual(text, (m.BodySection as AmqpValue).Value));
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                Assert.IsNull(message.NMSDestination, "Should not yet have a NMSDestination");
+
+                producer.Send(destination, message);
+
+                Assert.AreEqual(destination, message.NMSDestination, "Should have had NMSDestination set");
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageSetsNMSTimestamp()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                // Create matcher to expect the absolute-expiry-time field of the properties section to
+                // be set to a value greater than 'now'+ttl, within a delta.
+
+                DateTime creationLower = DateTime.UtcNow;
+                DateTime creationUpper = creationLower + TimeSpan.FromMilliseconds(3000);
+
+                var text = "myMessage";
+                testPeer.ExpectTransfer(m =>
+                {
+                    Assert.IsTrue(m.Header.Durable);
+                    Assert.That(m.Properties.CreationTime.Ticks, Is.GreaterThanOrEqualTo(creationLower.Ticks).Within(TICKS_PER_MILLISECOND));
+                    Assert.That(m.Properties.CreationTime.Ticks, Is.LessThanOrEqualTo(creationUpper.Ticks).Within(TICKS_PER_MILLISECOND));
+                    Assert.AreEqual(text, (m.BodySection as AmqpValue).Value);
+                });
+
+                ITextMessage message = context.CreateTextMessage(text);
+                producer.Send(destination, message);
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageSetsNMSExpirationRelatedAbsoluteExpiryAndTtlFields()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                uint ttl = 100_000;
+                DateTime currentTime = DateTime.UtcNow;
+                DateTime expirationLower = currentTime + TimeSpan.FromMilliseconds(ttl);
+                DateTime expirationUpper = currentTime + TimeSpan.FromMilliseconds(ttl) + TimeSpan.FromMilliseconds(5000);
+
+                // Create matcher to expect the absolute-expiry-time field of the properties section to
+                // be set to a value greater than 'now'+ttl, within a delta.
+                string text = "myMessage";
+                testPeer.ExpectTransfer(m =>
+                {
+                    Assert.IsTrue(m.Header.Durable);
+                    Assert.AreEqual(ttl, m.Header.Ttl);
+                    Assert.That(m.Properties.AbsoluteExpiryTime.Ticks, Is.GreaterThanOrEqualTo(expirationLower.Ticks).Within(TICKS_PER_MILLISECOND));
+                    Assert.That(m.Properties.AbsoluteExpiryTime.Ticks, Is.LessThanOrEqualTo(expirationUpper.Ticks).Within(TICKS_PER_MILLISECOND));
+                    Assert.AreEqual(text, (m.BodySection as AmqpValue).Value);
+                });
+
+                ITextMessage message = context.CreateTextMessage(text);
+                producer.TimeToLive = TimeSpan.FromMilliseconds(ttl);
+                producer.Priority = NMSConstants.defaultPriority;
+                producer.DeliveryMode = NMSConstants.defaultDeliveryMode;
+                producer.Send(destination, message);
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestMessagesAreProducedWithProperDefaultPriorityWhenNoPrioritySpecified()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                byte priority = 4;
+
+                testPeer.ExpectTransfer(m => Assert.AreEqual(priority, m.Header.Priority));
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                ITextMessage message = context.CreateTextMessage();
+                Assert.AreEqual(MsgPriority.BelowNormal, message.NMSPriority);
+
+                producer.Send(destination, message);
+
+                Assert.AreEqual((MsgPriority) priority, message.NMSPriority);
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestNonDefaultPriorityProducesMessagesWithPriorityFieldAndSetsNMSPriority()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                byte priority = 9;
+
+                testPeer.ExpectTransfer(m => Assert.AreEqual(priority, m.Header.Priority));
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                ITextMessage message = context.CreateTextMessage();
+                Assert.AreEqual(MsgPriority.BelowNormal, message.NMSPriority);
+
+                producer.DeliveryMode = MsgDeliveryMode.Persistent;
+                producer.Priority = (MsgPriority) priority;
+                producer.TimeToLive = NMSConstants.defaultTimeToLive;
+                producer.Send(destination, message);
+
+                Assert.AreEqual((MsgPriority) priority, message.NMSPriority);
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageSetsNMSMessageId()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                string text = "myMessage";
+                string actualMessageId = null;
+                testPeer.ExpectTransfer(m =>
+                {
+                    Assert.IsTrue(m.Header.Durable);
+                    Assert.IsNotEmpty(m.Properties.MessageId);
+                    actualMessageId = m.Properties.MessageId;
+                });
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                ITextMessage message = context.CreateTextMessage(text);
+                Assert.IsNull(message.NMSMessageId, "NMSMessageId should not yet be set");
+
+                producer.Send(destination, message);
+
+                Assert.IsNotNull(message.NMSMessageId);
+                Assert.IsNotEmpty(message.NMSMessageId, "NMSMessageId should be set");
+                Assert.IsTrue(message.NMSMessageId.StartsWith("ID:"), "MMS 'ID:' prefix not found");
+
+                context.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");
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageWithDisableMessageIdHint()
+        {
+            DoSendingMessageWithDisableMessageIdHintTestImpl(false);
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageWithDisableMessageIdHintAndExistingMessageId()
+        {
+            DoSendingMessageWithDisableMessageIdHintTestImpl(true);
+        }
+
+        private void DoSendingMessageWithDisableMessageIdHintTestImpl(bool existingId)
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                string text = "myMessage";
+                testPeer.ExpectTransfer(m =>
+                {
+                    Assert.IsTrue(m.Header.Durable);
+                    Assert.IsNull(m.Properties.MessageId); // Check there is no message-id value;
+                    Assert.AreEqual(text, (m.BodySection as AmqpValue).Value);
+                });
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                ITextMessage message = context.CreateTextMessage(text);
+
+                Assert.IsNull(message.NMSMessageId, "NMSMessageId should not yet be set");
+
+                if (existingId)
+                {
+                    string existingMessageId = "ID:this-should-be-overwritten-in-send";
+                    message.NMSMessageId = existingMessageId;
+                    Assert.AreEqual(existingMessageId, message.NMSMessageId, "NMSMessageId should now be se");
+                }
+
+                producer.DisableMessageID = true;
+
+                producer.Send(destination, message);
+
+                Assert.IsNull(message.NMSMessageId, "NMSMessageID should be null");
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(2000);
+            }
+        }
+
+        // TODO No connection listener in nms context
+        // [Test, Timeout(20_000)]
+        // public void TestRemotelyCloseProducer()
+        // {
+        //     string breadCrumb = "ErrorMessageBreadCrumb";
+        //
+        //     ManualResetEvent producerClosed = new ManualResetEvent(false);
+        //     Mock<INmsConnectionListener> mockConnectionListener = new Mock<INmsConnectionListener>();
+        //     mockConnectionListener
+        //         .Setup(listener => listener.OnProducerClosed(It.IsAny<NmsMessageProducer>(), It.IsAny<Exception>()))
+        //         .Callback(() => { producerClosed.Set(); });
+        //
+        //     using (TestAmqpPeer testPeer = new TestAmqpPeer())
+        //     {
+        //         NmsContext context = (NmsContext) EstablishNMSContext(testPeer);
+        //         context.AddConnectionListener(mockConnectionListener.Object);
+        //
+        //         testPeer.ExpectBegin();
+        //         ISession session = context.CreateSession(AcknowledgementMode.AutoAcknowledge);
+        //
+        //         // Create a producer, then remotely end it afterwards.
+        //         testPeer.ExpectSenderAttach();
+        //         testPeer.RemotelyDetachLastOpenedLinkOnLastOpenedSession(expectDetachResponse: true, closed: true, errorType: AmqpError.RESOURCE_DELETED, breadCrumb, delayBeforeSend: 10);
+        //
+        //         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();
+        //     }
+        // }
+
+        [Test, Timeout(20_000)]
+        public void TestSendWhenLinkCreditIsZeroAndTimeout()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer, optionsString: "nms.sendTimeout=500");
+                testPeer.ExpectBegin();
+
+                IQueue queue = context.GetQueue("myQueue");
+
+                ITextMessage message = context.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.ExpectEnd();
+                testPeer.ExpectClose();
+
+                var producer = context.CreateProducer();
+
+                Assert.Catch<Exception>(() => producer.Send(queue, message), "Send should time out.");
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendTimesOutWhenNoDispositionArrives()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer, optionsString: "nms.sendTimeout=500");
+                testPeer.ExpectBegin();
+
+                IQueue queue = context.GetQueue("myQueue");
+
+                ITextMessage message = context.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);
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                var producer = context.CreateProducer();
+
+                Assert.Catch<Exception>(() => producer.Send(queue, message), "Send should time out.");
+
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendWorksWhenConnectionNotStarted()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                testPeer.ExpectTransfer(Assert.IsNotNull);
+
+                producer.Send(destination, context.CreateMessage());
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+                producer.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendWorksAfterConnectionStopped()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+
+                testPeer.ExpectTransfer(Assert.IsNotNull);
+
+                context.Stop();
+
+                producer.Send(destination, context.CreateMessage());
+
+                testPeer.ExpectDetach(expectClosed: true, sendResponse: true, replyClosed: true);
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+
+                producer.Close();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessagePersistentSetsBatchableFalse()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+                testPeer.ExpectTransfer(messageMatcher: Assert.IsNotNull,
+                    stateMatcher: Assert.IsNull,
+                    settled: false,
+                    sendResponseDisposition: true,
+                    responseState: new Accepted(),
+                    responseSettled: true,
+                    batchable: false);
+
+                IMessage message = context.CreateMessage();
+                producer.DeliveryMode = MsgDeliveryMode.Persistent;
+                producer.Priority = MsgPriority.Normal;
+                producer.TimeToLive = NMSConstants.defaultTimeToLive;
+                producer.Send(destination, message);
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+        [Test, Timeout(20_000)]
+        public void TestSendingMessageNonPersistentSetsBatchableFalse()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                var context = EstablishNMSContext(testPeer);
+                context.Start();
+
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                IQueue destination = context.GetQueue("myQueue");
+                var producer = context.CreateProducer();
+                testPeer.ExpectTransfer(messageMatcher: Assert.IsNotNull,
+                    stateMatcher: Assert.IsNull,
+                    settled: false,
+                    sendResponseDisposition: true,
+                    responseState: new Accepted(),
+                    responseSettled: true,
+                    batchable: false);
+
+                IMessage message = context.CreateMessage();
+                producer.DeliveryMode = MsgDeliveryMode.NonPersistent;
+                producer.Priority = MsgPriority.Normal;
+                producer.TimeToLive = NMSConstants.defaultTimeToLive;
+                producer.Send(destination, message);
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+
+                testPeer.ExpectEnd();
+                testPeer.ExpectClose();
+                context.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationAsyncTest.cs b/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationAsyncTest.cs
new file mode 100644
index 0000000..88d05df
--- /dev/null
+++ b/test/Apache-NMS-AMQP-Test/Integration/ProducerIntegrationAsyncTest.cs
@@ -0,0 +1,135 @@
+/*
+ * 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 System;
+using System.Threading.Tasks;
+using Amqp.Framing;
+using Apache.NMS;
+using NMS.AMQP.Test.TestAmqp;
+using NUnit.Framework;
+
+namespace NMS.AMQP.Test.Integration
+{
+    [TestFixture]
+    public class ProducerIntegrationAsyncTest : IntegrationTestFixture
+    {
+        [Test, Timeout(20_000)]
+        public void TestSentAsyncIsAsynchronous()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                IConnection connection = base.EstablishConnection(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                IQueue queue = session.GetQueue("myQueue");
+                IMessageProducer producer = session.CreateProducer(queue);
+
+                // Create and transfer a new message
+                String text = "myMessage";
+                testPeer.ExpectTransfer(messageMatcher: m => Assert.AreEqual(text, (m.BodySection as AmqpValue).Value),
+                    settled: false,
+                    sendResponseDisposition: true,
+                    responseState: new Accepted(),
+                    responseSettled: true,
+                    stateMatcher: Assert.IsNull,
+                    dispositionDelay: 10); // 10ms should be enough
+                testPeer.ExpectClose();
+
+                ITextMessage message = session.CreateTextMessage(text);
+                var sendTask = producer.SendAsync(message);
+                // Instantly check if its not completed yet, we want async, so it should not be completed right after 
+                Assert.AreEqual(false, sendTask.IsCompleted);
+                
+                // And now wait for task to complete
+                sendTask.Wait(20_000);
+
+                connection.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+        
+        [Test, Timeout(20_000)]
+        public async Task TestSentAsync()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                IConnection connection = base.EstablishConnection(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                IQueue queue = session.GetQueue("myQueue");
+                IMessageProducer producer = session.CreateProducer(queue);
+
+                // Create and transfer a new message
+                String text = "myMessage";
+                testPeer.ExpectTransfer(messageMatcher: m => Assert.AreEqual(text, (m.BodySection as AmqpValue).Value),
+                    settled: false,
+                    sendResponseDisposition: true,
+                    responseState: new Accepted(),
+                    responseSettled: true,
+                    stateMatcher: Assert.IsNull,
+                    dispositionDelay: 10); // 10ms should be enough
+                testPeer.ExpectClose();
+
+                ITextMessage message = session.CreateTextMessage(text);
+                await producer.SendAsync(message);
+              
+                connection.Close();
+
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+       
+
+        [Test, Timeout(20_000)]
+        public async Task TestProducerWorkWithAsyncAwait()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                IConnection connection = base.EstablishConnection(testPeer);
+                testPeer.ExpectBegin();
+                testPeer.ExpectSenderAttach();
+
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                IQueue queue = session.GetQueue("myQueue");
+                IMessageProducer producer = session.CreateProducer(queue);
+
+                // Create and transfer a new message, explicitly setting the deliveryMode on the
+                // message (which applications shouldn't) to NON_PERSISTENT and sending it to check
+                // that the producer ignores this value and sends the message as PERSISTENT(/durable)
+                testPeer.ExpectTransfer(message => Assert.IsTrue(message.Header.Durable));
+                testPeer.ExpectClose();
+
+                ITextMessage textMessage = session.CreateTextMessage();
+                textMessage.NMSDeliveryMode = MsgDeliveryMode.NonPersistent;
+                Assert.AreEqual(MsgDeliveryMode.NonPersistent, textMessage.NMSDeliveryMode);
+
+                await producer.SendAsync(textMessage);
+
+                Assert.AreEqual(MsgDeliveryMode.Persistent, textMessage.NMSDeliveryMode);
+
+                connection.Close();
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
+
+    }
+}
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/Integration/SessionIntegrationTest.cs b/test/Apache-NMS-AMQP-Test/Integration/SessionIntegrationTest.cs
index 4fbdc29..da336e8 100644
--- a/test/Apache-NMS-AMQP-Test/Integration/SessionIntegrationTest.cs
+++ b/test/Apache-NMS-AMQP-Test/Integration/SessionIntegrationTest.cs
@@ -227,5 +227,62 @@
                 testPeer.WaitForAllMatchersToComplete(1000);
             }
         }
+       
+        [Test, Timeout(20_000)]
+        public void TestCreateSharedConsumer()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                IConnection connection = EstablishConnection(testPeer);
+                connection.Start();
+
+                testPeer.ExpectBegin();
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                
+                string topicName = "myTopic";
+                ITopic topic = session.GetTopic(topicName);
+                string subscriptionName = "mySubscription";
+
+                testPeer.ExpectSharedSubscriberAttach(topicName, subscriptionName);
+                testPeer.ExpectLinkFlow();
+                
+                IMessageConsumer durableConsumer = session.CreateSharedConsumer(topic, subscriptionName, null);//, false);
+                // IMessageConsumer durableConsumer = session.CreateDurableConsumer(topic, subscriptionName, null, false);
+                Assert.NotNull(durableConsumer, "MessageConsumer object was null");
+                
+                testPeer.ExpectClose();
+                connection.Close();
+                
+                testPeer.WaitForAllMatchersToComplete(20000);
+            }
+        }
+        
+        [Test, Timeout(20_000)]
+        public void TestCreateSharedDurableConsumer()
+        {
+            using (TestAmqpPeer testPeer = new TestAmqpPeer())
+            {
+                IConnection connection = EstablishConnection(testPeer);
+                connection.Start();
+
+                testPeer.ExpectBegin();
+                ISession session = connection.CreateSession(AcknowledgementMode.AutoAcknowledge);
+                
+                string topicName = "myTopic";
+                ITopic topic = session.GetTopic(topicName);
+                string subscriptionName = "mySubscription";
+
+                testPeer.ExpectSharedDurableSubscriberAttach(topicName, subscriptionName);
+                testPeer.ExpectLinkFlow();
+                
+                IMessageConsumer durableConsumer = session.CreateSharedDurableConsumer(topic, subscriptionName, null); //, false);
+                Assert.NotNull(durableConsumer, "MessageConsumer object was null");
+                
+                testPeer.ExpectClose();
+                connection.Close();
+                
+                testPeer.WaitForAllMatchersToComplete(1000);
+            }
+        }
     }
 }
\ No newline at end of file
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/NLogAdapter.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/NLogAdapter.cs
index e8c3f6f..b5688d6 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/NLogAdapter.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/NLogAdapter.cs
@@ -20,7 +20,7 @@
 
 namespace NMS.AMQP.Test.TestAmqp
 {
-    class NLogAdapter : ITrace
+    public class NLogAdapter : ITrace
     {
         private static readonly Logger Logger = LogManager.GetCurrentClassLogger();
 
diff --git a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
index 164f141..eccf3e9 100644
--- a/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
+++ b/test/Apache-NMS-AMQP-Test/TestAmqp/TestAmqpPeer.cs
@@ -555,6 +555,7 @@
                         Handle = context.Command.Handle,
                         LinkName = context.Command.LinkName,
                         Target = context.Command.Target,
+                        OfferedCapabilities = new []{ SymbolUtil.OPEN_CAPABILITY_SHARED_SUBS }
                     };
 
                     if (refuseLink)
@@ -605,7 +606,49 @@
 
             ExpectReceiverAttach(linkNameMatcher: linkNameMatcher, sourceMatcher: sourceMatcher, targetMatcher: targetMatcher, settled: false, errorType: null, errorMessage: null);
         }
+        
+        public void ExpectSharedDurableSubscriberAttach(string topicName, string subscriptionName)
+        {
+            Action<string> linkNameMatcher = linkName => Assert.AreEqual(subscriptionName, linkName);
 
+            Action<object> sourceMatcher = o =>
+            {
+                Assert.IsInstanceOf<Source>(o);
+                var source = (Source) o;
+                Assert.AreEqual(topicName, source.Address);
+                Assert.IsFalse(source.Dynamic);
+                Assert.AreEqual(TerminusDurability.UNSETTLED_STATE, (TerminusDurability) source.Durable);
+                Assert.AreEqual(TerminusExpiryPolicy.NEVER, source.ExpiryPolicy);
+                CollectionAssert.Contains(source.Capabilities, SymbolUtil.ATTACH_CAPABILITIES_TOPIC);
+                CollectionAssert.Contains(source.Capabilities, SymbolUtil.SHARED);
+            };
+
+            Action<Target> targetMatcher = Assert.IsNotNull;
+
+            ExpectReceiverAttach(linkNameMatcher: linkNameMatcher, sourceMatcher: sourceMatcher, targetMatcher: targetMatcher, settled: false, errorType: null, errorMessage: null);
+        }
+
+        public void ExpectSharedSubscriberAttach(string topicName, string subscriptionName)
+        {
+            Action<string> linkNameMatcher = linkName => Assert.AreEqual(subscriptionName+"|volatile1", linkName);
+
+            Action<object> sourceMatcher = o =>
+            {
+                Assert.IsInstanceOf<Source>(o);
+                var source = (Source) o;
+                Assert.AreEqual(topicName, source.Address);
+                Assert.IsFalse(source.Dynamic);
+                // Assert.AreEqual(TerminusDurability.UNSETTLED_STATE, (TerminusDurability) source.Durable);
+                // Assert.AreEqual(TerminusExpiryPolicy.NEVER, source.ExpiryPolicy);
+                CollectionAssert.Contains(source.Capabilities, SymbolUtil.ATTACH_CAPABILITIES_TOPIC);
+                CollectionAssert.Contains(source.Capabilities, SymbolUtil.SHARED);
+            };
+
+            Action<Target> targetMatcher = Assert.IsNotNull;
+
+            ExpectReceiverAttach(linkNameMatcher: linkNameMatcher, sourceMatcher: sourceMatcher, targetMatcher: targetMatcher, settled: false, errorType: null, errorMessage: null);
+        }
+        
         public void ExpectDetach(bool expectClosed, bool sendResponse, bool replyClosed, Symbol errorType = null, String errorMessage = null)
         {
             var detachMatcher = new FrameMatcher<Detach>()