Attachment 617569 Details for Bug 860704 – Quick patch to work-around the discovered issues

[patch] Quick patch to work-around the discovered issues

qpid-4285.patch (text/plain), 6.27 KB, created by Jason Dillaman on 2012-09-26 14:25:28 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Jason Dillaman

Created: 2012-09-26 14:25:28 UTC

Size: 6.27 KB

patch

obsolete

>diff --git a/qpid/cpp/src/qpid/broker/Link.cpp b/qpid/cpp/src/qpid/broker/Link.cpp
>index 157b75c..eb38795 100644
>--- a/qpid/cpp/src/qpid/broker/Link.cpp
>+++ b/qpid/cpp/src/qpid/broker/Link.cpp
>@@ -367,6 +367,7 @@ void Link::closed(int, std::string text)
>         created.push_back(*i);
>     }
>     active.clear();
>+    cancellations.clear();
> 
>     if (state != STATE_FAILED && state != STATE_PASSIVE)
>     {
>@@ -457,9 +458,14 @@ void Link::ioThreadProcessing()
>             active.begin(), active.end(), boost::bind(&Bridge::isDetached, _1));
>         for (Bridges::iterator i = removed; i != active.end(); ++i) {
>             Bridge::shared_ptr  bridge = *i;
>-            bridge->closed();
>-            bridge->cancel(*connection);
>             created.push_back(bridge);
>+            try {
>+                bridge->closed();
>+                bridge->cancel(*connection);
>+            } catch(const std::exception& e) {
>+                QPID_LOG(error, "Cleanup of bridge '" << bridge->getName() << "' failed: "
>+                         << e.what());
>+            }
>         }
>         active.erase(removed, active.end());
>     }
>@@ -475,8 +481,14 @@ void Link::ioThreadProcessing()
>     }
>     if (!created.empty()) {
>         for (Bridges::iterator i = created.begin(); i != created.end(); ++i) {
>-            active.push_back(*i);
>-            (*i)->create(*connection);
>+            Bridge::shared_ptr bridge = *i;
>+            active.push_back(bridge);
>+            try {
>+                bridge->create(*connection);
>+            } catch(const std::exception& e) {
>+                QPID_LOG(error, "Creation of bridge '" << bridge->getName() << "' failed: "
>+                         << e.what());
>+            }
>         }
>         created.clear();
>     }
>diff --git a/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp b/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
>index b73f8a0..0c6d791 100644
>--- a/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
>+++ b/qpid/cpp/src/qpid/ha/BrokerReplicator.cpp
>@@ -238,6 +238,8 @@ void BrokerReplicator::initializeBridge(Bridge& bridge, SessionHandler& sessionH
>     peer.getMessage().subscribe(queueName, args.i_dest, 1, 0, false, "", 0, FieldTable());
>     peer.getMessage().flow(args.i_dest, 0, 0xFFFFFFFF);
>     peer.getMessage().flow(args.i_dest, 1, 0xFFFFFFFF);
>+    
>+    broker.getExchanges().eachExchange(boost::bind(&BrokerReplicator::recordQueueReplicator, this, _1));
> 
>     // Issue a query request for queues, exchanges, bindings and the habroker
>     // using event queue as the reply-to address
>@@ -292,9 +294,15 @@ void BrokerReplicator::route(Deliverable& msg) {
>                 else if (type == BINDING) doResponseBind(values);
>                 else if (type == HA_BROKER) doResponseHaBroker(values);
>             }
>-            if (messageProperties->getCorrelationId() == EXCHANGE && !headers->isSet(PARTIAL)) {
>-                // We have received all of the exchange response.
>-                alternates.clear();
>+            if (!headers->isSet(PARTIAL)) {
>+                if (messageProperties->getCorrelationId() == QUEUE) {
>+                    // Stop any queue replicators for queues that no longer exist in the primary
>+                    stopQueueReplicators();
>+                }
>+                else if (messageProperties->getCorrelationId() == EXCHANGE) {
>+                    // We have received all of the exchange response.
>+                    alternates.clear();
>+                }
>             }
>         }
>     } catch (const std::exception& e) {
>@@ -453,6 +461,7 @@ void BrokerReplicator::doResponseQueue(Variant::Map& values) {
>         return;
>     string name(values[NAME].asString());
>     QPID_LOG(debug, logPrefix << "Queue response: " << name);
>+    queueReplicatorNames.erase(name);
>     framing::FieldTable args;
>     amqp_0_10::translate(argsMap, args);
>     boost::shared_ptr<Queue> queue =
>@@ -561,6 +570,17 @@ void BrokerReplicator::stopQueueReplicator(const std::string& name) {
>     }
> }
> 
>+void BrokerReplicator::stopQueueReplicators() {
>+    std::for_each(queueReplicatorNames.begin(), queueReplicatorNames.end(), boost::bind(&BrokerReplicator::stopQueueReplicator, this, _1));
>+    queueReplicatorNames.clear();
>+}
>+
>+void BrokerReplicator::recordQueueReplicator(Exchange::shared_ptr ex) {
>+    if (boost::dynamic_pointer_cast<QueueReplicator>(ex)) {
>+        queueReplicatorNames.insert(ex->getName());
>+    }
>+}
>+
> boost::shared_ptr<Queue> BrokerReplicator::createQueue(
>     const std::string& name,
>     bool durable,
>diff --git a/qpid/cpp/src/qpid/ha/BrokerReplicator.h b/qpid/cpp/src/qpid/ha/BrokerReplicator.h
>index 69653b8..269113a 100644
>--- a/qpid/cpp/src/qpid/ha/BrokerReplicator.h
>+++ b/qpid/cpp/src/qpid/ha/BrokerReplicator.h
>@@ -31,6 +31,7 @@
> #include "qpid/management/ManagementObject.h"
> #include <boost/shared_ptr.hpp>
> #include <boost/enable_shared_from_this.hpp>
>+#include <set>
> 
> namespace qpid {
> 
>@@ -78,6 +79,7 @@ class BrokerReplicator : public broker::Exchange,
> 
>   private:
>     typedef boost::shared_ptr<QueueReplicator> QueueReplicatorPtr;
>+    typedef std::set<std::string> QueueReplicatorNames;
> 
>     void initializeBridge(broker::Bridge&, broker::SessionHandler&);
> 
>@@ -98,6 +100,9 @@ class BrokerReplicator : public broker::Exchange,
>     void startQueueReplicator(const boost::shared_ptr<broker::Queue>&);
>     void stopQueueReplicator(const std::string& name);
> 
>+    void stopQueueReplicators();
>+    void recordQueueReplicator(broker::Exchange::shared_ptr);
>+
>     boost::shared_ptr<broker::Queue> createQueue(
>         const std::string& name,
>         bool durable,
>@@ -120,6 +125,7 @@ class BrokerReplicator : public broker::Exchange,
>     boost::shared_ptr<broker::Link> link;
>     bool initialized;
>     AlternateExchangeSetter alternates;
>+    QueueReplicatorNames queueReplicatorNames;
>     qpid::Address primary;
> };
> }} // namespace qpid::broker
>diff --git a/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp b/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp
>index 93cd2fa..75d4fba 100644
>--- a/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp
>+++ b/qpid/cpp/src/qpid/ha/ReplicatingSubscription.cpp
>@@ -105,6 +105,7 @@ class GetPositionConsumer : public Consumer
>     void acknowledged(const broker::QueuedMessage&) {}
>     bool browseAcquired() const { return true; }
>     broker::OwnershipToken* getSession() { return 0; }
>+    bool hideDeletedError() { return true; }
> };
> 
>

Actions: View | Diff

Attachments on bug 860704: 617569