Rework pjlib++

git-svn-id: http://svn.pjsip.org/repos/pjproject/main@36 74dad513-b988-da41-8d7b-12977e46ad98

1 files changed, 438 insertions, 45 deletions

diff --git a/pjlib/include/pj++/proactor.hpp b/pjlib/include/pj++/proactor.hpp
index cae9cf43..7021a5a0 100644
--- a/pjlib/include/pj++/proactor.hpp
+++ b/pjlib/include/pj++/proactor.hpp
@@ -1,88 +1,481 @@
 /* $Id$
- *
  */

-#ifndef __PJPP_EVENT_HANDLER_H__

-#define __PJPP_EVENT_HANDLER_H__

+#ifndef __PJPP_PROACTOR_H__

+#define __PJPP_PROACTOR_H__

 

 #include <pj/ioqueue.h>

 #include <pj++/pool.hpp>

 #include <pj++/sock.hpp>

 #include <pj++/timer.hpp>

+#include <pj/errno.h>

 

-class PJ_Proactor;

+class Pj_Proactor;

+class Pj_Event_Handler;

 

 

-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();

+    //

+    // 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_ );

+    }

 

-    virtual pj_oshandle_t get_handle() = 0;

+    //

+    // 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));

+    }

 

-    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);

 #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

+    //////////////////

+

+    //

+    // Timeout callback.

+    //

+    virtual void on_timeout(int data) 

+    {

+    }

+

     //

-    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) {}

+    // 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);

+    //

+    // 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);

+    }

 

-    void destroy();

+    //

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

+        }

 

-    bool register_handler(PJ_Pool *pool, PJ_Event_Handler *handler);

-    void unregister_handler(PJ_Event_Handler *handler);

+        /* Poll events in ioqueue. */

+        int ioqueue_count;

 

-    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);

+        ioqueue_count = pj_ioqueue_poll(ioq_, &timeout);

+        if (ioqueue_count < 0)

+	    return ioqueue_count;

 

-    bool handle_events(PJ_Time_Val *timeout);

+        return ioqueue_count + timer_count;

+    }

 

-    pj_ioqueue_t *get_io_queue();

-    pj_timer_heap_t *get_timer_heap();

+    //

+    // 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_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);

+    }

 

-    PJ_Proactor() {}

 };

 

-#endif	/* __PJPP_EVENT_HANDLER_H__ */

+#endif	/* __PJPP_PROACTOR_H__ */