QueueBuf is updated, and everything else uses it now, including Client.

Unfortunately this breaks some programs that accessed the client internal
buffer directly.  Overall it's much, much more efficient, so it's worth it,
maybe we'll find a good workaround later.

7 files changed, 80 insertions, 93 deletions

diff --git a/src/client.cpp b/src/client.cpp
index 95008f9..88eb49d 100644
--- a/src/client.cpp
+++ b/src/client.cpp
@@ -20,7 +20,6 @@ Bu::Client::Client( Bu::Socket *pSocket, class Bu::ClientLinkFactory *pfLink ) :
         pTopStream( pSocket ),
         pSocket( pSocket ),
         pProto( NULL ),
-        nRBOffset( 0 ),
         bWantsDisconnect( false ),
         pfLink( pfLink )
 {
@@ -54,7 +53,7 @@ void Bu::Client::processInput()
                         else
                         {
                                 nTotal += nRead;
-                                sReadBuf.append( buf, nRead );
+                                qbRead.write( buf, nRead );
                                 if( !pTopStream->canRead() )
                                         break;
                         }
@@ -81,13 +80,14 @@ void Bu::Client::processInput()
 
 void Bu::Client::processOutput()
 {
-        if( sWriteBuf.getSize() > 0 )
+        if( qbWrite.getSize() > 0 )
         {
-                int nAmnt = (sWriteBuf.getSize()<2048)?(sWriteBuf.getSize()):(2048);
-                int nReal = pTopStream->write( sWriteBuf.getStr(), nAmnt );
+                int nAmnt = RBS;
+                char *buf = new char[nAmnt];
+                nAmnt = qbWrite.peek( buf, nAmnt );
+                int nReal = pTopStream->write( buf, nAmnt );
+                qbWrite.seek( nReal );
                 pTopStream->flush();
-                sWriteBuf.trimFront( nReal );
-                //sWriteBuf.clear();
         }
 }
 
@@ -106,7 +106,7 @@ void Bu::Client::clearProtocol()
 {
         pProto = NULL;
 }
-
+/*
 Bu::FString &Bu::Client::getInput()
 {
         return sReadBuf;
@@ -116,6 +116,7 @@ Bu::FString &Bu::Client::getOutput()
 {
         return sWriteBuf;
 }
+*/
 
 bool Bu::Client::isOpen()
 {
@@ -125,118 +126,77 @@ bool Bu::Client::isOpen()
 
 void Bu::Client::write( const Bu::FString &sData )
 {
-        sWriteBuf += sData;
+        qbWrite.write( sData.getStr(), sData.getSize() );
 }
 
 void Bu::Client::write( const void *pData, int nBytes )
 {
-        sWriteBuf.append( (const char *)pData, nBytes );
+        qbWrite.write( pData, nBytes );
 }
 
 void Bu::Client::write( int8_t nData )
 {
-        sWriteBuf.append( (const char *)&nData, sizeof(nData) );
+        qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 void Bu::Client::write( int16_t nData )
 {
-        sWriteBuf.append( (const char *)&nData, sizeof(nData) );
+        qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 void Bu::Client::write( int32_t nData )
 {
-        sWriteBuf.append( (const char *)&nData, sizeof(nData) );
+        qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 void Bu::Client::write( int64_t nData )
 {
-        sWriteBuf.append( (const char *)&nData, sizeof(nData) );
+        qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 void Bu::Client::write( uint8_t nData )
 {
-        sWriteBuf.append( (const char *)&nData, sizeof(nData) );
+        qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 void Bu::Client::write( uint16_t nData )
 {
-        sWriteBuf.append( (const char *)&nData, sizeof(nData) );
+        qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 void Bu::Client::write( uint32_t nData )
 {
-        sWriteBuf.append( (const char *)&nData, sizeof(nData) );
+        qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 void Bu::Client::write( uint64_t nData )
 {
-        sWriteBuf.append( (const char *)&nData, sizeof(nData) );
+        qbWrite.write( (const char *)&nData, sizeof(nData) );
 }
 
 int Bu::Client::read( void *pData, int nBytes )
 {
-        if( nBytes > sReadBuf.getSize()-nRBOffset )
-        {
-                nBytes = sReadBuf.getSize()-nRBOffset;
-                if( nBytes <= 0 )
-                        return 0;
-        }
-        memcpy( pData, sReadBuf.getStr()+nRBOffset, nBytes );
-        nRBOffset += nBytes;
-        if( sReadBuf.getSize()-nRBOffset == 0 )
-        {
-                sReadBuf.clear();
-                nRBOffset = 0;
-        }
-        // This is an experimental threshold, maybe I'll make this configurable
-        // later on.
-        else if(
-                (sReadBuf.getSize() >= 1024 && nRBOffset >= sReadBuf.getSize()/2) ||
-                (nRBOffset >= sReadBuf.getSize()/4)
-                )
-        {
-                sReadBuf.trimFront( nRBOffset );
-                nRBOffset = 0;
-        }
-
-        return nBytes;
+        return qbRead.read( pData, nBytes );
 }
 
 int Bu::Client::peek( void *pData, int nBytes, int nOffset )
 {
-        if( nBytes+nOffset > sReadBuf.getSize()-nRBOffset )
-        {
-                nBytes = sReadBuf.getSize()-nRBOffset-nOffset;
-                if( nBytes <= 0 )
-                        return 0;
-        }
-        memcpy( pData, sReadBuf.getStr()+nRBOffset+nOffset, nBytes );
-        return nBytes;
+        return qbRead.peek( pData, nBytes, nOffset );
 }
 
 void Bu::Client::seek( int nBytes )
 {
-        nRBOffset += nBytes;
-        if( sReadBuf.getSize()-nRBOffset == 0 )
-        {
-                sReadBuf.clear();
-                nRBOffset = 0;
-        }
-        // This is an experimental threshold, maybe I'll make this configurable
-        // later on.
-        else if(
-                (sReadBuf.getSize() >= 1024 && nRBOffset >= sReadBuf.getSize()/2) ||
-                (nRBOffset >= sReadBuf.getSize()/4)
-                )
-        {
-                sReadBuf.trimFront( nRBOffset );
-                nRBOffset = 0;
-        }
+        return qbRead.seek( nBytes );
 }
 
 long Bu::Client::getInputSize()
 {
-        return sReadBuf.getSize()-nRBOffset;
+        return qbRead.getSize();
+}
+
+long Bu::Client::getOutputSize()
+{
+        return qbWrite.getSize();
 }
 
 const Bu::Socket *Bu::Client::getSocket() const
diff --git a/src/client.h b/src/client.h
index aecb16c..5a933ca 100644
--- a/src/client.h
+++ b/src/client.h
@@ -11,6 +11,7 @@
 #include <stdint.h>
 
 #include "bu/fstring.h"
+#include "bu/queuebuf.h"
 
 namespace Bu
 {
@@ -31,8 +32,8 @@ namespace Bu
                 void processInput();
                 void processOutput();
 
-                Bu::FString &getInput();
-                Bu::FString &getOutput();
+                //Bu::FString &getInput();
+                //Bu::FString &getOutput();
                 void write( const Bu::FString &sData );
                 void write( const void *pData, int nBytes );
                 void write( int8_t nData );
@@ -47,6 +48,7 @@ namespace Bu
                 int peek( void *pData, int nBytes, int nOffset=0 );
                 void seek( int nBytes );
                 long getInputSize();
+                long getOutputSize();
 
                 void setProtocol( Protocol *pProto );
                 Bu::Protocol *getProtocol();
@@ -65,7 +67,8 @@ namespace Bu
 
                 void onMessage( const Bu::FString &sMsg );
 
-                bool hasOutput() { return !sWriteBuf.isEmpty(); }
+                bool hasOutput() { return qbWrite.getSize() > 0; }
+                bool hasInput() { return qbRead.getSize() > 0; }
 
                 template<typename filter>
                 void pushFilter()
@@ -97,9 +100,8 @@ namespace Bu
                 Bu::Stream *pTopStream;
                 Bu::Socket *pSocket;
                 Bu::Protocol *pProto;
-                Bu::FString sReadBuf;
-                int nRBOffset;
-                Bu::FString sWriteBuf;
+                Bu::QueueBuf qbRead;
+                Bu::QueueBuf qbWrite;
                 bool bWantsDisconnect;
                 class Bu::ClientLinkFactory *pfLink;
         };
diff --git a/src/queuebuf.cpp b/src/queuebuf.cpp
index 9577793..01d92f8 100644
--- a/src/queuebuf.cpp
+++ b/src/queuebuf.cpp
@@ -76,7 +76,12 @@ size_t Bu::QueueBuf::read( void *pRawBuf, size_t nBytes )
         return nBytes - iLeft;
 }
 
-size_t Bu::QueueBuf::peek( void *pRawBuf, size_t nBytes )
+size_t Bu::QueueBuf::peek( void *pBuf, size_t nBytes )
+{
+        return peek( pBuf, nBytes, 0 );
+}
+
+size_t Bu::QueueBuf::peek( void *pRawBuf, size_t nBytes, size_t nSkip )
 {
         if( nBytes <= 0 )
                 return 0;
@@ -87,12 +92,16 @@ size_t Bu::QueueBuf::peek( void *pRawBuf, size_t nBytes )
         size_t iLeft = nBytes;
         char *pBuf = (char *)pRawBuf;
 
-        int iTmpReadOffset = iReadOffset;
+        int iTmpReadOffset = iReadOffset + nSkip;
         size_t iTmpRemSize = iTotalSize;
         BlockList::iterator iBlock = lBlocks.begin();
+        while( iTmpReadOffset > iBlockSize )
+        {
+                iTmpReadOffset -= iBlockSize;
+                iBlock++;
+        }
         while( iLeft > 0 && iTmpRemSize > 0 )
         {
-                // Switching to use temp variables instead of iReadOffset and iTotalSize
                 if( iTmpReadOffset == iBlockSize )
                 {
                         iBlock++;
diff --git a/src/queuebuf.h b/src/queuebuf.h
index 3591959..382863d 100644
--- a/src/queuebuf.h
+++ b/src/queuebuf.h
@@ -29,6 +29,7 @@ namespace Bu
                 virtual void close();
                 virtual size_t read( void *pBuf, size_t nBytes );
                 virtual size_t peek( void *pBuf, size_t nBytes );
+                virtual size_t peek( void *pBuf, size_t nBytes, size_t nSkip );
                 virtual size_t write( const void *pBuf, size_t nBytes );
                 virtual long tell();
                 virtual void seek( long offset );
diff --git a/src/tests/itoserver.cpp b/src/tests/itoserver.cpp
index cf5f6d1..5f6e4fa 100644
--- a/src/tests/itoserver.cpp
+++ b/src/tests/itoserver.cpp
@@ -33,8 +33,12 @@ public:
         virtual void onNewData( Bu::Client *pClient )
         {
                 TRACE();
-                pClient->write( pClient->getInput().getStr(), pClient->getInputSize() );
-                pClient->seek( pClient->getInputSize() );
+                char buf[1024];
+                while( pClient->hasInput() )
+                {
+                        int iAmnt = pClient->read( buf, 1024 );
+                        pClient->write( buf, iAmnt );
+                }
         }
 };
 
diff --git a/src/tests/multiserver.cpp b/src/tests/multiserver.cpp
index 22ce94b..85971b5 100644
--- a/src/tests/multiserver.cpp
+++ b/src/tests/multiserver.cpp
@@ -19,8 +19,12 @@ public:
 
         virtual void onNewData( Bu::Client *pClient )
         {
-                pClient->write( pClient->getInput() );
-                pClient->seek( pClient->getInputSize() );
+                char buf[1024];
+                while( pClient->hasInput() )
+                {
+                        int iAmnt = pClient->read( buf, 1024 );
+                        pClient->write( buf, iAmnt );
+                }
         }
 };
 
@@ -34,20 +38,23 @@ public:
 
         virtual void onNewData( Bu::Client *pClient )
         {
-                Bu::FString sTmp = pClient->getInput();
-                for( int j = 0; j < sTmp.getSize(); j++ )
+                while( pClient->hasInput() )
                 {
-                        if( sTmp[j] >= 'a' && sTmp[j] <= 'z' )
-                        {
-                                sTmp[j] = ((sTmp[j]-'a'+13)%26) + 'a';
-                        }
-                        else if( sTmp[j] >= 'A' && sTmp[j] <= 'Z' )
+                        char sTmp[1024];
+                        int iAmnt = pClient->read( sTmp, 1024 );
+                        for( int j = 0; j < iAmnt; j++ )
                         {
-                                sTmp[j] = ((sTmp[j]-'A'+13)%26) + 'A';
+                                if( sTmp[j] >= 'a' && sTmp[j] <= 'z' )
+                                {
+                                        sTmp[j] = ((sTmp[j]-'a'+13)%26) + 'a';
+                                }
+                                else if( sTmp[j] >= 'A' && sTmp[j] <= 'Z' )
+                                {
+                                        sTmp[j] = ((sTmp[j]-'A'+13)%26) + 'A';
+                                }
                         }
+                        pClient->write( sTmp, iAmnt );
                 }
-                pClient->write( sTmp );
-                pClient->seek( pClient->getInputSize() );
         }
 };
 
diff --git a/src/tests/rot13.cpp b/src/tests/rot13.cpp
index 2326888..03ba385 100644
--- a/src/tests/rot13.cpp
+++ b/src/tests/rot13.cpp
@@ -67,8 +67,12 @@ public:
 
         void onNewData( Bu::Client *pClient )
         {
-                pClient->write( pClient->getInput().getStr(), pClient->getInputSize() );
-                pClient->seek( pClient->getInputSize() );
+                char buf[1024];
+                while( pClient->hasInput() )
+                {
+                        int iAmnt = pClient->read( buf, 1024 );
+                        pClient->write( buf, iAmnt );
+                }
         }
 };