Bug fixes: corrected fragmentation of large messages in client, set buffer size on server to be the same as the max frame size (this still needs some work to be better managed and configurable).

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

2 files changed, 13 insertions, 11 deletions

diff --git a/cpp/src/qpid/apr/LFSessionContext.cpp b/cpp/src/qpid/apr/LFSessionContext.cpp
index 1399025c32..7dc70ca245 100644
--- a/cpp/src/qpid/apr/LFSessionContext.cpp
+++ b/cpp/src/qpid/apr/LFSessionContext.cpp
@@ -33,8 +33,8 @@ LFSessionContext::LFSessionContext(apr_pool_t* _pool, apr_socket_t* _socket,
     debug(_debug),
     socket(_socket),
     initiated(false),
-    in(32768),
-    out(32768),
+    in(65536),
+    out(65536),
     processor(_processor),
     processing(false),
     closing(false)
diff --git a/cpp/src/qpid/client/Channel.cpp b/cpp/src/qpid/client/Channel.cpp
index 6901407072..ab4ea5b787 100644
--- a/cpp/src/qpid/client/Channel.cpp
+++ b/cpp/src/qpid/client/Channel.cpp
@@ -220,19 +220,21 @@ void Channel::publish(Message& msg, const Exchange& exchange, const std::string&
     AMQBody::shared_ptr body(static_pointer_cast<AMQBody, AMQHeaderBody>(msg.header));
     out->send(new AMQFrame(id, body));
     
-    int data_length = data.length();
+    u_int64_t data_length = data.length();
     if(data_length > 0){
-        //TODO fragmentation of messages, need to know max frame size for connection
-        int frag_size = con->getMaxFrameSize() - 4;
+        u_int32_t frag_size = con->getMaxFrameSize() - 8;//frame itself uses 8 bytes
         if(data_length < frag_size){
             out->send(new AMQFrame(id, new AMQContentBody(data)));
         }else{
-            int frag_count = data_length / frag_size;
-            for(int i = 0; i < frag_count; i++){
-                int pos = i*frag_size;
-                int len = i < frag_count - 1 ? frag_size : data_length - pos;
-                string frag(data.substr(pos, len));
-                out->send(new AMQFrame(id, new AMQContentBody(frag)));          
+            u_int32_t offset = 0;
+            u_int32_t remaining = data_length - offset;
+            while (remaining > 0) {
+                u_int32_t length = remaining > frag_size ? frag_size : remaining;
+                string frag(data.substr(offset, length));
+                out->send(new AMQFrame(id, new AMQContentBody(frag)));                          
+                
+                offset += length;
+                remaining = data_length - offset;
             }
         }
     }