diff --git a/src/NMS.AMQP/NmsDurableTopicSubscriber.cs b/src/NMS.AMQP/NmsDurableTopicSubscriber.cs
new file mode 100644
index 0000000..f39d7ea
--- /dev/null
+++ b/src/NMS.AMQP/NmsDurableTopicSubscriber.cs
@@ -0,0 +1,17 @@
+using Apache.NMS.AMQP.Util;
+
+namespace Apache.NMS.AMQP
+{
+    public class NmsDurableTopicSubscriber : NmsMessageConsumer
+    {
+        public NmsDurableTopicSubscriber(Id consumerId, NmsSession session, IDestination destination, string selector, bool noLocal) : base(consumerId, session, destination, selector, noLocal)
+        {
+        }
+
+        public NmsDurableTopicSubscriber(Id consumerId, NmsSession session, IDestination destination, string name, string selector, bool noLocal) : base(consumerId, session, destination, name, selector, noLocal)
+        {
+        }
+
+        protected override bool IsDurableSubscription => true;
+    }
+}
\ No newline at end of file
diff --git a/src/NMS.AMQP/NmsMessageConsumer.cs b/src/NMS.AMQP/NmsMessageConsumer.cs
index 13a4cf0..34afd0c 100644
--- a/src/NMS.AMQP/NmsMessageConsumer.cs
+++ b/src/NMS.AMQP/NmsMessageConsumer.cs
@@ -40,7 +40,7 @@
         {
         }
 
-        public NmsMessageConsumer(Id consumerId, NmsSession session, IDestination destination, string name, string selector, bool noLocal)
+        protected NmsMessageConsumer(Id consumerId, NmsSession session, IDestination destination, string name, string selector, bool noLocal)
         {
             Session = session;
             acknowledgementMode = session.AcknowledgementMode;
@@ -56,9 +56,14 @@
                 Selector = selector,
                 NoLocal = noLocal,
                 SubscriptionName = name,
-                LocalMessageExpiry = Session.Connection.ConnectionInfo.LocalMessageExpiry
+                LocalMessageExpiry = Session.Connection.ConnectionInfo.LocalMessageExpiry,
+                IsDurable = IsDurableSubscription
             };
             deliveryTask = new MessageDeliveryTask(this);
+            
+            Session.Connection.CreateResource(Info).ConfigureAwait(false).GetAwaiter().GetResult();
+            
+            Session.Add(this);
 
             if (Session.IsStarted)
                 Start();
@@ -68,6 +73,8 @@
         public ConsumerInfo Info { get; }
         public IDestination Destination => Info.Destination;
 
+        protected virtual bool IsDurableSubscription => false;
+
         public void Dispose()
         {
             try
@@ -186,10 +193,9 @@
 
         private event MessageListener Listener;
 
-        public async Task Init()
+        public Task Init()
         {
-            await Session.Connection.CreateResource(Info);
-            await Session.Connection.StartResource(Info);
+            return Session.Connection.StartResource(Info);
         }
 
         public void OnInboundMessage(InboundMessageDispatch envelope)
@@ -417,10 +423,10 @@
         {
             if (closed.CompareAndSet(false, true))
             {
-                messageQueue.Dispose();
                 failureCause = exception;
                 Session.Remove(this);
                 started.Set(false);
+                messageQueue.Dispose();
             }
         }
 
@@ -500,9 +506,9 @@
         {
             try
             {
-                Session.Connection.StartResource(Info);
+                Session.Connection.StartResource(Info).ConfigureAwait(false).GetAwaiter().GetResult();
             }
-            catch (NMSException ex)
+            catch (NMSException)
             {
                 Session.Remove(this);
                 throw;
diff --git a/src/NMS.AMQP/NmsMessageProducer.cs b/src/NMS.AMQP/NmsMessageProducer.cs
index feca455..c5da8e2 100644
--- a/src/NMS.AMQP/NmsMessageProducer.cs
+++ b/src/NMS.AMQP/NmsMessageProducer.cs
@@ -43,6 +43,10 @@
             {
                 Destination = destination
             };
+            
+            session.Connection.CreateResource(Info).ConfigureAwait(false).GetAwaiter().GetResult();
+            
+            session.Add(this);
         }
 
         public ProducerInfo Info { get; }
diff --git a/src/NMS.AMQP/NmsSession.cs b/src/NMS.AMQP/NmsSession.cs
index 5843606..c9f008d 100644
--- a/src/NMS.AMQP/NmsSession.cs
+++ b/src/NMS.AMQP/NmsSession.cs
@@ -20,6 +20,7 @@
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
+using Amqp;
 using Apache.NMS.AMQP.Message;
 using Apache.NMS.AMQP.Meta;
 using Apache.NMS.AMQP.Provider;
@@ -99,10 +100,7 @@
 
         public IMessageProducer CreateProducer(IDestination destination)
         {
-            NmsMessageProducer messageProducer = new NmsMessageProducer(producerIdGenerator.GenerateId(), this, destination);
-            messageProducer.Init().ConfigureAwait(false).GetAwaiter().GetResult();
-            producers.TryAdd(messageProducer.Info.Id, messageProducer);
-            return messageProducer;
+            return new NmsMessageProducer(producerIdGenerator.GenerateId(), this, destination);
         }
 
         public IMessageConsumer CreateConsumer(IDestination destination)
@@ -121,8 +119,6 @@
 
             NmsMessageConsumer messageConsumer = new NmsMessageConsumer(consumerIdGenerator.GenerateId(), this, destination, selector, noLocal);
             messageConsumer.Init().ConfigureAwait(false).GetAwaiter().GetResult();
-
-            consumers.TryAdd(messageConsumer.Info.Id, messageConsumer);
             
             return messageConsumer;
         }
@@ -131,10 +127,8 @@
         {
             CheckClosed();
 
-            NmsMessageConsumer messageConsumer = new NmsMessageConsumer(consumerIdGenerator.GenerateId(), this, destination, name, selector, noLocal);
-            messageConsumer.Info.IsDurable = true;
+            NmsMessageConsumer messageConsumer = new NmsDurableTopicSubscriber(consumerIdGenerator.GenerateId(), this, destination, name, selector, noLocal);
             messageConsumer.Init().ConfigureAwait(false).GetAwaiter().GetResult();
-            consumers.TryAdd(messageConsumer.Info.Id, messageConsumer);
 
             return messageConsumer;
         }
@@ -474,14 +468,24 @@
             return producer;
         }
 
+        public void Add(NmsMessageConsumer messageConsumer)
+        {
+            consumers.TryAdd(messageConsumer.Info.Id, messageConsumer);
+        }
+        
+        public void Add(NmsMessageProducer messageProducer)
+        {
+            producers.TryAdd(messageProducer.Info.Id, messageProducer);
+        }
+
         public void Remove(NmsMessageConsumer messageConsumer)
         {
-            consumers.TryRemove(messageConsumer.Info.Id, out messageConsumer);
+            consumers.TryRemove(messageConsumer.Info.Id, out _);
         }
 
         public void Remove(NmsMessageProducer messageProducer)
         {
-            producers.TryRemove(messageProducer.Info.Id, out messageProducer);
+            producers.TryRemove(messageProducer.Info.Id, out _);
         }
 
         public void Shutdown(NMSException exception = null)
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs b/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
index 03ca15e..3410eb6 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpConnection.cs
@@ -74,7 +74,7 @@
             underlyingConnection.AddClosedCallback((sender, error) => Provider.OnConnectionClosed(error));
             
             // Wait for connection to be opened
-            await tsc.Task;
+            await tsc.Task.ConfigureAwait(false);
 
             // Create a Session for this connection that is used for Temporary Destinations
             // and perhaps later on management and advisory monitoring.
@@ -130,7 +130,7 @@
         public async Task CreateSession(SessionInfo sessionInfo)
         {
             var amqpSession = new AmqpSession(this, sessionInfo);
-            await amqpSession.Start();
+            await amqpSession.Start().ConfigureAwait(false);
             sessions.TryAdd(sessionInfo.Id, amqpSession);
         }
 
@@ -154,8 +154,7 @@
             {
                 return session;
             }
-
-            throw new Exception();
+            throw new InvalidOperationException($"Amqp Session {sessionId} doesn't exist and cannot be retrieved.");
         }
 
         public void RemoveSession(Id sessionId)
diff --git a/src/NMS.AMQP/Provider/Amqp/AmqpSession.cs b/src/NMS.AMQP/Provider/Amqp/AmqpSession.cs
index d20ce76..ee81348 100644
--- a/src/NMS.AMQP/Provider/Amqp/AmqpSession.cs
+++ b/src/NMS.AMQP/Provider/Amqp/AmqpSession.cs
@@ -69,7 +69,6 @@
                     SessionInfo.remoteChannel = begin.RemoteChannel;
                     tcs.TrySetResult(true);
                 });
-            UnderlyingSession.AddClosedCallback((sender, error) => { tcs.TrySetException(ExceptionSupport.GetException(error)); });
             UnderlyingSession.AddClosedCallback((sender, error) =>
             {
                 if (!tcs.TrySetException(ExceptionSupport.GetException(error)))
diff --git a/src/NMS.AMQP/Provider/Failover/FailoverProvider.cs b/src/NMS.AMQP/Provider/Failover/FailoverProvider.cs
index b338555..923ab83 100644
--- a/src/NMS.AMQP/Provider/Failover/FailoverProvider.cs
+++ b/src/NMS.AMQP/Provider/Failover/FailoverProvider.cs
@@ -128,8 +128,8 @@
                             {
                                 Tracer.Debug($"Connection attempt:[{reconnectAttempts}] to: {target.Scheme}://{target.Host}:{target.Port} in-progress");
                                 provider = ProviderFactory.Create(target);
-                                await provider.Connect(connectionInfo);
-                                await InitializeNewConnection(provider);
+                                await provider.Connect(connectionInfo).ConfigureAwait(false);
+                                await InitializeNewConnection(provider).ConfigureAwait(false);
                                 return;
                             }
                             catch (Exception e)
@@ -171,7 +171,7 @@
                         }
                         else
                         {
-                            await reconnectControl.ScheduleReconnect(Reconnect);
+                            await reconnectControl.ScheduleReconnect(Reconnect).ConfigureAwait(false);
                         }
                     }
                 }
@@ -218,21 +218,21 @@
                 Tracer.Debug($"Signalling connection recovery: {provider}");
 
                 // Allow listener to recover its resources
-                await listener.OnConnectionRecovery(provider);
+                await listener.OnConnectionRecovery(provider).ConfigureAwait(false);
 
                 // Restart consumers, send pull commands, etc.
-                await listener.OnConnectionRecovered(provider);
+                await listener.OnConnectionRecovered(provider).ConfigureAwait(false);
 
                 // Let the client know that connection has restored.
                 listener.OnConnectionRestored(connectedUri);
 
                 // If we try to run pending requests right after the connection is reestablished 
                 // it will result in timeout on the first send request
-                await Task.Delay(50);
+                await Task.Delay(50).ConfigureAwait(false);
 
                 foreach (FailoverRequest request in GetPendingRequests())
                 {
-                    await request.Run();
+                    await request.Run().ConfigureAwait(false);
                 }
 
                 reconnectControl.ConnectionEstablished();
diff --git a/src/NMS.AMQP/Provider/Failover/FailoverRequest.cs b/src/NMS.AMQP/Provider/Failover/FailoverRequest.cs
index b54a20f..b6a5684 100644
--- a/src/NMS.AMQP/Provider/Failover/FailoverRequest.cs
+++ b/src/NMS.AMQP/Provider/Failover/FailoverRequest.cs
@@ -61,7 +61,7 @@
             {
                 try
                 {
-                    await this.DoTask(activeProvider);
+                    await this.DoTask(activeProvider).ConfigureAwait(false);
                     this.taskCompletionSource.TrySetResult(true);
                     this.failoverProvider.RemoveFailoverRequest(this);
                     this.cancellationTokenSource?.Dispose();