1 files changed, 33 insertions, 151 deletions

diff --git a/src/stable/client.cpp b/src/stable/client.cpp
index 56c5094..d2d48d7 100644
--- a/src/stable/client.cpp
+++ b/src/stable/client.cpp
@@ -6,7 +6,6 @@
  */
 
 #include "bu/client.h"
-#include "bu/tcpsocket.h"
 #include <stdlib.h>
 #include <errno.h>
 #include "bu/protocol.h"
@@ -24,72 +23,23 @@
 #define BU_PROFILE_END( x ) (void)0
 #endif
 
-Bu::Client::Client( Bu::TcpSocket *pSocket,
-        class Bu::ClientLinkFactory *pfLink ) :
-    pTopStream( pSocket ),
-    pSocket( pSocket ),
+Bu::Client::Client( class Bu::ClientLinkFactory *pfLink ) :
     pProto( NULL ),
     bWantsDisconnect( false ),
     pfLink( pfLink )
 {
-    Bu::ReadWriteMutex::WriteLocker lGlobal( mGlobal );
-    lFilts.prepend( pSocket );
 }
 
 Bu::Client::~Client()
 {
-    Bu::ReadWriteMutex::WriteLocker lGlobal( mGlobal );
-    for( FilterList::iterator i = lFilts.begin(); i; i++ )
-    {
-        delete *i;
-    }
-    pTopStream = pSocket = NULL;
     delete pfLink;
 }
 
 void Bu::Client::processInput()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    mRead.lock();
-    char buf[RBS];
-    Bu::size nRead, nTotal=0;
+    Bu::MutexLocker l( mProto );
 
-    BU_PROFILE_START("client.read");
-    for(;;)
-    {
-        try
-        {
-            nRead = pTopStream->read( buf, RBS );
-
-            if( nRead == 0 )
-            {
-                break;
-            }
-            else
-            {
-                nTotal += nRead;
-                qbRead.write( buf, nRead );
-                if( !pTopStream->canRead() )
-                    break;
-            }
-        }
-        catch( Bu::TcpSocketException &e )
-        {
-            pTopStream->close();
-            bWantsDisconnect = true;
-            break;
-        }
-    }
-    BU_PROFILE_END("client.read");
-    mRead.unlock();
-
-    if( nTotal == 0 )
-    {
-        pTopStream->close();
-        bWantsDisconnect = true;
-    }
-
-    if( pProto && nTotal )
+    if( pProto && getInputSize() > 0 )
     {
         BU_PROFILE_START("client.process");
         pProto->onNewData( this );
@@ -97,197 +47,133 @@ void Bu::Client::processInput()
     }
 }
 
-void Bu::Client::processOutput()
-{
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    mWrite.lock();
-    char buf[RBS];
-    if( qbWrite.getSize() > 0 )
-    {
-        int nAmnt = RBS;
-        nAmnt = qbWrite.peek( buf, nAmnt );
-        int nReal = pTopStream->write( buf, nAmnt );
-        qbWrite.seek( nReal );
-        pTopStream->flush();
-    }
-    mWrite.unlock();
-}
-
 void Bu::Client::setProtocol( Protocol *pProto )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
+    Bu::MutexLocker l( mProto );
     this->pProto = pProto;
     this->pProto->onNewConnection( this );
 }
 
 Bu::Protocol *Bu::Client::getProtocol()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
+    Bu::MutexLocker l( mProto );
     return pProto;
 }
 
 void Bu::Client::clearProtocol()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
+    Bu::MutexLocker l( mProto );
     pProto = NULL;
 }
-/*
-Bu::String &Bu::Client::getInput()
-{
-    return sReadBuf;
-}
-
-Bu::String &Bu::Client::getOutput()
-{
-    return sWriteBuf;
-}
-*/
 
 bool Bu::Client::isOpen()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    if( !pTopStream ) return false;
-    return pTopStream->isOpen();
+    return true;
 }
 
 Bu::size Bu::Client::write( const Bu::String &sData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( sData.getStr(), sData.getSize() );
+    return cbBuffer.client().write( sData.getStr(), sData.getSize() );
 }
 
 Bu::size Bu::Client::write( const void *pData, Bu::size nBytes )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( pData, nBytes );
+    return cbBuffer.client().write( pData, nBytes );
 }
 
 Bu::size Bu::Client::write( int8_t nData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( (const char *)&nData, sizeof(nData) );
+    return cbBuffer.client().write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( int16_t nData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( (const char *)&nData, sizeof(nData) );
+    return cbBuffer.client().write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( int32_t nData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( (const char *)&nData, sizeof(nData) );
+    return cbBuffer.client().write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( int64_t nData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( (const char *)&nData, sizeof(nData) );
+    return cbBuffer.client().write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( uint8_t nData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( (const char *)&nData, sizeof(nData) );
+    return cbBuffer.client().write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( uint16_t nData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( (const char *)&nData, sizeof(nData) );
+    return cbBuffer.client().write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( uint32_t nData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( (const char *)&nData, sizeof(nData) );
+    return cbBuffer.client().write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::write( uint64_t nData )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.write( (const char *)&nData, sizeof(nData) );
+    return cbBuffer.client().write( (const char *)&nData, sizeof(nData) );
 }
 
 Bu::size Bu::Client::read( void *pData, Bu::size nBytes )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbRead.read( pData, nBytes );
+    return cbBuffer.client().read( pData, nBytes );
 }
 
+
 Bu::size Bu::Client::peek( void *pData, int nBytes, int nOffset )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbRead.peek( pData, nBytes, nOffset );
+    return cbBuffer.client().peek( pData, nBytes, nOffset );
 }
 
+
 Bu::size Bu::Client::getInputSize()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbRead.getSize();
+    return cbBuffer.client().getSize();
 }
 
 Bu::size Bu::Client::getOutputSize()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mWrite );
-    return qbWrite.getSize();
-}
-
-const Bu::TcpSocket *Bu::Client::getSocket() const
-{
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    return pSocket;
+    return cbBuffer.server().getSize();
 }
 
 void Bu::Client::disconnect()
 {
-    Bu::ReadWriteMutex::WriteLocker lGlobal( mGlobal );
+    Bu::MutexLocker l( mDisconnect );
     bWantsDisconnect = true;
 }
 
 bool Bu::Client::wantsDisconnect()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
+    Bu::MutexLocker l( mDisconnect );
     return bWantsDisconnect;
 }
 
 void Bu::Client::close()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    pTopStream->close();
 }
 
 Bu::ClientLink *Bu::Client::getLink()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
     return pfLink->createLink( this );
 }
 
 void Bu::Client::onMessage( const Bu::String &sMsg )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
+    Bu::MutexLocker l( mProto );
     if( pProto )
         pProto->onMessage( this, sMsg );
 }
 
 void Bu::Client::tick()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
+    Bu::MutexLocker l( mProto );
     if( pProto )
         pProto->onTick( this );
 }
@@ -299,9 +185,7 @@ Bu::size Bu::Client::tell()
 
 void Bu::Client::seek( Bu::size offset )
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mRead );
-    return qbRead.seek( offset );
+    return cbBuffer.client().seek( offset );
 }
 
 void Bu::Client::setPos( Bu::size )
@@ -321,15 +205,11 @@ bool Bu::Client::isEos()
 
 void Bu::Client::flush()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    processOutput();
 }
 
 bool Bu::Client::canRead()
 {
-    Bu::ReadWriteMutex::ReadLocker lGlobal( mGlobal );
-    Bu::MutexLocker l( mRead );
-    return qbRead.getSize() > 0;
+    return cbBuffer.client().getSize() > 0;
 }
 
 bool Bu::Client::canWrite()
@@ -374,11 +254,13 @@ Bu::size Bu::Client::getSize() const
 
 Bu::size Bu::Client::getBlockSize() const
 {
-    return pSocket->getBlockSize();
+    return 0;
+    //return pSocket->getBlockSize();
 }
 
 Bu::String Bu::Client::getLocation() const
 {
-    return pSocket->getLocation();
+    return "???";
+    //return pSocket->getLocation();
 }