Bu::Conduit now works exactly as it was advertised some time ago, it uses

Bu::QueueBuf and creates a really slick blocking inter-thread I/O system.

1 files changed, 226 insertions, 0 deletions

diff --git a/src/conduit.cpp b/src/conduit.cpp
index bb99526..cfa93d8 100644
--- a/src/conduit.cpp
+++ b/src/conduit.cpp
@@ -5,3 +5,229 @@
  * terms of the license contained in the file LICENSE.
  */
 
+#include "bu/conduit.h"
+
+Bu::Conduit::Conduit( int iBlockSize ) :
+        qb( iBlockSize ),
+        bBlocking( true ),
+        bOpen( true )
+{
+}
+
+Bu::Conduit::~Conduit()
+{
+}
+
+void Bu::Conduit::close()
+{
+        im.lock();
+//      qb.close();
+        bOpen = false;
+        
+        cBlock.signal();
+        im.unlock();
+}
+
+#include <stdio.h>
+Bu::size Bu::Conduit::read( void *pBuf, Bu::size nBytes )
+{
+        if( !isOpen() )
+        {
+                return 0;
+        }
+        im.lock();
+        if( bBlocking )
+        {
+                im.unlock();
+                cBlock.lock();
+                for(;;)
+                {
+                        im.lock();
+                        if( qb.getSize() == 0 && bOpen == false )
+                        {
+                                im.unlock();
+                                cBlock.unlock();
+                                return 0;
+                        }
+                        else if( qb.getSize() > 0 )
+                        {
+                                im.unlock();
+                                break;
+                        }
+                        im.unlock();
+
+                        cBlock.wait();
+                }
+
+                im.lock();
+                Bu::size iRet = qb.read( pBuf, nBytes );
+                im.unlock();
+
+                cBlock.unlock();
+                return iRet;
+        }
+        else
+        {
+                Bu::size iRet = qb.read( pBuf, nBytes );
+                im.unlock();
+
+                return iRet;
+        }
+}
+
+Bu::size Bu::Conduit::peek( void *pBuf, Bu::size nBytes )
+{
+        im.lock();
+        Bu::size iRet = qb.peek( pBuf, nBytes );
+        im.unlock();
+
+        return iRet;
+}
+
+Bu::size Bu::Conduit::peek( void *pBuf, Bu::size nBytes, Bu::size nSkip )
+{
+        im.lock();
+        Bu::size iRet = qb.peek( pBuf, nBytes, nSkip );
+        im.unlock();
+
+        return iRet;
+}
+
+Bu::size Bu::Conduit::write( const void *pBuf, Bu::size nBytes )
+{
+        im.lock();
+        if( bOpen == false )
+        {
+                im.unlock();
+                return 0;
+        }
+        Bu::size sRet = qb.write( pBuf, nBytes );
+        cBlock.signal();
+        im.unlock();
+
+        return sRet;
+}
+
+Bu::size Bu::Conduit::tell()
+{
+        im.lock();
+        Bu::size sRet = qb.tell();
+        im.unlock();
+        return sRet;
+}
+
+void Bu::Conduit::seek( Bu::size offset )
+{
+}
+
+void Bu::Conduit::setPos( Bu::size pos )
+{
+}
+
+void Bu::Conduit::setPosEnd( Bu::size pos )
+{
+}
+
+bool Bu::Conduit::isEos()
+{
+        im.lock();
+        bool bRet = qb.isEos();
+        im.unlock();
+        return bRet;
+}
+
+bool Bu::Conduit::isOpen()
+{
+        im.lock();
+        bool bRet = bOpen || (qb.getSize() > 0);
+        im.unlock();
+        return bRet;
+}
+
+void Bu::Conduit::flush()
+{
+}
+
+bool Bu::Conduit::canRead()
+{
+        im.lock();
+        bool bRet = qb.canRead();
+        im.unlock();
+        return bRet;
+}
+
+bool Bu::Conduit::canWrite()
+{
+        im.lock();
+        bool bRet = qb.canWrite();
+        im.unlock();
+        return bRet;
+}
+
+bool Bu::Conduit::isReadable()
+{
+        im.lock();
+        bool bRet = qb.isReadable();
+        im.unlock();
+        return bRet;
+}
+
+bool Bu::Conduit::isWritable()
+{
+        im.lock();
+        bool bRet = qb.isWritable();
+        im.unlock();
+        return bRet;
+}
+
+bool Bu::Conduit::isSeekable()
+{
+        im.lock();
+        bool bRet = qb.isSeekable();
+        im.unlock();
+        return bRet;
+}
+
+bool Bu::Conduit::isBlocking()
+{
+        im.lock();
+        bool bRet = bBlocking;
+        im.unlock();
+        return bRet;
+}
+
+void Bu::Conduit::setBlocking( bool bBlocking )
+{
+        im.lock();
+        this->bBlocking = bBlocking;
+        im.unlock();
+}
+
+void Bu::Conduit::setSize( Bu::size iSize )
+{
+}
+
+Bu::size Bu::Conduit::getSize() const
+{
+        im.lock();
+        Bu::size sRet = qb.getSize();
+        im.unlock();
+        return sRet;
+}
+
+Bu::size Bu::Conduit::getBlockSize() const
+{
+        im.lock();
+        Bu::size sRet = qb.getBlockSize();
+        im.unlock();
+        return sRet;
+}
+
+Bu::String Bu::Conduit::getLocation() const
+{
+        im.lock();
+        Bu::String sRet = qb.getLocation();
+        im.unlock();
+        return sRet;
+}
+