r1152@fuschia: andrew | 2007-02-17 21:14:42 +0000

 More support for references (and transfers of reference content)
 r1220@fuschia:  andrew | 2007-02-21 00:22:53 +0000
 Working version of delivering Message Transfers by reference


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/qpid.0-9@509834 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 40 insertions, 5 deletions

diff --git a/cpp/lib/broker/BrokerMessageMessage.cpp b/cpp/lib/broker/BrokerMessageMessage.cpp
index f51722f2da..e1be57fad7 100644
--- a/cpp/lib/broker/BrokerMessageMessage.cpp
+++ b/cpp/lib/broker/BrokerMessageMessage.cpp
@@ -22,12 +22,14 @@
 #include "BrokerMessageMessage.h"
 #include "ChannelAdapter.h"
 #include "MessageTransferBody.h"
+#include "MessageOpenBody.h"
+#include "MessageCloseBody.h"
 #include "MessageAppendBody.h"
 #include "Reference.h"
 #include "framing/FieldTable.h"
 #include "framing/BasicHeaderProperties.h"
 
-#include <iostream>
+#include <algorithm>
 
 using namespace std;
 using namespace qpid::framing;
@@ -36,21 +38,50 @@ namespace qpid {
 namespace broker {
 	
 MessageMessage::MessageMessage(
-    ConnectionToken* publisher, TransferPtr transfer_
+    ConnectionToken* publisher, RequestId requestId_, TransferPtr transfer_
 ) : Message(publisher, transfer_->getDestination(),
             transfer_->getRoutingKey(),
             transfer_->getMandatory(),
             transfer_->getImmediate(),
             transfer_),
+    requestId(requestId_),
     transfer(transfer_)
 {}
 
+MessageMessage::MessageMessage(
+    ConnectionToken* publisher, RequestId requestId_, TransferPtr transfer_,
+    ReferencePtr reference_
+) : Message(publisher, transfer_->getDestination(),
+            transfer_->getRoutingKey(),
+            transfer_->getMandatory(),
+            transfer_->getImmediate(),
+            transfer_),
+    requestId(requestId_),
+    transfer(transfer_),
+    reference(reference_)
+{}
+
 void MessageMessage::deliver(
     framing::ChannelAdapter& channel, 
     const std::string& consumerTag, 
     u_int64_t /*deliveryTag*/, 
     u_int32_t /*framesize*/)
 {
+	const framing::Content& body = transfer->getBody();
+	
+	// Send any reference data
+	if (!body.isInline()){
+		// Open
+		channel.send(new MessageOpenBody(channel.getVersion(), reference->getId()));
+		// Appends
+		for(Reference::Appends::const_iterator a = reference->getAppends().begin();
+			a != reference->getAppends().end();
+			++a) {
+			channel.send(new MessageAppendBody(*a->get()));
+		}
+	}
+	
+	// The the transfer
     channel.send(
     	new MessageTransferBody(channel.getVersion(), 
                                 transfer->getTicket(),
@@ -74,8 +105,13 @@ void MessageMessage::deliver(
                                 transfer->getTransactionId(),
                                 transfer->getSecurityToken(),
                                 transfer->getApplicationHeaders(),
-                                transfer->getBody(),
+                                body,
                                 transfer->getMandatory()));
+	// Close any reference data
+	if (!body.isInline()){
+		// Close
+		channel.send(new MessageCloseBody(channel.getVersion(), reference->getId()));
+	}
 }
 
 void MessageMessage::sendGetOk(
@@ -120,11 +156,10 @@ bool MessageMessage::isComplete()
 
 u_int64_t MessageMessage::contentSize() const
 {
-	// FIXME astitcher 2007-2-7 only works for inline content
 	if (transfer->getBody().isInline())
 	    return transfer->getBody().getValue().size();
 	else
-    	THROW_QPID_ERROR(INTERNAL_ERROR, "Unfinished");		 
+    	return reference->getSize();		 
 }
 
 qpid::framing::BasicHeaderProperties* MessageMessage::getHeaderProperties()