correction for Active-Active clustering, allowing late joining nodes in the cluster to sync counter values for sequenced messages

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

1 files changed, 54 insertions, 37 deletions

diff --git a/qpid/cpp/src/tests/ExchangeTest.cpp b/qpid/cpp/src/tests/ExchangeTest.cpp
index c2f6078b33..0946d3115d 100644
--- a/qpid/cpp/src/tests/ExchangeTest.cpp
+++ b/qpid/cpp/src/tests/ExchangeTest.cpp
@@ -179,48 +179,65 @@ QPID_AUTO_TEST_CASE(testSequenceOptions)
 {
     FieldTable args;
     args.setInt("qpid.msg_sequence",1);
-    
-    DirectExchange direct("direct1", false, args);
-    
-    intrusive_ptr<Message> msg1 = cmessage("e", "A");
-    intrusive_ptr<Message> msg2 = cmessage("e", "B");
-    intrusive_ptr<Message> msg3 = cmessage("e", "C");
+    char* buff = new char[10000];
+    framing::Buffer buffer(buff,10000);
+    {
+        DirectExchange direct("direct1", false, args);
 
-    DeliverableMessage dmsg1(msg1);
-    DeliverableMessage dmsg2(msg2);
-    DeliverableMessage dmsg3(msg3);
+        intrusive_ptr<Message> msg1 = cmessage("e", "A");
+        intrusive_ptr<Message> msg2 = cmessage("e", "B");
+        intrusive_ptr<Message> msg3 = cmessage("e", "C");
 
-    direct.route(dmsg1, "abc", 0);
-    direct.route(dmsg2, "abc", 0);
-    direct.route(dmsg3, "abc", 0);
+        DeliverableMessage dmsg1(msg1);
+        DeliverableMessage dmsg2(msg2);
+        DeliverableMessage dmsg3(msg3);
 
-    BOOST_CHECK_EQUAL(1, msg1->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
-    BOOST_CHECK_EQUAL(2, msg2->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
-    BOOST_CHECK_EQUAL(3, msg3->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
-    
-    FanOutExchange fanout("fanout1", false, args);
-    HeadersExchange header("headers1", false, args);
-    TopicExchange topic ("topic1", false, args);
-    
-    // check other exchanges, that they preroute
-    intrusive_ptr<Message> msg4 = cmessage("e", "A");
-    intrusive_ptr<Message> msg5 = cmessage("e", "B");
-    intrusive_ptr<Message> msg6 = cmessage("e", "C");
-
-    DeliverableMessage dmsg4(msg4);
-    DeliverableMessage dmsg5(msg5);
-    DeliverableMessage dmsg6(msg6);
-    
-    fanout.route(dmsg4, "abc", 0);
-    BOOST_CHECK_EQUAL(1, msg4->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
-    
-    FieldTable headers;
-    header.route(dmsg5, "abc", &headers);
-    BOOST_CHECK_EQUAL(1, msg5->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
+        direct.route(dmsg1, "abc", 0);
+        direct.route(dmsg2, "abc", 0);
+        direct.route(dmsg3, "abc", 0);
+
+        BOOST_CHECK_EQUAL(1, msg1->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
+        BOOST_CHECK_EQUAL(2, msg2->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
+        BOOST_CHECK_EQUAL(3, msg3->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
+
+        FanOutExchange fanout("fanout1", false, args);
+        HeadersExchange header("headers1", false, args);
+        TopicExchange topic ("topic1", false, args);
+
+        // check other exchanges, that they preroute
+        intrusive_ptr<Message> msg4 = cmessage("e", "A");
+        intrusive_ptr<Message> msg5 = cmessage("e", "B");
+        intrusive_ptr<Message> msg6 = cmessage("e", "C");
+
+        DeliverableMessage dmsg4(msg4);
+        DeliverableMessage dmsg5(msg5);
+        DeliverableMessage dmsg6(msg6);
 
-    topic.route(dmsg6, "abc", 0);
-    BOOST_CHECK_EQUAL(1, msg6->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
+        fanout.route(dmsg4, "abc", 0);
+        BOOST_CHECK_EQUAL(1, msg4->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
 
+        FieldTable headers;
+        header.route(dmsg5, "abc", &headers);
+        BOOST_CHECK_EQUAL(1, msg5->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
+
+        topic.route(dmsg6, "abc", 0);
+        BOOST_CHECK_EQUAL(1, msg6->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
+        direct.encode(buffer);
+    }
+    {
+        
+        ExchangeRegistry exchanges;
+        buffer.reset();
+        DirectExchange::shared_ptr exch_dec = Exchange::decode(exchanges, buffer);
+        
+        intrusive_ptr<Message> msg1 = cmessage("e", "A");
+        DeliverableMessage dmsg1(msg1);
+        exch_dec->route(dmsg1, "abc", 0);
+
+        BOOST_CHECK_EQUAL(4, msg1->getApplicationHeaders()->getAsInt64("qpid.msg_sequence"));
+    
+    }
+    delete [] buff;
 }
 
 QPID_AUTO_TEST_CASE(testIVEOption)