Attachment 641618 Details for Bug 873883 – Patch to auto-unbind propagated bindings when destination broker disconnects

[patch] Patch to auto-unbind propagated bindings when destination broker disconnects

bz873883.patch (text/plain), 6.03 KB, created by Jason Dillaman on 2012-11-09 15:19:38 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Jason Dillaman

Created: 2012-11-09 15:19:38 UTC

Size: 6.03 KB

patch

obsolete

>diff --git a/qpid/cpp/src/qpid/broker/Bridge.cpp b/qpid/cpp/src/qpid/broker/Bridge.cpp
>index 7d73460..c6403c2 100644
>--- a/qpid/cpp/src/qpid/broker/Bridge.cpp
>+++ b/qpid/cpp/src/qpid/broker/Bridge.cpp
>@@ -331,6 +331,7 @@ void Bridge::propagateBinding(const string& key, const string& tagList,
>          }
>          string newTagList(tagList + string(tagList.empty() ? "" : ",") + localTag);
> 
>+         bindArgs.setString("x-scope", "session");
>          bindArgs.setString(qpidFedOp, op);
>          bindArgs.setString(qpidFedTags, newTagList);
>          if (origin.empty())
>diff --git a/qpid/cpp/src/qpid/broker/SemanticState.cpp b/qpid/cpp/src/qpid/broker/SemanticState.cpp
>index 3f35758..a3c02e5 100644
>--- a/qpid/cpp/src/qpid/broker/SemanticState.cpp
>+++ b/qpid/cpp/src/qpid/broker/SemanticState.cpp
>@@ -41,6 +41,7 @@
> 
> #include <boost/bind.hpp>
> #include <boost/format.hpp>
>+#include <boost/tuple/tuple_comparison.hpp>
> 
> #include <iostream>
> #include <sstream>
>@@ -96,6 +97,12 @@ void SemanticState::closed() {
>         for (ConsumerImplMap::iterator i = consumers.begin(); i != consumers.end(); i++) {
>             cancel(i->second);
>         }
>+        
>+        //unbind session-scoped bindings
>+        for (Bindings::iterator i = bindings.begin(); i != bindings.end(); i++) {
>+            session.getBroker().unbind(i->get<0>(), i->get<1>(), i->get<2>(),
>+                                       userID, connectionId);
>+        }
>         closeComplete = true;
>     }
> }
>@@ -845,4 +852,16 @@ void SemanticState::detached()
>     }
> }
> 
>+void SemanticState::addBinding(const string& queueName, const string& exchangeName, 
>+                               const string& routingKey)
>+{
>+    bindings.insert(boost::make_tuple(queueName, exchangeName, routingKey));
>+}
>+
>+void SemanticState::removeBinding(const string& queueName, const string& exchangeName, 
>+                                  const string& routingKey)
>+{
>+    bindings.erase(boost::make_tuple(queueName, exchangeName, routingKey));
>+}
>+
> }} // namespace qpid::broker
>diff --git a/qpid/cpp/src/qpid/broker/SemanticState.h b/qpid/cpp/src/qpid/broker/SemanticState.h
>index 7b3a873..36830c2 100644
>--- a/qpid/cpp/src/qpid/broker/SemanticState.h
>+++ b/qpid/cpp/src/qpid/broker/SemanticState.h
>@@ -49,6 +49,7 @@
> #include <boost/enable_shared_from_this.hpp>
> #include <boost/intrusive_ptr.hpp>
> #include <boost/cast.hpp>
>+#include <boost/tuple/tuple.hpp>
> 
> namespace qpid {
> namespace broker {
>@@ -167,6 +168,8 @@ class SemanticState : private boost::noncopyable {
> 
>   private:
>     typedef std::map<std::string, ConsumerImpl::shared_ptr> ConsumerImplMap;
>+    typedef boost::tuple<std::string, std::string, std::string> Binding;
>+    typedef std::set<Binding> Bindings;
> 
>     SessionContext& session;
>     DeliveryAdapter& deliveryAdapter;
>@@ -185,6 +188,8 @@ class SemanticState : private boost::noncopyable {
>     //needed for queue delete events in auto-delete:
>     const std::string connectionId;
> 
>+    Bindings bindings;
>+
>     void route(boost::intrusive_ptr<Message> msg, Deliverable& strategy);
>     void checkDtxTimeout();
> 
>@@ -267,6 +272,11 @@ class SemanticState : private boost::noncopyable {
>     void setAccumulatedAck(const framing::SequenceSet& s) { accumulatedAck = s; }
>     void record(const DeliveryRecord& delivery);
>     DtxBufferMap& getSuspendedXids() { return suspendedXids; }
>+    
>+    void addBinding(const std::string& queueName, const std::string& exchangeName, 
>+                   const std::string& routingKey);
>+    void removeBinding(const std::string& queueName, const std::string& exchangeName, 
>+                      const std::string& routingKey);
> };
> 
> }} // namespace qpid::broker
>diff --git a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
>index ec4e02e..3976885 100644
>--- a/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
>+++ b/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
>@@ -38,6 +38,11 @@
> #include <boost/cast.hpp>
> #include <boost/bind.hpp>
> 
>+namespace {
>+const std::string X_SCOPE("x-scope");
>+const std::string SESSION("session");
>+}
>+
> namespace qpid {
> namespace broker {
> 
>@@ -165,12 +170,16 @@ void SessionAdapter::ExchangeHandlerImpl::bind(const string& queueName,
> {
>     getBroker().bind(queueName, exchangeName, routingKey, arguments,
>                      getConnection().getUserId(), getConnection().getUrl());
>+    if (arguments.getAsString(X_SCOPE) == SESSION) {
>+        state.addBinding(queueName, exchangeName, routingKey);
>+    }
> }
> 
> void SessionAdapter::ExchangeHandlerImpl::unbind(const string& queueName,
>                                                  const string& exchangeName,
>                                                  const string& routingKey)
> {
>+    state.removeBinding(queueName, exchangeName, routingKey);
>     getBroker().unbind(queueName, exchangeName, routingKey,
>                        getConnection().getUserId(), getConnection().getUrl());
> }
>diff --git a/qpid/cpp/src/tests/MessagingSessionTests.cpp b/qpid/cpp/src/tests/MessagingSessionTests.cpp
>index edb50fc..3d5a606 100644
>--- a/qpid/cpp/src/tests/MessagingSessionTests.cpp
>+++ b/qpid/cpp/src/tests/MessagingSessionTests.cpp
>@@ -1196,6 +1196,27 @@ QPID_AUTO_TEST_CASE(testBrowseOnly)
>     fix.session.acknowledge();
> }
> 
>+QPID_AUTO_TEST_CASE(testLinkBindingCleanup)
>+{
>+    MessagingFixture fix;
>+
>+    Sender sender = fix.session.createSender("test.ex;{create:always,node:{type:topic}}");
>+
>+    Connection connection = fix.newConnection();
>+    connection.open();
>+
>+    Session session(connection.createSession());
>+	Receiver receiver1 = session.createReceiver("test.q;{create:always, node:{type:queue},link:{x-bindings:[{exchange:test.ex,queue:test.q,key:#,arguments:{x-scope:session}}]}}");
>+    Receiver receiver2 = fix.session.createReceiver("test.q;{create:never, delete:always}");
>+    connection.close();
>+
>+    sender.send(Message("test-message"), true);
>+
>+    // The session-scoped binding should be removed when receiver1's network connection is lost
>+    Message in;
>+    BOOST_CHECK(!receiver2.fetch(in, Duration::IMMEDIATE));
>+}
>+
> QPID_AUTO_TEST_SUITE_END()
> 
> }} // namespace qpid::tests

Actions: View | Diff

Attachments on bug 873883: 641618 | 642320