QPID-3858: Updated branch - merged from trunk r.1368650

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

1 files changed, 203 insertions, 133 deletions

diff --git a/cpp/src/qpid/broker/Link.cpp b/cpp/src/qpid/broker/Link.cpp
index f21c861149..84dd163ac3 100644
--- a/cpp/src/qpid/broker/Link.cpp
+++ b/cpp/src/qpid/broker/Link.cpp
@@ -125,18 +125,20 @@ boost::shared_ptr<Exchange> Link::linkExchangeFactory( const std::string& _name
     return Exchange::shared_ptr(new LinkExchange(_name));
 }
 
-Link::Link(LinkRegistry*  _links,
-           MessageStore*  _store,
+Link::Link(const string&  _name,
+           LinkRegistry*  _links,
            const string&        _host,
            uint16_t       _port,
            const string&        _transport,
+           DestroyedListener    l,
            bool           _durable,
            const string&        _authMechanism,
            const string&        _username,
            const string&        _password,
            Broker*        _broker,
-           Manageable*    parent)
-    : links(_links), store(_store),
+           Manageable*    parent,
+           bool failover_)
+    : name(_name), links(_links),
       configuredTransport(_transport), configuredHost(_host), configuredPort(_port),
       host(_host), port(_port), transport(_transport),
       durable(_durable),
@@ -149,7 +151,9 @@ Link::Link(LinkRegistry*  _links,
       channelCounter(1),
       connection(0),
       agent(0),
+      listener(l),
       timerTask(new LinkTimerTask(*this, broker->getTimer())),
+      failover(failover_),
       failoverChannel(0)
 {
     if (parent != 0 && broker != 0)
@@ -157,7 +161,10 @@ Link::Link(LinkRegistry*  _links,
         agent = broker->getManagementAgent();
         if (agent != 0)
         {
-            mgmtObject = new _qmf::Link(agent, this, parent, _host, _port, _transport, _durable);
+            mgmtObject = new _qmf::Link(agent, this, parent, name, durable);
+            mgmtObject->set_host(host);
+            mgmtObject->set_port(port);
+            mgmtObject->set_transport(transport);
             agent->addObject(mgmtObject, 0, durable);
         }
     }
@@ -169,13 +176,15 @@ Link::Link(LinkRegistry*  _links,
     }
     broker->getTimer().add(timerTask);
 
-    stringstream _name;
-    _name << "qpid.link." << transport << ":" << host << ":" << port;
-    std::pair<Exchange::shared_ptr, bool> rc = broker->getExchanges().declare(_name.str(),
-                                                                              exchangeTypeName);
-    failoverExchange = boost::static_pointer_cast<LinkExchange>(rc.first);
-    assert(failoverExchange);
-    failoverExchange->setLink(this);
+    if (failover) {
+        stringstream exchangeName;
+        exchangeName << "qpid.link." << name;
+        std::pair<Exchange::shared_ptr, bool> rc =
+            broker->getExchanges().declare(exchangeName.str(), exchangeTypeName);
+        failoverExchange = boost::static_pointer_cast<LinkExchange>(rc.first);
+        assert(failoverExchange);
+        failoverExchange->setLink(this);
+    }
 }
 
 Link::~Link ()
@@ -187,7 +196,8 @@ Link::~Link ()
     if (mgmtObject != 0)
         mgmtObject->resourceDestroy ();
 
-    broker->getExchanges().destroy(failoverExchange->getName());
+    if (failover)
+        broker->getExchanges().destroy(failoverExchange->getName());
 }
 
 void Link::setStateLH (int newState)
@@ -239,16 +249,19 @@ void Link::established(Connection* c)
 
     if (!hideManagement() && agent)
         agent->raiseEvent(_qmf::EventBrokerLinkUp(addr.str()));
-
-    Mutex::ScopedLock mutex(lock);
-    setStateLH(STATE_OPERATIONAL);
-    currentInterval = 1;
-    visitCount      = 0;
-    connection = c;
-    if (closing)
+    bool isClosing = false;
+    {
+        Mutex::ScopedLock mutex(lock);
+        setStateLH(STATE_OPERATIONAL);
+        currentInterval = 1;
+        visitCount      = 0;
+        connection = c;
+        isClosing = closing;
+    }
+    if (isClosing)
         destroy();
     else // Process any IO tasks bridges added before established.
-        connection->requestIOProcessing (boost::bind(&Link::ioThreadProcessing, this));
+        c->requestIOProcessing (boost::bind(&Link::ioThreadProcessing, this));
 }
 
 
@@ -261,16 +274,26 @@ void Link::setUrl(const Url& u) {
 
 
 namespace {
-    /** invoked when session used to subscribe to remote's amq.failover exchange detaches */
-    void sessionDetached(Link *link) {
-        QPID_LOG(debug, "detached from 'amq.failover' for link: " << link->getName());
-    }
+class DetachedCallback : public SessionHandler::ErrorListener {
+  public:
+    DetachedCallback(const Link& link) : name(link.getName()) {}
+    void connectionException(framing::connection::CloseCode, const std::string&) {}
+    void channelException(framing::session::DetachCode, const std::string&) {}
+    void executionException(framing::execution::ErrorCode, const std::string&) {}
+    void detach() {}
+  private:
+    const std::string name;
+};
 }
 
-
 void Link::opened() {
     Mutex::ScopedLock mutex(lock);
     if (!connection) return;
+
+    if (!hideManagement() && connection->GetManagementObject()) {
+        mgmtObject->set_connectionRef(connection->GetManagementObject()->getObjectId());
+    }
+
     // Get default URL from known-hosts if not already set
     if (url.empty()) {
         const std::vector<Url>& known = connection->getKnownHosts();
@@ -282,80 +305,82 @@ void Link::opened() {
         QPID_LOG(debug, "Known hosts for peer of inter-broker link: " << url);
     }
 
-    //
-    // attempt to subscribe to failover exchange for updates from remote
-    //
-
-    const std::string queueName = "qpid.link." + framing::Uuid(true).str();
-    failoverChannel = nextChannel();
-
-    SessionHandler& sessionHandler = connection->getChannel(failoverChannel);
-    sessionHandler.setDetachedCallback( boost::bind(&sessionDetached, this) );
-    failoverSession = queueName;
-    sessionHandler.attachAs(failoverSession);
-
-    framing::AMQP_ServerProxy remoteBroker(sessionHandler.out);
-
-    remoteBroker.getQueue().declare(queueName,
-                                    "",         // alt-exchange
-                                    false,      // passive
-                                    false,      // durable
-                                    true,       // exclusive
-                                    true,       // auto-delete
-                                    FieldTable());
-    remoteBroker.getExchange().bind(queueName,
-                                    FAILOVER_EXCHANGE,
-                                    "",     // no key
-                                    FieldTable());
-    remoteBroker.getMessage().subscribe(queueName,
-                                        failoverExchange->getName(),
-                                        1,           // implied-accept mode
-                                        0,           // pre-acquire mode
-                                        false,       // exclusive
-                                        "",          // resume-id
-                                        0,           // resume-ttl
+    if (failover) {
+        //
+        // attempt to subscribe to failover exchange for updates from remote
+        //
+
+        const std::string queueName = "qpid.link." + framing::Uuid(true).str();
+        failoverChannel = nextChannel();
+
+        SessionHandler& sessionHandler = connection->getChannel(failoverChannel);
+        sessionHandler.setErrorListener(
+            boost::shared_ptr<SessionHandler::ErrorListener>(new DetachedCallback(*this)));
+        failoverSession = queueName;
+        sessionHandler.attachAs(failoverSession);
+
+        framing::AMQP_ServerProxy remoteBroker(sessionHandler.out);
+
+        remoteBroker.getQueue().declare(queueName,
+                                        "",         // alt-exchange
+                                        false,      // passive
+                                        false,      // durable
+                                        true,       // exclusive
+                                        true,       // auto-delete
+                                        FieldTable());
+        remoteBroker.getExchange().bind(queueName,
+                                        FAILOVER_EXCHANGE,
+                                        "",     // no key
                                         FieldTable());
-    remoteBroker.getMessage().flow(failoverExchange->getName(), 0, 0xFFFFFFFF);
-    remoteBroker.getMessage().flow(failoverExchange->getName(), 1, 0xFFFFFFFF);
+        remoteBroker.getMessage().subscribe(queueName,
+                                            failoverExchange->getName(),
+                                            1,           // implied-accept mode
+                                            0,           // pre-acquire mode
+                                            false,       // exclusive
+                                            "",          // resume-id
+                                            0,           // resume-ttl
+                                            FieldTable());
+        remoteBroker.getMessage().flow(failoverExchange->getName(), 0, 0xFFFFFFFF);
+        remoteBroker.getMessage().flow(failoverExchange->getName(), 1, 0xFFFFFFFF);
+    }
 }
 
 void Link::closed(int, std::string text)
 {
-    bool isClosing = false;
-    {
-        Mutex::ScopedLock mutex(lock);
-        QPID_LOG (info, "Inter-broker link disconnected from " << host << ":" << port << " " << text);
+    Mutex::ScopedLock mutex(lock);
+    QPID_LOG (info, "Inter-broker link disconnected from " << host << ":" << port << " " << text);
 
-        connection = 0;
-        if (state == STATE_OPERATIONAL) {
+    connection = 0;
+
+    if (!hideManagement()) {
+        mgmtObject->set_connectionRef(qpid::management::ObjectId());
+        if (state == STATE_OPERATIONAL && agent) {
             stringstream addr;
             addr << host << ":" << port;
-            if (!hideManagement() && agent)
-                agent->raiseEvent(_qmf::EventBrokerLinkDown(addr.str()));
+            agent->raiseEvent(_qmf::EventBrokerLinkDown(addr.str()));
         }
+    }
 
-        for (Bridges::iterator i = active.begin(); i != active.end(); i++) {
-            (*i)->closed();
-            created.push_back(*i);
-        }
-        active.clear();
+    for (Bridges::iterator i = active.begin(); i != active.end(); i++) {
+        (*i)->closed();
+        created.push_back(*i);
+    }
+    active.clear();
 
-        if (state != STATE_FAILED && state != STATE_PASSIVE)
-        {
-            setStateLH(STATE_WAITING);
-            if (!hideManagement())
-                mgmtObject->set_lastError (text);
-        }
+    if (state != STATE_FAILED && state != STATE_PASSIVE)
+    {
+        setStateLH(STATE_WAITING);
+        if (!hideManagement())
+            mgmtObject->set_lastError (text);
     }
-    // Call destroy outside of the lock, don't want to be deleted with lock held.
-    if (isClosing)
-        destroy();
 }
 
-// Called in connection IO thread.
+// Called in connection IO thread, cleans up the connection before destroying Link
 void Link::destroy ()
 {
     Bridges toDelete;
+
+    timerTask->cancel();    // call prior to locking so maintenance visit can finish
     {
         Mutex::ScopedLock mutex(lock);
 
@@ -374,14 +399,13 @@ void Link::destroy ()
         for (Bridges::iterator i = created.begin(); i != created.end(); i++)
             toDelete.push_back(*i);
         created.clear();
-
-        timerTask->cancel();
     }
+
     // Now delete all bridges on this link (don't hold the lock for this).
     for (Bridges::iterator i = toDelete.begin(); i != toDelete.end(); i++)
-        (*i)->destroy();
+        (*i)->close();
     toDelete.clear();
-    links->destroy (configuredHost, configuredPort);
+    listener(this); // notify LinkRegistry that this Link has been destroyed
 }
 
 void Link::add(Bridge::shared_ptr bridge)
@@ -423,7 +447,7 @@ void Link::ioThreadProcessing()
 {
     Mutex::ScopedLock mutex(lock);
 
-    if (state != STATE_OPERATIONAL)
+    if (state != STATE_OPERATIONAL || closing)
         return;
 
     // check for bridge session errors and recover
@@ -460,7 +484,7 @@ void Link::ioThreadProcessing()
 void Link::maintenanceVisit ()
 {
     Mutex::ScopedLock mutex(lock);
-
+    if (closing) return;
     if (state == STATE_WAITING)
     {
         visitCount++;
@@ -476,21 +500,27 @@ void Link::maintenanceVisit ()
             }
         }
     }
-    else if (state == STATE_OPERATIONAL && (!active.empty() || !created.empty() || !cancellations.empty()) && connection != 0)
+    else if (state == STATE_OPERATIONAL &&
+             (!active.empty() || !created.empty() || !cancellations.empty()) &&
+             connection && connection->isOpen())
         connection->requestIOProcessing (boost::bind(&Link::ioThreadProcessing, this));
-    }
+}
 
 void Link::reconnectLH(const Address& a)
 {
     host = a.host;
     port = a.port;
     transport = a.protocol;
-    startConnectionLH();
+
     if (!hideManagement()) {
         stringstream errorString;
-        errorString << "Failed over to " << a;
+        errorString << "Failing over to " << a;
         mgmtObject->set_lastError(errorString.str());
+        mgmtObject->set_host(host);
+        mgmtObject->set_port(port);
+        mgmtObject->set_transport(transport);
     }
+    startConnectionLH();
 }
 
 bool Link::tryFailoverLH() {
@@ -499,15 +529,14 @@ bool Link::tryFailoverLH() {
     if (url.empty()) return false;
     Address next = url[reconnectNext++];
     if (next.host != host || next.port != port || next.protocol != transport) {
-        links->changeAddress(Address(transport, host, port), next);
-        QPID_LOG(debug, "Inter-broker link failing over to " << next.host << ":" << next.port);
+        QPID_LOG(notice, "Inter-broker link '" << name << "' failing over to " << next);
         reconnectLH(next);
         return true;
     }
     return false;
 }
 
-// Management updates for a linke are inconsistent in a cluster, so they are
+// Management updates for a link are inconsistent in a cluster, so they are
 // suppressed.
 bool Link::hideManagement() const {
     return !mgmtObject || ( broker && broker->isInCluster());
@@ -536,18 +565,34 @@ void Link::setPersistenceId(uint64_t id) const
 
 const string& Link::getName() const
 {
-    return configuredHost;
+    return name;
+}
+
+const std::string Link::ENCODED_IDENTIFIER("link.v2");
+const std::string Link::ENCODED_IDENTIFIER_V1("link");
+
+bool Link::isEncodedLink(const std::string& key)
+{
+    return key == ENCODED_IDENTIFIER || key == ENCODED_IDENTIFIER_V1;
 }
 
 Link::shared_ptr Link::decode(LinkRegistry& links, Buffer& buffer)
 {
+    string kind;
+    buffer.getShortString(kind);
+
     string   host;
     uint16_t port;
     string   transport;
     string   authMechanism;
     string   username;
     string   password;
+    string   name;
 
+    if (kind == ENCODED_IDENTIFIER) {
+        // newer version provides a link name.
+        buffer.getShortString(name);
+    }
     buffer.getShortString(host);
     port = buffer.getShort();
     buffer.getShortString(transport);
@@ -556,12 +601,21 @@ Link::shared_ptr Link::decode(LinkRegistry& links, Buffer& buffer)
     buffer.getShortString(username);
     buffer.getShortString(password);
 
-    return links.declare(host, port, transport, durable, authMechanism, username, password).first;
+    if (kind == ENCODED_IDENTIFIER_V1) {
+        /** previous versions identified the Link by host:port, there was no name
+         * assigned.  So create a name for the new Link.
+         */
+        name = createName(transport, host, port);
+    }
+
+    return links.declare(name, host, port, transport, durable, authMechanism,
+                         username, password).first;
 }
 
 void Link::encode(Buffer& buffer) const
 {
-    buffer.putShortString(string("link"));
+    buffer.putShortString(ENCODED_IDENTIFIER);
+    buffer.putShortString(name);
     buffer.putShortString(configuredHost);
     buffer.putShort(configuredPort);
     buffer.putShortString(configuredTransport);
@@ -573,8 +627,9 @@ void Link::encode(Buffer& buffer) const
 
 uint32_t Link::encodedSize() const
 {
-    return configuredHost.size() + 1 // short-string (host)
-        + 5                // short-string ("link")
+    return ENCODED_IDENTIFIER.size() + 1 // +1 byte length
+        + name.size() + 1
+        + configuredHost.size() + 1 // short-string (host)
         + 2                // port
         + configuredTransport.size() + 1 // short-string(transport)
         + 1                // durable
@@ -589,6 +644,7 @@ ManagementObject* Link::GetManagementObject (void) const
 }
 
 void Link::close() {
+    QPID_LOG(debug, "Link::close(), link=" << name );
     Mutex::ScopedLock mutex(lock);
     if (!closing) {
         closing = true;
@@ -609,36 +665,31 @@ Manageable::status_t Link::ManagementMethod (uint32_t op, Args& args, string& te
         return Manageable::STATUS_OK;
 
     case _qmf::Link::METHOD_BRIDGE :
+        /* TBD: deprecate this interface in favor of the Broker::create() method.  The
+         * Broker::create() method allows the user to assign a name to the bridge.
+         */
+        QPID_LOG(info, "The Link::bridge() method will be removed in a future release of QPID."
+                 " Please use the Broker::create() method with type='bridge' instead.");
         _qmf::ArgsLinkBridge& iargs = (_qmf::ArgsLinkBridge&) args;
-        QPID_LOG(debug, "Link::bridge() request received");
-
-        // Durable bridges are only valid on durable links
-        if (iargs.i_durable && !durable) {
-            text = "Can't create a durable route on a non-durable link";
-            return Manageable::STATUS_USER;
-        }
-
-        if (iargs.i_dynamic) {
-            Exchange::shared_ptr exchange = getBroker()->getExchanges().get(iargs.i_src);
-            if (exchange.get() == 0) {
-                text = "Exchange not found";
-                return Manageable::STATUS_USER;
-            }
-            if (!exchange->supportsDynamicBinding()) {
-                text = "Exchange type does not support dynamic routing";
-                return Manageable::STATUS_USER;
+        QPID_LOG(debug, "Link::bridge() request received; src=" << iargs.i_src <<
+                 "; dest=" << iargs.i_dest << "; key=" << iargs.i_key);
+
+        // Does a bridge already exist that has the src/dest/key?  If so, re-use the
+        // existing bridge - this behavior is backward compatible with previous releases.
+        Bridge::shared_ptr bridge = links->getBridge(*this, iargs.i_src, iargs.i_dest, iargs.i_key);
+        if (!bridge) {
+            // need to create a new bridge on this link.
+            std::pair<Bridge::shared_ptr, bool> rc =
+              links->declare( Bridge::createName(name, iargs.i_src, iargs.i_dest, iargs.i_key),
+                              *this, iargs.i_durable,
+                              iargs.i_src, iargs.i_dest, iargs.i_key, iargs.i_srcIsQueue,
+                              iargs.i_srcIsLocal, iargs.i_tag, iargs.i_excludes,
+                              iargs.i_dynamic, iargs.i_sync);
+            if (!rc.first) {
+                text = "invalid parameters";
+                return Manageable::STATUS_PARAMETER_INVALID;
             }
         }
-
-        std::pair<Bridge::shared_ptr, bool> result =
-            links->declare (configuredHost, configuredPort, iargs.i_durable, iargs.i_src,
-                            iargs.i_dest, iargs.i_key, iargs.i_srcIsQueue,
-                            iargs.i_srcIsLocal, iargs.i_tag, iargs.i_excludes,
-                            iargs.i_dynamic, iargs.i_sync);
-
-        if (result.second && iargs.i_durable)
-            store->create(*result.first);
-
         return Manageable::STATUS_OK;
     }
 
@@ -666,11 +717,13 @@ void Link::closeConnection( const std::string& reason)
 {
     if (connection != 0) {
         // cancel our subscription to the failover exchange
-        SessionHandler& sessionHandler = connection->getChannel(failoverChannel);
-        if (sessionHandler.getSession()) {
-            framing::AMQP_ServerProxy remoteBroker(sessionHandler.out);
-            remoteBroker.getMessage().cancel(failoverExchange->getName());
-            remoteBroker.getSession().detach(failoverSession);
+        if (failover) {
+            SessionHandler& sessionHandler = connection->getChannel(failoverChannel);
+            if (sessionHandler.getSession()) {
+                framing::AMQP_ServerProxy remoteBroker(sessionHandler.out);
+                remoteBroker.getMessage().cancel(failoverExchange->getName());
+                remoteBroker.getSession().detach(failoverSession);
+            }
         }
         connection->close(CLOSE_CODE_CONNECTION_FORCED, reason);
         connection = 0;
@@ -716,6 +769,23 @@ void Link::setState(const framing::FieldTable& state)
     }
 }
 
+std::string Link::createName(const std::string& transport,
+                             const std::string& host,
+                             uint16_t  port)
+{
+    stringstream linkName;
+    linkName << QPID_NAME_PREFIX << transport << std::string(":")
+             << host << std::string(":") << port;
+    return linkName.str();
+}
+
+
+bool Link::pendingConnection(const std::string& _host, uint16_t _port) const
+{
+    Mutex::ScopedLock mutex(lock);
+    return (isConnecting() && _port == port && _host == host);
+}
+
 
 const std::string Link::exchangeTypeName("qpid.LinkExchange");