ChangeLogTag:Tue Aug 19 10:07:47 2003 Douglas C. Schmidt <schmidt@ace.cs.wustl.edu>

12 files changed, 239 insertions, 152 deletions

diff --git a/docs/tutorials/015/Client_i.cpp b/docs/tutorials/015/Client_i.cpp
index 6415fed1966..5039fde992e 100644
--- a/docs/tutorials/015/Client_i.cpp
+++ b/docs/tutorials/015/Client_i.cpp
@@ -39,7 +39,7 @@ int Client::open( void )
         // to ensure that our data is in the correct format when
         // received by the server.  Thus, we open the stream and
         // transfer ownership of the peer.
-    return stream().open( peer() );
+    return stream().open( peer(), 0, true);
 }
 
 // The remainder of the functions just delegate to the stream.
diff --git a/docs/tutorials/015/Compressor.cpp b/docs/tutorials/015/Compressor.cpp
index bbe76eb1a69..1cb1bf8bbf9 100644
--- a/docs/tutorials/015/Compressor.cpp
+++ b/docs/tutorials/015/Compressor.cpp
@@ -1,101 +1,60 @@
-
 // $Id$
 
+#include "ZlibCompressor.h"
 #include "Compressor.h"
 #include "ace/SOCK_Stream.h"
 
-Compressor::Compressor( void )
-        : Protocol_Task()
+#include <zlib.h>
+
+Compressor::Compressor(AlgorithmCode algorithm)
 {
-    ;
+    switch(algorithm)
+    {
+    case COMPRESSION_ZLIB:
+        compressor_ = new ZlibCompressor();
+        break;
+
+    default:
+        compressor_ = 0;
+        break;
+    }
 }
 
 Compressor::~Compressor(void)
 {
-    ;
+    delete compressor_;
 }
 
-/* This is where you insert your compression code.  Most compressors
-   want to work on a block of data instead of a byte-stream.
-   Fortunately the message block has a block that can be compressed.
-   Take a look at libz for a quick way to add compression to your
-   apps
- */
-int Compressor::send(ACE_Message_Block *message, ACE_Time_Value *timeout)
+int Compressor::send (ACE_Message_Block *message, ACE_Time_Value *timeout)
 {
-    ACE_UNUSED_ARG(message);
-    ACE_UNUSED_ARG(timeout);
-
-    ACE_DEBUG ((LM_INFO, "(%P|%t) Compressor::send() compressing (%s)\n", message->rd_ptr() ));
-
-        // Create a block to hold the compressed data.  I belive libz
-        // recommends a buffer about 10-20% larger than the source.
-        // Other libraries/algorithms may have their own quirks.
-    ACE_Message_Block * compressed = new ACE_Message_Block(
-        message->size() +16 );
+    if (compressor_ == 0)
+        return -1;
 
-        // Perform a bogus compression algorithm.  'CD' just tells me
-        // that this is compressed data and when we "decompress" we'll
-        // look for this signature to validate the data received.
-    ACE_OS::sprintf( compressed->wr_ptr(), "CD:%s", message->rd_ptr() );
-    compressed->wr_ptr( strlen(compressed->wr_ptr())+1 );
+    ACE_Message_Block* compressed = 0;
+    int retval = compressor_->send(message, timeout, compressed);
 
-        // Send the compressed data down the stream to the next module
+    // Send the compressed data down the stream to the next module
     this->put_next( compressed );
 
-        // We're done here.
+    // We're done here.
     message->release();
 
-    return( 0 );
+    return retval;
 }
 
-/* And here's the decompression side.  We've written Xmit/Recv so that
-   we're guaranteed to get an entire block of compressed data.  If
-   we'd used recv() in the Recv object then we might have gotten a
-   partial block and that may not decompress very nicely.
- */
-int Compressor::recv(ACE_Message_Block *message, ACE_Time_Value *timeout)
+int Compressor::recv (ACE_Message_Block *message, ACE_Time_Value *timeout)
 {
-    ACE_UNUSED_ARG(message);
-    ACE_UNUSED_ARG(timeout);
+    if (compressor_ == 0)
+        return -1;
 
-    ACE_DEBUG ((LM_INFO, "(%P|%t) Compress::recv() decompressing (%s)\n", message->rd_ptr() ));
+    ACE_Message_Block* decompressed = 0;
+    int retval = compressor_->recv(message, timeout, decompressed);
 
-        // Room for the decompressed data.  In the real world you
-        // would probably want to send the original (uncompressed)
-        // data size in the message.  You can predict the maximum
-        // possible decompression size but it's cheap and easy just to
-        // send that along.  Look again at how I do exacly that
-        // between Xmit and Recv.
-    ACE_Message_Block * decompressed = new ACE_Message_Block(
-        message->size() + 16 );
-
-        // Check for our signature.  Even when you use a real
-        // compression algorithm you may want to include your own
-        // signature so that you can verify the block.  It pays to be
-        // paranoid!
-    if( ACE_OS::strncmp( message->rd_ptr(), "CD:", 3  ) )
-    {
-        ACE_DEBUG ((LM_INFO, "(%P|%t) Improperly encompressed data.\n" ));
-        message->release();
-        return(-1);
-    }
-
-        // Skip past the signature before going any further.
-    message->rd_ptr( 3 );
-
-        // Perform a bogus decompression algorithm.  This is where you
-        // would feed to libz or your favorite decompressor.  (It's
-        // costly but you could invoke popen() on gzip!)
-    ACE_OS::sprintf( decompressed->wr_ptr(), "%s", message->rd_ptr() );
-    decompressed->wr_ptr( strlen(decompressed->wr_ptr())+1 );
-
-        // Recv the decompressed data down the stream to the next module
+    // Recv the decompressed data down the stream to the next module
     this->put_next( decompressed );
 
-        // We're done here.
+    // We're done here.
     message->release();
 
-    return( 0 );
+    return retval;
 }
-
diff --git a/docs/tutorials/015/Compressor.h b/docs/tutorials/015/Compressor.h
index cb4c7248cce..cf86f0caedc 100644
--- a/docs/tutorials/015/Compressor.h
+++ b/docs/tutorials/015/Compressor.h
@@ -1,35 +1,40 @@
 // $Id$
 
 #ifndef COMPRESSOR_H
-#define COMPRESSOR_h
+#define COMPRESSOR_H
 
 #include "Protocol_Task.h"
 
-/* A reallly dumb compression object.  (It actually adds 3 bytes to
-   every message block.)
-*/
+#include "CompressorBase.h"
+
 class Compressor : public Protocol_Task
 {
 public:
+    enum AlgorithmCode
+    {
+        COMPRESSION_NONE   = 0
+      , COMPRESSION_ZLIB   = 1
+      , COMPRESSION_RLE    = 2 // for future expansion
+      , COMPRESSION_BZIP2  = 3 // for future expansion
+    };
 
-  typedef Protocol_Task inherited;
-
-  Compressor (void);
+    Compressor(AlgorithmCode algorithm);
 
-  ~Compressor (void);
+    virtual ~Compressor (void);
 
 protected:
 
-  // This is called when the compressor is on the downstream side.
-  // We'll take the message, compress it and move it along to the next
-  // module.
-  int send (ACE_Message_Block *message,
-            ACE_Time_Value *timeout);
+    // This is called when the compressor is on the
+    // downstream side. We'll take the message, compress it
+    // and move it along to the next module.
+    virtual int send (ACE_Message_Block *message, ACE_Time_Value *timeout);
+
+    // This one is called on the upstream side.  No surprise: we
+    // decompress the data and send it on up the stream.
+    virtual int recv (ACE_Message_Block *message, ACE_Time_Value *timeout);
 
-  // This one is called on the upstream side.  No surprise: we
-  // decompress the data and send it on up the stream.
-  int recv (ACE_Message_Block *message,
-            ACE_Time_Value *timeout);
+private:
+    CompressorBase* compressor_;
 };
 
 #endif /* COMPRESSOR_H */
diff --git a/docs/tutorials/015/Handler.cpp b/docs/tutorials/015/Handler.cpp
index e940a8a8ce7..201d3605c41 100644
--- a/docs/tutorials/015/Handler.cpp
+++ b/docs/tutorials/015/Handler.cpp
@@ -9,7 +9,7 @@
    more when we talk about the stream in detail.  For now it's enough
    to know that Handler_Task::recv() will be invoked by the stream
    after data from the client has been received and processed (eg --
-   decrypted, uncompressed, and whatever else the protocol requires.)
+   uncompressed and whatever else the protocol requires.)
 */
 class Handler_Task : public Protocol_Task
 {
@@ -66,7 +66,7 @@ int Handler::open (void *)
         // read client requests and send our responses.  We also
         // provide a Handler_Task instance that will ultimately be
         // responsible for processing any client data we receive.
-    int rval = stream().open( this->peer(), new Handler_Task() );
+    int rval = stream().open( this->peer(), new Handler_Task(), false );
 
         // Of course, we have to account for the chance that the
         // stream's open() may fail.
@@ -161,7 +161,7 @@ int Handler_Task::recv(ACE_Message_Block * message,
     ACE_DEBUG ((LM_INFO, "(%P|%t) Handler_Task::recv() got (%s)\n", message->rd_ptr() ));
 
         // Create a response message to send to the client
-    ACE_Message_Block * response = new ACE_Message_Block( 128 );
+    ACE_Message_Block * response = new ACE_Message_Block( 1024 );
 
         // Nothing very original about this I'm afraid...
     ACE_OS::sprintf( response->wr_ptr(), "You Said:  (%s)", message->rd_ptr() );
diff --git a/docs/tutorials/015/Makefile.client b/docs/tutorials/015/Makefile.client
index 20680aea15b..440b813cfb3 100644
--- a/docs/tutorials/015/Makefile.client
+++ b/docs/tutorials/015/Makefile.client
@@ -11,8 +11,10 @@ FILES += Protocol_Stream
 FILES += Protocol_Task
 FILES += Xmit
 FILES += Recv
+FILES += CompressionSocket
 FILES += Compressor
-FILES += Crypt
+FILES += CompressorBase
+FILES += ZlibCompressor
 FILES += Client_i
 
 BUILD  = $(VBIN)
diff --git a/docs/tutorials/015/Makefile.server b/docs/tutorials/015/Makefile.server
index 109cecc8e90..bc44c1d3867 100644
--- a/docs/tutorials/015/Makefile.server
+++ b/docs/tutorials/015/Makefile.server
@@ -11,9 +11,10 @@ FILES += Protocol_Stream
 FILES += Protocol_Task
 FILES += Xmit
 FILES += Recv
+FILES += CompressionSocket
 FILES += Compressor
-FILES += Crypt
-
+FILES += CompressorBase
+FILES += ZlibCompressor
 FILES += Handler
 FILES += Server_i
 
diff --git a/docs/tutorials/015/Protocol_Stream.cpp b/docs/tutorials/015/Protocol_Stream.cpp
index e1c210ec119..6b7fd05c3cc 100644
--- a/docs/tutorials/015/Protocol_Stream.cpp
+++ b/docs/tutorials/015/Protocol_Stream.cpp
@@ -7,15 +7,12 @@
 #include "Recv.h"
 
 #include "Compressor.h"
-#include "Crypt.h"
 
 #include "ace/Stream_Modules.h"
 
 /* You can choose at compile time to include/exclude the protocol
    pieces.
 */
-#define ENABLE_COMPRESSION
-#define ENABLE_ENCRYPTION
 
 // The usual typedefs to make things easier to type.
 typedef ACE_Module<ACE_MT_SYNCH> Module;
@@ -49,7 +46,8 @@ Protocol_Stream::~Protocol_Stream (void)
  */
 int
 Protocol_Stream::open (ACE_SOCK_Stream &peer,
-                       Protocol_Task *reader)
+                       Protocol_Task *reader,
+                       bool isOriginator)
 {
   // Initialize our peer() to read/write the socket we're given
   peer_.set_handle (peer.get_handle ());
@@ -73,7 +71,7 @@ Protocol_Stream::open (ACE_SOCK_Stream &peer,
 
   // Add any other protocol tasks to the stream.  Each one is added at
   // the head.  The net result is that Xmit/Recv are at the tail.
-  if (this->open () == -1)
+  if (this->open (isOriginator) == -1)
     return -1;
 
   // If a reader task was provided then push that in as the upstream
@@ -96,32 +94,115 @@ Protocol_Stream::open (ACE_SOCK_Stream &peer,
   return 0;
 }
 
-/* Add the necessary protocol objects to the stream.  The way we're
-   pushing things on we will compress the data before encrypting it.
+/* Add the necessary protocol objects to the stream.
 */
+
 int
-Protocol_Stream::open (void)
+Protocol_Stream::open (bool isOriginator)
 {
-#if defined (ENABLE_ENCRYPTION)
-  if (stream ().push (new Module ("crypt",
-                                  new Crypt (),
-                                  new Crypt ())) == -1)
-    ACE_ERROR_RETURN ((LM_ERROR,
-                       "%p\n",
-                       "stream().push(crypt)"),
-                      -1);
-#endif /* ENABLE_ENCRYPTION */
+    // We are protocol version 1, we only support ZLIB compression currently.
 
-#if defined (ENABLE_COMPRESSION)
-  if (stream ().push (new Module ("compress",
-                                  new Compressor (),
-                                  new Compressor ())) == -1)
-    ACE_ERROR_RETURN ((LM_ERROR,
-                       "%p\n",
-                       "stream().push(comprssor)"),
-                      -1);
-#endif /* ENABLE_COMPRESSION */
-  return 0;
+    ProtocolVersion pversion = PROTOCOL_VERSION_1;
+    Compressor::AlgorithmCode acode = Compressor::COMPRESSION_ZLIB;
+
+     // If we are the originator, we put a message asking for
+    // what compression is to be used.
+
+    if (isOriginator)
+    {
+        ACE_Message_Block * message = new ACE_Message_Block( 1024 );
+
+        CompressionNegotiation neg;
+        neg.protocolVersion = pversion;
+        neg.algorithmCount = 1;
+        neg.algorithm[0] = acode;
+
+        memcpy(message->wr_ptr(), &neg, sizeof(neg));
+        message->wr_ptr(sizeof(neg));
+
+        if (put(message, 0) < 0)
+        {
+            return -1;
+        }
+
+        ACE_Message_Block * response;
+        if (get(response, 0) < 0)
+        {
+            return -1;
+        }
+
+        CompressionNegotiationReply negResponse;
+        memcpy(&negResponse, response->rd_ptr(), sizeof(negResponse));
+
+        response->release();
+        if (negResponse.errorCode != NEGOTIATE_OK)
+        {
+            return -1;
+        }
+    }
+    else
+    {
+        // Read the message to see what algorithm the originator speaks.
+
+        ACE_Message_Block* message;
+        if (get(message, 0) < 0)
+        {
+            return -1;
+        }
+
+        CompressionNegotiation neg;
+        memcpy(&neg, message->rd_ptr(), sizeof(neg));
+        message->release();
+
+        // Reply to handshake message.
+
+        CompressionNegotiationReply negResponse;
+        negResponse.errorCode = NEGOTIATE_OK;
+        negResponse.protocolVersion = PROTOCOL_VERSION_1;
+
+        if (neg.protocolVersion == PROTOCOL_VERSION_1)
+        {
+            // Ensure that incoming algorithms include the one we support.
+
+            negResponse.algorithm = Compressor::COMPRESSION_NONE;
+            for (int i = 0; i < neg.algorithmCount; i++)
+            {
+                if (neg.algorithm[i] == acode)
+                {
+                    negResponse.algorithm = acode;
+                    break;
+                }
+            }
+            if (negResponse.algorithm == Compressor::COMPRESSION_NONE)
+            {
+                negResponse.errorCode = NEGOTIATE_UNSUPPORTED_ALGORITHM;
+            }
+        }
+        else
+        {
+            negResponse.errorCode = NEGOTIATE_UNSUPPORTED_VERSION;
+        }
+
+        ACE_Message_Block* reply = new ACE_Message_Block( 1024 );
+        memcpy(reply->wr_ptr(), &negResponse, sizeof(negResponse));
+        reply->wr_ptr(sizeof(neg));
+
+        if (put(reply, 0) < 0)
+        {
+            return -1;
+        }
+    }
+
+    // Add the compression component to the pipeline.
+
+    if (stream ().push (new Module ("compress",
+                                  new Compressor (acode),
+                                  new Compressor (acode))) == -1)
+        ACE_ERROR_RETURN ((LM_ERROR,
+                          "%p\n",
+                           "stream().push(compressor)"),
+                         -1);
+    return 0;
 }
 
 // Closing the Protocol_Stream is as simple as closing the ACE_Stream.
diff --git a/docs/tutorials/015/Protocol_Stream.h b/docs/tutorials/015/Protocol_Stream.h
index 5209ed68907..f846f9efb54 100644
--- a/docs/tutorials/015/Protocol_Stream.h
+++ b/docs/tutorials/015/Protocol_Stream.h
@@ -33,7 +33,8 @@ public:
   // reader task just below the stream head so that it can process
   // data read from the peer.
   int open (ACE_SOCK_Stream &peer,
-            Protocol_Task *reader = 0);
+            Protocol_Task *reader,
+            bool isOriginator);
 
   // Close the stream.  All of the tasks & modules will also be
   // closed.
@@ -64,6 +65,32 @@ public:
   }
 
 private:
+    enum ProtocolVersion
+    {
+        PROTOCOL_VERSION_1 = 1
+    };
+
+    enum CompressionNegotiationErrorCode
+    {
+        NEGOTIATE_OK = 0
+      , NEGOTIATE_UNSUPPORTED_VERSION = -1
+      , NEGOTIATE_UNSUPPORTED_ALGORITHM = -2
+    };
+
+    struct CompressionNegotiation
+    {
+        unsigned char protocolVersion;
+        unsigned char algorithmCount;
+        unsigned char algorithm[5]; // as of protocol version 1.
+    };
+
+    struct CompressionNegotiationReply
+    {
+        signed char errorCode;
+        unsigned char protocolVersion;
+        unsigned char algorithm;
+    };
+
   // Our peer connection
   ACE_SOCK_Stream peer_;
 
@@ -71,16 +98,18 @@ private:
   Stream stream_;
 
   // A task which is capable of receiving data on a socket.
-  // Note that this is only useful by server-side applications.
+  // Note that this is only useful by client-side applications.
   Recv *recv_;
 
+  bool isHandshakeComplete_;
+
   Stream &stream (void)
   {
     return this->stream_;
   }
 
   // Install the protocol tasks into the stream.
-  int open (void);
+  int open (bool isOriginator);
 };
 
 #endif /* PROTOCOL_STREAM_H */
diff --git a/docs/tutorials/015/Protocol_Task.cpp b/docs/tutorials/015/Protocol_Task.cpp
index 6060fb1ddfa..da202a50b46 100644
--- a/docs/tutorials/015/Protocol_Task.cpp
+++ b/docs/tutorials/015/Protocol_Task.cpp
@@ -46,7 +46,8 @@ int Protocol_Task::svc(void)
    we're moving data upstream or downstream and invoke the appropriate
    virtual function to handle it.
 */
-int Protocol_Task::process(ACE_Message_Block * message, ACE_Time_Value *timeout)
+int Protocol_Task::process(ACE_Message_Block * message, 
+                           ACE_Time_Value *timeout)
 {
     if( this->is_writer() )
     {
diff --git a/docs/tutorials/015/Xmit.cpp b/docs/tutorials/015/Xmit.cpp
index 28be01b1d25..4f2ed4f274f 100644
--- a/docs/tutorials/015/Xmit.cpp
+++ b/docs/tutorials/015/Xmit.cpp
@@ -47,10 +47,10 @@ int Xmit::send(ACE_Message_Block *message, ACE_Time_Value *timeout)
 {
     int rval;
 
-    ACE_DEBUG ((LM_INFO, "(%P|%t) Xmit::send() sending (%s)(%d)\n", message->rd_ptr(), message->length() ));
+    ACE_DEBUG ((LM_INFO, "(%P|%t) Xmit::send() sending buff of len %d\n", message->length() ));
 
      /* Since we're going to be sending data that may have been
-        compressed and encrypted it's probably important for the
+        compressed it's probably important for the
         receiver to get an entire "block" instead of having a
         partial read.
 
@@ -75,7 +75,8 @@ int Xmit::send(ACE_Message_Block *message, ACE_Time_Value *timeout)
         containing msize and the message data and send it all at
         once.
      */
-    rval = this->peer().send_n( message->rd_ptr(), message->length(), 0, timeout );
+    rval = this->peer().send_n(message->rd_ptr(), 
+                               message->length(), 0, timeout);
 
      // Release the message block since we're done with it.
     message->release();
diff --git a/docs/tutorials/015/client.cpp b/docs/tutorials/015/client.cpp
index 93041dcdfdd..af4433d4a23 100644
--- a/docs/tutorials/015/client.cpp
+++ b/docs/tutorials/015/client.cpp
@@ -8,42 +8,51 @@
 
 int main(int argc, char *argv[])
 {
-        // How many messages will we send?
+    // How many messages will we send?
     int mcount = argc > 1 ? ACE_OS::atoi(argv[1]) : 3;
 
-        // Construct a Client with our desired endpoint.
+    // Construct a Client with our desired endpoint.
     Client client(ACE_DEFAULT_SERVER_PORT,ACE_DEFAULT_SERVER_HOST);
 
-        // Attempt to open the connection to the server.
+    // Attempt to open the connection to the server.
     if( client.open() == -1 )
     {
         ACE_ERROR_RETURN ((LM_ERROR, "%p\n", "Client::open()"), -1);
     }
 
-        // Send a few messages to the server and get some responses...
+    // Send a few messages to the server and get some responses...
     for( int i = 0 ; i < mcount ; ++i )
     {
-            // Since we'll be using a Protocol Stream (even though we
-            // don't know that at this level) we require the use of
-            // ACE_Message_Block objects to send/receive data.
-        ACE_Message_Block * message = new ACE_Message_Block( 128 );
-
-            // Construct a silly message to send to the server.
-            // Notice that we're careful to add one to the strlen() so
-            // that we also send the end-of-string NULL character.
-        ACE_OS::sprintf (message->wr_ptr (), "This is message %d.", i);
+        // Since we'll be using a Protocol Stream (even though we
+        // don't know that at this level) we require the use of
+        // ACE_Message_Block objects to send/receive data.
+        ACE_Message_Block * message = new ACE_Message_Block( 1024 );
+
+        // Construct a silly message to send to the server.
+        // Notice that we're careful to add one to the strlen() so
+        // that we also send the end-of-string NULL character.
+        ACE_OS::sprintf (message->wr_ptr (),
+                         "This is message %d. It is a bit verbose because "
+                          "we want there to be an opportunity for some compression to occur. "
+                          "That's why it has a whofflly style and goes on and on and on.",
+                         i);
         message->wr_ptr (strlen (message->rd_ptr ())+1);
 
-            // client will take ownership of the message block so that
-            // we don't have to remember to release().  We *do* have
-            // to remember not to use it after put() since it may be
-            // released almost immediately.
-        client.put( message );
+        // client will take ownership of the message block so that
+        // we don't have to remember to release().  We *do* have
+        // to remember not to use it after put() since it may be
+        // released almost immediately.
+        if (client.put( message ) < 0)
+        {
+            fprintf(stderr, "%s:%d: client put failed.\n", __FILE__, __LINE__);
+            return -1;
+        }
 
         ACE_Message_Block * response;
 
-            // get() takes an ACE_Message_Block pointer reference.  We then
-            // assume ownership of it and must release() when we're done.
+        // get() takes an ACE_Message_Block pointer reference.  We then
+        // assume ownership of it and must release() when we're done.
+
         if( client.get( response ) == -1 )
         {
             ACE_DEBUG ((LM_INFO, "(%P|%t) Failed to get response from server\n" ));
diff --git a/docs/tutorials/015/page01.html b/docs/tutorials/015/page01.html
index 0709a089c19..7d66f966e77 100644
--- a/docs/tutorials/015/page01.html
+++ b/docs/tutorials/015/page01.html
@@ -98,8 +98,7 @@ now from the previous tutorials.
 etc.)
 </UL>
 <P>
-<font size=-1>* Ok, I didn't really implement encryption and
-        compression objects.  I'll leave that as a thought
+<font size=-1>* Ok, I didn't really implement encryption objects.  I'll leave that as a thought
         exercise!</font>
 <P><HR WIDTH="100%">
 <CENTER>[<A HREF="../online-tutorials.html">Tutorial Index</A>] [<A HREF="page02.html">Continue This Tutorial</A>]</CENTER>