Index: pjproject/trunk/pjsip/src/pjsip/sip_transport_tcp.c
===================================================================
--- pjproject/trunk/pjsip/src/pjsip/sip_transport_tcp.c	(revision 2180)
+++ pjproject/trunk/pjsip/src/pjsip/sip_transport_tcp.c	(revision 2188)
@@ -22,6 +22,6 @@
 #include <pj/compat/socket.h>
 #include <pj/addr_resolv.h>
+#include <pj/activesock.h>
 #include <pj/assert.h>
-#include <pj/ioqueue.h>
 #include <pj/lock.h>
 #include <pj/log.h>
@@ -47,23 +47,4 @@
 
 /*
- * This structure is "descendant" of pj_ioqueue_op_key_t, and it is used to
- * track pending/asynchronous accept() operation. TCP transport may have
- * more than one pending accept() operations, depending on the value of
- * async_cnt.
- */
-struct pending_accept
-{
-    pj_ioqueue_op_key_t	     op_key;
-    struct tcp_listener	    *listener;
-    unsigned		     index;
-    pj_pool_t		    *pool;
-    pj_sock_t		     new_sock;
-    int			     addr_len;
-    pj_sockaddr_in	     local_addr;
-    pj_sockaddr_in	     remote_addr;
-};
-
-
-/*
  * This is the TCP listener, which is a "descendant" of pjsip_tpfactory (the
  * SIP transport factory).
@@ -75,8 +56,5 @@
     pjsip_endpoint	    *endpt;
     pjsip_tpmgr		    *tpmgr;
-    pj_sock_t		     sock;
-    pj_ioqueue_key_t	    *key;
-    unsigned		     async_cnt;
-    struct pending_accept   *accept_op[MAX_ASYNC_CNT];
+    pj_activesock_t	    *asock;
 };
 
@@ -115,5 +93,5 @@
     pj_status_t		     close_reason;
     pj_sock_t		     sock;
-    pj_ioqueue_key_t	    *key;
+    pj_activesock_t	    *asock;
     pj_bool_t		     has_pending_connect;
 
@@ -140,14 +118,8 @@
 
 /* This callback is called when pending accept() operation completes. */
-static void on_accept_complete(	pj_ioqueue_key_t *key, 
-				pj_ioqueue_op_key_t *op_key, 
-				pj_sock_t sock, 
-				pj_status_t status);
-
-/* Handle accept() completion */
-static pj_status_t handle_accept(pj_ioqueue_key_t *key, 
-				 pj_ioqueue_op_key_t *op_key, 
-				 pj_sock_t sock, 
-				 pj_status_t status);
+static pj_bool_t on_accept_complete(pj_activesock_t *asock,
+				    pj_sock_t newsock,
+				    const pj_sockaddr_t *src_addr,
+				    int src_addr_len);
 
 /* This callback is called by transport manager to destroy listener */
@@ -186,9 +158,8 @@
 				   const pj_sockaddr_in *addr )
 {
-    enum { M = 48 };
-    host_port->host.ptr = (char*) pj_pool_alloc(pool, M);
-    host_port->host.slen = pj_ansi_snprintf( host_port->host.ptr, M, "%s", 
-					    pj_inet_ntoa(addr->sin_addr));
-    host_port->port = pj_ntohs(addr->sin_port);
+    host_port->host.ptr = (char*) pj_pool_alloc(pool, PJ_INET6_ADDRSTRLEN+4);
+    pj_sockaddr_print(addr, host_port->host.ptr, PJ_INET6_ADDRSTRLEN+4, 2);
+    host_port->host.slen = pj_ansi_strlen(host_port->host.ptr);
+    host_port->port = pj_sockaddr_get_port(addr);
 }
 
@@ -210,9 +181,10 @@
 {
     pj_pool_t *pool;
+    pj_sock_t sock = PJ_INVALID_SOCKET;
     struct tcp_listener *listener;
-    pj_ioqueue_callback listener_cb;
+    pj_activesock_cfg asock_cfg;
+    pj_activesock_cb listener_cb;
     pj_sockaddr_in *listener_addr;
     int addr_len;
-    unsigned i;
     pj_status_t status;
 
@@ -245,5 +217,4 @@
     listener->factory.flag = 
 	pjsip_transport_get_flag_from_type(PJSIP_TRANSPORT_TCP);
-    listener->sock = PJ_INVALID_SOCKET;
 
     pj_ansi_strcpy(listener->factory.obj_name, "tcplis");
@@ -256,6 +227,5 @@
 
     /* Create and bind socket */
-    status = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, 
-			    &listener->sock);
+    status = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0, &sock);
     if (status != PJ_SUCCESS)
 	goto on_error;
@@ -268,6 +238,5 @@
     }
 
-    status = pj_sock_bind(listener->sock, listener_addr, 
-			  sizeof(pj_sockaddr_in));
+    status = pj_sock_bind(sock, listener_addr, sizeof(pj_sockaddr_in));
     if (status != PJ_SUCCESS)
 	goto on_error;
@@ -275,5 +244,5 @@
     /* Retrieve the bound address */
     addr_len = sizeof(pj_sockaddr_in);
-    status = pj_sock_getsockname(listener->sock, listener_addr, &addr_len);
+    status = pj_sock_getsockname(sock, listener_addr, &addr_len);
     if (status != PJ_SUCCESS)
 	goto on_error;
@@ -321,17 +290,20 @@
 
     /* Start listening to the address */
-    status = pj_sock_listen(listener->sock, PJSIP_TCP_TRANSPORT_BACKLOG);
+    status = pj_sock_listen(sock, PJSIP_TCP_TRANSPORT_BACKLOG);
     if (status != PJ_SUCCESS)
 	goto on_error;
 
 
-    /* Register socket to ioqeuue */
+    /* Create active socket */
+    if (async_cnt > MAX_ASYNC_CNT) async_cnt = MAX_ASYNC_CNT;
+    pj_activesock_cfg_default(&asock_cfg);
+    asock_cfg.async_cnt = async_cnt;
+
     pj_bzero(&listener_cb, sizeof(listener_cb));
     listener_cb.on_accept_complete = &on_accept_complete;
-    status = pj_ioqueue_register_sock(pool, pjsip_endpt_get_ioqueue(endpt),
-				      listener->sock, listener,
-				      &listener_cb, &listener->key);
-    if (status != PJ_SUCCESS)
-	goto on_error;
+    status = pj_activesock_create(pool, sock, pj_SOCK_STREAM(), &asock_cfg,
+				  pjsip_endpt_get_ioqueue(endpt), 
+				  &listener_cb, listener,
+				  &listener->asock);
 
     /* Register to transport manager */
@@ -348,32 +320,8 @@
     }
 
-
     /* Start pending accept() operations */
-    if (async_cnt > MAX_ASYNC_CNT) async_cnt = MAX_ASYNC_CNT;
-    listener->async_cnt = async_cnt;
-
-    for (i=0; i<async_cnt; ++i) {
-	pj_pool_t *pool;
-
-	pool = pjsip_endpt_create_pool(endpt, "tcps%p", POOL_TP_INIT, 
-				       POOL_TP_INIT);
-	if (!pool) {
-	    status = PJ_ENOMEM;
-	    goto on_error;
-	}
-
-	listener->accept_op[i] = PJ_POOL_ZALLOC_T(pool, 
-						  struct pending_accept);
-	pj_ioqueue_op_key_init(&listener->accept_op[i]->op_key, 
-				sizeof(listener->accept_op[i]->op_key));
-	listener->accept_op[i]->pool = pool;
-	listener->accept_op[i]->listener = listener;
-	listener->accept_op[i]->index = i;
-
-	status = handle_accept(listener->key, &listener->accept_op[i]->op_key,
-			       listener->sock, PJ_EPENDING);
-	if (status != PJ_SUCCESS)
-	    goto on_error;
-    }
+    status = pj_activesock_start_accept(listener->asock, pool);
+    if (status != PJ_SUCCESS)
+	goto on_error;
 
     PJ_LOG(4,(listener->factory.obj_name, 
@@ -389,4 +337,6 @@
 
 on_error:
+    if (listener->asock==NULL && sock!=PJ_INVALID_SOCKET)
+	pj_sock_close(sock);
     lis_destroy(&listener->factory);
     return status;
@@ -411,5 +361,4 @@
 {
     struct tcp_listener *listener = (struct tcp_listener *)factory;
-    unsigned i;
 
     if (listener->is_registered) {
@@ -418,13 +367,7 @@
     }
 
-    if (listener->key) {
-	pj_ioqueue_unregister(listener->key);
-	listener->key = NULL;
-	listener->sock = PJ_INVALID_SOCKET;
-    }
-
-    if (listener->sock != PJ_INVALID_SOCKET) {
-	pj_sock_close(listener->sock);
-	listener->sock = PJ_INVALID_SOCKET;
+    if (listener->asock) {
+	pj_activesock_close(listener->asock);
+	listener->asock = NULL;
     }
 
@@ -432,12 +375,4 @@
 	pj_lock_destroy(listener->factory.lock);
 	listener->factory.lock = NULL;
-    }
-
-    for (i=0; i<PJ_ARRAY_SIZE(listener->accept_op); ++i) {
-	if (listener->accept_op[i] && listener->accept_op[i]->pool) {
-	    pj_pool_t *pool = listener->accept_op[i]->pool;
-	    listener->accept_op[i]->pool = NULL;
-	    pj_pool_release(pool);
-	}
     }
 
@@ -481,17 +416,19 @@
 			       pj_status_t reason);
 
-/* Callback from ioqueue on incoming packet */
-static void on_read_complete(pj_ioqueue_key_t *key, 
-                             pj_ioqueue_op_key_t *op_key, 
-                             pj_ssize_t bytes_read);
-
-/* Callback from ioqueue when packet is sent */
-static void on_write_complete(pj_ioqueue_key_t *key, 
-                              pj_ioqueue_op_key_t *op_key, 
-                              pj_ssize_t bytes_sent);
-
-/* Callback from ioqueue when connect completes */
-static void on_connect_complete(pj_ioqueue_key_t *key, 
-                                pj_status_t status);
+/* Callback on incoming data */
+static pj_bool_t on_data_read(pj_activesock_t *asock,
+			      void *data,
+			      pj_size_t size,
+			      pj_status_t status,
+			      pj_size_t *remainder);
+
+/* Callback when packet is sent */
+static pj_bool_t on_data_sent(pj_activesock_t *asock,
+			      pj_ioqueue_op_key_t *send_key,
+			      pj_ssize_t sent);
+
+/* Callback when connect completes */
+static pj_bool_t on_connect_complete(pj_activesock_t *asock,
+				     pj_status_t status);
 
 /* TCP keep-alive timer callback */
@@ -511,5 +448,6 @@
     struct tcp_transport *tcp;
     pj_ioqueue_t *ioqueue;
-    pj_ioqueue_callback tcp_callback;
+    pj_activesock_cfg asock_cfg;
+    pj_activesock_cb tcp_callback;
     const pj_str_t ka_pkt = PJSIP_TCP_KEEP_ALIVE_DATA;
     pj_status_t status;
@@ -529,6 +467,6 @@
      */
     tcp = PJ_POOL_ZALLOC_T(pool, struct tcp_transport);
+    tcp->is_server = is_server;
     tcp->sock = sock;
-    tcp->is_server = is_server;
     /*tcp->listener = listener;*/
     pj_list_init(&tcp->delayed_list);
@@ -570,13 +508,16 @@
 
 
-    /* Register socket to ioqueue */
-    pj_bzero(&tcp_callback, sizeof(pj_ioqueue_callback));
-    tcp_callback.on_read_complete = &on_read_complete;
-    tcp_callback.on_write_complete = &on_write_complete;
+    /* Create active socket */
+    pj_activesock_cfg_default(&asock_cfg);
+    asock_cfg.async_cnt = 1;
+
+    pj_bzero(&tcp_callback, sizeof(tcp_callback));
+    tcp_callback.on_data_read = &on_data_read;
+    tcp_callback.on_data_sent = &on_data_sent;
     tcp_callback.on_connect_complete = &on_connect_complete;
 
     ioqueue = pjsip_endpt_get_ioqueue(listener->endpt);
-    status = pj_ioqueue_register_sock(pool, ioqueue, sock, 
-				      tcp, &tcp_callback, &tcp->key);
+    status = pj_activesock_create(pool, sock, pj_SOCK_STREAM(), &asock_cfg,
+				  ioqueue, &tcp_callback, tcp, &tcp->asock);
     if (status != PJ_SUCCESS) {
 	goto on_error;
@@ -628,11 +569,10 @@
 	op_key = (pj_ioqueue_op_key_t*)pending_tx->tdata_op_key;
 
-	/* send to ioqueue! */
+	/* send! */
 	size = tdata->buf.cur - tdata->buf.start;
-	status = pj_ioqueue_send(tcp->key, op_key,
-				 tdata->buf.start, &size, 0);
-
+	status = pj_activesock_send(tcp->asock, op_key, tdata->buf.start, 
+				    &size, 0);
 	if (status != PJ_EPENDING) {
-	    on_write_complete(tcp->key, op_key, size);
+	    on_data_sent(tcp->asock, op_key, size);
 	}
 
@@ -694,5 +634,5 @@
 	op_key = (pj_ioqueue_op_key_t*)pending_tx->tdata_op_key;
 
-	on_write_complete(tcp->key, op_key, -reason);
+	on_data_sent(tcp->asock, op_key, -reason);
     }
 
@@ -702,11 +642,9 @@
     }
 
-    if (tcp->key) {
-	pj_ioqueue_unregister(tcp->key);
-	tcp->key = NULL;
+    if (tcp->asock) {
+	pj_activesock_close(tcp->asock);
+	tcp->asock = NULL;
 	tcp->sock = PJ_INVALID_SOCKET;
-    }
-
-    if (tcp->sock != PJ_INVALID_SOCKET) {
+    } else if (tcp->sock != PJ_INVALID_SOCKET) {
 	pj_sock_close(tcp->sock);
 	tcp->sock = PJ_INVALID_SOCKET;
@@ -760,4 +698,5 @@
     pj_ssize_t size;
     pj_sockaddr_in *rem_addr;
+    void *readbuf[1];
     pj_status_t status;
 
@@ -788,9 +727,10 @@
 
     size = sizeof(tcp->rdata.pkt_info.packet);
-    status = pj_ioqueue_recv(tcp->key, &tcp->rdata.tp_info.op_key.op_key,
-			     tcp->rdata.pkt_info.packet, &size,
-			     PJ_IOQUEUE_ALWAYS_ASYNC);
+    readbuf[0] = tcp->rdata.pkt_info.packet;
+    status = pj_activesock_start_read2(tcp->asock, tcp->base.pool, size,
+				       readbuf, 0);
     if (status != PJ_SUCCESS && status != PJ_EPENDING) {
-	PJ_LOG(4, (tcp->base.obj_name, "ioqueue recv() error, status=%d", 
+	PJ_LOG(4, (tcp->base.obj_name, 
+		   "pj_activesock_start_read() error, status=%d", 
 		   status));
 	return status;
@@ -862,5 +802,6 @@
     /* Start asynchronous connect() operation */
     tcp->has_pending_connect = PJ_TRUE;
-    status = pj_ioqueue_connect(tcp->key, rem_addr, sizeof(pj_sockaddr_in));
+    status = pj_activesock_start_connect(tcp->asock, tcp->base.pool, rem_addr,
+					 sizeof(pj_sockaddr_in));
     if (status == PJ_SUCCESS) {
 	tcp->has_pending_connect = PJ_FALSE;
@@ -874,5 +815,5 @@
      */
     addr_len = sizeof(pj_sockaddr_in);
-    if (pj_sock_getsockname(tcp->sock, &local_addr, &addr_len)==PJ_SUCCESS) {
+    if (pj_sock_getsockname(sock, &local_addr, &addr_len)==PJ_SUCCESS) {
 	pj_sockaddr_in *tp_addr = (pj_sockaddr_in*)&tcp->base.local_addr;
 
@@ -909,143 +850,58 @@
 
 /*
- * This callback is called by ioqueue when pending accept() operation has
- * completed.
- */
-static void on_accept_complete(	pj_ioqueue_key_t *key, 
-				pj_ioqueue_op_key_t *op_key, 
-				pj_sock_t sock, 
-				pj_status_t status)
-{
-    handle_accept(key, op_key, sock, status);
-}
-
-
-/* Handle accept() completion */
-static pj_status_t handle_accept(pj_ioqueue_key_t *key, 
-				 pj_ioqueue_op_key_t *op_key, 
-				 pj_sock_t sock, 
-				 pj_status_t status)
+ * This callback is called by active socket when pending accept() operation
+ * has completed.
+ */
+static pj_bool_t on_accept_complete(pj_activesock_t *asock,
+				    pj_sock_t sock,
+				    const pj_sockaddr_t *src_addr,
+				    int src_addr_len)
 {
     struct tcp_listener *listener;
     struct tcp_transport *tcp;
-    struct pending_accept *accept_op;
-    int err_cnt = 0;
-
-    listener = (struct tcp_listener*) pj_ioqueue_get_user_data(key);
-    accept_op = (struct pending_accept*) op_key;
-
-    /*
-     * Loop while there is immediate connection or when there is error.
+    char addr[PJ_INET6_ADDRSTRLEN+10];
+    pj_status_t status;
+
+    PJ_UNUSED_ARG(src_addr_len);
+
+    listener = (struct tcp_listener*) pj_activesock_get_user_data(asock);
+
+    PJ_ASSERT_RETURN(sock != PJ_INVALID_SOCKET, PJ_TRUE);
+
+    PJ_LOG(4,(listener->factory.obj_name, 
+	      "TCP listener %.*s:%d: got incoming TCP connection "
+	      "from %s, sock=%d",
+	      (int)listener->factory.addr_name.host.slen,
+	      listener->factory.addr_name.host.ptr,
+	      listener->factory.addr_name.port,
+	      pj_sockaddr_print(src_addr, addr, sizeof(addr), 3),
+	      sock));
+
+    /* 
+     * Incoming connection!
+     * Create TCP transport for the new socket.
      */
-    do {
-	if (status == PJ_EPENDING) {
-	    /*
-	     * This can only happen when this function is called during
-	     * initialization to kick off asynchronous accept().
-	     */
-
-	} else if (status != PJ_SUCCESS) {
-
-	    /*
-	     * Error in accept().
-	     */
-	    tcp_perror(listener->factory.obj_name, "Error in accept()", 
-		       status);
-
-	    /*
-	     * Prevent endless accept() error loop by limiting the
-	     * number of consecutive errors. Once the number of errors
-	     * is equal to maximum, we treat this as permanent error, and
-	     * we stop the accept() operation.
-	     */
-	    ++err_cnt;
-	    if (err_cnt >= 20) {
-		PJ_LOG(1, (listener->factory.obj_name, 
-			   "Too many errors, LISTENER IS STOPPING!"));
-		return status;
+    status = tcp_create( listener, NULL, sock, PJ_TRUE,
+			 (const pj_sockaddr_in*)&listener->factory.local_addr,
+			 (const pj_sockaddr_in*)src_addr, &tcp);
+    if (status == PJ_SUCCESS) {
+	status = tcp_start_read(tcp);
+	if (status != PJ_SUCCESS) {
+	    PJ_LOG(3,(tcp->base.obj_name, "New transport cancelled"));
+	    tcp_destroy(&tcp->base, status);
+	} else {
+	    /* Start keep-alive timer */
+	    if (PJSIP_TCP_KEEP_ALIVE_INTERVAL) {
+		pj_time_val delay = {PJSIP_TCP_KEEP_ALIVE_INTERVAL, 0};
+		pjsip_endpt_schedule_timer(listener->endpt, 
+					   &tcp->ka_timer, 
+					   &delay);
+		tcp->ka_timer.id = PJ_TRUE;
+		pj_gettimeofday(&tcp->last_activity);
 	    }
-
-	} else {
-	    pj_pool_t *pool;
-	    struct pending_accept *new_op;
-
-	    if (sock == PJ_INVALID_SOCKET) {
-		sock = accept_op->new_sock;
-	    }
-
-	    if (sock == PJ_INVALID_SOCKET) {
-		pj_assert(!"Should not happen. status should be error");
-		goto next_accept;
-	    }
-
-	    PJ_LOG(4,(listener->factory.obj_name, 
-		      "TCP listener %.*s:%d: got incoming TCP connection "
-		      "from %s:%d, sock=%d",
-		      (int)listener->factory.addr_name.host.slen,
-		      listener->factory.addr_name.host.ptr,
-		      listener->factory.addr_name.port,
-		      pj_inet_ntoa(accept_op->remote_addr.sin_addr),
-		      pj_ntohs(accept_op->remote_addr.sin_port),
-		      sock));
-
-	    /* Create new accept_opt */
-	    pool = pjsip_endpt_create_pool(listener->endpt, "tcps%p", 
-					   POOL_TP_INIT, POOL_TP_INC);
-	    new_op = PJ_POOL_ZALLOC_T(pool, struct pending_accept);
-	    new_op->pool = pool;
-	    new_op->listener = listener;
-	    new_op->index = accept_op->index;
-	    pj_ioqueue_op_key_init(&new_op->op_key, sizeof(new_op->op_key));
-	    listener->accept_op[accept_op->index] = new_op;
-
-	    /* 
-	     * Incoming connections!
-	     * Create TCP transport for the new socket.
-	     */
-	    status = tcp_create( listener, accept_op->pool, sock, PJ_TRUE,
-				 &accept_op->local_addr, 
-				 &accept_op->remote_addr, &tcp);
-	    if (status == PJ_SUCCESS) {
-		status = tcp_start_read(tcp);
-		if (status != PJ_SUCCESS) {
-		    PJ_LOG(3,(tcp->base.obj_name, "New transport cancelled"));
-		    tcp_destroy(&tcp->base, status);
-		} else {
-		    /* Start keep-alive timer */
-		    if (PJSIP_TCP_KEEP_ALIVE_INTERVAL) {
-			pj_time_val delay = {PJSIP_TCP_KEEP_ALIVE_INTERVAL, 0};
-			pjsip_endpt_schedule_timer(listener->endpt, 
-						   &tcp->ka_timer, 
-						   &delay);
-			tcp->ka_timer.id = PJ_TRUE;
-			pj_gettimeofday(&tcp->last_activity);
-		    }
-		}
-	    }
-
-	    accept_op = new_op;
 	}
-
-next_accept:
-	/*
-	 * Start the next asynchronous accept() operation.
-	 */
-	accept_op->addr_len = sizeof(pj_sockaddr_in);
-	accept_op->new_sock = PJ_INVALID_SOCKET;
-
-	status = pj_ioqueue_accept(listener->key, 
-				   &accept_op->op_key,
-				   &accept_op->new_sock,
-				   &accept_op->local_addr,
-				   &accept_op->remote_addr,
-				   &accept_op->addr_len);
-
-	/*
-	 * Loop while we have immediate connection or when there is error.
-	 */
-
-    } while (status != PJ_EPENDING);
-
-    return PJ_SUCCESS;
+    }
+
+    return PJ_TRUE;
 }
 
@@ -1054,10 +910,10 @@
  * Callback from ioqueue when packet is sent.
  */
-static void on_write_complete(pj_ioqueue_key_t *key, 
-                              pj_ioqueue_op_key_t *op_key, 
-                              pj_ssize_t bytes_sent)
+static pj_bool_t on_data_sent(pj_activesock_t *asock,
+			      pj_ioqueue_op_key_t *op_key,
+			      pj_ssize_t bytes_sent)
 {
     struct tcp_transport *tcp = (struct tcp_transport*) 
-    				pj_ioqueue_get_user_data(key);
+    				pj_activesock_get_user_data(asock);
     pjsip_tx_data_op_key *tdata_op_key = (pjsip_tx_data_op_key*)op_key;
 
@@ -1079,4 +935,5 @@
 	/* Mark last activity time */
 	pj_gettimeofday(&tcp->last_activity);
+
     }
 
@@ -1092,6 +949,9 @@
 	if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
 	pjsip_transport_shutdown(&tcp->base);
-    }
-
+
+	return PJ_FALSE;
+    }
+
+    return PJ_TRUE;
 }
 
@@ -1167,7 +1027,7 @@
 	 */
 	size = tdata->buf.cur - tdata->buf.start;
-	status = pj_ioqueue_send(tcp->key, 
-				 (pj_ioqueue_op_key_t*)&tdata->op_key,
-				 tdata->buf.start, &size, 0);
+	status = pj_activesock_send(tcp->asock, 
+				    (pj_ioqueue_op_key_t*)&tdata->op_key,
+				    tdata->buf.start, &size, 0);
 
 	if (status != PJ_EPENDING) {
@@ -1213,142 +1073,81 @@
  * Callback from ioqueue that an incoming data is received from the socket.
  */
-static void on_read_complete(pj_ioqueue_key_t *key, 
-                             pj_ioqueue_op_key_t *op_key, 
-                             pj_ssize_t bytes_read)
+static pj_bool_t on_data_read(pj_activesock_t *asock,
+			      void *data,
+			      pj_size_t size,
+			      pj_status_t status,
+			      pj_size_t *remainder)
 {
     enum { MAX_IMMEDIATE_PACKET = 10 };
-    pjsip_rx_data_op_key *rdata_op_key = (pjsip_rx_data_op_key*) op_key;
-    pjsip_rx_data *rdata = rdata_op_key->rdata;
-    struct tcp_transport *tcp = 
-	(struct tcp_transport*)rdata->tp_info.transport;
-    int i;
-    pj_status_t status;
+    struct tcp_transport *tcp;
+    pjsip_rx_data *rdata;
+
+    PJ_UNUSED_ARG(data);
+
+    tcp = (struct tcp_transport*) pj_activesock_get_user_data(asock);
+    rdata = &tcp->rdata;
 
     /* Don't do anything if transport is closing. */
     if (tcp->is_closing) {
 	tcp->is_closing++;
-	return;
-    }
-
-    /*
-     * The idea of the loop is to process immediate data received by
-     * pj_ioqueue_recv(), as long as i < MAX_IMMEDIATE_PACKET. When
-     * i is >= MAX_IMMEDIATE_PACKET, we force the recv() operation to
-     * complete asynchronously, to allow other sockets to get their data.
+	return PJ_FALSE;
+    }
+
+    /* Houston, we have packet! Report the packet to transport manager
+     * to be parsed.
      */
-    for (i=0;; ++i) {
-	pj_uint32_t flags;
-
-	/* Houston, we have packet! Report the packet to transport manager
-	 * to be parsed.
+    if (status == PJ_SUCCESS) {
+	pj_size_t size_eaten;
+
+	/* Mark this as an activity */
+	pj_gettimeofday(&tcp->last_activity);
+
+	pj_assert((void*)rdata->pkt_info.packet == data);
+
+	/* Init pkt_info part. */
+	rdata->pkt_info.len = size;
+	rdata->pkt_info.zero = 0;
+	pj_gettimeofday(&rdata->pkt_info.timestamp);
+
+	/* Report to transport manager.
+	 * The transport manager will tell us how many bytes of the packet
+	 * have been processed (as valid SIP message).
 	 */
-	if (bytes_read > 0) {
-	    pj_size_t size_eaten;
-
-	    /* Mark this as an activity */
-	    pj_gettimeofday(&tcp->last_activity);
-
-	    /* Init pkt_info part. */
-	    rdata->pkt_info.len += bytes_read;
-	    rdata->pkt_info.zero = 0;
-	    pj_gettimeofday(&rdata->pkt_info.timestamp);
-
-	    /* Report to transport manager.
-	     * The transport manager will tell us how many bytes of the packet
-	     * have been processed (as valid SIP message).
-	     */
-	    size_eaten = 
-		pjsip_tpmgr_receive_packet(rdata->tp_info.transport->tpmgr, 
-					   rdata);
-
-	    pj_assert(size_eaten <= (pj_size_t)rdata->pkt_info.len);
-
-	    /* Move unprocessed data to the front of the buffer */
-	    if (size_eaten>0 && size_eaten<(pj_size_t)rdata->pkt_info.len) {
-		pj_memmove(rdata->pkt_info.packet,
-			   rdata->pkt_info.packet + size_eaten,
-			   rdata->pkt_info.len - size_eaten);
-	    }
-	    
-	    rdata->pkt_info.len -= size_eaten;
-
-	} else if (bytes_read == 0) {
-
-	    /* Transport is closed */
-	    PJ_LOG(4,(tcp->base.obj_name, "TCP connection closed"));
-	    
-	    /* We can not destroy the transport since high level objects may
-	     * still keep reference to this transport. So we can only 
-	     * instruct transport manager to gracefully start the shutdown
-	     * procedure for this transport.
-	     */
-	    if (tcp->close_reason==PJ_SUCCESS) 
-		tcp->close_reason = PJ_RETURN_OS_ERROR(OSERR_ENOTCONN);
-	    pjsip_transport_shutdown(&tcp->base);
-
-	    return;
-
-	//} else if (bytes_read < 0)  {
-	} else if (-bytes_read != PJ_STATUS_FROM_OS(OSERR_EWOULDBLOCK) &&
-		   -bytes_read != PJ_STATUS_FROM_OS(OSERR_EINPROGRESS) && 
-		   -bytes_read != PJ_STATUS_FROM_OS(OSERR_ECONNRESET)) 
-	{
-
-	    /* Socket error. */
-	    PJ_LOG(4,(tcp->base.obj_name, "TCP connection reset"));
-
-	    /* We can not destroy the transport since high level objects may
-	     * still keep reference to this transport. So we can only 
-	     * instruct transport manager to gracefully start the shutdown
-	     * procedure for this transport.
-	     */
-	    if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = -bytes_read;
-	    pjsip_transport_shutdown(&tcp->base);
-
-	    return;
+	size_eaten = 
+	    pjsip_tpmgr_receive_packet(rdata->tp_info.transport->tpmgr, 
+				       rdata);
+
+	pj_assert(size_eaten <= (pj_size_t)rdata->pkt_info.len);
+
+	/* Move unprocessed data to the front of the buffer */
+	*remainder = size - size_eaten;
+	if (*remainder > 0 && *remainder != size) {
+	    pj_memmove(rdata->pkt_info.packet,
+		       rdata->pkt_info.packet + size_eaten,
+		       *remainder);
 	}
 
-	if (i >= MAX_IMMEDIATE_PACKET) {
-	    /* Receive quota reached. Force ioqueue_recv() to 
-	     * return PJ_EPENDING 
-	     */
-	    flags = PJ_IOQUEUE_ALWAYS_ASYNC;
-	} else {
-	    flags = 0;
-	}
-
-	/* Reset pool. */
-	pj_pool_reset(rdata->tp_info.pool);
-
-	/* Read next packet. */
-	bytes_read = sizeof(rdata->pkt_info.packet) - rdata->pkt_info.len;
-	rdata->pkt_info.src_addr_len = sizeof(pj_sockaddr_in);
-	status = pj_ioqueue_recv(key, op_key, 
-				 rdata->pkt_info.packet+rdata->pkt_info.len,
-				 &bytes_read, flags);
-
-	if (status == PJ_SUCCESS) {
-
-	    /* Continue loop. */
-	    pj_assert(i < MAX_IMMEDIATE_PACKET);
-
-	} else if (status == PJ_EPENDING) {
-	    break;
-
-	} else {
-	    /* Socket error */
-	    PJ_LOG(4,(tcp->base.obj_name, "TCP connection reset"));
-
-	    /* We can not destroy the transport since high level objects may
-	     * still keep reference to this transport. So we can only 
-	     * instruct transport manager to gracefully start the shutdown
-	     * procedure for this transport.
-	     */
-	    if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
-	    pjsip_transport_shutdown(&tcp->base);
-
-	    return;
-	}
-    }
+    } else {
+
+	/* Transport is closed */
+	PJ_LOG(4,(tcp->base.obj_name, "TCP connection closed"));
+	
+	/* We can not destroy the transport since high level objects may
+	 * still keep reference to this transport. So we can only 
+	 * instruct transport manager to gracefully start the shutdown
+	 * procedure for this transport.
+	 */
+	if (tcp->close_reason==PJ_SUCCESS) 
+	    tcp->close_reason = status;
+	pjsip_transport_shutdown(&tcp->base);
+
+	return PJ_FALSE;
+
+    }
+
+    /* Reset pool. */
+    pj_pool_reset(rdata->tp_info.pool);
+
+    return PJ_TRUE;
 }
 
@@ -1357,6 +1156,6 @@
  * Callback from ioqueue when asynchronous connect() operation completes.
  */
-static void on_connect_complete(pj_ioqueue_key_t *key, 
-                                pj_status_t status)
+static pj_bool_t on_connect_complete(pj_activesock_t *asock,
+				     pj_status_t status)
 {
     struct tcp_transport *tcp;
@@ -1364,5 +1163,5 @@
     int addrlen;
 
-    tcp = (struct tcp_transport*) pj_ioqueue_get_user_data(key);
+    tcp = (struct tcp_transport*) pj_activesock_get_user_data(asock);
 
     /* Mark that pending connect() operation has completed. */
@@ -1384,5 +1183,5 @@
 	    op_key = (pj_ioqueue_op_key_t*)pending_tx->tdata_op_key;
 
-	    on_write_complete(tcp->key, op_key, -status);
+	    on_data_sent(tcp->asock, op_key, -status);
 	}
 
@@ -1394,5 +1193,5 @@
 	if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
 	pjsip_transport_shutdown(&tcp->base);
-	return;
+	return PJ_FALSE;
     }
 
@@ -1433,5 +1232,5 @@
 	if (tcp->close_reason==PJ_SUCCESS) tcp->close_reason = status;
 	pjsip_transport_shutdown(&tcp->base);
-	return;
+	return PJ_FALSE;
     }
 
@@ -1447,4 +1246,6 @@
 	pj_gettimeofday(&tcp->last_activity);
     }
+
+    return PJ_TRUE;
 }
 
@@ -1483,6 +1284,6 @@
     /* Send the data */
     size = tcp->ka_pkt.slen;
-    status = pj_ioqueue_send(tcp->key, &tcp->ka_op_key.key,
-			     tcp->ka_pkt.ptr, &size, 0);
+    status = pj_activesock_send(tcp->asock, &tcp->ka_op_key.key,
+				tcp->ka_pkt.ptr, &size, 0);
 
     if (status != PJ_SUCCESS && status != PJ_EPENDING) {
