qpid-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From acon...@apache.org
Subject svn commit: r1487579 - in /qpid/trunk/qpid/cpp/src/qpid/broker: MessageInterceptor.h Queue.cpp Queue.h
Date Wed, 29 May 2013 18:04:00 GMT
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: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org


Mime
View raw message