QPID-4728: add 'credit' parameter to Federation Bridge configuration.

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

7 files changed, 93 insertions, 48 deletions

diff --git a/cpp/src/qpid/broker/Bridge.cpp b/cpp/src/qpid/broker/Bridge.cpp
index 68bdf6d474..ae83e4415b 100644
--- a/cpp/src/qpid/broker/Bridge.cpp
+++ b/cpp/src/qpid/broker/Bridge.cpp
@@ -54,6 +54,8 @@ namespace _qmf = qmf::org::apache::qpid::broker;
 namespace {
 const std::string QPID_REPLICATE("qpid.replicate");
 const std::string NONE("none");
+const uint8_t EXPLICIT_ACK(0); // msg.accept required to be sent
+const uint8_t IMPLIED_ACK(1);  // msg.accept assumed, not sent
 }
 
 namespace qpid {
@@ -76,12 +78,19 @@ Bridge::Bridge(const std::string& _name, Link* _link, framing::ChannelId _id,
     useExistingQueue(!_queueName.empty()),
     sessionName("qpid.bridge_session_" + name + "_" + link->getBroker()->getFederationTag())
 {
+    // If both acks (i_sync) and limited credit is configured, then we'd
+    // better be able to sync before running out of credit or we
+    // may stall (note: i_credit==0 means "unlimited")
+    if (args.i_credit && args.i_sync && args.i_sync > args.i_credit)
+        throw Exception("The credit value must be greater than configured sync (ack) interval.");
+
     ManagementAgent* agent = link->getBroker()->getManagementAgent();
     if (agent != 0) {
         mgmtObject = _qmf::Bridge::shared_ptr(new _qmf::Bridge
             (agent, this, link, name, args.i_durable, args.i_src, args.i_dest,
              args.i_key, args.i_srcIsQueue, args.i_srcIsLocal,
-             args.i_tag, args.i_excludes, args.i_dynamic, args.i_sync));
+             args.i_tag, args.i_excludes, args.i_dynamic, args.i_sync,
+             args.i_credit));
         mgmtObject->set_channelId(channel);
         agent->addObject(mgmtObject);
     }
@@ -98,8 +107,7 @@ void Bridge::create(Connection& c)
     detached = false;           // Reset detached in case we are recovering.
     connState = &c;
     conn = &c;
-    FieldTable options;
-    if (args.i_sync) options.setInt("qpid.sync_frequency", args.i_sync);
+
     SessionHandler& sessionHandler = c.getChannel(channel);
     sessionHandler.setErrorListener(shared_from_this());
     if (args.i_srcIsLocal) {
@@ -121,51 +129,75 @@ void Bridge::create(Connection& c)
     }
 
     if (args.i_srcIsLocal) sessionHandler.getSession()->disableReceiverTracking();
-    if (initialize) initialize(*this, sessionHandler);
-    else if (args.i_srcIsQueue) {
-        peer->getMessage().subscribe(args.i_src, args.i_dest, args.i_sync ? 0 : 1, 0, false, "", 0, options);
-        peer->getMessage().flow(args.i_dest, 0, args.i_sync ? 2 * args.i_sync : 0xFFFFFFFF);
-        peer->getMessage().flow(args.i_dest, 1, 0xFFFFFFFF);
-        QPID_LOG(debug, "Activated bridge " << name << " for route from queue " << args.i_src << " to " << args.i_dest);
-    } else {
-        if (!useExistingQueue) {
-            FieldTable queueSettings;
 
-            if (args.i_tag.size()) {
-                queueSettings.setString("qpid.trace.id", args.i_tag);
-            } else {
-                const string& peerTag = c.getFederationPeerTag();
-                if (peerTag.size())
-                    queueSettings.setString("qpid.trace.id", peerTag);
+    if (initialize) {
+        initialize(*this, sessionHandler);  // custom subscription initializer supplied
+    } else {
+        // will a temp queue be created for this bridge?
+        const bool temp_queue = !args.i_srcIsQueue && !useExistingQueue;
+        // UI convention: user specifies 0 for infinite credit
+        const uint32_t credit = (args.i_credit == 0) ? LinkRegistry::INFINITE_CREDIT : args.i_credit;
+        // use explicit acks only for non-temp queues, useless for temp queues since they are
+        // destroyed when the session drops (can't resend unacked msgs)
+        const uint8_t ack_mode = (args.i_sync && !temp_queue) ? EXPLICIT_ACK : IMPLIED_ACK;
+
+        // configure command.sync frequency
+        FieldTable options;
+        uint32_t freq = 0;
+        if (ack_mode == EXPLICIT_ACK) {  // user explicitly configured syncs
+            freq = uint32_t(args.i_sync);
+        } else if (credit && credit != LinkRegistry::INFINITE_CREDIT) {
+            // force occasional sync to keep from stalling due to lack of credit
+            freq = (credit + 1)/2;
+        }
+        if (freq)
+            options.setInt("qpid.sync_frequency", freq);
+
+        // create a subscription on the remote
+        if (args.i_srcIsQueue) {
+            peer->getMessage().subscribe(args.i_src, args.i_dest, ack_mode, 0, false, "", 0, options);
+            peer->getMessage().flow(args.i_dest, 0, credit);  // message credit
+            peer->getMessage().flow(args.i_dest, 1, LinkRegistry::INFINITE_CREDIT);     // byte credit
+            QPID_LOG(debug, "Activated bridge " << name << " for route from queue " << args.i_src << " to " << args.i_dest);
+        } else {
+            if (!useExistingQueue) {
+                FieldTable queueSettings;
+
+                if (args.i_tag.size()) {
+                    queueSettings.setString("qpid.trace.id", args.i_tag);
+                } else {
+                    const string& peerTag = c.getFederationPeerTag();
+                    if (peerTag.size())
+                        queueSettings.setString("qpid.trace.id", peerTag);
+                }
+
+                if (args.i_excludes.size()) {
+                    queueSettings.setString("qpid.trace.exclude", args.i_excludes);
+                } else {
+                    const string& localTag = link->getBroker()->getFederationTag();
+                    if (localTag.size())
+                        queueSettings.setString("qpid.trace.exclude", localTag);
+                }
+
+                bool durable = false;//should this be an arg, or would we use srcIsQueue for durable queues?
+                bool exclusive = true;  // only exclusive if the queue is owned by the bridge
+                bool autoDelete = exclusive && !durable;//auto delete transient queues?
+                peer->getQueue().declare(queueName, altEx, false, durable, exclusive, autoDelete, queueSettings);
             }
-
-            if (args.i_excludes.size()) {
-                queueSettings.setString("qpid.trace.exclude", args.i_excludes);
+            if (!args.i_dynamic)
+                peer->getExchange().bind(queueName, args.i_src, args.i_key, FieldTable());
+            peer->getMessage().subscribe(queueName, args.i_dest, ack_mode, 0, false, "", 0, options);
+            peer->getMessage().flow(args.i_dest, 0, credit);
+            peer->getMessage().flow(args.i_dest, 1, LinkRegistry::INFINITE_CREDIT);
+            if (args.i_dynamic) {
+                Exchange::shared_ptr exchange = link->getBroker()->getExchanges().get(args.i_src);
+                if (exchange.get() == 0)
+                    throw Exception("Exchange not found for dynamic route");
+                exchange->registerDynamicBridge(this);
+                QPID_LOG(debug, "Activated bridge " << name << " for dynamic route for exchange " << args.i_src);
             } else {
-                const string& localTag = link->getBroker()->getFederationTag();
-                if (localTag.size())
-                    queueSettings.setString("qpid.trace.exclude", localTag);
+                QPID_LOG(debug, "Activated bridge " << name << " for static route from exchange " << args.i_src << " to " << args.i_dest);
             }
-
-            bool durable = false;//should this be an arg, or would we use srcIsQueue for durable queues?
-            bool exclusive = true;  // only exclusive if the queue is owned by the bridge
-            bool autoDelete = exclusive && !durable;//auto delete transient queues?
-            peer->getQueue().declare(queueName, altEx, false, durable, exclusive, autoDelete, queueSettings);
-        }
-        if (!args.i_dynamic)
-            peer->getExchange().bind(queueName, args.i_src, args.i_key, FieldTable());
-        peer->getMessage().subscribe(queueName, args.i_dest, (useExistingQueue && args.i_sync) ? 0 : 1, 0, false, "", 0, options);
-        peer->getMessage().flow(args.i_dest, 0, (useExistingQueue && args.i_sync) ? 2 * args.i_sync : 0xFFFFFFFF);
-        peer->getMessage().flow(args.i_dest, 1, 0xFFFFFFFF);
-
-        if (args.i_dynamic) {
-            Exchange::shared_ptr exchange = link->getBroker()->getExchanges().get(args.i_src);
-            if (exchange.get() == 0)
-                throw Exception("Exchange not found for dynamic route");
-            exchange->registerDynamicBridge(this);
-            QPID_LOG(debug, "Activated bridge " << name << " for dynamic route for exchange " << args.i_src);
-        } else {
-            QPID_LOG(debug, "Activated bridge " << name << " for static route from exchange " << args.i_src << " to " << args.i_dest);
         }
     }
     if (args.i_srcIsLocal) sessionHandler.getSession()->enableReceiverTracking();
@@ -260,6 +292,7 @@ Bridge::shared_ptr Bridge::decode(LinkRegistry& links, Buffer& buffer)
     buffer.getShortString(excludes);
     bool dynamic(buffer.getOctet());
     uint16_t sync = buffer.getShort();
+    uint32_t credit = buffer.getLong();
 
     if (kind == ENCODED_IDENTIFIER_V1) {
         /** previous versions did not provide a name for the bridge, so create one
@@ -268,7 +301,7 @@ Bridge::shared_ptr Bridge::decode(LinkRegistry& links, Buffer& buffer)
     }
 
     return links.declare(name, *link, durable, src, dest, key, is_queue,
-                         is_local, id, excludes, dynamic, sync).first;
+                         is_local, id, excludes, dynamic, sync, credit).first;
 }
 
 void Bridge::encode(Buffer& buffer) const
@@ -286,6 +319,7 @@ void Bridge::encode(Buffer& buffer) const
     buffer.putShortString(args.i_excludes);
     buffer.putOctet(args.i_dynamic ? 1 : 0);
     buffer.putShort(args.i_sync);
+    buffer.putLong(args.i_credit);
 }
 
 uint32_t Bridge::encodedSize() const
@@ -302,7 +336,8 @@ uint32_t Bridge::encodedSize() const
         + args.i_tag.size() + 1
         + args.i_excludes.size() + 1
         + 1               // dynamic
-        + 2;              // sync
+        + 2               // sync
+        + 4;              // credit
 }
 
 management::ManagementObject::shared_ptr Bridge::GetManagementObject(void) const
diff --git a/cpp/src/qpid/broker/Broker.cpp b/cpp/src/qpid/broker/Broker.cpp
index 098ffe34c4..ab5398a0e8 100644
--- a/cpp/src/qpid/broker/Broker.cpp
+++ b/cpp/src/qpid/broker/Broker.cpp
@@ -624,6 +624,7 @@ const std::string SRC_IS_QUEUE("srcIsQueue");
 const std::string SRC_IS_LOCAL("srcIsLocal");
 const std::string DYNAMIC("dynamic");
 const std::string SYNC("sync");
+const std::string CREDIT("credit");
 
 // parameters for deleting a Queue object
 const std::string IF_EMPTY("if_empty");
@@ -840,6 +841,7 @@ void Broker::createObject(const std::string& type, const std::string& name,
         bool srcIsLocal = false;
         bool dynamic = false;
         uint16_t sync = 0;
+        uint32_t credit = LinkRegistry::INFINITE_CREDIT;
 
         for (Variant::Map::const_iterator i = properties.begin(); i != properties.end(); ++i) {
 
@@ -853,6 +855,7 @@ void Broker::createObject(const std::string& type, const std::string& name,
             else if (i->first == SRC_IS_LOCAL) srcIsLocal = bool(i->second);
             else if (i->first == DYNAMIC) dynamic = bool(i->second);
             else if (i->first == SYNC) sync = i->second.asUint16();
+            else if (i->first == CREDIT) credit = i->second.asUint32();
             else if (i->first == DURABLE) durable = bool(i->second);
             else if (i->first == QUEUE_NAME) queueName = i->second.asString();
             else {
@@ -867,7 +870,7 @@ void Broker::createObject(const std::string& type, const std::string& name,
         }
         std::pair<Bridge::shared_ptr, bool> rc =
           links.declare(name, *link, durable, src, dest, key, srcIsQueue, srcIsLocal, id, excludes,
-                        dynamic, sync,
+                        dynamic, sync, credit,
                         0,
                         queueName);
 
diff --git a/cpp/src/qpid/broker/Link.cpp b/cpp/src/qpid/broker/Link.cpp
index 5d01a567b5..3380708c0e 100644
--- a/cpp/src/qpid/broker/Link.cpp
+++ b/cpp/src/qpid/broker/Link.cpp
@@ -733,7 +733,7 @@ Manageable::status_t Link::ManagementMethod (uint32_t op, Args& args, string& te
                               *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);
+                              iargs.i_dynamic, iargs.i_sync, iargs.i_credit);
             if (!rc.first) {
                 text = "invalid parameters";
                 return Manageable::STATUS_PARAMETER_INVALID;
diff --git a/cpp/src/qpid/broker/LinkRegistry.cpp b/cpp/src/qpid/broker/LinkRegistry.cpp
index 5af6053943..8642294d06 100644
--- a/cpp/src/qpid/broker/LinkRegistry.cpp
+++ b/cpp/src/qpid/broker/LinkRegistry.cpp
@@ -168,6 +168,7 @@ pair<Bridge::shared_ptr, bool> LinkRegistry::declare(const std::string& name,
                                                      const std::string& excludes,
                                                      bool         dynamic,
                                                      uint16_t     sync,
+                                                     uint32_t     credit,
                                                      Bridge::InitializeCallback init,
                                                      const std::string& queueName,
                                                      const std::string& altExchange
@@ -209,6 +210,7 @@ pair<Bridge::shared_ptr, bool> LinkRegistry::declare(const std::string& name,
         args.i_excludes   = excludes;
         args.i_dynamic    = dynamic;
         args.i_sync       = sync;
+        args.i_credit     = credit;
 
         bridge = Bridge::shared_ptr
           (new Bridge (name, &link, link.nextChannel(),
diff --git a/cpp/src/qpid/broker/LinkRegistry.h b/cpp/src/qpid/broker/LinkRegistry.h
index 21e8ddec81..e5b1c40781 100644
--- a/cpp/src/qpid/broker/LinkRegistry.h
+++ b/cpp/src/qpid/broker/LinkRegistry.h
@@ -108,10 +108,13 @@ namespace broker {
                 const std::string& excludes,
                 bool         dynamic,
                 uint16_t     sync,
+                uint32_t     credit,
                 Bridge::InitializeCallback=0,
                 const std::string& queueName="",
                 const std::string& altExchange=""
         );
+        QPID_BROKER_EXTERN static const uint32_t INFINITE_CREDIT = 0xFFFFFFFF;
+
         /** determine if Bridge exists */
         QPID_BROKER_EXTERN Bridge::shared_ptr
           getBridge(const std::string&  name);
diff --git a/cpp/src/qpid/ha/BrokerReplicator.cpp b/cpp/src/qpid/ha/BrokerReplicator.cpp
index 42e54a0125..076bcac63f 100644
--- a/cpp/src/qpid/ha/BrokerReplicator.cpp
+++ b/cpp/src/qpid/ha/BrokerReplicator.cpp
@@ -326,6 +326,7 @@ void BrokerReplicator::initialize() {
         "",                 // excludes
         false,              // dynamic
         0,                  // sync?
+        LinkRegistry::INFINITE_CREDIT,
         // shared_ptr keeps this in memory until outstanding connected
         // calls are run.
         boost::bind(&BrokerReplicator::connected, shared_from_this(), _1, _2)
diff --git a/cpp/src/qpid/ha/QueueReplicator.cpp b/cpp/src/qpid/ha/QueueReplicator.cpp
index 6fe49bc1af..dece9dd045 100644
--- a/cpp/src/qpid/ha/QueueReplicator.cpp
+++ b/cpp/src/qpid/ha/QueueReplicator.cpp
@@ -137,6 +137,7 @@ void QueueReplicator::activate() {
         "",                 // excludes
         false,              // dynamic
         0,                  // sync?
+        LinkRegistry::INFINITE_CREDIT,
         // Include shared_ptr to self to ensure we are not deleted
         // before initializeBridge is called.
         boost::bind(&QueueReplicator::initializeBridge, shared_from_this(), _1, _2)