Switched ito* to synchro*, except the server, I'm thinking of takeing the core

in a different direction anyway.

Added the Deflate class, it uses zlib, and can do raw (headerless) deflate
streams, zlib format, or gzip format.  It's easy to use and quite versitile.

13 files changed, 404 insertions, 45 deletions

diff --git a/default.bld b/default.bld
index 8f98db0..6666fe2 100644
--- a/default.bld
+++ b/default.bld
@@ -167,6 +167,11 @@ target ["tests/bzip2", "tests/streamstack"]
         LDFLAGS += "-lbz2";
 }
 
+target ["tests/deflate"]
+{
+        LDFLAGS += "-lz";
+}
+
 target ["tests/itoserver", "tests/socketblock", "tests/itoheap",
         "tests/itoqueue1", "tests/itoqueue2", "tests/conduit"]
 {
diff --git a/src/bzip2.cpp b/src/bzip2.cpp
index 5c35a26..0ff5444 100644
--- a/src/bzip2.cpp
+++ b/src/bzip2.cpp
@@ -33,7 +33,7 @@ void Bu::BZip2::start()
         bzState.bzfree = NULL;
         bzState.opaque = NULL;
 
-        nBufSize = 50000;
+        nBufSize = 64*1024;
         pBuf = new char[nBufSize];
 }
 
diff --git a/src/conduit.h b/src/conduit.h
index 72b8d52..9babaaf 100644
--- a/src/conduit.h
+++ b/src/conduit.h
@@ -11,8 +11,8 @@
 #include "bu/stream.h"
 #include "bu/string.h"
 #include "bu/queuebuf.h"
-#include "bu/itomutex.h"
-#include "bu/itocondition.h"
+#include "bu/mutex.h"
+#include "bu/condition.h"
 
 namespace Bu
 {
@@ -54,8 +54,8 @@ namespace Bu
 
         private:
                 QueueBuf qb;
-                mutable ItoMutex im;
-                ItoCondition cBlock;
+                mutable Mutex im;
+                Condition cBlock;
                 bool bBlocking;
                 bool bOpen;
         };
diff --git a/src/deflate.cpp b/src/deflate.cpp
new file mode 100644
index 0000000..aec2a18
--- /dev/null
+++ b/src/deflate.cpp
@@ -0,0 +1,242 @@
+/*
+ * Copyright (C) 2007-2011 Xagasoft, All rights reserved.
+ *
+ * This file is part of the libbu++ library and is released under the
+ * terms of the license contained in the file LICENSE.
+ */
+
+#include "bu/deflate.h"
+#include "bu/trace.h"
+
+using namespace Bu;
+
+Bu::Deflate::Deflate( Bu::Stream &rNext, int nCompression, Format eFmt ) :
+        Bu::Filter( rNext ),
+        nCompression( nCompression ),
+        sTotalOut( 0 ),
+        eFmt( eFmt ),
+        bEos( false )
+{
+        TRACE( nCompression );
+        start();
+}
+
+Bu::Deflate::~Deflate()
+{
+        TRACE();
+        stop();
+}
+
+void Bu::Deflate::start()
+{
+        TRACE();
+        zState.zalloc = NULL;
+        zState.zfree = NULL;
+        zState.opaque = NULL;
+        zState.state = NULL;
+
+        nBufSize = 64*1024;
+        pBuf = new char[nBufSize];
+}
+
+Bu::size Bu::Deflate::stop()
+{
+        TRACE();
+        if( zState.state )
+        {
+                if( bReading )
+                {
+                        inflateEnd( &zState );
+                        delete[] pBuf;
+                        pBuf = NULL;
+                        return 0;
+                }
+                else
+                {
+                        for(;;)
+                        {
+                                zState.next_in = NULL;
+                                zState.avail_in = 0;
+                                zState.avail_out = nBufSize;
+                                zState.next_out = (Bytef *)pBuf;
+                                int res = deflate( &zState, Z_FINISH );
+                                if( zState.avail_out < nBufSize )
+                                {
+                                        sTotalOut += rNext.write( pBuf, nBufSize-zState.avail_out );
+                                }
+                                if( res == Z_STREAM_END )
+                                        break;
+                        }
+                        deflateEnd( &zState );
+                        delete[] pBuf;
+                        pBuf = NULL;
+                        return sTotalOut;
+                }
+        }
+        return 0;
+}
+
+void Bu::Deflate::zError( int code )
+{
+        TRACE( code );
+        switch( code )
+        {
+                case Z_OK:
+                case Z_STREAM_END:
+                case Z_NEED_DICT:
+                        return;
+
+                case Z_ERRNO:
+                        throw ExceptionBase("Deflate: Errno - %s", zState.msg );
+
+                case Z_STREAM_ERROR:
+                        throw ExceptionBase("Deflate: Stream Error - %s", zState.msg );
+
+                case Z_DATA_ERROR:
+                        throw ExceptionBase("Deflate: Data Error - %s", zState.msg );
+
+                case Z_MEM_ERROR:
+                        throw ExceptionBase("Deflate: Mem Error - %s", zState.msg );
+
+                case Z_BUF_ERROR:
+                        throw ExceptionBase("Deflate: Buf Error - %s", zState.msg );
+
+                case Z_VERSION_ERROR:
+                        throw ExceptionBase("Deflate: Version Error - %s", zState.msg );
+
+                default:
+                        throw ExceptionBase("Deflate: Unknown error encountered - %s.", zState.msg );
+        
+        }
+}
+
+Bu::size Bu::Deflate::read( void *pData, Bu::size nBytes )
+{
+        TRACE( pData, nBytes );
+        if( !zState.state )
+        {
+                bReading = true;
+                if( eFmt&AutoDetect )
+                        inflateInit2( &zState, 32+15 ); // Auto-detect, large window
+                else if( eFmt == Raw )
+                        inflateInit2( &zState, -15 ); // Raw
+                else if( eFmt == Zlib )
+                        inflateInit2( &zState, 15 ); // Zlib
+                else if( eFmt == Gzip )
+                        inflateInit2( &zState, 16+15 ); // GZip
+                else
+                        throw Bu::ExceptionBase("Format mode for deflate read.");
+                zState.next_in = (Bytef *)pBuf;
+                zState.avail_in = 0;
+        }
+        if( bReading == false )
+                throw ExceptionBase("This deflate filter is in writing mode, you can't read.");
+
+        int nRead = 0;
+        int nReadTotal = zState.total_out;
+        zState.next_out = (Bytef *)pData;
+        zState.avail_out = nBytes;
+        for(;;)
+        {
+                int ret = inflate( &zState, Z_NO_FLUSH );
+                printf("inflate returned %d; avail in=%d, out=%d\n", ret,
+                                zState.avail_in, zState.avail_out );
+        
+                nReadTotal += nRead-zState.avail_out;
+
+                if( ret == Z_STREAM_END )
+                {
+                        bEos = true;
+                        if( zState.avail_in > 0 )
+                        {
+                                if( rNext.isSeekable() )
+                                {
+                                        rNext.seek( -zState.avail_in );
+                                }
+                        }
+                        return nBytes-zState.avail_out;
+                }
+                if( ret != Z_BUF_ERROR )
+                        zError( ret );
+
+                if( zState.avail_out )
+                {
+                        if( zState.avail_in == 0 )
+                        {
+                                nRead = rNext.read( pBuf, nBufSize );
+                                if( nRead == 0 && rNext.isEos() )
+                                {
+                                        throw Bu::ExceptionBase("Premature end of underlying "
+                                                        "stream found reading deflate stream.");
+                                }
+                                zState.next_in = (Bytef *)pBuf;
+                                zState.avail_in = nRead;
+                        }
+                }
+                else
+                {
+                        return nBytes-zState.avail_out;
+                }
+        }
+        return 0;
+}
+
+Bu::size Bu::Deflate::write( const void *pData, Bu::size nBytes )
+{
+        TRACE( pData, nBytes );
+        if( !zState.state )
+        {
+                bReading = false;
+                int iFmt = eFmt&Gzip;
+                if( iFmt == Raw )
+                        deflateInit2( &zState, nCompression, Z_DEFLATED, -15, 9,
+                                Z_DEFAULT_STRATEGY );
+                else if( iFmt == Zlib )
+                        deflateInit2( &zState, nCompression, Z_DEFLATED, 15, 9,
+                                Z_DEFAULT_STRATEGY );
+                else if( iFmt == Gzip )
+                        deflateInit2( &zState, nCompression, Z_DEFLATED, 16+15, 9,
+                                Z_DEFAULT_STRATEGY );
+                else
+                        throw Bu::ExceptionBase("Invalid format for deflate.");
+        }
+        if( bReading == true )
+                throw ExceptionBase("This deflate filter is in reading mode, you can't write.");
+
+        zState.next_in = (Bytef *)pData;
+        zState.avail_in = nBytes;
+        for(;;)
+        {
+                zState.avail_out = nBufSize;
+                zState.next_out = (Bytef *)pBuf;
+
+                zError( deflate( &zState, Z_NO_FLUSH ) );
+
+                if( zState.avail_out < nBufSize )
+                {
+                        sTotalOut += rNext.write( pBuf, nBufSize-zState.avail_out );
+                }
+                if( zState.avail_in == 0 )
+                        break;
+        }
+
+        return nBytes;
+}
+
+bool Bu::Deflate::isOpen()
+{
+        TRACE();
+        return (zState.state != NULL);
+}
+
+bool Bu::Deflate::isEos()
+{
+        TRACE();
+        return bEos;
+}
+
+Bu::size Bu::Deflate::getCompressedSize()
+{
+        return sTotalOut;
+}
+
diff --git a/src/deflate.h b/src/deflate.h
new file mode 100644
index 0000000..cab9b51
--- /dev/null
+++ b/src/deflate.h
@@ -0,0 +1,63 @@
+/*
+ * Copyright (C) 2007-2011 Xagasoft, All rights reserved.
+ *
+ * This file is part of the libbu++ library and is released under the
+ * terms of the license contained in the file LICENSE.
+ */
+
+#ifndef BU_DEFLATE_H
+#define BU_DEFLATE_H
+
+#include <stdint.h>
+#include <zlib.h>
+
+#include "bu/filter.h"
+
+namespace Bu
+{
+        /**
+         *
+         *@ingroup Streams
+         */
+        class Deflate : public Bu::Filter
+        {
+        public:
+                enum Format
+                {
+                        Raw                     = 0x01,
+                        Zlib            = 0x02,
+                        Gzip            = 0x03,
+                        AutoDetect      = 0x04,
+
+                        AutoRaw         = 0x04|0x01,
+                        AutoZlib        = 0x04|0x02,
+                        AutoGzip        = 0x04|0x03
+                };
+
+                Deflate( Bu::Stream &rNext, int nCompression=9, Format eFmt=AutoRaw );
+                virtual ~Deflate();
+
+                virtual void start();
+                virtual Bu::size stop();
+                virtual Bu::size read( void *pBuf, Bu::size nBytes );
+                virtual Bu::size write( const void *pBuf, Bu::size nBytes );
+
+                virtual bool isOpen();
+                virtual bool isEos();
+
+                Bu::size getCompressedSize();
+
+        private:
+                void zError( int code );
+                z_stream zState;
+                bool bReading;
+                int nCompression;
+                char *pBuf;
+                uint32_t nBufSize;
+                Bu::size sTotalOut;
+                Format eFmt;
+                bool bEos;
+        };
+}
+
+#endif
diff --git a/src/itoserver.h b/src/itoserver.h
index 75b3349..b1f5479 100644
--- a/src/itoserver.h
+++ b/src/itoserver.h
@@ -18,7 +18,7 @@
 #include "bu/list.h"
 #include "bu/thread.h"
 #include "bu/mutex.h"
-#include "bu/itoqueue.h"
+#include "bu/synchroqueue.h"
 #include "bu/set.h"
 
 #include "bu/clientlink.h"
@@ -82,7 +82,7 @@ namespace Bu
                                         int nTimeoutSec, int nTimeoutUSec );
                         virtual ~ItoClient();
 
-                        typedef ItoQueue<Bu::String *> StringQueue;
+                        typedef SynchroQueue<Bu::String *> StringQueue;
                         StringQueue qMsg;
 
                 protected:
@@ -129,7 +129,7 @@ namespace Bu
                 typedef Hash<int,TcpServerSocket *> ServerHash;
                 ServerHash hServers;
                 typedef Hash<int,ItoClient *> ClientHash;
-                typedef ItoQueue<ItoClient *> ClientQueue;
+                typedef SynchroQueue<ItoClient *> ClientQueue;
                 ClientHash hClients;
                 ClientQueue qClientCleanup;
                 Mutex imClients;
diff --git a/src/itoatom.h b/src/synchroatom.h
index 3659f4e..fb02054 100644
--- a/src/itoatom.h
+++ b/src/synchroatom.h
@@ -5,13 +5,12 @@
  * terms of the license contained in the file LICENSE.
  */
 
-#ifndef BU_ITO_ATOM_H
-#define BU_ITO_ATOM_H
+#ifndef BU_SYNCHRO_ATOM_H
+#define BU_SYNCHRO_ATOM_H
 
 #include <pthread.h>
 
-#include "itomutex.h"
-#include "itocondition.h"
+#include "bu/mutex.h"
 
 namespace Bu
 {
@@ -20,22 +19,22 @@ namespace Bu
          *@ingroup Threading
          */
         template <class T>
-        class ItoAtom
+        class SynchroAtom
         {
         public:
                 /**
                  * Construct an empty queue.
                  */
-                ItoAtom()
+                SynchroAtom()
                 {
                 }
 
-                ItoAtom( const T &src ) :
+                SynchroAtom( const T &src ) :
                         data( src )
                 {
                 }
                 
-                ~ItoAtom()
+                ~SynchroAtom()
                 {
                 }
 
@@ -57,7 +56,7 @@ namespace Bu
         private:
                 T data;
 
-                ItoMutex mOperate;      /**< The master mutex, used on all operations. */
+                Mutex mOperate; /**< The master mutex, used on all operations. */
         };
 };
 
diff --git a/src/itocounter.cpp b/src/synchrocounter.cpp
index 0c6e06c..48bbe21 100644
--- a/src/itocounter.cpp
+++ b/src/synchrocounter.cpp
@@ -5,4 +5,4 @@
  * terms of the license contained in the file LICENSE.
  */
 
-#include "bu/itocounter.h"
+#include "bu/synchrocounter.h"
diff --git a/src/itocounter.h b/src/synchrocounter.h
index 10126a5..d201bee 100644
--- a/src/itocounter.h
+++ b/src/synchrocounter.h
@@ -5,10 +5,10 @@
  * terms of the license contained in the file LICENSE.
  */
 
-#ifndef BU_ITO_COUNTER_H
-#define BU_ITO_COUNTER_H
+#ifndef BU_SYNCHRO_COUNTER_H
+#define BU_SYNCHRO_COUNTER_H
 
-#include "mutex.h"
+#include "bu/mutex.h"
 
 namespace Bu
 {
@@ -18,15 +18,15 @@ namespace Bu
          *@ingroup Threading Containers
          */
         template <class T>
-        class ItoCounter
+        class SynchroCounter
         {
         public:
-                ItoCounter() :
+                SynchroCounter() :
                         tCounter( 0 )
                 {
                 }
                 
-                virtual ~ItoCounter()
+                virtual ~SynchroCounter()
                 {
                 }
 
diff --git a/src/itoheap.cpp b/src/synchroheap.cpp
index 21ccef8..5dcce33 100644
--- a/src/itoheap.cpp
+++ b/src/synchroheap.cpp
@@ -5,5 +5,5 @@
  * terms of the license contained in the file LICENSE.
  */
 
-#include "bu/itoheap.h"
+#include "bu/synchroheap.h"
 
diff --git a/src/itoheap.h b/src/synchroheap.h
index a5aad05..4dd898d 100644
--- a/src/itoheap.h
+++ b/src/synchroheap.h
@@ -5,28 +5,25 @@
  * terms of the license contained in the file LICENSE.
  */
 
-#ifndef BU_ITO_HEAP_H
-#define BU_ITO_HEAP_H
+#ifndef BU_SYNCHRO_HEAP_H
+#define BU_SYNCHRO_HEAP_H
 
 #include "bu/heap.h"
-#include "bu/itomutex.h"
-#include "bu/itocondition.h"
+#include "bu/mutex.h"
+#include "bu/condition.h"
 
 namespace Bu
 {
-        class ItoMutex;
-        class ItoCondition;
-
         template<typename item, typename cmpfunc=__basicLTCmp<item>,
                 typename itemalloc=std::allocator<item> >
-        class ItoHeap
+        class SynchroHeap
         {
         public:
-                ItoHeap()
+                SynchroHeap()
                 {
                 }
 
-                virtual ~ItoHeap()
+                virtual ~SynchroHeap()
                 {
                 }
 
@@ -145,8 +142,8 @@ namespace Bu
 
         private:
                 Heap< item, cmpfunc, itemalloc > hData;
-                ItoMutex imData;
-                ItoCondition icBlock;
+                Mutex imData;
+                Condition icBlock;
         };
 };
 
diff --git a/src/itoqueue.h b/src/synchroqueue.h
index 039e09c..79d5e49 100644
--- a/src/itoqueue.h
+++ b/src/synchroqueue.h
@@ -5,19 +5,19 @@
  * terms of the license contained in the file LICENSE.
  */
 
-#ifndef BU_ITO_QUEUE_H
-#define BU_ITO_QUEUE_H
+#ifndef BU_SYNCHRO_QUEUE_H
+#define BU_SYNCHRO_QUEUE_H
 
 #include <pthread.h>
 
-#include "mutex.h"
-#include "condition.h"
+#include "bu/mutex.h"
+#include "bu/condition.h"
 
 namespace Bu
 {
         /**
          * A thread-safe queue class.  This class is a very simple queue with some
-         * cool extra functionality for use with the Ito system.  The main extra
+         * cool extra functionality for use with the Synchro system.  The main extra
          * that it provides is the option to either dequeue without blocking, with
          * infinite blocking, or with timed blocking, which will return a value if
          * something is enqueued within the specified time limit, or NULL if the
@@ -25,7 +25,7 @@ namespace Bu
          *@ingroup Threading Containers
          */
         template <class T>
-        class ItoQueue
+        class SynchroQueue
         {
         private:
                 /**
@@ -41,7 +41,7 @@ namespace Bu
                 /**
                  * Construct an empty queue.
                  */
-                ItoQueue() :
+                SynchroQueue() :
                         pStart( NULL ),
                         pEnd( NULL ),
                         nSize( 0 )
@@ -54,7 +54,7 @@ namespace Bu
                  * pointers without cleaning up the memory they pointed to.  Make sure
                  * you're queue is empty before allowing it to be destroyed!
                  */
-                ~ItoQueue()
+                ~SynchroQueue()
                 {
                         Item *pCur = pStart;
                         while( pCur )
diff --git a/src/tests/deflate.cpp b/src/tests/deflate.cpp
new file mode 100644
index 0000000..9796408
--- /dev/null
+++ b/src/tests/deflate.cpp
@@ -0,0 +1,53 @@
+/*
+ * Copyright (C) 2007-2011 Xagasoft, All rights reserved.
+ *
+ * This file is part of the libbu++ library and is released under the
+ * terms of the license contained in the file LICENSE.
+ */
+
+#include "bu/deflate.h"
+#include "bu/file.h"
+
+int main( int argc, char *argv[] )
+{
+        if( argc < 3 )
+        {
+                printf("usage:  %s <in> <out>\n", argv[0] );
+                return -1;
+        }
+
+        char buf[1024];
+        size_t nRead;
+
+        /*
+        Bu::File fin( argv[1], Bu::File::Read );
+        fin.seek( 4 );
+        Bu::Deflate def( fin );
+        
+        Bu::File f( argv[2], Bu::File::WriteNew );
+
+        for(;;)
+        {
+                nRead = def.read( buf, 1024 );
+                if( nRead > 0 )
+                        f.write( buf, nRead );
+                if( def.isEos() )
+                        break;
+        }
+        */
+
+        Bu::File fin( argv[1], Bu::File::Read );
+        
+        Bu::File f( argv[2], Bu::File::WriteNew );
+        Bu::Deflate def( f, 9, Bu::Deflate::Gzip );
+
+        for(;;)
+        {
+                nRead = fin.read( buf, 1024 );
+                if( nRead > 0 )
+                        def.write( buf, nRead );
+                if( fin.isEos() )
+                        break;
+        }
+}
+