Inherit offsets between autorecovering channel delegates

Author: michaelklishin

projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringConnection.cs

@@ -388,7 +388,8 @@ public IModel CreateModel()
             AutorecoveringModel m;
             lock(this)
             {
-                m = new AutorecoveringModel(this, (Model)this.CreateNonRecoveringModel());
+                m = new AutorecoveringModel(this,
+                                            (RecoveryAwareModel)this.CreateNonRecoveringModel());
                 m_models.Add(m);
             }
             return m;
@@ -402,10 +403,10 @@ public void UnregisterModel(AutorecoveringModel model)
             }
         }
 
-        protected IModel CreateNonRecoveringModel()
+        public RecoveryAwareModel CreateNonRecoveringModel()
         {
-            ISession session  = m_delegate.CreateSession();
-            IFullModel result = (IFullModel)(new RecoveryAwareModel(session));
+            var session = m_delegate.CreateSession();
+            var result  = new RecoveryAwareModel(session);
             result._Private_ChannelOpen("");
             return result;
         }

projects/client/RabbitMQ.Client/src/client/impl/AutorecoveringModel.cs

@@ -51,7 +51,7 @@ namespace RabbitMQ.Client.Impl
     public class AutorecoveringModel : IFullModel, IRecoverable
     {
         protected AutorecoveringConnection m_connection;
-        protected Model m_delegate;
+        protected RecoveryAwareModel m_delegate;
 
         public readonly object m_eventLock = new object();
 
@@ -73,7 +73,7 @@ public class AutorecoveringModel : IFullModel, IRecoverable
         protected bool usesPublisherConfirms   = false;
         protected bool usesTransactions        = false;
 
-        public AutorecoveringModel(AutorecoveringConnection conn, Model _delegate)
+        public AutorecoveringModel(AutorecoveringConnection conn, RecoveryAwareModel _delegate)
         {
             this.m_connection = conn;
             this.m_delegate   = _delegate;
@@ -82,8 +82,10 @@ public AutorecoveringModel(AutorecoveringConnection conn, Model _delegate)
         public void AutomaticallyRecover(AutorecoveringConnection conn, IConnection connDelegate)
         {
             this.m_connection = conn;
-            this.m_delegate   = (Model)connDelegate.CreateModel();
-            // TODO: inherit ack offset
+            var defunctModel  = this.m_delegate;
+
+            this.m_delegate   = conn.CreateNonRecoveringModel();
+            this.m_delegate.InheritOffsetFrom(defunctModel);
 
             this.RecoverModelShutdownHandlers();
             this.RecoverState();

projects/client/RabbitMQ.Client/src/client/impl/RecoveryAwareModel.cs

@@ -49,6 +49,16 @@ public class RecoveryAwareModel : Model, IFullModel, IRecoverable
         private ulong maxSeenDeliveryTag = 0;
         private ulong activeDeliveryTagOffset = 0;
 
+        public ulong MaxSeenDeliveryTag
+        {
+            get { return maxSeenDeliveryTag; }
+        }
+
+        public ulong ActiveDeliveryTagOffset
+        {
+            get { return activeDeliveryTagOffset; }
+        }
+
         public RecoveryAwareModel(ISession session) : base(session) {}
 
         public override void HandleBasicDeliver(string consumerTag,
@@ -101,6 +111,11 @@ public override void BasicNack(ulong deliveryTag,
             }
         }
 
+        public void InheritOffsetFrom(RecoveryAwareModel other)
+        {
+            this.activeDeliveryTagOffset = other.ActiveDeliveryTagOffset + other.MaxSeenDeliveryTag;
+            this.maxSeenDeliveryTag = 0;
+        }
 
         protected ulong OffsetDeliveryTag(ulong deliveryTag)
         {