Index: pjproject/main/pjlib/include/pj++/proactor.hpp
===================================================================
--- pjproject/main/pjlib/include/pj++/proactor.hpp	(revision 29)
+++ pjproject/main/pjlib/include/pj++/proactor.hpp	(revision 36)
@@ -1,7 +1,6 @@
 /* $Id$
- *
  */
-#ifndef __PJPP_EVENT_HANDLER_H__
-#define __PJPP_EVENT_HANDLER_H__
+#ifndef __PJPP_PROACTOR_H__
+#define __PJPP_PROACTOR_H__
 
 #include <pj/ioqueue.h>
@@ -9,80 +8,474 @@
 #include <pj++/sock.hpp>
 #include <pj++/timer.hpp>
-
-class PJ_Proactor;
-
-
-class PJ_Event_Handler
+#include <pj/errno.h>
+
+class Pj_Proactor;
+class Pj_Event_Handler;
+
+
+//////////////////////////////////////////////////////////////////////////////
+// Asynchronous operation key.
+//
+// Applications may inheric this class to put their application
+// specific data.
+//
+class Pj_Async_Op : public pj_ioqueue_op_key_t
 {
-    friend class PJ_Proactor;
 public:
-    PJ_Event_Handler();
-    virtual ~PJ_Event_Handler();
-
-    virtual pj_oshandle_t get_handle() = 0;
-
-    bool read(void *buf, pj_size_t len);
-    bool recvfrom(void *buf, pj_size_t len, PJ_INET_Addr *addr);
-    bool write(const void *data, pj_size_t len);
-    bool sendto(const void *data, pj_size_t len, const PJ_INET_Addr &addr);
+    //
+    // Constructor.
+    //
+    explicit Pj_Async_Op(Pj_Event_Handler *handler)
+        : handler_(handler)
+    {
+        pj_memset(this, 0, sizeof(pj_ioqueue_op_key_t));
+    }
+
+    //
+    // Check whether operation is still pending for this key.
+    //
+    bool is_pending();
+
+    //
+    // Cancel the operation.
+    //
+    bool cancel(pj_ssize_t bytes_status=-PJ_ECANCELLED);
+
+protected:
+    Pj_Event_Handler *handler_;
+};
+
+
+//////////////////////////////////////////////////////////////////////////////
+// Event handler.
+//
+// Applications should inherit this class to receive various event
+// notifications.
+//
+// Applications should implement get_socket_handle().
+//
+class Pj_Event_Handler : public Pj_Object
+{
+    friend class Pj_Proactor;
+public:
+    //
+    // Default constructor.
+    //
+    Pj_Event_Handler()
+        : key_(NULL)
+    {
+        pj_memset(&timer_, 0, sizeof(timer_));
+        timer_.user_data = this;
+        timer_.cb = &timer_callback;
+    }
+    
+    //
+    // Destroy.
+    //
+    virtual ~Pj_Event_Handler()
+    {
+        unregister();
+    }
+
+    //
+    // Unregister this handler from the ioqueue.
+    //
+    void unregister()
+    {
+        if (key_) {
+            pj_ioqueue_unregister(key_);
+            key_ = NULL;
+        }
+    }
+
+    //
+    // Get socket handle associated with this.
+    //
+    virtual pj_sock_t get_socket_handle()
+    {
+        return PJ_INVALID_SOCKET;
+    }
+
+    //
+    // Receive data.
+    //
+    pj_status_t recv( Pj_Async_Op *op_key, 
+                      void *buf, pj_ssize_t *len, 
+                      unsigned flags)
+    {
+        return pj_ioqueue_recv( key_, op_key,
+                                buf, len, flags);
+    }
+
+    //
+    // Recvfrom()
+    //
+    pj_status_t recvfrom( Pj_Async_Op *op_key, 
+                          void *buf, pj_ssize_t *len, unsigned flags,
+                          Pj_Inet_Addr *addr)
+    {
+        addr->addrlen_ = sizeof(Pj_Inet_Addr);
+        return pj_ioqueue_recvfrom( key_, op_key, buf, len, flags,
+                                    addr, &addr->addrlen_ );
+    }
+
+    //
+    // send()
+    //
+    pj_status_t send( Pj_Async_Op *op_key, 
+                      const void *data, pj_ssize_t *len, 
+                      unsigned flags)
+    {
+        return pj_ioqueue_send( key_, op_key, data, len, flags);
+    }
+
+    //
+    // sendto()
+    //
+    pj_status_t sendto( Pj_Async_Op *op_key,
+                        const void *data, pj_ssize_t *len, unsigned flags,
+                        const Pj_Inet_Addr &addr)
+    {
+        return pj_ioqueue_sendto(key_, op_key, data, len, flags,
+                                 &addr, sizeof(addr));
+    }
+
 #if PJ_HAS_TCP
-    bool connect(const PJ_INET_Addr &addr);
-    bool accept(PJ_Socket *sock, PJ_INET_Addr *local=NULL, PJ_INET_Addr *remote=NULL);
+    //
+    // connect()
+    //
+    pj_status_t connect(const Pj_Inet_Addr &addr)
+    {
+        return pj_ioqueue_connect(key_, &addr, sizeof(addr));
+    }
+
+    //
+    // accept.
+    //
+    pj_status_t accept( Pj_Async_Op *op_key,
+                        Pj_Socket *sock, 
+                        Pj_Inet_Addr *local = NULL, 
+                        Pj_Inet_Addr *remote = NULL)
+    {
+        int *addrlen = local ? &local->addrlen_ : NULL;
+        return pj_ioqueue_accept( key_, op_key, &sock->sock_,
+                                  local, remote, addrlen );
+    }
+
 #endif
 
 protected:
-    //
+    //////////////////
     // Overridables
-    //
-    virtual void on_timeout(int data) {}
-    virtual void on_read_complete(pj_ssize_t bytes_read) {}
-    virtual void on_write_complete(pj_ssize_t bytes_sent) {}
+    //////////////////
+
+    //
+    // Timeout callback.
+    //
+    virtual void on_timeout(int data) 
+    {
+    }
+
+    //
+    // On read complete callback.
+    //
+    virtual void on_read_complete( Pj_Async_Op *op_key,
+                                   pj_ssize_t bytes_read) 
+    {
+    }
+
+    //
+    // On write complete callback.
+    //
+    virtual void on_write_complete( Pj_Async_Op *op_key, 
+                                    pj_ssize_t bytes_sent) 
+    {
+    }
+
 #if PJ_HAS_TCP
-    virtual void on_connect_complete(int status) {}
-    virtual void on_accept_complete(int status) {}
+    //
+    // On connect complete callback.
+    //
+    virtual void on_connect_complete(pj_status_t status) 
+    {
+    }
+
+    //
+    // On new connection callback.
+    //
+    virtual void on_accept_complete( Pj_Async_Op *op_key,
+                                     pj_sock_t new_sock,
+                                     pj_status_t status) 
+    {
+    }
+
 #endif
 
+
 private:
-    PJ_Proactor	     *proactor_;
     pj_ioqueue_key_t *key_;
     pj_timer_entry    timer_;
-    int		      tmp_recvfrom_addr_len;
-
-public:
-    // Internal IO Queue/timer callback.
-    static void timer_callback( pj_timer_heap_t *timer_heap, struct pj_timer_entry *entry);
-    static void read_complete_cb(pj_ioqueue_key_t *key, pj_ssize_t bytes_read);
-    static void write_complete_cb(pj_ioqueue_key_t *key, pj_ssize_t bytes_sent);
-    static void accept_complete_cb(pj_ioqueue_key_t *key, int status);
-    static void connect_complete_cb(pj_ioqueue_key_t *key, int status);
+
+    friend class Pj_Proactor;
+    friend class Pj_Async_Op;
+
+    //
+    // Static timer callback.
+    //
+    static void timer_callback( pj_timer_heap_t *timer_heap, 
+                                struct pj_timer_entry *entry)
+    {
+        Pj_Event_Handler *handler = 
+            (Pj_Event_Handler*) entry->user_data;
+
+        handler->on_timeout(entry->id);
+    }
 };
 
-class PJ_Proactor
+inline bool Pj_Async_Op::is_pending()
+{
+    return pj_ioqueue_is_pending(handler_->key_, this) != 0;
+}
+
+inline bool Pj_Async_Op::cancel(pj_ssize_t bytes_status)
+{
+    return pj_ioqueue_post_completion(handler_->key_, this, 
+                                      bytes_status) == PJ_SUCCESS;
+}
+
+//////////////////////////////////////////////////////////////////////////////
+// Proactor
+//
+class Pj_Proactor : public Pj_Object
 {
 public:
-    static PJ_Proactor *create(PJ_Pool *pool, pj_size_t max_fd, 
-			       pj_size_t timer_entry_count, unsigned timer_flags=0);
-
-    void destroy();
-
-    bool register_handler(PJ_Pool *pool, PJ_Event_Handler *handler);
-    void unregister_handler(PJ_Event_Handler *handler);
-
-    static bool schedule_timer( pj_timer_heap_t *timer, PJ_Event_Handler *handler,
-				const PJ_Time_Val &delay, int id=-1);
-    bool schedule_timer(PJ_Event_Handler *handler, const PJ_Time_Val &delay, int id=-1);
-    bool cancel_timer(PJ_Event_Handler *handler);
-
-    bool handle_events(PJ_Time_Val *timeout);
-
-    pj_ioqueue_t *get_io_queue();
-    pj_timer_heap_t *get_timer_heap();
+    //
+    // Default constructor, initializes to NULL.
+    //
+    Pj_Proactor()
+        : ioq_(NULL), th_(NULL)
+    {
+        cb_.on_read_complete    = &read_complete_cb;
+        cb_.on_write_complete   = &write_complete_cb;
+        cb_.on_accept_complete  = &accept_complete_cb;
+        cb_.on_connect_complete = &connect_complete_cb;
+    }
+
+    //
+    // Construct proactor.
+    //
+    Pj_Proactor( Pj_Pool *pool, pj_size_t max_fd,
+                 pj_size_t max_timer_entries )
+    : ioq_(NULL), th_(NULL)
+    {
+        cb_.on_read_complete    = &read_complete_cb;
+        cb_.on_write_complete   = &write_complete_cb;
+        cb_.on_accept_complete  = &accept_complete_cb;
+        cb_.on_connect_complete = &connect_complete_cb;
+    }
+
+    //
+    // Destructor.
+    //
+    ~Pj_Proactor()
+    {
+        destroy();
+    }
+
+    //
+    // Create proactor.
+    //
+    pj_status_t create( Pj_Pool *pool, pj_size_t max_fd, 
+			pj_size_t timer_entry_count)
+    {
+        pj_status_t status;
+
+        destroy();
+
+        status = pj_ioqueue_create(pool->pool_(), max_fd, &ioq_);
+        if (status != PJ_SUCCESS) 
+            return status;
+        
+        status = pj_timer_heap_create(pool->pool_(), 
+                                      timer_entry_count, &th_);
+        if (status != PJ_SUCCESS) {
+            pj_ioqueue_destroy(ioq_);
+            ioq_ = NULL;
+            return NULL;
+        }
+        
+        status;
+    }
+
+    //
+    // Destroy proactor.
+    //
+    void destroy()
+    {
+        if (ioq_) {
+            pj_ioqueue_destroy(ioq_);
+            ioq_ = NULL;
+        }
+        if (th_) {
+            pj_timer_heap_destroy(th_);
+            th_ = NULL;
+        }
+    }
+
+    //
+    // Register handler.
+    // This will call handler->get_socket_handle()
+    //
+    pj_status_t register_socket_handler(Pj_Pool *pool, 
+                                        Pj_Event_Handler *handler)
+    {
+        return   pj_ioqueue_register_sock( pool->pool_(), ioq_,
+                                           handler->get_socket_handle(),
+                                           handler, &cb_, &handler->key_ );
+    }
+
+    //
+    // Unregister handler.
+    //
+    static void unregister_handler(Pj_Event_Handler *handler)
+    {
+        if (handler->key_) {
+            pj_ioqueue_unregister( handler->key_ );
+            handler->key_ = NULL;
+        }
+    }
+
+    //
+    // Scheduler timer.
+    //
+    bool schedule_timer( Pj_Event_Handler *handler, 
+                         const Pj_Time_Val &delay, 
+                         int id=-1)
+    {
+        return schedule_timer(th_, handler, delay, id);
+    }
+
+    //
+    // Cancel timer.
+    //
+    bool cancel_timer(Pj_Event_Handler *handler)
+    {
+        return pj_timer_heap_cancel(th_, &handler->timer_) == 1;
+    }
+
+    //
+    // Handle events.
+    //
+    int handle_events(Pj_Time_Val *max_timeout)
+    {
+        Pj_Time_Val timeout(0, 0);
+        int timer_count;
+
+        timer_count = pj_timer_heap_poll( th_, &timeout );
+
+        if (timeout.get_sec() < 0) 
+            timeout.sec = PJ_MAXINT32;
+
+        /* If caller specifies maximum time to wait, then compare the value 
+         * with the timeout to wait from timer, and use the minimum value.
+         */
+        if (max_timeout && timeout >= *max_timeout) {
+	    timeout = *max_timeout;
+        }
+
+        /* Poll events in ioqueue. */
+        int ioqueue_count;
+
+        ioqueue_count = pj_ioqueue_poll(ioq_, &timeout);
+        if (ioqueue_count < 0)
+	    return ioqueue_count;
+
+        return ioqueue_count + timer_count;
+    }
+
+    //
+    // Get the internal ioqueue object.
+    //
+    pj_ioqueue_t *get_io_queue()
+    {
+        return ioq_;
+    }
+
+    //
+    // Get the internal timer heap object.
+    //
+    pj_timer_heap_t *get_timer_heap()
+    {
+        return th_;
+    }
 
 private:
     pj_ioqueue_t *ioq_;
     pj_timer_heap_t *th_;
-
-    PJ_Proactor() {}
+    pj_ioqueue_callback cb_;
+
+    static bool schedule_timer( pj_timer_heap_t *timer, 
+                                Pj_Event_Handler *handler,
+				const Pj_Time_Val &delay, 
+                                int id=-1)
+    {
+        handler->timer_.id = id;
+        return pj_timer_heap_schedule(timer, &handler->timer_, &delay) == 0;
+    }
+
+
+    //
+    // Static read completion callback.
+    //
+    static void read_complete_cb( pj_ioqueue_key_t *key, 
+                                  pj_ioqueue_op_key_t *op_key, 
+                                  pj_ssize_t bytes_read)
+    {
+        Pj_Event_Handler *handler = 
+	    (Pj_Event_Handler*) pj_ioqueue_get_user_data(key);
+
+        handler->on_read_complete((Pj_Async_Op*)op_key, bytes_read);
+    }
+
+    //
+    // Static write completion callback.
+    //
+    static void write_complete_cb(pj_ioqueue_key_t *key, 
+                                  pj_ioqueue_op_key_t *op_key,
+                                  pj_ssize_t bytes_sent)
+    {
+        Pj_Event_Handler *handler = 
+	    (Pj_Event_Handler*) pj_ioqueue_get_user_data(key);
+
+        handler->on_write_complete((Pj_Async_Op*)op_key, bytes_sent);
+    }
+
+    //
+    // Static accept completion callback.
+    //
+    static void accept_complete_cb(pj_ioqueue_key_t *key, 
+                                   pj_ioqueue_op_key_t *op_key,
+                                   pj_sock_t new_sock,
+                                   pj_status_t status)
+    {
+        Pj_Event_Handler *handler = 
+	    (Pj_Event_Handler*) pj_ioqueue_get_user_data(key);
+
+        handler->on_accept_complete((Pj_Async_Op*)op_key, new_sock, status);
+    }
+
+    //
+    // Static connect completion callback.
+    //
+    static void connect_complete_cb(pj_ioqueue_key_t *key, 
+                                    pj_status_t status)
+    {
+        Pj_Event_Handler *handler = 
+	    (Pj_Event_Handler*) pj_ioqueue_get_user_data(key);
+
+        handler->on_connect_complete(status);
+    }
+
 };
 
-#endif	/* __PJPP_EVENT_HANDLER_H__ */
+#endif	/* __PJPP_PROACTOR_H__ */
