Initial changes to record and replay frames on fail over.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@480157 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 17 insertions, 31 deletions

diff --git a/dotnet/Qpid.Client/Client/AmqChannel.cs b/dotnet/Qpid.Client/Client/AmqChannel.cs
index 77d5529126..6b0661b3be 100644
--- a/dotnet/Qpid.Client/Client/AmqChannel.cs
+++ b/dotnet/Qpid.Client/Client/AmqChannel.cs
@@ -71,6 +71,8 @@ namespace Qpid.Client
         /// </summary>
         private Hashtable _consumers = Hashtable.Synchronized(new Hashtable());
 
+        private ArrayList _replayFrames = new ArrayList();
+
         /// <summary>
         /// The counter of the _next producer id. This id is generated by the session and used only to allow the
         /// producer to identify itself to the session when deregistering itself.
@@ -709,32 +711,13 @@ namespace Qpid.Client
          * Resubscribes all producers and consumers. This is called when performing failover.
          * @throws AMQException
          */
-        internal void Resubscribe()
-        {
-            ResubscribeProducers();
-            ResubscribeConsumers();
-        }
-
-        private void ResubscribeProducers()
-        {
-            // FIXME: This needs to Replay DeclareExchange method calls.
-            
-//            ArrayList producers = new ArrayList(_producers.Values);
-//            _logger.Debug(String.Format("Resubscribing producers = {0} producers.size={1}", producers, producers.Count));
-//            foreach (BasicMessageProducer producer in producers)
-//            {
-//                producer.Resubscribe();
-//            }
-        }
-
-        private void ResubscribeConsumers()
+        internal void ReplayOnFailOver()
         {
-            ArrayList consumers = new ArrayList(_consumers.Values);
-            _consumers.Clear();
-
-            foreach (BasicMessageConsumer consumer in consumers)
+            _logger.Debug(string.Format("Replaying frames for channel {0}", _channelId));
+            foreach (AMQFrame frame in _replayFrames)
             {
-                RegisterConsumer(consumer);                
+                _logger.Debug(string.Format("Replaying frame=[{0}]", frame));
+                _connection.ProtocolWriter.Write(frame);
             }
         }
 
@@ -769,6 +752,8 @@ namespace Qpid.Client
             AMQFrame queueBind = QueueBindBody.CreateAMQFrame(_channelId, 0,
                                                               queueName, exchangeName,
                                                               routingKey, true, args);
+            _replayFrames.Add(queueBind);
+
             _connection.ProtocolWriter.Write(queueBind);
         }
 
@@ -945,6 +930,8 @@ namespace Qpid.Client
                                                                     false, isDurable, isExclusive,
                                                                     isAutoDelete, true, null);
 
+            _replayFrames.Add(queueDeclare);
+
             _connection.ProtocolWriter.Write(queueDeclare);
         }
 
@@ -963,19 +950,18 @@ namespace Qpid.Client
             _logger.Debug(String.Format("DeclareExchange channelId={0} exchangeName={1} exchangeClass={2}",
                                             _channelId, exchangeName, exchangeClass));
 
-            AMQFrame exchangeDeclareFrame = ExchangeDeclareBody.CreateAMQFrame(
+            AMQFrame declareExchange = ExchangeDeclareBody.CreateAMQFrame(
                 channelId, ticket, exchangeName, exchangeClass, passive, durable, autoDelete, xinternal, noWait, args);
+
+            _replayFrames.Add(declareExchange);
             
-            // FIXME: Probably need to record the exchangeDeclareBody for later replay.
-            ExchangeDeclareBody exchangeDeclareBody = (ExchangeDeclareBody)exchangeDeclareFrame.BodyFrame;
-//            Console.WriteLine(string.Format("XXX AMQP:DeclareExchangeBody=[{0}]", exchangeDeclareBody));
-            if (exchangeDeclareBody.Nowait)
+            if (noWait)
             {
-                _connection.ProtocolWriter.Write(exchangeDeclareFrame);
+                _connection.ProtocolWriter.Write(declareExchange);
             }
             else
             {
-                _connection.ConvenientProtocolWriter.SyncWrite(exchangeDeclareFrame, typeof (ExchangeDeclareOkBody));
+                _connection.ConvenientProtocolWriter.SyncWrite(declareExchange, typeof (ExchangeDeclareOkBody));
             }
         }
     }