Author: aconway
Date: Wed May 29 18:03:59 2013
New Revision: 1487579
URL: http://svn.apache.org/r1487579
Log:
QPID-4886: Pass non-const reference to Message in QueueObserver functions.
Instead of modifying QueueObserver, a new class MessageInterceptor was
introduced to allow messages to be modified.
Added:
qpid/trunk/qpid/cpp/src/qpid/broker/MessageInterceptor.h (with props)
Modified:
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h
Added: qpid/trunk/qpid/cpp/src/qpid/broker/MessageInterceptor.h
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/MessageInterceptor.h?rev=1487579&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/MessageInterceptor.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/MessageInterceptor.h Wed May 29
18:03:59 2013
@@ -0,0 +1,53 @@
+#ifndef QPID_BROKER_MESSAGEINTERCEPTOR_H
+#define QPID_BROKER_MESSAGEINTERCEPTOR_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 "Observers.h"
+
+namespace qpid {
+namespace broker {
+
+class Message;
+
+/**
+ * Interface for classes that want to modify a message as it is processed by
the queue.
+ */
+class MessageInterceptor
+{
+ public:
+ virtual ~MessageInterceptor() {}
+
+ /** Modify a message as it is being published onto the queue. */
+ virtual void publish(Message&) = 0;
+};
+
+class MessageInterceptors : public Observers<MessageInterceptor> {
+ public:
+ void publish(Message& m) {
+ each(boost::bind(&MessageInterceptor::publish, _1, boost::ref(m)));
+ }
+};
+
+}} // namespace qpid::broker
+
+#endif /*!QPID_BROKER_MESSAGEINTERCEPTOR_H*/
Propchange: qpid/trunk/qpid/cpp/src/qpid/broker/MessageInterceptor.h
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: qpid/trunk/qpid/cpp/src/qpid/broker/MessageInterceptor.h
------------------------------------------------------------------------------
svn:keywords = Rev Date
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=1487579&r1=1487578&r2=1487579&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.cpp Wed May 29 18:03:59 2013
@@ -429,7 +429,8 @@ bool Queue::getNextMessage(Message& m, C
continue; //try another message
}
}
- QPID_LOG(debug, "Message retrieved from '" << name << "'");
+ QPID_LOG(debug, "Message " << msg->getSequence() << "
retrieved from '"
+ << name << "'");
m = *msg;
return true;
} else {
@@ -767,6 +768,7 @@ void Queue::push(Message& message, bool
Mutex::ScopedLock locker(messageLock);
message.setSequence(++sequence);
if (settings.sequencing) message.addAnnotation(settings.sequenceKey,
(uint32_t)sequence);
+ interceptors.publish(message);
messages->publish(message);
listeners.populate(copy);
observeEnqueue(message, locker);
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=1487579&r1=1487578&r2=1487579&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Queue.h Wed May 29 18:03:59 2013
@@ -27,6 +27,7 @@
#include "qpid/broker/Consumer.h"
#include "qpid/broker/Message.h"
#include "qpid/broker/Messages.h"
+#include "qpid/broker/MessageInterceptor.h"
#include "qpid/broker/PersistableQueue.h"
#include "qpid/broker/QueueBindings.h"
#include "qpid/broker/QueueListeners.h"
@@ -164,6 +165,7 @@ class Queue : public boost::enable_share
sys::AtomicValue<uint32_t> dequeueSincePurge; // Count dequeues since last
purge.
int eventMode;
Observers observers;
+ MessageInterceptors interceptors;
std::string seqNoKey;
Broker* broker;
bool deleted;
@@ -414,6 +416,8 @@ class Queue : public boost::enable_share
QPID_BROKER_EXTERN void addObserver(boost::shared_ptr<QueueObserver>);
QPID_BROKER_EXTERN void removeObserver(boost::shared_ptr<QueueObserver>);
QPID_BROKER_EXTERN void insertSequenceNumbers(const std::string& key);
+
+ QPID_BROKER_EXTERN MessageInterceptors& getMessageInterceptors() { return
interceptors; }
/**
* Notify queue that recovery has completed.
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
