ChangeLogTag:Fri Apr 14 15:53:21 UTC 2006 Steve Huston <shuston@riverace.com>

6 files changed, 704 insertions, 117 deletions

diff --git a/ace/SOCK_Stream.h b/ace/SOCK_Stream.h
index 92000947271..9537162a4d3 100644
--- a/ace/SOCK_Stream.h
+++ b/ace/SOCK_Stream.h
@@ -81,8 +81,22 @@ public:
   /// Destructor.
   ~ACE_SOCK_Stream (void);
 
-  // = I/O functions.
-
+  /** @name Counted send/receive methods
+   *
+   * The counted send/receive methods attempt to send a specified number of
+   * bytes even if they must block and retry the operation in order to
+   * transfer the entire amount. The time spent blocking for the entire
+   * transfer can be limited by a specified ACE_Time_Value object which is
+   * a relative time (i.e., a fixed amount of time, not an absolute time
+   * of day). These methods return the count of transferred bytes, or -1
+   * if an error occurs or the operation times out. In error or timeout
+   * situations it's possible that some data was transferred before the error
+   * or timeout. The @c bytes_transferred parameter is used to obtain the
+   * count of bytes transferred before the error or timeout occurred. If the
+   * total specified number of bytes is transferred without error, the
+   * method return value should equal the value of @c bytes_transferred.
+   */
+  //@{
   /// Try to recv exactly @a len bytes into @a buf from the connected socket.
   ssize_t recv_n (void *buf,
                   size_t len,
@@ -128,6 +142,8 @@ public:
                    const ACE_Time_Value *timeout = 0,
                    size_t *bytes_transferred = 0) const;
 
+  //@}
+
   // = Send/receive ``urgent'' data (see TCP specs...).
   ssize_t send_urg (const void *ptr,
                     size_t len = sizeof (char),
diff --git a/ace/SSL/SSL_Asynch_Stream.cpp b/ace/SSL/SSL_Asynch_Stream.cpp
index 6ed1282b82d..23ab69059fd 100644
--- a/ace/SSL/SSL_Asynch_Stream.cpp
+++ b/ace/SSL/SSL_Asynch_Stream.cpp
@@ -8,25 +8,9 @@ ACE_RCSID (ACE_SSL,
 #if OPENSSL_VERSION_NUMBER > 0x0090581fL && ((defined (ACE_WIN32) && !defined (ACE_HAS_WINCE)) || (defined (ACE_HAS_AIO_CALLS)))
 
 #if defined (ACE_WIN32)
-
-# define A_RESULT     ACE_WIN32_Asynch_Result
-# define ARS_RESULT   ACE_WIN32_Asynch_Read_Stream_Result
-# define AWS_RESULT   ACE_WIN32_Asynch_Write_Stream_Result
-
-# define ERR_CANCELED ERROR_OPERATION_ABORTED
-
 # include "ace/WIN32_Proactor.h"
-
 #else
-
-# define A_RESULT     ACE_POSIX_Asynch_Result
-# define ARS_RESULT   ACE_POSIX_Asynch_Read_Stream_Result
-# define AWS_RESULT   ACE_POSIX_Asynch_Write_Stream_Result
-
-# define ERR_CANCELED ECANCELED
-
 # include "ace/POSIX_Proactor.h"
-
 #endif  /* ACE_WIN32 */
 
 #include "ace/OS_NS_string.h"
@@ -36,28 +20,6 @@ ACE_RCSID (ACE_SSL,
 
 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
 
-// ************************************************************
-//  SSL Asynchronous Write Result
-// ************************************************************
-
-class ACE_SSL_Asynch_Write_Stream_Result
-  : public AWS_RESULT
-{
-  /// Factory class will have special permissions.
-  friend class ACE_SSL_Asynch_Stream;
-
-protected:
-
-  ACE_SSL_Asynch_Write_Stream_Result (ACE_Handler &handler,
-                                      ACE_HANDLE handle,
-                                      ACE_Message_Block &message_block,
-                                      size_t bytes_to_read,
-                                      const void* act,
-                                      ACE_HANDLE event,
-                                      int priority,
-                                      int signal_number);
-};
-
 ACE_SSL_Asynch_Write_Stream_Result::ACE_SSL_Asynch_Write_Stream_Result
   (ACE_Handler &       handler,
    ACE_HANDLE          handle,
@@ -80,29 +42,6 @@ ACE_SSL_Asynch_Write_Stream_Result::ACE_SSL_Asynch_Write_Stream_Result
 {
 }
 
-// ************************************************************
-//  SSL Asynchronous Read Result
-// ************************************************************
-class ACE_SSL_Asynch_Read_Stream_Result
-  : public ARS_RESULT
-{
-  /// Factory class will have special permissions.
-  friend class ACE_SSL_Asynch_Stream;
-
-protected:
-
-  ACE_SSL_Asynch_Read_Stream_Result (ACE_Handler &handler,
-                                     ACE_HANDLE handle,
-                                     ACE_Message_Block &message_block,
-                                     size_t bytes_to_read,
-                                     const void* act,
-                                     ACE_HANDLE event,
-                                     int priority,
-                                     int signal_number);
-};
-
-
-
 ACE_SSL_Asynch_Read_Stream_Result::ACE_SSL_Asynch_Read_Stream_Result
   (ACE_Handler &        handler,
    ACE_HANDLE           handle,
@@ -125,23 +64,7 @@ ACE_SSL_Asynch_Read_Stream_Result::ACE_SSL_Asynch_Read_Stream_Result
 {
 }
 
-
-// ************************************************************
-//   Faked Result. It is used for close notification
-// ************************************************************
-class ACE_SSL_Asynch_Result : public A_RESULT
-{
-public:
-    ACE_SSL_Asynch_Result (ACE_Handler & handler);
-
-    void complete (size_t bytes_transferred,
-                   int    success,
-                   const  void * completion_key,
-                   u_long error);
-};
-
-ACE_SSL_Asynch_Result::ACE_SSL_Asynch_Result
-  (ACE_Handler & handler)
+ACE_SSL_Asynch_Result::ACE_SSL_Asynch_Result (ACE_Handler & handler)
   : A_RESULT (handler.proxy (),
               0,          // act,
               ACE_INVALID_HANDLE,
@@ -163,7 +86,7 @@ ACE_SSL_Asynch_Result::complete (size_t /* bytes_transferred */,
 }
 
 // ************************************************************
-//  ACE_SSL_Asynch_Stream Constructor / Desctructor
+//  ACE_SSL_Asynch_Stream Constructor / Destructor
 // ************************************************************
 ACE_SSL_Asynch_Stream::ACE_SSL_Asynch_Stream (
   ACE_SSL_Asynch_Stream::Stream_Type s_type,
@@ -383,13 +306,6 @@ ACE_SSL_Asynch_Stream::open (ACE_Handler & handler,
   return 0;
 }
 
-void
-ACE_SSL_Asynch_Stream::open (ACE_HANDLE new_handle,
-                             ACE_Message_Block &block)
-{
-  ACE_Service_Handler::open (new_handle,
-                             block);
-}
 
 // ************************************************************
 //  Asynch_Operation interface
diff --git a/ace/SSL/SSL_Asynch_Stream.h b/ace/SSL/SSL_Asynch_Stream.h
index 57007f5013d..671cca46c24 100644
--- a/ace/SSL/SSL_Asynch_Stream.h
+++ b/ace/SSL/SSL_Asynch_Stream.h
@@ -29,13 +29,101 @@
 #include "ace/Synch_Traits.h"
 #include "ace/Thread_Mutex.h"
 
+/*
+ * This facility doesn't follow the normal ACE asynch I/O support classes'
+ * interface/implementation arrangement. It's not needed because rather than
+ * branching off to platform-specific APIs, all platforms use the OpenSSL
+ * API. Thus, you can think of this class as the implementation class (for
+ * OpenSSL) and there's no separate interface class.
+ * Also, since both read and write operations are defined in one I/O
+ * factory, there's no single Result class defined as there is for
+ * ACE_Asynch_Read_Stream, et al. There are separate result classes defined
+ * for read and write operations.
+ */
+
+#if defined (ACE_WIN32)
+#  include "ace/WIN32_Asynch_IO.h"
+typedef ACE_WIN32_Asynch_Result              A_RESULT;
+typedef ACE_WIN32_Asynch_Read_Stream_Result  ARS_RESULT;
+typedef ACE_WIN32_Asynch_Write_Stream_Result AWS_RESULT;
+
+# define ERR_CANCELED ERROR_OPERATION_ABORTED
+
+#else
+#  include "ace/POSIX_Asynch_IO.h"
+typedef ACE_POSIX_Asynch_Result              A_RESULT;
+typedef ACE_POSIX_Asynch_Read_Stream_Result  ARS_RESULT;
+typedef ACE_POSIX_Asynch_Write_Stream_Result AWS_RESULT;
+
+# define ERR_CANCELED ECANCELED
+
+#endif  /* ACE_WIN32 */
+
 
 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
 
-/// Forward declarations
-class ACE_SSL_Asynch_Result;
-class ACE_SSL_Asynch_Read_Stream_Result;
-class ACE_SSL_Asynch_Write_Stream_Result;
+class ACE_SSL_Asynch_Stream;     // Forward decl for use in result class def.
+
+/**
+ * @class ACE_SSL_Asynch_Read_Stream_Result
+ *
+ * Result class that communicates result of read operations initiated on
+ * an ACE_SSL_Asynch_Stream object.
+ */
+class ACE_SSL_Asynch_Read_Stream_Result : public ARS_RESULT
+{
+  /// Factory class will have special permissions.
+  friend class ACE_SSL_Asynch_Stream;
+
+protected:
+  ACE_SSL_Asynch_Read_Stream_Result (ACE_Handler &handler,
+                                     ACE_HANDLE handle,
+                                     ACE_Message_Block &message_block,
+                                     size_t bytes_to_read,
+                                     const void* act,
+                                     ACE_HANDLE event,
+                                     int priority,
+                                     int signal_number);
+};
+
+/**
+ * @class ACE_SSL_Asynch_Write_Stream_Result
+ *
+ * Result class that communicates result of write operations initiated on
+ * an ACE_SSL_Asynch_Stream object.
+ */
+class ACE_SSL_Asynch_Write_Stream_Result : public AWS_RESULT
+{
+  /// Factory class will have special permissions.
+  friend class ACE_SSL_Asynch_Stream;
+
+protected:
+  ACE_SSL_Asynch_Write_Stream_Result (ACE_Handler &handler,
+                                      ACE_HANDLE handle,
+                                      ACE_Message_Block &message_block,
+                                      size_t bytes_to_read,
+                                      const void* act,
+                                      ACE_HANDLE event,
+                                      int priority,
+                                      int signal_number);
+};
+
+
+/**
+ * @class ACE_SSL_Asynch_Result
+ *
+ * Result class that is used internally for socket close notifications.
+ */
+class ACE_SSL_Asynch_Result : public A_RESULT
+{
+public:
+    ACE_SSL_Asynch_Result (ACE_Handler &handler);
+
+    void complete (size_t bytes_transferred,
+                   int    success,
+                   const  void * completion_key,
+                   u_long error);
+};
 
 
 // Only provide forward declarations to prevent possible abuse of the
@@ -45,19 +133,17 @@ struct ACE_SSL_Asynch_Stream_Accessor;
 /**
  * @class ACE_SSL_Asynch_Stream
  *
- * @brief This class is a factory for starting off asynchronous reads
- * on a stream. This class forwards all methods to its
- * implementation class.
- * @par
- * Once open() is called, multiple asynchronous read()s can
- * started using this class.  An ACE_SSL_Asynch_Stream::Result
- * will be passed back to the @param handler when the asynchronous
- * reads completes through the ACE_Handler::handle_read_stream
- * callback.
+ * @brief This class is a factory for initiating asynchronous reads
+ * and writes on an SSL stream.
+ *
+ * Once open() is called, multiple asynchronous read and write operations
+ * can be started using this class.  The handler object (derived from
+ * ACE_Handler) specified in open() will receive completion events for the
+ * operations initiated via this class.
  */
 class ACE_SSL_Export ACE_SSL_Asynch_Stream
   : public ACE_Asynch_Operation,
-    public ACE_Service_Handler
+    public ACE_Handler
 {
 public:
 
@@ -79,13 +165,15 @@ public:
       ST_SERVER = 0x0002
     };
 
-  /// The constructor.
+  /// Constructor.
   /**
-   * @param context Pointer to @c ACE_SSL_Context instance containing
-   *                the OpenSSL @c SSL data structure to be associated
-   *                with this @c ACE_SSL_SOCK_Stream.  The @c SSL data
-   *                structure will be copied to make it at least
-   *                logically independent of the supplied @a context.
+   * @arg context  Pointer to an ACE_SSL_Context instance containing
+   *               the OpenSSL information to be associated with this
+   *               ACE_SSL_Asynch_Stream.  The needed SSL data will be
+   *               copied before return. Therefore, this object can be
+   *               reused, modified, or deleted upon return. If a 0 pointer
+   *               is passed, the ACE_SSL_Context::instance() method will
+   *               be called to get access to a singleton.
    */
   ACE_SSL_Asynch_Stream (Stream_Type s_type = ST_SERVER,
                          ACE_SSL_Context * context = 0);
@@ -93,28 +181,92 @@ public:
   /// Destructor
   virtual ~ACE_SSL_Asynch_Stream (void);
 
-  int cancel(void);
+  int cancel (void);
 
   int close (void);
 
+  /**
+   * Initializes the factory with information which will be used with
+   * each asynchronous call.
+   *
+   * @arg handler The ACE_Handler that will be called to handle completions
+   *              for operations initiated using this factory.
+   * @arg handle  The handle that future read/write operations will use.
+   *
+   * @retval 0    for success.
+   * @retval -1   for failure; consult @c errno for further information.
+   */
   int open (ACE_Handler &handler,
             ACE_HANDLE handle = ACE_INVALID_HANDLE,
             const void *completion_key = 0,
             ACE_Proactor *proactor = 0);
 
-  /// NOTE: This method has been specifically put in place so that
-  /// compilers like the borland doesnt get confused between the above
-  /// open () call with the one in the ACE_Service_Handler, from which
-  /// this class is derived from..
-  void open (ACE_HANDLE new_handle,
-             ACE_Message_Block &message_block);
-
+  /**
+   * Initiates an asynchronous read. If the operation is successfully
+   * initiated, the handle_read_stream() method will be called on the
+   * ACE_Handler object passed to open() when the operation completes.
+   * Data is read into the specified ACE_Message_Block beginning at its
+   * write pointer; the block's write pointer is updated to reflect any
+   * added data when the operation completes.
+   *
+   * @arg message_block      The specified ACE_Message_Block will receive any
+   *                         data that is read. Data will be read into the
+   *                         block beginning at the block's write pointer.
+   * @arg num_bytes_to_read  The maximum number of bytes to read. The actual
+   *                         amount read may be less.
+   * @arg act                ACT which is passed to the completion handler in
+   *                         the result object.
+   * @arg priority           Specifies the operation priority. This has an
+   *                         affect on POSIX only. Works like @i nice in Unix.
+   *                         Negative values are not allowed. 0 means priority
+   *                         of the operation same as the process priority.
+   *                         1 means priority of the operation is one less than
+   *                         process, and so forth. This parameter has no
+   *                         affect on Win32.
+   * @arg signal_number      The POSIX4 real-time signal number to be used
+   *                         for the operation. signal_number ranges from
+   *                         ACE_SIGRTMIN to ACE_SIGRTMAX. This argument is
+   *                         unused on non-POSIX4 systems.
+   *
+   * @retval 0    for success.
+   * @retval -1   for failure; consult @c errno for further information.
+   */
   int read (ACE_Message_Block &message_block,
             size_t num_bytes_to_read,
             const void *act = 0,
             int priority = 0,
             int signal_number = ACE_SIGRTMIN);
 
+  /**
+   * Initiates an asynchronous write. If the operation is successfully
+   * initiated, the handle_write_stream() method will be called on the
+   * ACE_Handler object passed to open() when the operation completes.
+   * Data is taken from the specified ACE_Message_Block beginning at its
+   * read pointer; the block's read pointer is updated to reflect any
+   * data successfully sent when the operation completes.
+   *
+   * @arg message_block      The specified ACE_Message_Block is the source of
+   *                         data that is written. Data will be taken from the
+   *                         block beginning at the block's read pointer.
+   * @arg bytes_to_write     The maximum number of bytes to write. The actual
+   *                         amount written may be less.
+   * @arg act                ACT which is passed to the completion handler in
+   *                         the result object.
+   * @arg priority           Specifies the operation priority. This has an
+   *                         affect on POSIX only. Works like @i nice in Unix.
+   *                         Negative values are not allowed. 0 means priority
+   *                         of the operation same as the process priority.
+   *                         1 means priority of the operation is one less than
+   *                         process, and so forth. This parameter has no
+   *                         affect on Win32.
+   * @arg signal_number      The POSIX4 real-time signal number to be used
+   *                         for the operation. signal_number ranges from
+   *                         ACE_SIGRTMIN to ACE_SIGRTMAX. This argument is
+   *                         unused on non-POSIX4 systems.
+   *
+   * @retval 0    for success.
+   * @retval -1   for failure; consult @c errno for further information.
+   */
   int write (ACE_Message_Block &message_block,
              size_t bytes_to_write,
              const void *act = 0,
@@ -122,8 +274,11 @@ public:
              int signal_number = ACE_SIGRTMIN);
 
 protected:
+  /// Virtual from ACE_Asynch_Operation. Since this class is essentially an
+  /// implementation class, simply return 0.
+  virtual ACE_Asynch_Operation_Impl *implementation (void) const { return 0; }
 
-  /// virtual from ACE_Service_Handler
+  /// virtual from ACE_Handler
 
   /// This method is called when BIO write request is completed. It
   /// processes the IO completion and calls do_SSL_state_machine().
diff --git a/tests/SSL/Makefile.am b/tests/SSL/Makefile.am
index d4dcc3ab929..86283019e5e 100644
--- a/tests/SSL/Makefile.am
+++ b/tests/SSL/Makefile.am
@@ -31,6 +31,21 @@ Thread_Pool_Reactor_SSL_Test_SOURCES = \
 Thread_Pool_Reactor_SSL_Test_LDADD = \
  $(top_builddir)/ace/SSL/libACE_SSL.la $(top_builddir)/tests/libTest_Output.la $(top_builddir)/ace/libACE.la @ACE_TLS_LDFLAGS@ @ACE_TLS_LIBS@
 
+noinst_PROGRAMS += SSL_Asynch_Stream_Test
+
+SSL_Asynch_Stream_Test_CPPFLAGS = \
+  -I$(ACE_ROOT) \
+  -I$(ACE_BUILDDIR) \
+  @ACE_TLS_CPPFLAGS@ \
+  -DACE_HAS_SSL=1
+
+SSL_Asynch_Stream_Test_SOURCES = \
+  $(ACE_ROOT)/tests/Main.cpp \
+  SSL_Asynch_Stream_Test.cpp
+
+SSL_Asynch_Stream_Test_LDADD = \
+ $(top_builddir)/ace/SSL/libACE_SSL.la $(top_builddir)/tests/libTest_Output.la $(top_builddir)/ace/libACE.la @ACE_TLS_LDFLAGS@ @ACE_TLS_LIBS@
+
 endif BUILD_SSL
 
 ## Clean up template repositories, etc.
diff --git a/tests/SSL/SSL_Asynch_Stream_Test.cpp b/tests/SSL/SSL_Asynch_Stream_Test.cpp
new file mode 100644
index 00000000000..386dcc7917c
--- /dev/null
+++ b/tests/SSL/SSL_Asynch_Stream_Test.cpp
@@ -0,0 +1,478 @@
+// $Id$
+
+// ============================================================================
+//
+// = LIBRARY
+//    tests/SSL
+//
+// = FILENAME
+//    SSL_Asynch_Stream_Test.cpp
+//
+// = DESCRIPTION
+//      This program is a functionality test of ACE_SSL_Asynch_Stream.
+//      It demonstrates one proper use case of ACE_SSL_Asynch_Stream in the
+//      Proactor framework and validates its basic functionality.
+//
+//      Usage: SSL_Asynch_Stream_Test [-r <hostname:port#>]
+//                [-t <num threads>] [-d <delay>]
+//                [-i <client conn attempt#>] [-n <client request# per conn>]
+//
+//      Default value:
+//          <hostname:port#>:       ACE_DEFAULT_SERVER_HOST:ACE_DEFAULT_PORT
+//          <num threads>:          ACE_MAX_THREADS
+//          <client conn attempt#>: ACE_MAX_ITERATIONS
+//          <client req# per conn>: 20
+//          <delay>:                0 usec
+//
+// = AUTHOR
+//      Steve Huston <shuston@riverace.com>
+//
+// ============================================================================
+
+#include "tests/test_config.h"
+#include "ace/Default_Constants.h"
+#include "ace/OS_NS_string.h"
+#include "ace/Event_Handler.h"
+#include "ace/Get_Opt.h"
+#include "ace/Proactor.h"
+#include "ace/Reactor.h"
+#include "ace/Thread_Manager.h"
+#include "ace/INET_Addr.h"
+#include "ace/SSL/SSL_Asynch_Stream.h"
+#include "ace/SSL/SSL_SOCK_Connector.h"
+#include "ace/SSL/SSL_SOCK_Acceptor.h"
+#include "ace/SSL/SSL_SOCK_Stream.h"
+
+ACE_RCSID(tests, SSL_Asynch_Stream_Test, "$Id$")
+
+#if defined (ACE_HAS_THREADS)
+
+class Client_Handler : public ACE_Handler
+{
+public:
+  Client_Handler ()
+    : msgs_sent_ (0),
+      stream_ (ACE_SSL_Asynch_Stream::ST_CLIENT),
+      block_ (1024) {}
+  ~Client_Handler ();
+
+  int open (ACE_HANDLE);
+
+private:
+  virtual void handle_write_stream (const ACE_SSL_Asynch_Write_Stream_Result &result);
+
+private:
+  size_t                  msgs_sent_;
+  ACE_SSL_Asynch_Stream   stream_;
+  ACE_Message_Block       block_;
+};
+
+class Server_Handler : public ACE_Handler
+{
+public:
+  Server_Handler ()
+    : msgs_rcvd_ (0),
+      stream_ (ACE_SSL_Asynch_Stream::ST_SERVER),
+      block_ (1024) {}
+  ~Server_Handler ();
+
+  int open (ACE_HANDLE);
+
+private:
+  virtual void handle_read_stream (const ACE_SSL_Asynch_Read_Stream_Result &result);
+
+private:
+  size_t                  msgs_rcvd_;
+  ACE_SSL_Asynch_Stream   stream_;
+  ACE_Message_Block       block_;
+};
+
+class Server_Acceptor : public ACE_Event_Handler
+{
+public:
+  int open (const ACE_INET_Addr &listen_addr);
+
+  // Called when a new connection is ready to accept.
+  virtual int handle_input (ACE_HANDLE fd = ACE_INVALID_HANDLE);
+
+  virtual int handle_close (ACE_HANDLE handle,
+                            ACE_Reactor_Mask close_mask);
+
+private:
+  ACE_SSL_SOCK_Acceptor acceptor_;
+};
+
+// Accepting end point.  This is actually "localhost:10010", but some
+// platform couldn't resolve the name so we use the IP address
+// directly here.
+static const ACE_TCHAR *rendezvous = \
+  ACE_DEFAULT_SERVER_HOST ACE_TEXT (":") ACE_DEFAULT_SERVER_PORT_STR;
+
+// Total number of proactor threads.
+static size_t num_threads = ACE_MAX_THREADS;
+
+#if defined (CHORUS) // Add platforms that can't handle too many
+                     // connection simultaneously here.
+#define ACE_LOAD_FACTOR /2
+#else
+#define ACE_LOAD_FACTOR
+#endif
+
+// Number of client connections to attempt.
+static size_t cli_conn_no = ACE_MAX_ITERATIONS ACE_LOAD_FACTOR;
+
+// Number of requests each client connection sends.
+static size_t cli_req_no = ACE_MAX_THREADS ACE_LOAD_FACTOR;
+
+// Delay before a thread sending the next request (in msec.)
+static int req_delay = 0;
+
+// This is the string sent from client to server.
+static const char *test_string = "SSL_Asynch_Stream_Test!";
+
+static void
+parse_args (int argc, ACE_TCHAR *argv[])
+{
+  ACE_Get_Opt getopt (argc, argv, ACE_TEXT ("r:t:d:i:n:"));
+
+  int c;
+
+  while ((c = getopt ()) != -1)
+    {
+      switch (c)
+        {
+        case 'r':                   // hostname:port
+          rendezvous = getopt.opt_arg ();
+          break;
+        case 't':
+          num_threads = ACE_OS::atoi (getopt.opt_arg ());
+          break;
+        case 'd':
+          req_delay = ACE_OS::atoi (getopt.opt_arg ());
+          break;
+        case 'i':
+          cli_conn_no = ACE_OS::atoi (getopt.opt_arg ());
+          break;
+        case 'n':
+          cli_req_no = ACE_OS::atoi (getopt.opt_arg ());
+          break;
+        default:
+          ACE_ERROR ((LM_ERROR,
+                      ACE_TEXT ("Usage: %s [-r <hostname:port#>]")
+                      ACE_TEXT ("\t[-t <nr threads>] [-d <delay>]")
+                      ACE_TEXT ("\t[-i <client conn attempt#>]")
+                      ACE_TEXT ("\t[-n <client request# per conn>]\n"),
+                      argv[0]));
+          break;
+        }
+    }
+}
+
+Client_Handler::~Client_Handler ()
+{
+  if (this->stream_.handle () != ACE_INVALID_HANDLE)
+    {
+      if (this->msgs_sent_ != cli_req_no)
+        ACE_ERROR ((LM_ERROR,
+                    ACE_TEXT ("(%t) Client handle %d sent %d messages; ")
+                    ACE_TEXT ("expected %d\n"),
+                    this->stream_.handle (),
+                    this->msgs_sent_,
+                    cli_req_no));
+      else
+        ACE_DEBUG ((LM_DEBUG,
+                    ACE_TEXT ("(%t) Client handle %d sent %d messages; ")
+                    ACE_TEXT ("closing connection\n"),
+                    this->stream_.handle (),
+                    cli_req_no));
+    }
+  this->stream_.close ();
+}
+
+int
+Client_Handler::open (ACE_HANDLE handle)
+{
+  if (this->stream_.open (*this, handle) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%t) Client_Handler: %p\n"),
+                       ACE_TEXT ("open")),
+                      -1);
+  this->block_.copy (test_string);
+  if (this->stream_.write (this->block_, this->block_.length ()) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%t) Client_Handler: %p\n"),
+                       ACE_TEXT ("initiate write")),
+                      -1);
+  return 0;
+}
+
+void
+Client_Handler::handle_write_stream
+  (const ACE_SSL_Asynch_Write_Stream_Result &result)
+{
+  if (!result.success ())
+    {
+      errno = result.error ();
+      ACE_ERROR ((LM_ERROR,
+                  ACE_TEXT ("(%t) Client handle %d: %p\n"),
+                  this->stream_.handle (),
+                  ACE_TEXT ("write")));
+      delete this;
+      return;
+    }
+  ACE_Message_Block &b = result.message_block ();
+  bool send_again = true;
+  if (b.length () == 0)
+    {
+      // All block's data sent; rewind the read pointer and send it again
+      // until we've sent the configured number of times.
+      ++this->msgs_sent_;
+      if (this->msgs_sent_ == cli_req_no)
+        send_again = false;   // All done
+      else
+        b.rd_ptr (b.base ());
+    }
+
+  if (send_again)
+    {
+      if (this->stream_.write (this->block_, this->block_.length ()) == -1)
+        {
+          ACE_ERROR ((LM_ERROR,
+                      ACE_TEXT ("(%t) Client_Handler: %p\n"),
+                      ACE_TEXT ("initiate write")));
+          delete this;
+        }
+    }
+  else
+    {
+      ACE_DEBUG ((LM_DEBUG,
+                  ACE_TEXT ("(%t) Client handle %d done sending\n"),
+                  this->stream_.handle ()));
+      delete this;
+    }
+  return;
+}
+
+Server_Handler::~Server_Handler ()
+{
+  if (this->stream_.handle () != ACE_INVALID_HANDLE)
+    {
+      if (this->msgs_rcvd_ != cli_req_no)
+        ACE_ERROR ((LM_ERROR,
+                    ACE_TEXT ("(%t) Server handle %d received %d messages; ")
+                    ACE_TEXT ("expected %d\n"),
+                    this->stream_.handle (),
+                    this->msgs_rcvd_,
+                    cli_req_no));
+      else
+        ACE_DEBUG ((LM_DEBUG,
+                    ACE_TEXT ("(%t) Server handle %d received %d messages; ")
+                    ACE_TEXT ("closing connection\n"),
+                    this->stream_.handle (),
+                    cli_req_no));
+    }
+  this->stream_.close ();
+}
+
+int
+Server_Handler::open (ACE_HANDLE handle)
+{
+  if (this->stream_.open (*this, handle) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%t) Server_Handler: %p\n"),
+                       ACE_TEXT ("open")),
+                      -1);
+  if (this->stream_.read (this->block_, this->block_.space () - 1) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%t) Server_Handler: %p\n"),
+                       ACE_TEXT ("read")),
+                      -1);
+  return 0;
+}
+
+void
+Server_Handler::handle_read_stream
+  (const ACE_SSL_Asynch_Read_Stream_Result &result)
+{
+  if (!result.success ())
+    {
+      errno = result.error ();
+      ACE_ERROR ((LM_ERROR,
+                  ACE_TEXT ("(%t) Server handle %d: %p\n"),
+                  this->stream_.handle (),
+                  ACE_TEXT ("read")));
+      delete this;
+      return;
+    }
+  if (result.bytes_transferred () == 0)
+    {
+      ACE_DEBUG ((LM_DEBUG,
+                  ACE_TEXT ("(%t) Server handle %d closed by peer\n"),
+                  this->stream_.handle ()));
+      delete this;
+      return;
+    }
+  
+  // Scan through the received data for the expected string. There may be
+  // multiples and/or partials. Count up how many arrive before the connection
+  // is closed.
+  // The read operation left one byte space at the end so we can insert a
+  // nul terminator to ease scanning.
+  ACE_Message_Block &b = result.message_block ();
+  *(b.wr_ptr ()) = '\0';
+  size_t test_string_len = ACE_OS::strlen (test_string);
+  while (b.length () >= test_string_len)
+    {
+      if (0 != ACE_OS::strncmp (b.rd_ptr (), test_string, test_string_len))
+        ACE_ERROR_BREAK ((LM_ERROR,
+                          ACE_TEXT ("(%t) Read string: %C; expected: %C\n"),
+                          b.rd_ptr (),
+                          test_string));
+      b.rd_ptr (test_string_len);
+    }
+  b.crunch ();
+  if (this->stream_.read (b, b.space () - 1) == -1)
+    {
+      ACE_ERROR ((LM_ERROR,
+                  ACE_TEXT ("(%t) Server_Handler: %p\n"),
+                  ACE_TEXT ("read")));
+      delete this;
+    }
+  return;
+}
+
+
+int
+Server_Acceptor::open (const ACE_INET_Addr &listen_addr)
+{
+  if (this->acceptor_.open (listen_addr) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("%p\n"),
+                       ACE_TEXT ("listen")),
+                      -1);
+  return 0;
+}
+
+int
+Server_Acceptor::handle_input (ACE_HANDLE)
+{
+  ACE_SSL_SOCK_Stream new_stream;
+  if (this->acceptor_.accept (new_stream) == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%t) %p\n"),
+                       ACE_TEXT ("accept")),
+                      -1);
+  Server_Handler *new_handler = 0;
+  ACE_NEW_RETURN (new_handler, Server_Handler, -1);
+  if (new_handler->open (new_stream.get_handle ()) != 0)
+    delete new_handler;
+
+  return 0;
+}
+
+int
+Server_Acceptor::handle_close (ACE_HANDLE, ACE_Reactor_Mask)
+{
+  this->acceptor_.close ();
+  return 0;
+}
+
+
+static ACE_THR_FUNC_RETURN
+proactor_loop (void *)
+{
+  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Start handling events.\n")));
+
+  int result =
+    ACE_Proactor::instance ()->proactor_run_event_loop ();
+  if (result == -1)
+    ACE_ERROR_RETURN ((LM_ERROR,
+                       ACE_TEXT ("(%t) %p\n"),
+                       ACE_TEXT ("Error handling events")),
+                      0);
+
+  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Done handling events.\n")));
+
+  return 0;
+}
+
+static ACE_THR_FUNC_RETURN
+start_clients (void *arg)
+{
+  // Client thread function.
+  ACE_INET_Addr addr (rendezvous);
+  ACE_SSL_SOCK_Stream stream;
+  ACE_SSL_SOCK_Connector connect;
+  ACE_Time_Value delay (0, req_delay);
+  size_t len = * reinterpret_cast<ACE_TCHAR *> (arg);
+
+  for (size_t i = 0 ; i < cli_conn_no; i++)
+    {
+      if (connect.connect (stream, addr) < 0)
+        {
+          ACE_ERROR ((LM_ERROR,
+                      ACE_TEXT ("(%t) %p\n"),
+                      ACE_TEXT ("connect")));
+          continue;
+        }
+
+      Client_Handler *new_handler = 0;
+      ACE_NEW_RETURN (new_handler, Client_Handler, -1);
+      if (new_handler->open (stream.get_handle ()) != 0)
+        delete new_handler;
+      stream.set_handle (ACE_INVALID_HANDLE);
+    }
+
+  return 0;
+}
+
+int
+run_main (int argc, ACE_TCHAR *argv[])
+{
+  ACE_START_TEST (ACE_TEXT ("SSL_Asynch_Stream_Test"));
+
+  ACE_SSL_Context *context = ACE_SSL_Context::instance ();
+  // Note - the next two strings are naked on purpose... the arguments to
+  // the ACE_SSL_Context methods are const char *, not ACE_TCHAR *.
+  context->certificate ("dummy.pem", SSL_FILETYPE_PEM);
+  context->private_key ("key.pem", SSL_FILETYPE_PEM);
+
+  parse_args (argc, argv);
+
+  Server_Acceptor acceptor;
+  ACE_INET_Addr accept_addr (rendezvous);
+
+  if (acceptor.open (accept_addr) == -1)
+    return 1;
+
+  ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("(%t) Listening at %s\n"), rendezvous));
+  ACE_DEBUG ((LM_DEBUG,
+              ACE_TEXT ("(%t) Spawning %d proactor threads\n"),
+              num_threads));
+  ACE_Thread_Manager::instance ()->spawn_n (num_threads, proactor_loop);
+  ACE_Thread_Manager::instance ()->spawn (start_clients);
+
+  ACE_Time_Value loop_limit (20);
+  ACE_Reactor::instance ()->run_reactor_event_loop (loop_limit);
+  ACE_Thread_Manager::instance ()->wait ();
+
+  // Check for num connections up/down.
+
+  ACE_END_TEST;
+  return 0;
+}
+
+#else
+int
+ACE_TMAIN (int, ACE_TCHAR *[])
+{
+  ACE_START_TEST (ACE_TEXT ("SSL_Asynch_Stream_Test"));
+
+  ACE_ERROR ((LM_INFO,
+              ACE_TEXT ("This test requires threads which are not supported ")
+              ACE_TEXT ("on this platform\n"));
+
+  ACE_END_TEST;
+  return 0;
+}
+#endif /* ACE_HAS_THREADS */
diff --git a/tests/SSL/tests.mpc b/tests/SSL/tests.mpc
index f27c4041968..c49e4d746a5 100644
--- a/tests/SSL/tests.mpc
+++ b/tests/SSL/tests.mpc
@@ -1,6 +1,13 @@
 //  -*- MPC -*-
 // $Id$
 
+project(SSL Asynch_Stream Test) : acetest, ssl {
+  exename = SSL_Asynch_Stream_Test
+  Source_Files {
+    SSL_Asynch_Stream_Test.cpp
+  }
+}
+
 project(Thread Pool Reactor SSL Test) : acetest, ssl {
   exename = Thread_Pool_Reactor_SSL_Test
   Source_Files {