Set protocol versions correctly in cluster code.

Cluster code was broken by a recent checkin to validate protocol
versions.  The cluster was not correctly setting the version on both
sides of a connection.


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

5 files changed, 29 insertions, 12 deletions

diff --git a/cpp/src/qpid/amqp_0_10/Connection.cpp b/cpp/src/qpid/amqp_0_10/Connection.cpp
index c1de5e2dec..96d5146c30 100644
--- a/cpp/src/qpid/amqp_0_10/Connection.cpp
+++ b/cpp/src/qpid/amqp_0_10/Connection.cpp
@@ -31,7 +31,8 @@ namespace amqp_0_10 {
 using sys::Mutex;
 
 Connection::Connection(sys::OutputControl& o, const std::string& id, bool _isClient)
-    : frameQueueClosed(false), output(o), identifier(id), initialized(false), isClient(_isClient), buffered(0)
+    : frameQueueClosed(false), output(o), identifier(id), initialized(false),
+      isClient(_isClient), buffered(0), version(0,10)
 {}
 
 void Connection::setInputHandler(std::auto_ptr<sys::ConnectionInputHandler> c) {
@@ -44,7 +45,9 @@ size_t  Connection::decode(const char* buffer, size_t size) {
         //read in protocol header
         framing::ProtocolInitiation pi;
         if (pi.decode(in)) {
-            //TODO: check the version is correct
+            if(!(pi==version))
+                throw Exception(QPID_MSG("Unsupported version: " << pi
+                                         << " supported version " << version));
             QPID_LOG(trace, "RECV " << identifier << " INIT(" << pi << ")");
         }
         initialized = true;
@@ -128,7 +131,11 @@ void Connection::send(framing::AMQFrame& f) {
 }
 
 framing::ProtocolVersion Connection::getVersion() const {
-    return framing::ProtocolVersion(0,10);
+    return version;
+}
+
+void Connection::setVersion(const framing::ProtocolVersion& v)  {
+    version = v;
 }
 
 size_t Connection::getBuffered() const {
diff --git a/cpp/src/qpid/amqp_0_10/Connection.h b/cpp/src/qpid/amqp_0_10/Connection.h
index 32aadff105..6fd51381fc 100644
--- a/cpp/src/qpid/amqp_0_10/Connection.h
+++ b/cpp/src/qpid/amqp_0_10/Connection.h
@@ -55,6 +55,7 @@ class Connection  : public sys::ConnectionCodec,
     bool initialized;
     bool isClient;
     size_t buffered;
+    framing::ProtocolVersion version;
 
   public:
     QPID_BROKER_EXTERN Connection(sys::OutputControl&, const std::string& id, bool isClient);
@@ -71,6 +72,10 @@ class Connection  : public sys::ConnectionCodec,
     void send(framing::AMQFrame&);
     framing::ProtocolVersion getVersion() const;
     size_t getBuffered() const;
+
+    /** Used by cluster code to set a special version on "update" connections. */
+    // FIXME aconway 2009-07-30: find a cleaner mechanism for this.
+    void setVersion(const framing::ProtocolVersion&);
 };
 
 }} // namespace qpid::amqp_0_10
diff --git a/cpp/src/qpid/client/Connector.cpp b/cpp/src/qpid/client/Connector.cpp
index dca35faacf..f69032b26d 100644
--- a/cpp/src/qpid/client/Connector.cpp
+++ b/cpp/src/qpid/client/Connector.cpp
@@ -362,7 +362,8 @@ size_t TCPConnector::decode(const char* buffer, size_t size)
         if (protocolInit.decode(in)) {
             QPID_LOG(debug, "RECV " << identifier << " INIT(" << protocolInit << ")");
             if(!(protocolInit==version)){
-                throw Exception(QPID_MSG("Unsupported version: " << protocolInit));
+                throw Exception(QPID_MSG("Unsupported version: " << protocolInit
+                                         << " supported version " << version));
             }
         }
         initiated = true;
diff --git a/cpp/src/qpid/cluster/ConnectionCodec.cpp b/cpp/src/qpid/cluster/ConnectionCodec.cpp
index 0c791cdf44..4ff8b0a4a3 100644
--- a/cpp/src/qpid/cluster/ConnectionCodec.cpp
+++ b/cpp/src/qpid/cluster/ConnectionCodec.cpp
@@ -38,24 +38,27 @@ using namespace framing;
 sys::ConnectionCodec*
 ConnectionCodec::Factory::create(ProtocolVersion v, sys::OutputControl& out, const std::string& id) {
     if (v == ProtocolVersion(0, 10))
-        return new ConnectionCodec(out, id, cluster, false, false);
-    else if (v == ProtocolVersion(0x80 + 0, 0x80 + 10))
-        return new ConnectionCodec(out, id, cluster, true, false); // Catch-up connection
+        return new ConnectionCodec(v, out, id, cluster, false, false);
+    else if (v == ProtocolVersion(0x80 + 0, 0x80 + 10)) // Catch-up connection
+        return new ConnectionCodec(v, out, id, cluster, true, false); 
     return 0;
 }
 
 // Used for outgoing Link connections
 sys::ConnectionCodec*
 ConnectionCodec::Factory::create(sys::OutputControl& out, const std::string& logId) {
-    return new ConnectionCodec(out, logId, cluster, false, true);
+    return new ConnectionCodec(ProtocolVersion(0,10), out, logId, cluster, false, true);
 }
 
-ConnectionCodec::ConnectionCodec(sys::OutputControl& out, const std::string& logId, Cluster& cluster, bool catchUp, bool isLink)
-    : codec(out, logId, isLink),
-      interceptor(new Connection(cluster, codec, logId, cluster.getId(), catchUp, isLink))
+ConnectionCodec::ConnectionCodec(
+    const ProtocolVersion& v, sys::OutputControl& out,
+    const std::string& logId, Cluster& cluster, bool catchUp, bool isLink
+) : codec(out, logId, isLink),
+    interceptor(new Connection(cluster, codec, logId, cluster.getId(), catchUp, isLink))
 {
     std::auto_ptr<sys::ConnectionInputHandler> ih(new ProxyInputHandler(interceptor));
     codec.setInputHandler(ih);
+    codec.setVersion(v);
 }
 
 ConnectionCodec::~ConnectionCodec() {}
diff --git a/cpp/src/qpid/cluster/ConnectionCodec.h b/cpp/src/qpid/cluster/ConnectionCodec.h
index ea01b7abb9..4ff738b603 100644
--- a/cpp/src/qpid/cluster/ConnectionCodec.h
+++ b/cpp/src/qpid/cluster/ConnectionCodec.h
@@ -56,7 +56,8 @@ class ConnectionCodec : public sys::ConnectionCodec {
         sys::ConnectionCodec* create(sys::OutputControl&, const std::string& id);
     };
 
-    ConnectionCodec(sys::OutputControl& out, const std::string& logId, Cluster& c, bool catchUp, bool isLink);
+    ConnectionCodec(const framing::ProtocolVersion&, sys::OutputControl& out,
+                    const std::string& logId, Cluster& c, bool catchUp, bool isLink);
     ~ConnectionCodec();
 
     // ConnectionCodec functions.