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;
+        }
+}
+