Modifications to allow loading of message data in chunks, refragmentation of messages, plus some related refactoring and tests.

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

1 files changed, 13 insertions, 8 deletions

diff --git a/cpp/src/qpid/broker/MessageBuilder.cpp b/cpp/src/qpid/broker/MessageBuilder.cpp
index b4efd3d001..1a58523c08 100644
--- a/cpp/src/qpid/broker/MessageBuilder.cpp
+++ b/cpp/src/qpid/broker/MessageBuilder.cpp
@@ -20,25 +20,23 @@
  */
 #include <qpid/broker/MessageBuilder.h>
 
+#include <qpid/broker/InMemoryContent.h>
+#include <qpid/broker/LazyLoadedContent.h>
+
 using namespace qpid::broker;
 using namespace qpid::framing;
+using std::auto_ptr;
 
 MessageBuilder::MessageBuilder(CompletionHandler* _handler, MessageStore* const _store, u_int64_t _stagingThreshold) : 
     handler(_handler),
     store(_store),
-    stagingThreshold(_stagingThreshold),
-    staging(false)
+    stagingThreshold(_stagingThreshold)
 {}
 
 void MessageBuilder::route(){
-    if (staging && store) {
-        store->stage(message);
-        message->releaseContent();
-    }
     if (message->isComplete()) {
         if (handler) handler->complete(message);
         message.reset();
-        staging = false;
     }
 }
 
@@ -54,7 +52,14 @@ void MessageBuilder::setHeader(AMQHeaderBody::shared_ptr& header){
         THROW_QPID_ERROR(PROTOCOL_ERROR + 504, "Invalid message sequence: got header before publish.");
     }
     message->setHeader(header);
-    staging = stagingThreshold && header->getContentSize() >= stagingThreshold;
+    if (stagingThreshold && header->getContentSize() >= stagingThreshold) {
+        store->stage(message);
+        auto_ptr<Content> content(new LazyLoadedContent(store, message->getPersistenceId(), message->expectedContentSize()));
+        message->setContent(content);
+    } else {
+        auto_ptr<Content> content(new InMemoryContent());
+        message->setContent(content);
+    }
     route();
 }