QPID-4339: simple paged queue implementation

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

1 files changed, 16 insertions, 3 deletions

diff --git a/cpp/src/qpid/broker/Protocol.cpp b/cpp/src/qpid/broker/Protocol.cpp
index e236698142..a98160e502 100644
--- a/cpp/src/qpid/broker/Protocol.cpp
+++ b/cpp/src/qpid/broker/Protocol.cpp
@@ -20,6 +20,7 @@
  */
 #include "Protocol.h"
 #include "qpid/sys/ConnectionCodec.h"
+#include "qpid/broker/RecoverableMessageImpl.h"
 #include "qpid/broker/amqp_0_10/MessageTransfer.h"
 #include "qpid/log/Statement.h"
 
@@ -47,13 +48,25 @@ boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> ProtocolReg
 }
 boost::shared_ptr<RecoverableMessage> ProtocolRegistry::recover(qpid::framing::Buffer& b)
 {
-    boost::shared_ptr<RecoverableMessage> msg;
-    for (Protocols::const_iterator i = protocols.begin(); !msg && i != protocols.end(); ++i) {
-        msg = i->second->recover(b);
+    uint32_t position = b.getPosition();
+    for (Protocols::const_iterator i = protocols.begin(); i != protocols.end(); ++i) {
+        boost::shared_ptr<RecoverableMessage> msg = i->second->recover(b);
+        if (msg) return msg;
+        else b.setPosition(position);
     }
+    boost::intrusive_ptr<qpid::broker::amqp_0_10::MessageTransfer> transfer(new qpid::broker::amqp_0_10::MessageTransfer());
+    transfer->decodeHeader(b);
+    boost::shared_ptr<RecoverableMessage> msg(new RecoverableMessageImpl(Message(transfer, transfer)));
     return msg;
 }
 
+Message ProtocolRegistry::decode(qpid::framing::Buffer& buffer)
+{
+    boost::shared_ptr<RecoverableMessage> r = recover(buffer);
+    r->decodeContent(buffer);
+    return r->getMessage();
+}
+
 ProtocolRegistry::~ProtocolRegistry()
 {
     for (Protocols::const_iterator i = protocols.begin(); i != protocols.end(); ++i) {