You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2010/01/22 11:58:21 UTC
svn commit: r902055 - in /qpid/trunk/qpid: cpp/src/ cpp/src/qpid/amqp_0_10/
cpp/src/qpid/broker/ cpp/src/qpid/client/ cpp/src/tests/ python/qpid/tests/
python/tests_0-10/
Author: gsim
Date: Fri Jan 22 10:58:20 2010
New Revision: 902055
URL: http://svn.apache.org/viewvc?rev=902055&view=rev
Log:
QPID-2347: Signal deletion of queue to active subscribers via a resource-deleted exception.
Added:
qpid/trunk/qpid/cpp/src/qpid/broker/SessionOutputException.h
Modified:
qpid/trunk/qpid/cpp/src/Makefile.am
qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.h
qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h
qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp
qpid/trunk/qpid/python/qpid/tests/messaging.py
qpid/trunk/qpid/python/tests_0-10/exchange.py
Modified: qpid/trunk/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/Makefile.am?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/Makefile.am Fri Jan 22 10:58:20 2010
@@ -610,6 +610,7 @@
qpid/broker/SessionManager.cpp \
qpid/broker/SessionManager.h \
qpid/broker/SessionManager.h \
+ qpid/broker/SessionOutputException.h \
qpid/broker/SessionState.cpp \
qpid/broker/SessionState.h \
qpid/broker/SignalHandler.cpp \
Modified: qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.cpp Fri Jan 22 10:58:20 2010
@@ -119,6 +119,16 @@
}
}
+void SessionHandler::handleException(const qpid::SessionException& e)
+{
+ QPID_LOG(error, "Execution exception (during output): " << e.what());
+ executionException(e.code, e.what()); // Let subclass handle this first.
+ framing::AMQP_AllProxy::Execution execution(channel);
+ execution.exception(e.code, 0, 0, 0, 0, e.what(), FieldTable());
+ detaching();
+ sendDetach();
+}
+
namespace {
bool isControl(const AMQFrame& f) {
return f.getMethod() && f.getMethod()->type() == framing::SEGMENT_TYPE_CONTROL;
Modified: qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.h?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/amqp_0_10/SessionHandler.h Fri Jan 22 10:58:20 2010
@@ -30,6 +30,7 @@
namespace qpid {
+struct SessionException;
namespace amqp_0_10 {
@@ -62,6 +63,7 @@
QPID_COMMON_EXTERN void sendTimeout(uint32_t t);
QPID_COMMON_EXTERN void sendFlush();
QPID_COMMON_EXTERN void markReadyToSend();//TODO: only needed for inter-broker bridge; cleanup
+ QPID_COMMON_EXTERN void handleException(const qpid::SessionException& e);
/** True if the handler is ready to send and receive */
bool ready() const;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Connection.cpp Fri Jan 22 10:58:20 2010
@@ -19,6 +19,7 @@
*
*/
#include "qpid/broker/Connection.h"
+#include "qpid/broker/SessionOutputException.h"
#include "qpid/broker/SessionState.h"
#include "qpid/broker/Bridge.h"
#include "qpid/broker/Broker.h"
@@ -278,6 +279,9 @@
//then do other output as needed:
return outputTasks.doOutput();
}
+ }catch(const SessionOutputException& e){
+ getChannel(e.channel).handleException(e);
+ return true;
}catch(ConnectionException& e){
close(e.code, e.getMessage());
}catch(std::exception& e){
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Fri Jan 22 10:58:20 2010
@@ -101,7 +101,8 @@
eventMode(0),
eventMgr(0),
insertSeqNo(0),
- broker(b)
+ broker(b),
+ deleted(false)
{
if (parent != 0 && broker != 0)
{
@@ -291,6 +292,7 @@
bool Queue::getNextMessage(QueuedMessage& m, Consumer::shared_ptr c)
{
+ checkNotDeleted();
if (c->preAcquires()) {
switch (consumeNextMessage(m, c)) {
case CONSUMED:
@@ -869,6 +871,17 @@
}
}
+void Queue::notifyDeleted()
+{
+ QueueListeners::ListenerSet set;
+ {
+ Mutex::ScopedLock locker(messageLock);
+ listeners.snapshot(set);
+ deleted = true;
+ }
+ set.notifyAll();
+}
+
void Queue::bound(const string& exchange, const string& key,
const FieldTable& args)
{
@@ -1102,3 +1115,10 @@
}
QueueListeners& Queue::getListeners() { return listeners; }
+
+void Queue::checkNotDeleted()
+{
+ if (deleted) {
+ throw ResourceDeletedException(QPID_MSG("Queue " << getName() << " has been deleted."));
+ }
+}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Fri Jan 22 10:58:20 2010
@@ -107,6 +107,7 @@
bool insertSeqNo;
std::string seqNoKey;
Broker* broker;
+ bool deleted;
void push(boost::intrusive_ptr<Message>& msg, bool isRecovery=false);
void setPolicy(std::auto_ptr<QueuePolicy> policy);
@@ -150,6 +151,7 @@
}
Messages::iterator findAt(framing::SequenceNumber pos);
+ void checkNotDeleted();
public:
@@ -173,6 +175,7 @@
QPID_BROKER_EXTERN void configure(const qpid::framing::FieldTable& settings,
bool recovering = false);
void destroy();
+ void notifyDeleted();
QPID_BROKER_EXTERN void bound(const string& exchange,
const string& key,
const qpid::framing::FieldTable& args);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.cpp Fri Jan 22 10:58:20 2010
@@ -78,4 +78,15 @@
std::find(consumers.begin(), consumers.end(), c) != consumers.end();
}
+void QueueListeners::ListenerSet::notifyAll()
+{
+ std::for_each(listeners.begin(), listeners.end(), boost::mem_fn(&Consumer::notify));
+}
+
+void QueueListeners::snapshot(ListenerSet& set)
+{
+ set.listeners.insert(set.listeners.end(), consumers.begin(), consumers.end());
+ set.listeners.insert(set.listeners.end(), browsers.begin(), browsers.end());
+}
+
}} // namespace qpid::broker
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueListeners.h Fri Jan 22 10:58:20 2010
@@ -52,10 +52,21 @@
friend class QueueListeners;
};
+ class ListenerSet
+ {
+ public:
+ void notifyAll();
+ private:
+ Listeners listeners;
+ friend class QueueListeners;
+ };
+
void addListener(Consumer::shared_ptr);
void removeListener(Consumer::shared_ptr);
void populate(NotificationSet&);
+ void snapshot(ListenerSet&);
bool contains(Consumer::shared_ptr c) const;
+ void notifyAll();
template <class F> void eachListener(F f) {
std::for_each(browsers.begin(), browsers.end(), f);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SemanticState.cpp Fri Jan 22 10:58:20 2010
@@ -27,6 +27,7 @@
#include "qpid/broker/Message.h"
#include "qpid/broker/Queue.h"
#include "qpid/broker/SessionContext.h"
+#include "qpid/broker/SessionOutputException.h"
#include "qpid/broker/TxAccept.h"
#include "qpid/broker/TxPublish.h"
#include "qpid/framing/reply_exceptions.h"
@@ -671,7 +672,11 @@
bool SemanticState::ConsumerImpl::doOutput()
{
- return haveCredit() && queue->dispatch(shared_from_this());
+ try {
+ return haveCredit() && queue->dispatch(shared_from_this());
+ } catch (const SessionException& e) {
+ throw SessionOutputException(e, parent->session.getChannel());
+ }
}
void SemanticState::ConsumerImpl::enableNotify()
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SessionAdapter.cpp Fri Jan 22 10:58:20 2010
@@ -439,6 +439,7 @@
ManagementAgent* agent = getBroker().getManagementAgent();
if (agent)
agent->raiseEvent(_qmf::EventQueueDelete(getConnection().getUrl(), getConnection().getUserId(), queue));
+ q->notifyDeleted();
}
}
Added: qpid/trunk/qpid/cpp/src/qpid/broker/SessionOutputException.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/SessionOutputException.h?rev=902055&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/SessionOutputException.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/SessionOutputException.h Fri Jan 22 10:58:20 2010
@@ -0,0 +1,47 @@
+#ifndef QPID_BROKER_SESSIONOUTPUTEXCEPTION_H
+#define QPID_BROKER_SESSIONOUTPUTEXCEPTION_H
+
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+
+#include "qpid/Exception.h"
+
+namespace qpid {
+namespace broker {
+
+/**
+ * This exception is used to signal 'session' exceptions (aka
+ * execution exceptions in AMQP 0-10 terms) that occur during output
+ * processing. It simply allows the channel of the session to be
+ * specified in addition to the other details. Special treatment is
+ * required at present because the output processing chain is
+ * different from that which handles incoming commands (specifically
+ * AggregateOutput cannot reasonably handle exceptions as it has no
+ * context).
+ */
+struct SessionOutputException : qpid::SessionException
+{
+ const uint16_t channel;
+ SessionOutputException(const qpid::SessionException& e, uint16_t c) : qpid::SessionException(e.code, e.getMessage()), channel(c) {}
+};
+}} // namespace qpid::broker
+
+#endif /*!QPID_BROKER_SESSIONOUTPUTEXCEPTION_H*/
Modified: qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/client/SessionImpl.cpp Fri Jan 22 10:58:20 2010
@@ -558,6 +558,7 @@
setState(DETACHED);
QPID_LOG(info, "Session detached by peer: " << id);
proxy.detached(_name, DETACH_CODE_NORMAL);
+ handleClosed();
}
void SessionImpl::detached(const std::string& _name, uint8_t _code) {
Modified: qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/ClientSessionTest.cpp Fri Jan 22 10:58:20 2010
@@ -629,6 +629,19 @@
BOOST_CHECK(!fix.subs.get(got, "a"));
}
+
+QPID_AUTO_TEST_CASE(testQueueDeleted)
+{
+ ClientSessionFixture fix;
+ fix.session.queueDeclare(arg::queue="my-queue");
+ LocalQueue queue;
+ fix.subs.subscribe(queue, "my-queue");
+
+ ScopedSuppressLogging sl;
+ fix.session.queueDelete(arg::queue="my-queue");
+ BOOST_CHECK_THROW(queue.get(1*qpid::sys::TIME_SEC), qpid::framing::ResourceDeletedException);
+}
+
QPID_AUTO_TEST_SUITE_END()
}} // namespace qpid::tests
Modified: qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/MessagingSessionTests.cpp Fri Jan 22 10:58:20 2010
@@ -180,6 +180,7 @@
~MultiQueueFixture()
{
+ connection.close();
for (const_iterator i = queues.begin(); i != queues.end(); ++i) {
admin.deleteQueue(*i);
}
Modified: qpid/trunk/qpid/python/qpid/tests/messaging.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/python/qpid/tests/messaging.py?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/python/qpid/tests/messaging.py (original)
+++ qpid/trunk/qpid/python/qpid/tests/messaging.py Fri Jan 22 10:58:20 2010
@@ -239,6 +239,12 @@
pass
assert msgs == fetched, "expecting %s, got %s" % (msgs, fetched)
self.ssn.acknowledge()
+ #we set the capacity to 0 to prevent the deletion of the queue -
+ #triggered the deletion policy when the first receiver is closed -
+ #resulting in session exceptions being issued for the remaining
+ #active subscriptions:
+ for r in [rcv1, rcv2, rcv3]:
+ r.capacity = 0
# XXX, we need a convenient way to assert that required queues are
# empty on setup, and possibly also to drain queues on teardown
Modified: qpid/trunk/qpid/python/tests_0-10/exchange.py
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/python/tests_0-10/exchange.py?rev=902055&r1=902054&r2=902055&view=diff
==============================================================================
--- qpid/trunk/qpid/python/tests_0-10/exchange.py (original)
+++ qpid/trunk/qpid/python/tests_0-10/exchange.py Fri Jan 22 10:58:20 2010
@@ -35,9 +35,12 @@
TestBase010.setUp(self)
self.queues = []
self.exchanges = []
+ self.subscriptions = []
def tearDown(self):
try:
+ for s in self.subscriptions:
+ self.session.message_cancel(destination=s)
for ssn, q in self.queues:
ssn.queue_delete(queue=q)
for ssn, ex in self.exchanges:
@@ -110,6 +113,7 @@
self.session.message_subscribe(queue=queueName, destination=consumer_tag)
self.session.message_flow(destination=consumer_tag, unit=self.session.credit_unit.message, value=0xFFFFFFFFL)
self.session.message_flow(destination=consumer_tag, unit=self.session.credit_unit.byte, value=0xFFFFFFFFL)
+ self.subscriptions.append(consumer_tag)
return self.session.incoming(consumer_tag)
---------------------------------------------------------------------
Apache Qpid - AMQP Messaging Implementation
Project: http://qpid.apache.org
Use/Interact: mailto:commits-subscribe@qpid.apache.org