Identify transports by name

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

5 files changed, 30 insertions, 25 deletions

diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index 4074723c28..5ffac46474 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -370,34 +370,35 @@ Manageable::status_t Broker::ManagementMethod (uint32_t methodId,
     return status;
 }
 
-boost::shared_ptr<ProtocolFactory> Broker::getProtocolFactory() const {
-    assert(protocolFactories.size() > 0);
-    return protocolFactories[0];
+boost::shared_ptr<ProtocolFactory> Broker::getProtocolFactory(const std::string& name) const {
+    ProtocolFactoryMap::const_iterator i = protocolFactories.find(name);
+    if (i == protocolFactories.end()) return boost::shared_ptr<ProtocolFactory>();
+    else return i->second;
 }
 
-void Broker::registerProtocolFactory(ProtocolFactory::shared_ptr protocolFactory) {
-    protocolFactories.push_back(protocolFactory);
-}
-
-// TODO: This can only work if there is only one protocolFactory
+//TODO: should this allow choosing the port by transport name?
 uint16_t Broker::getPort() const  {
     return getProtocolFactory()->getPort();
 }
 
-// TODO: This should iterate over all protocolFactories
-void Broker::accept() {
-    for (unsigned int i = 0; i < protocolFactories.size(); ++i)
-        protocolFactories[i]->accept(poller, factory.get());
+void Broker::registerProtocolFactory(const std::string& name, ProtocolFactory::shared_ptr protocolFactory) {
+    protocolFactories[name] = protocolFactory;
 }
 
+void Broker::accept() {
+    for (ProtocolFactoryMap::const_iterator i = protocolFactories.begin(); i != protocolFactories.end(); i++) {
+        i->second->accept(poller, factory.get());
+    }
+}
 
-// TODO: How to chose the protocolFactory to use for the connection
 void Broker::connect(
-    const std::string& host, uint16_t port, bool /*useSsl*/,
+    const std::string& host, uint16_t port, const std::string& transport,
     boost::function2<void, int, std::string> failed,
     sys::ConnectionCodec::Factory* f)
 {
-    getProtocolFactory()->connect(poller, host, port, f ? f : factory.get(), failed);
+    boost::shared_ptr<ProtocolFactory> pf = getProtocolFactory(transport);
+    if (pf) pf->connect(poller, host, port, f ? f : factory.get(), failed);
+    else throw Exception(QPID_MSG("Unsupported transport type: " << transport));
 }
 
 void Broker::connect(
@@ -407,7 +408,7 @@ void Broker::connect(
 {
     url.throwIfEmpty();
     TcpAddress addr=boost::get<TcpAddress>(url[0]);
-    connect(addr.host, addr.port, false, failed, f);
+    connect(addr.host, addr.port, TCP_TRANSPORT, failed, f);
 }
 
 uint32_t Broker::queueMoveMessages( 
@@ -436,7 +437,7 @@ Broker::getKnownBrokersImpl()
   return knownBrokers;
 }
 
-
+const std::string Broker::TCP_TRANSPORT("tcp");
 
 }} // namespace qpid::broker
 
diff --git a/cpp/src/qpid/broker/Broker.h b/cpp/src/qpid/broker/Broker.h
index a15440bc0e..4b77c05209 100644
--- a/cpp/src/qpid/broker/Broker.h
+++ b/cpp/src/qpid/broker/Broker.h
@@ -93,10 +93,12 @@ class Broker : public sys::Runnable, public Plugin::Target,
     };
  
   private:
+    typedef std::map<std::string, boost::shared_ptr<sys::ProtocolFactory> > ProtocolFactoryMap;
+
     boost::shared_ptr<sys::Poller> poller;
     Options config;
     management::ManagementAgent::Singleton managementAgentSingleton;
-    std::vector< boost::shared_ptr<sys::ProtocolFactory> > protocolFactories;
+    ProtocolFactoryMap protocolFactories;
     MessageStore* store;
     AclModule* acl;
     DataDir dataDir;
@@ -166,13 +168,14 @@ class Broker : public sys::Runnable, public Plugin::Target,
                                                         std::string& text);
     
     /** Add to the broker's protocolFactorys */
-    void registerProtocolFactory(boost::shared_ptr<sys::ProtocolFactory>);
+    void registerProtocolFactory(const std::string& name, boost::shared_ptr<sys::ProtocolFactory>);
 
     /** Accept connections */
     void accept();
 
     /** Create a connection to another broker. */
-    void connect(const std::string& host, uint16_t port, bool useSsl,
+    void connect(const std::string& host, uint16_t port, 
+                 const std::string& transport,
                  boost::function2<void, int, std::string> failed,
                  sys::ConnectionCodec::Factory* =0);
     /** Create a connection to another broker. */
@@ -189,7 +192,7 @@ class Broker : public sys::Runnable, public Plugin::Target,
 
     // TODO: There isn't a single ProtocolFactory so the use of the following needs to be fixed
     // For the present just return the first ProtocolFactory registered.
-    boost::shared_ptr<sys::ProtocolFactory> getProtocolFactory() const;
+    boost::shared_ptr<sys::ProtocolFactory> getProtocolFactory(const std::string& name = TCP_TRANSPORT) const;
 
     /** Expose poller so plugins can register their descriptors. */
     boost::shared_ptr<sys::Poller> getPoller();
@@ -200,7 +203,8 @@ class Broker : public sys::Runnable, public Plugin::Target,
     Timer& getTimer() { return timer; }
 
     boost::function<std::vector<Url> ()> getKnownBrokers;
-    
+
+    static const std::string TCP_TRANSPORT;
 };
 
 }}
diff --git a/cpp/src/qpid/broker/Link.cpp b/cpp/src/qpid/broker/Link.cpp
index 12cbf48986..f80e6078de 100644
--- a/cpp/src/qpid/broker/Link.cpp
+++ b/cpp/src/qpid/broker/Link.cpp
@@ -107,7 +107,7 @@ void Link::startConnectionLH ()
         // Set the state before calling connect.  It is possible that connect
         // will fail synchronously and call Link::closed before returning.
         setStateLH(STATE_CONNECTING);
-        broker->connect (host, port, useSsl,
+        broker->connect (host, port, useSsl ? "ssl" : Broker::TCP_TRANSPORT,
                          boost::bind (&Link::closed, this, _1, _2));
     } catch(std::exception& e) {
         setStateLH(STATE_WAITING);
diff --git a/cpp/src/qpid/sys/RdmaIOPlugin.cpp b/cpp/src/qpid/sys/RdmaIOPlugin.cpp
index 6ea006df9d..9a4452cc17 100644
--- a/cpp/src/qpid/sys/RdmaIOPlugin.cpp
+++ b/cpp/src/qpid/sys/RdmaIOPlugin.cpp
@@ -217,7 +217,7 @@ static class RdmaIOPlugin : public Plugin {
             const broker::Broker::Options& opts = broker->getOptions();
             ProtocolFactory::shared_ptr protocol(new RdmaIOProtocolFactory(opts.port, opts.connectionBacklog));
             QPID_LOG(info, "Listening on RDMA port " << protocol->getPort());
-            broker->registerProtocolFactory(protocol);
+            broker->registerProtocolFactory("rdma", protocol);
         }
     }
 } rdmaPlugin;
diff --git a/cpp/src/qpid/sys/TCPIOPlugin.cpp b/cpp/src/qpid/sys/TCPIOPlugin.cpp
index f38cf88e45..40edfa816f 100644
--- a/cpp/src/qpid/sys/TCPIOPlugin.cpp
+++ b/cpp/src/qpid/sys/TCPIOPlugin.cpp
@@ -67,7 +67,7 @@ static class TCPIOPlugin : public Plugin {
             const broker::Broker::Options& opts = broker->getOptions();
             ProtocolFactory::shared_ptr protocol(new AsynchIOProtocolFactory(opts.port, opts.connectionBacklog, opts.tcpNoDelay));
             QPID_LOG(info, "Listening on TCP port " << protocol->getPort());
-            broker->registerProtocolFactory(protocol);
+            broker->registerProtocolFactory("tcp", protocol);
         }
     }
 } tcpPlugin;