QPID-7250: check that message can fit in a page before writing to disk

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1742020 13f79535-47bb-0310-9956-ffa450edef68

4 files changed, 23 insertions, 5 deletions

diff --git a/qpid/cpp/src/qpid/broker/Messages.h b/qpid/cpp/src/qpid/broker/Messages.h
index cd846a4973..5f51e69c11 100644
--- a/qpid/cpp/src/qpid/broker/Messages.h
+++ b/qpid/cpp/src/qpid/broker/Messages.h
@@ -94,7 +94,14 @@ class Messages
      * Apply, the functor to each message held
      */
     virtual void foreach(Functor) = 0;
-  private:
+
+    /**
+     * Allows implementation to perform optional checks before message
+     * is stored.
+     */
+    virtual void check(const Message&) {};
+
+ private:
 };
 }} // namespace qpid::broker
 
diff --git a/qpid/cpp/src/qpid/broker/PagedQueue.cpp b/qpid/cpp/src/qpid/broker/PagedQueue.cpp
index b5edfb89c0..b3b7f6a2d4 100644
--- a/qpid/cpp/src/qpid/broker/PagedQueue.cpp
+++ b/qpid/cpp/src/qpid/broker/PagedQueue.cpp
@@ -130,12 +130,17 @@ bool PagedQueue::deleted(const QueueCursor& cursor)
     }
 }
 
-void PagedQueue::publish(const Message& added)
+void PagedQueue::check(const Message& added)
 {
     if (encodedSize(added) > pageSize) {
         QPID_LOG(error, "Message is larger than page size for queue " << name);
         throw qpid::framing::PreconditionFailedException(QPID_MSG("Message is larger than page size for queue " << name));
     }
+}
+
+void PagedQueue::publish(const Message& added)
+{
+    check(added);
     Used::reverse_iterator i = used.rbegin();
     if (i != used.rend()) {
         if (!i->second.isLoaded()) load(i->second);
diff --git a/qpid/cpp/src/qpid/broker/PagedQueue.h b/qpid/cpp/src/qpid/broker/PagedQueue.h
index c8a9f13fc7..a53fd8b469 100644
--- a/qpid/cpp/src/qpid/broker/PagedQueue.h
+++ b/qpid/cpp/src/qpid/broker/PagedQueue.h
@@ -32,6 +32,8 @@
 namespace qpid {
 namespace broker {
 class ProtocolRegistry;
+
+
 /**
  *
  */
@@ -47,6 +49,7 @@ class PagedQueue : public Messages {
     Message* find(const framing::SequenceNumber&, QueueCursor*);
     Message* find(const QueueCursor&);
     void foreach(Functor);
+    void check(const Message& added);
   private:
     class Page {
       public:
diff --git a/qpid/cpp/src/qpid/broker/Queue.cpp b/qpid/cpp/src/qpid/broker/Queue.cpp
index 97b5a75e28..f8934d291a 100644
--- a/qpid/cpp/src/qpid/broker/Queue.cpp
+++ b/qpid/cpp/src/qpid/broker/Queue.cpp
@@ -274,10 +274,13 @@ bool Queue::accept(const Message& msg)
         //drop message
         QPID_LOG(info, "Dropping excluded message from " << getName());
         return false;
-    } else if (selector) {
-        return selector->filter(msg);
     } else {
-        return true;
+        messages->check(msg);
+        if (selector) {
+            return selector->filter(msg);
+        } else {
+            return true;
+        }
     }
 }