Updated message.transfer encoding to use header and content segments (including new structs).

Unified more between the basic and message classes messages.



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

1 files changed, 6 insertions, 19 deletions

diff --git a/cpp/src/qpid/broker/MessageHandlerImpl.cpp b/cpp/src/qpid/broker/MessageHandlerImpl.cpp
index ce1fa1e028..a4ceb77c12 100644
--- a/cpp/src/qpid/broker/MessageHandlerImpl.cpp
+++ b/cpp/src/qpid/broker/MessageHandlerImpl.cpp
@@ -22,7 +22,7 @@
 #include "qpid/framing/FramingContent.h"
 #include "Connection.h"
 #include "Broker.h"
-#include "BrokerMessageMessage.h"
+#include "MessageDelivery.h"
 #include "qpid/framing/MessageAppendBody.h"
 #include "qpid/framing/MessageTransferBody.h"
 #include "BrokerAdapter.h"
@@ -55,7 +55,7 @@ MessageHandlerImpl::open(const string& /*reference*/)
 }
 
 void
-MessageHandlerImpl::append(const framing::AMQMethodBody& )
+MessageHandlerImpl::append(const std::string& /*reference*/, const std::string& /*bytes*/)
 {
     throw ConnectionException(540, "References no longer supported");
 }
@@ -92,7 +92,7 @@ MessageHandlerImpl::subscribe(uint16_t /*ticket*/,
                             const string& destination,
                             bool noLocal,
                             u_int8_t confirmMode,
-                            u_int8_t /*acquireMode*/,//TODO: implement acquire modes
+                            u_int8_t acquireMode,//TODO: implement acquire modes
                             bool exclusive,
                             const framing::FieldTable& filter )
 {
@@ -101,7 +101,8 @@ MessageHandlerImpl::subscribe(uint16_t /*ticket*/,
         throw ConnectionException(530, "Consumer tags must be unique");
 
     string tag = destination;
-    channel.consume(MessageMessage::getToken(destination), tag, queue, noLocal, confirmMode == 1, exclusive, &filter);
+    channel.consume(MessageDelivery::getMessageDeliveryToken(destination, confirmMode, acquireMode), 
+                    tag, queue, noLocal, confirmMode == 1, exclusive, &filter);
     // Dispatch messages as there is now a consumer.
     queue->requestDispatch();
 }
@@ -115,7 +116,7 @@ MessageHandlerImpl::get(uint16_t /*ticket*/,
 {
     Queue::shared_ptr queue = getQueue(queueName);
     
-    if (channel.get(MessageMessage::getToken(destination), queue, !noAck)){
+    if (channel.get(MessageDelivery::getMessageDeliveryToken(destination, noAck ? 0 : 1, 0), queue, !noAck)){
         //don't send any response... rely on execution completion
     } else {
         //temporarily disabled:
@@ -160,20 +161,6 @@ MessageHandlerImpl::reject(const SequenceNumberSet& /*transfers*/, uint16_t /*co
     //TODO: implement
 }
 
-void
-MessageHandlerImpl::transfer(const framing::AMQMethodBody& context)
-{
-    const MessageTransferBody* transfer = boost::polymorphic_downcast<const MessageTransferBody*>(&context);
-    if (transfer->getBody().isInline()) {
-        MessageMessage::shared_ptr message(new MessageMessage(&connection, transfer));
-        channel.handleInlineTransfer(message);
-    } else { 
-        throw ConnectionException(540, "References no longer supported");
-    }
-}
-
-
-
 void MessageHandlerImpl::flow(const std::string& destination, u_int8_t unit, u_int32_t value)
 {