Client always collects at least an entire frameset into a single buffer

when possible. Based on patch from Gordon Sim.

 - Refactor Connector::writebuff, ::send as Connector::Writer
 - Collect frames up to EOF notifying AIO write.
 - Encode all available complete framesets into buffers as compactly as possible.
 - Logging buffer size and frames encoded per write for client and broker.
 - framing::Buffer added getPosition(), getSize(), default ctor, copy ctor.


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

1 files changed, 28 insertions, 2 deletions

diff --git a/cpp/src/qpid/client/Connector.h b/cpp/src/qpid/client/Connector.h
index b1d759569c..9897789901 100644
--- a/cpp/src/qpid/client/Connector.h
+++ b/cpp/src/qpid/client/Connector.h
@@ -45,6 +45,33 @@ namespace client {
 class Connector : public framing::OutputHandler, 
                   private sys::Runnable
 {
+    struct Buff;
+
+    /** Batch up frames for writing to aio. */
+    class Writer : public framing::FrameHandler {
+        typedef sys::AsynchIO::BufferBase BufferBase;
+        typedef std::vector<framing::AMQFrame> Frames;
+
+        sys::Mutex lock;
+        sys::AsynchIO* aio;
+        BufferBase* buffer;
+        Frames frames;
+        Frames::iterator lastEof; // Points after last EOF in frames
+        framing::Buffer encode;
+        size_t framesEncoded;
+        
+        void writeOne(const sys::Mutex::ScopedLock&);
+        void newBuffer(const sys::Mutex::ScopedLock&);
+
+      public:
+        
+        Writer();
+        ~Writer();
+        void setAio(sys::AsynchIO*);
+        void handle(framing::AMQFrame&);
+        void write(sys::AsynchIO&);
+    };
+    
     const bool debug;
     const int receive_buffer_size;
     const int send_buffer_size;
@@ -65,8 +92,7 @@ class Connector : public framing::OutputHandler,
     framing::InitiationHandler* initialiser;
     framing::OutputHandler* output;
 
-    sys::Mutex writeLock;
-    std::queue<framing::AMQFrame> writeFrameQueue;
+    Writer writer;
     
     sys::Thread receiver;