QPID-6167: ensure broker responds with a header for an enabled protocol

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

6 files changed, 36 insertions, 5 deletions

diff --git a/qpid/cpp/src/qpid/broker/Protocol.cpp b/qpid/cpp/src/qpid/broker/Protocol.cpp
index 2ef8c66445..e9e7892499 100644
--- a/qpid/cpp/src/qpid/broker/Protocol.cpp
+++ b/qpid/cpp/src/qpid/broker/Protocol.cpp
@@ -35,8 +35,10 @@ ProtocolRegistry::ProtocolRegistry(const std::set<std::string>& e, Broker* b) :
 
 qpid::sys::ConnectionCodec* ProtocolRegistry::create(const qpid::framing::ProtocolVersion& v, qpid::sys::OutputControl& o, const std::string& id, const qpid::sys::SecuritySettings& s)
 {
-    if (v == qpid::framing::ProtocolVersion(0, 10) && isEnabled(AMQP_0_10)) {
-        return create_0_10(o, id, s, false);
+    if (v == qpid::framing::ProtocolVersion(0, 10)) {
+        if (isEnabled(AMQP_0_10)) {
+            return create_0_10(o, id, s, false);
+        }
     }
     qpid::sys::ConnectionCodec* codec = 0;
     for (Protocols::const_iterator i = protocols.begin(); !codec && i != protocols.end(); ++i) {
@@ -51,7 +53,22 @@ qpid::sys::ConnectionCodec* ProtocolRegistry::create(qpid::sys::OutputControl& o
     return create_0_10(o, id, s, true);
 }
 
-bool ProtocolRegistry::isEnabled(const std::string& name)
+qpid::framing::ProtocolVersion ProtocolRegistry::supportedVersion() const
+{
+    if (isEnabled(AMQP_0_10)) {
+        return qpid::framing::ProtocolVersion(0,10);
+    } else {
+        for (Protocols::const_iterator i = protocols.begin(); i != protocols.end(); ++i) {
+            if (isEnabled(i->first)) {
+                return i->second->supportedVersion();
+            }
+        }
+    }
+    QPID_LOG(error, "No enabled protocols!");
+    return qpid::framing::ProtocolVersion(0,0);
+}
+
+bool ProtocolRegistry::isEnabled(const std::string& name) const
 {
     return enabled.empty()/*if nothing is explicitly enabled, assume everything is*/ || enabled.find(name) != enabled.end();
 }
diff --git a/qpid/cpp/src/qpid/broker/Protocol.h b/qpid/cpp/src/qpid/broker/Protocol.h
index 59a631848e..a7dbc98fff 100644
--- a/qpid/cpp/src/qpid/broker/Protocol.h
+++ b/qpid/cpp/src/qpid/broker/Protocol.h
@@ -61,6 +61,7 @@ class Protocol
     virtual qpid::sys::ConnectionCodec* create(const qpid::framing::ProtocolVersion&, qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&) = 0;
     virtual boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> translate(const Message&) = 0;
     virtual boost::shared_ptr<RecoverableMessage> recover(qpid::framing::Buffer&) = 0;
+    virtual qpid::framing::ProtocolVersion supportedVersion() const = 0;
 
   private:
 };
@@ -70,6 +71,7 @@ class ProtocolRegistry : public Protocol, public qpid::sys::ConnectionCodec::Fac
   public:
     QPID_BROKER_EXTERN qpid::sys::ConnectionCodec* create(const qpid::framing::ProtocolVersion&, qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&);
     QPID_BROKER_EXTERN qpid::sys::ConnectionCodec* create(qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&);
+    QPID_BROKER_EXTERN qpid::framing::ProtocolVersion supportedVersion() const;
     QPID_BROKER_EXTERN boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> translate(const Message&);
     QPID_BROKER_EXTERN boost::shared_ptr<RecoverableMessage> recover(qpid::framing::Buffer&);
     QPID_BROKER_EXTERN Message decode(qpid::framing::Buffer&);
@@ -86,7 +88,7 @@ class ProtocolRegistry : public Protocol, public qpid::sys::ConnectionCodec::Fac
     Broker* broker;
 
     qpid::sys::ConnectionCodec* create_0_10(qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&, bool);
-    bool isEnabled(const std::string&);
+    bool isEnabled(const std::string&) const;
 
 };
 }} // namespace qpid::broker
diff --git a/qpid/cpp/src/qpid/broker/amqp/Domain.cpp b/qpid/cpp/src/qpid/broker/amqp/Domain.cpp
index cc714c0730..c2d4782fc4 100644
--- a/qpid/cpp/src/qpid/broker/amqp/Domain.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/Domain.cpp
@@ -140,6 +140,10 @@ class InterconnectFactory : public BrokerContext, public qpid::sys::ConnectionCo
                         boost::shared_ptr<Domain>, BrokerContext&, boost::shared_ptr<Relay>);
     qpid::sys::ConnectionCodec* create(const framing::ProtocolVersion&, qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&);
     qpid::sys::ConnectionCodec* create(qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&);
+    qpid::framing::ProtocolVersion supportedVersion() const
+    {
+        return qpid::framing::ProtocolVersion(1, 0);
+    }
     bool connect();
     void failed(int, std::string);
   private:
diff --git a/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp b/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
index 2ea381e2bc..621f25f04b 100644
--- a/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
+++ b/qpid/cpp/src/qpid/broker/amqp/ProtocolPlugin.cpp
@@ -73,6 +73,7 @@ class ProtocolImpl : public BrokerContext, public Protocol
     qpid::sys::ConnectionCodec* create(const qpid::framing::ProtocolVersion&, qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&);
     boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> translate(const qpid::broker::Message&);
     boost::shared_ptr<RecoverableMessage> recover(qpid::framing::Buffer&);
+    qpid::framing::ProtocolVersion supportedVersion() const;
   private:
 };
 
@@ -158,5 +159,10 @@ boost::shared_ptr<RecoverableMessage> ProtocolImpl::recover(qpid::framing::Buffe
     }
 }
 
+qpid::framing::ProtocolVersion ProtocolImpl::supportedVersion() const
+{
+    return qpid::framing::ProtocolVersion(1,0);
+}
+
 
 }}} // namespace qpid::broker::amqp
diff --git a/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp b/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
index bb59530883..7be625a1a3 100644
--- a/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
+++ b/qpid/cpp/src/qpid/sys/AsynchIOHandler.cpp
@@ -150,7 +150,7 @@ void AsynchIOHandler::readbuff(AsynchIO& , AsynchIO::BufferBase* buff) {
                 if (!codec) {
                     //TODO: may still want to revise this...
                     //send valid version header & close connection.
-                    write(framing::ProtocolInitiation(framing::highestProtocolVersion));
+                    write(framing::ProtocolInitiation(factory->supportedVersion()));
                     readError = true;
                     aio->queueWriteClose();
                 } else {
diff --git a/qpid/cpp/src/qpid/sys/ConnectionCodec.h b/qpid/cpp/src/qpid/sys/ConnectionCodec.h
index 969a3877e3..8b5b69cdb4 100644
--- a/qpid/cpp/src/qpid/sys/ConnectionCodec.h
+++ b/qpid/cpp/src/qpid/sys/ConnectionCodec.h
@@ -60,6 +60,8 @@ class ConnectionCodec : public Codec {
         virtual ConnectionCodec* create(
             OutputControl&, const std::string& id, const SecuritySettings&
         ) = 0;
+
+        virtual framing::ProtocolVersion supportedVersion() const = 0;
     };
 };