Ticket #1056:

 - Added functions to set/unset transport state notification callback on specific transport.
 - Updated transaction to immediately terminate the transactions when their transport gets disconnected.
 - Minor update: renamed function pjsip_tpmgr_set/get_status_cb() to pjsip_tpmgr_set/get_state_cb().



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

7 files changed, 325 insertions, 48 deletions

diff --git a/pjsip/include/pjsip/sip_transaction.h b/pjsip/include/pjsip/sip_transaction.h
index febc53c9..199d313d 100644
--- a/pjsip/include/pjsip/sip_transaction.h
+++ b/pjsip/include/pjsip/sip_transaction.h
@@ -124,6 +124,8 @@ struct pjsip_transaction
     pjsip_tx_data	       *pending_tx;	/**< Tdata which caused
 						     pending transport flag
 						     to be set on tsx.	    */
+    pjsip_tp_state_listener_key *tp_st_key;     /**< Transport state listener
+						     key.		    */
 
     /*
      * Messages and timer.
diff --git a/pjsip/include/pjsip/sip_transport.h b/pjsip/include/pjsip/sip_transport.h
index 92c624fe..bdcc6db4 100644
--- a/pjsip/include/pjsip/sip_transport.h
+++ b/pjsip/include/pjsip/sip_transport.h
@@ -750,6 +750,8 @@ struct pjsip_transport
     pjsip_tpmgr		   *tpmgr;	    /**< Transport manager.	    */
     pj_timer_entry	    idle_timer;	    /**< Timer when ref cnt is zero.*/
 
+    void		   *data;	    /**< Internal transport data.   */
+
     /**
      * Function to be called by transport manager to send SIP message.
      *
@@ -1270,6 +1272,11 @@ typedef enum pjsip_transport_state
 
 
 /**
+ * Definition of transport state listener key.
+ */
+typedef void pjsip_tp_state_listener_key;
+
+/**
  * Structure of transport state info passed by #pjsip_tp_state_callback.
  */
 typedef struct pjsip_transport_state_info {
@@ -1282,6 +1289,13 @@ typedef struct pjsip_transport_state_info {
      * Optional extended info, the content is specific for each transport type.
      */
     void		*ext_info;
+
+    /**
+     * Optional user data. In global transport state notification, this will
+     * always be NULL.
+     */
+    void		*user_data;
+
 } pjsip_transport_state_info;
 
 
@@ -1301,29 +1315,69 @@ typedef void (*pjsip_tp_state_callback)(
 
 
 /**
- * Setting callback of transport state notification. The caller will be
- * notified whenever the state of transport is changed. The type of
- * events are defined in #pjsip_transport_state.
+ * Set callback of global transport state notification. The caller will be
+ * notified whenever the state of any transport is changed. The type of events
+ * are defined in #pjsip_transport_state.
+ *
+ * Note that this function will override the existing callback, if any, so
+ * application is recommended to keep the old callback and manually forward
+ * the notification to the old callback, otherwise other component that 
+ * concerns about the transport state will no longer receive transport state 
+ * events.
  * 
  * @param mgr	    Transport manager.
  * @param cb	    Callback to be called to notify caller about transport 
- *		    status changing.
+ *		    state changing.
  *
  * @return	    PJ_SUCCESS on success, or the appropriate error code.
  */
-PJ_DECL(pj_status_t) pjsip_tpmgr_set_status_cb(pjsip_tpmgr *mgr,
-					       pjsip_tp_state_callback cb);
+PJ_DECL(pj_status_t) pjsip_tpmgr_set_state_cb(pjsip_tpmgr *mgr,
+					      pjsip_tp_state_callback cb);
 
 
 /**
- * Getting the callback of transport state notification.
+ * Get the callback of global transport state notification.
  * 
  * @param mgr	    Transport manager.
  *
  * @return	    The transport state callback or NULL if it is not set.
  */
-PJ_DECL(pjsip_tp_state_callback) pjsip_tpmgr_get_status_cb(
-					       const pjsip_tpmgr *mgr);
+PJ_DECL(pjsip_tp_state_callback) pjsip_tpmgr_get_state_cb(
+					      const pjsip_tpmgr *mgr);
+
+
+/**
+ * Add a listener to the specified transport for transport state notification.
+ * 
+ * @param tp	    The transport.
+ * @param cb	    Callback to be called to notify listener about transport 
+ *		    state changing.
+ * @param user_data The user data.
+ * @param key	    Output key, used to remove this listener.
+ *
+ * @return	    PJ_SUCCESS on success, or the appropriate error code.
+ */
+PJ_DECL(pj_status_t) pjsip_transport_add_state_listener (
+					    pjsip_transport *tp,
+					    pjsip_tp_state_callback cb,
+					    void *user_data,
+					    pjsip_tp_state_listener_key **key);
+
+
+/**
+ * Remove a listener from the specified transport for transport state 
+ * notification.
+ * 
+ * @param tp	    The transport.
+ * @param key	    The listener key.
+ * @param user_data The user data, for validation purpose.
+ *
+ * @return	    PJ_SUCCESS on success, or the appropriate error code.
+ */
+PJ_DECL(pj_status_t) pjsip_transport_remove_state_listener (
+				    pjsip_transport *tp,
+				    pjsip_tp_state_listener_key *key,
+				    const void *user_data);
 
 
 /**
diff --git a/pjsip/src/pjsip/sip_transaction.c b/pjsip/src/pjsip/sip_transaction.c
index 282c41c3..4b71dce6 100644
--- a/pjsip/src/pjsip/sip_transaction.c
+++ b/pjsip/src/pjsip/sip_transaction.c
@@ -180,6 +180,10 @@ static pj_status_t tsx_on_state_destroyed(	pjsip_transaction *tsx,
 					        pjsip_event *event);
 static void        tsx_timer_callback( pj_timer_heap_t *theap, 
 			               pj_timer_entry *entry);
+static void	   tsx_tp_state_callback(
+				       pjsip_transport *tp,
+				       pjsip_transport_state state,
+				       const pjsip_transport_state_info *info);
 static pj_status_t tsx_create( pjsip_module *tsx_user,
 			       pjsip_transaction **p_tsx);
 static pj_status_t tsx_destroy( pjsip_transaction *tsx );
@@ -187,6 +191,8 @@ static void	   tsx_resched_retransmission( pjsip_transaction *tsx );
 static pj_status_t tsx_retransmit( pjsip_transaction *tsx, int resched);
 static int         tsx_send_msg( pjsip_transaction *tsx, 
                                  pjsip_tx_data *tdata);
+static void        tsx_update_transport( pjsip_transaction *tsx, 
+					 pjsip_transport *tp);
 
 
 /* State handlers for UAC, indexed by state */
@@ -985,11 +991,9 @@ static pj_status_t tsx_destroy( pjsip_transaction *tsx )
 {
     struct tsx_lock_data *lck;
 
-    /* Decrement transport reference counter. */
-    if (tsx->transport) {
-	pjsip_transport_dec_ref( tsx->transport );
-	tsx->transport = NULL;
-    }
+    /* Release the transport */
+    tsx_update_transport(tsx, NULL);
+
     /* Decrement reference counter in transport selector */
     pjsip_tpselector_dec_ref(&tsx->tp_sel);
 
@@ -1404,8 +1408,7 @@ PJ_DEF(pj_status_t) pjsip_tsx_create_uas( pjsip_module *tsx_user,
      * transport.
      */
     if (tsx->res_addr.transport) {
-	tsx->transport = tsx->res_addr.transport;
-	pjsip_transport_add_ref(tsx->transport);
+	tsx_update_transport(tsx, tsx->res_addr.transport);
 	pj_memcpy(&tsx->addr, &tsx->res_addr.addr, tsx->res_addr.addr_len);
 	tsx->addr_len = tsx->res_addr.addr_len;
 	tsx->is_reliable = PJSIP_TRANSPORT_IS_RELIABLE(tsx->transport);
@@ -1675,12 +1678,7 @@ static void send_msg_callback( pjsip_send_state *send_state,
 
 	if (tsx->transport != send_state->cur_transport) {
 	    /* Update transport. */
-	    if (tsx->transport) {
-		pjsip_transport_dec_ref(tsx->transport);
-		tsx->transport = NULL;
-	    }
-	    tsx->transport = send_state->cur_transport;
-	    pjsip_transport_add_ref(tsx->transport);
+	    tsx_update_transport(tsx, send_state->cur_transport);
 
 	    /* Update remote address. */
 	    tsx->addr_len = tdata->dest_info.addr.entry[tdata->dest_info.cur_addr].addr_len;
@@ -1729,12 +1727,8 @@ static void send_msg_callback( pjsip_send_state *send_state,
 	/* If transaction is using the same transport as the failed one, 
 	 * release the transport.
 	 */
-	if (send_state->cur_transport==tsx->transport &&
-	    tsx->transport != NULL)
-	{
-	    pjsip_transport_dec_ref(tsx->transport);
-	    tsx->transport = NULL;
-	}
+	if (send_state->cur_transport==tsx->transport)
+	    tsx_update_transport(tsx, NULL);
 
 	/* Also stop processing if transaction has been flagged with
 	 * pending destroy (http://trac.pjsip.org/repos/ticket/906)
@@ -1836,9 +1830,8 @@ static void transport_callback(void *token, pjsip_tx_data *tdata,
 
 	lock_tsx(tsx, &lck);
 
-	/* Dereference transport. */
-	pjsip_transport_dec_ref(tsx->transport);
-	tsx->transport = NULL;
+	/* Release transport. */
+	tsx_update_transport(tsx, NULL);
 
 	/* Terminate transaction. */
 	tsx_set_status_code(tsx, PJSIP_SC_TSX_TRANSPORT_ERROR, &err);
@@ -1849,6 +1842,40 @@ static void transport_callback(void *token, pjsip_tx_data *tdata,
    }
 }
 
+
+/*
+ * Callback when transport state changes.
+ */
+static void tsx_tp_state_callback( pjsip_transport *tp,
+				   pjsip_transport_state state,
+				   const pjsip_transport_state_info *info)
+{
+    if (state == PJSIP_TP_STATE_DISCONNECTED) {
+	pjsip_transaction *tsx;
+	struct tsx_lock_data lck;
+
+	pj_assert(tp && info && info->user_data);
+
+	tsx = (pjsip_transaction*)info->user_data;
+
+	lock_tsx(tsx, &lck);
+
+	/* Terminate transaction when transport disconnected */
+	if (tsx->state < PJSIP_TSX_STATE_TERMINATED) {
+	    pj_str_t err;
+	    char errmsg[PJ_ERR_MSG_SIZE];
+
+	    err = pj_strerror(info->status, errmsg, sizeof(errmsg));
+	    tsx_set_status_code(tsx, PJSIP_SC_TSX_TRANSPORT_ERROR, &err);
+	    tsx_set_state( tsx, PJSIP_TSX_STATE_TERMINATED, 
+			   PJSIP_EVENT_TRANSPORT_ERROR, NULL);
+	}
+
+	unlock_tsx(tsx, &lck);
+    }
+}
+
+
 /*
  * Send message to the transport.
  */
@@ -1886,10 +1913,8 @@ static pj_status_t tsx_send_msg( pjsip_transaction *tsx,
 	    /* On error, release transport to force using full transport
 	     * resolution procedure.
 	     */
-	    if (tsx->transport) {
-		pjsip_transport_dec_ref(tsx->transport);
-		tsx->transport = NULL;
-	    }
+	    tsx_update_transport(tsx, NULL);
+
 	    tsx->addr_len = 0;
 	    tsx->res_addr.transport = NULL;
 	    tsx->res_addr.addr_len = 0;
@@ -2097,6 +2122,26 @@ static pj_status_t tsx_retransmit( pjsip_transaction *tsx, int resched)
     return PJ_SUCCESS;
 }
 
+static void tsx_update_transport( pjsip_transaction *tsx, 
+				  pjsip_transport *tp)
+{
+    pj_assert(tsx);
+
+    if (tsx->transport) {
+	pjsip_transport_remove_state_listener(tsx->transport, 
+					       tsx->tp_st_key, tsx);
+	pjsip_transport_dec_ref( tsx->transport );
+	tsx->transport = NULL;
+    }
+
+    if (tp) {
+	tsx->transport = tp;
+	pjsip_transport_add_ref(tp);
+	pjsip_transport_add_state_listener(tp, &tsx_tp_state_callback, tsx,
+					    &tsx->tp_st_key);
+    }
+}
+
 
 /*
  * Handler for events in state Null.
diff --git a/pjsip/src/pjsip/sip_transport.c b/pjsip/src/pjsip/sip_transport.c
index 6041fdb7..e1e22999 100644
--- a/pjsip/src/pjsip/sip_transport.c
+++ b/pjsip/src/pjsip/sip_transport.c
@@ -33,6 +33,7 @@
 #include <pj/pool.h>
 #include <pj/assert.h>
 #include <pj/lock.h>
+#include <pj/list.h>
 
 
 #define THIS_FILE    "sip_transport.c"
@@ -90,10 +91,30 @@ struct pjsip_tpmgr
     void           (*on_rx_msg)(pjsip_endpoint*, pj_status_t, pjsip_rx_data*);
     pj_status_t	   (*on_tx_msg)(pjsip_endpoint*, pjsip_tx_data*);
     pjsip_tp_state_callback tp_state_cb;
-    void *tp_state_user_data;
 };
 
 
+/* Transport state listener list type */
+typedef struct tp_state_listener
+{
+    PJ_DECL_LIST_MEMBER(struct tp_state_listener);
+
+    pjsip_tp_state_callback  cb;
+    void *user_data;
+} tp_state_listener;
+
+
+/*
+ * Transport data.
+ */
+typedef struct transport_data
+{
+    /* Transport listeners */
+    tp_state_listener	    st_listeners;
+    tp_state_listener	    st_listeners_empty;
+} transport_data;
+
+
 /*****************************************************************************
  *
  * GENERAL TRANSPORT (NAMES, TYPES, ETC.)
@@ -178,6 +199,11 @@ struct transport_names_t
     },
 };
 
+static void tp_state_callback(pjsip_transport *tp,
+			      pjsip_transport_state state,
+			      const pjsip_transport_state_info *info);
+
+
 struct transport_names_t *get_tpname(pjsip_transport_type_e type)
 {
     unsigned i;
@@ -1091,6 +1117,11 @@ PJ_DEF(pj_status_t) pjsip_tpmgr_create( pj_pool_t *pool,
 	return status;
 #endif
 
+    /* Set transport state callback */
+    status = pjsip_tpmgr_set_state_cb(mgr, &tp_state_callback);
+    if (status != PJ_SUCCESS)
+	return status;
+
     PJ_LOG(5, (THIS_FILE, "Transport manager created."));
 
     *p_mgr = mgr;
@@ -1737,8 +1768,11 @@ PJ_DEF(void) pjsip_tpmgr_dump_transports(pjsip_tpmgr *mgr)
 #endif
 }
 
-PJ_DEF(pj_status_t) pjsip_tpmgr_set_status_cb(pjsip_tpmgr *mgr,
-					      pjsip_tp_state_callback cb)
+/**
+ * Set callback of global transport state notification.
+ */
+PJ_DEF(pj_status_t) pjsip_tpmgr_set_state_cb(pjsip_tpmgr *mgr,
+					     pjsip_tp_state_callback cb)
 {
     PJ_ASSERT_RETURN(mgr, PJ_EINVAL);
 
@@ -1747,10 +1781,152 @@ PJ_DEF(pj_status_t) pjsip_tpmgr_set_status_cb(pjsip_tpmgr *mgr,
     return PJ_SUCCESS;
 }
 
-PJ_DEF(pjsip_tp_state_callback) pjsip_tpmgr_get_status_cb(
-					      const pjsip_tpmgr *mgr)
+/**
+ * Get callback of global transport state notification.
+ */
+PJ_DEF(pjsip_tp_state_callback) pjsip_tpmgr_get_state_cb(
+					     const pjsip_tpmgr *mgr)
 {
     PJ_ASSERT_RETURN(mgr, NULL);
 
     return mgr->tp_state_cb;
 }
+
+
+/**
+ * Allocate and init transport data.
+ */
+static void init_tp_data(pjsip_transport *tp)
+{
+    transport_data *tp_data;
+
+    pj_assert(tp && !tp->data);
+
+    tp_data = PJ_POOL_ZALLOC_T(tp->pool, transport_data);
+    pj_list_init(&tp_data->st_listeners);
+    pj_list_init(&tp_data->st_listeners_empty);
+    tp->data = tp_data;
+}
+
+
+static void tp_state_callback(pjsip_transport *tp,
+			      pjsip_transport_state state,
+			      const pjsip_transport_state_info *info)
+{
+    transport_data *tp_data;
+
+    pj_lock_acquire(tp->lock);
+
+    tp_data = (transport_data*)tp->data;
+
+    /* Notify the transport state listeners, if any. */
+    if (!tp_data || pj_list_empty(&tp_data->st_listeners)) {
+	goto on_return;
+    } else {
+	pjsip_transport_state_info st_info;
+	tp_state_listener *st_listener = tp_data->st_listeners.next;
+
+	/* As we need to put the user data into the transport state info,
+	 * let's use a copy of transport state info.
+	 */
+	pj_memcpy(&st_info, info, sizeof(st_info));
+	while (st_listener != &tp_data->st_listeners) {
+	    st_info.user_data = st_listener->user_data;
+	    (*st_listener->cb)(tp, state, &st_info);
+
+	    st_listener = st_listener->next;
+	}
+    }
+
+on_return:
+    pj_lock_release(tp->lock);
+}
+
+
+/**
+ * Add a listener to the specified transport for transport state notification.
+ */
+PJ_DEF(pj_status_t) pjsip_transport_add_state_listener (
+					    pjsip_transport *tp,
+					    pjsip_tp_state_callback cb,
+					    void *user_data,
+					    pjsip_tp_state_listener_key **key)
+{
+    transport_data *tp_data;
+    tp_state_listener *entry;
+
+    PJ_ASSERT_RETURN(tp && cb && key, PJ_EINVAL);
+
+    pj_lock_acquire(tp->lock);
+
+    /* Init transport data, if it hasn't */
+    if (!tp->data)
+	init_tp_data(tp);
+
+    tp_data = (transport_data*)tp->data;
+
+    /* Init the new listener entry. Use available empty slot, if any,
+     * otherwise allocate it using the transport pool.
+     */
+    if (!pj_list_empty(&tp_data->st_listeners_empty)) {
+	entry = tp_data->st_listeners_empty.next;
+	pj_list_erase(entry);
+    } else {
+	entry = PJ_POOL_ZALLOC_T(tp->pool, tp_state_listener);
+    }
+    entry->cb = cb;
+    entry->user_data = user_data;
+
+    /* Add the new listener entry to the listeners list */
+    pj_list_push_back(&tp_data->st_listeners, entry);
+
+    *key = entry;
+
+    pj_lock_release(tp->lock);
+
+    return PJ_SUCCESS;
+}
+
+/**
+ * Remove a listener from the specified transport for transport state 
+ * notification.
+ */
+PJ_DEF(pj_status_t) pjsip_transport_remove_state_listener (
+				    pjsip_transport *tp,
+				    pjsip_tp_state_listener_key *key,
+				    const void *user_data)
+{
+    transport_data *tp_data;
+    tp_state_listener *entry;
+
+    PJ_ASSERT_RETURN(tp && key, PJ_EINVAL);
+
+    pj_lock_acquire(tp->lock);
+
+    tp_data = (transport_data*)tp->data;
+
+    /* Transport data is NULL or no registered listener? */
+    if (!tp_data || pj_list_empty(&tp_data->st_listeners)) {
+	pj_lock_release(tp->lock);
+	return PJ_ENOTFOUND;
+    }
+
+    entry = (tp_state_listener*)key;
+
+    /* Validate the user data */
+    if (entry->user_data != user_data) {
+	pj_assert(!"Invalid transport state listener key");
+	pj_lock_release(tp->lock);
+	return PJ_EBUG;
+    }
+
+    /* Reset the entry and move it to the empty list */
+    entry->cb = NULL;
+    entry->user_data = NULL;
+    pj_list_erase(entry);
+    pj_list_push_back(&tp_data->st_listeners_empty, entry);
+
+    pj_lock_release(tp->lock);
+
+    return PJ_SUCCESS;
+}
diff --git a/pjsip/src/pjsip/sip_transport_tcp.c b/pjsip/src/pjsip/sip_transport_tcp.c
index fa03f143..1ad8fedc 100644
--- a/pjsip/src/pjsip/sip_transport_tcp.c
+++ b/pjsip/src/pjsip/sip_transport_tcp.c
@@ -184,7 +184,7 @@ static void tcp_init_shutdown(struct tcp_transport *tcp, pj_status_t status)
     pjsip_transport_add_ref(&tcp->base);
 
     /* Notify application of transport disconnected state */
-    state_cb = pjsip_tpmgr_get_status_cb(tcp->base.tpmgr);
+    state_cb = pjsip_tpmgr_get_state_cb(tcp->base.tpmgr);
     if (state_cb) {
 	pjsip_transport_state_info state_info;
 
@@ -1009,7 +1009,7 @@ static pj_bool_t on_accept_complete(pj_activesock_t *asock,
 	    }
 
 	    /* Notify application of transport state accepted */
-	    state_cb = pjsip_tpmgr_get_status_cb(tcp->base.tpmgr);
+	    state_cb = pjsip_tpmgr_get_state_cb(tcp->base.tpmgr);
 	    if (state_cb) {
 		pjsip_transport_state_info state_info;
             
@@ -1336,7 +1336,7 @@ static pj_bool_t on_connect_complete(pj_activesock_t *asock,
     }
 
     /* Notify application of transport state connected */
-    state_cb = pjsip_tpmgr_get_status_cb(tcp->base.tpmgr);
+    state_cb = pjsip_tpmgr_get_state_cb(tcp->base.tpmgr);
     if (state_cb) {
 	pjsip_transport_state_info state_info;
     
diff --git a/pjsip/src/pjsip/sip_transport_tls.c b/pjsip/src/pjsip/sip_transport_tls.c
index fa5cd2b1..b9700d8f 100644
--- a/pjsip/src/pjsip/sip_transport_tls.c
+++ b/pjsip/src/pjsip/sip_transport_tls.c
@@ -191,7 +191,7 @@ static void tls_init_shutdown(struct tls_transport *tls, pj_status_t status)
     pjsip_transport_add_ref(&tls->base);
 
     /* Notify application of transport disconnected state */
-    state_cb = pjsip_tpmgr_get_status_cb(tls->base.tpmgr);
+    state_cb = pjsip_tpmgr_get_state_cb(tls->base.tpmgr);
     if (state_cb) {
 	pjsip_transport_state_info state_info;
 
@@ -1048,7 +1048,7 @@ static pj_bool_t on_accept_complete(pj_ssl_sock_t *ssock,
     }
 
     /* Notify transport state to application */
-    state_cb = pjsip_tpmgr_get_status_cb(tls->base.tpmgr);
+    state_cb = pjsip_tpmgr_get_state_cb(tls->base.tpmgr);
     if (state_cb) {
 	pjsip_transport_state_info state_info;
 	pjsip_tls_state_info tls_info;
@@ -1466,7 +1466,7 @@ static pj_bool_t on_connect_complete(pj_ssl_sock_t *ssock,
     }
 
     /* Notify transport state to application */
-    state_cb = pjsip_tpmgr_get_status_cb(tls->base.tpmgr);
+    state_cb = pjsip_tpmgr_get_state_cb(tls->base.tpmgr);
     if (state_cb) {
 	pjsip_transport_state_info state_info;
 	pjsip_tls_state_info tls_info;
diff --git a/pjsip/src/pjsua-lib/pjsua_core.c b/pjsip/src/pjsua-lib/pjsua_core.c
index f25d2e51..dd892373 100644
--- a/pjsip/src/pjsua-lib/pjsua_core.c
+++ b/pjsip/src/pjsua-lib/pjsua_core.c
@@ -1889,11 +1889,11 @@ PJ_DEF(pj_status_t) pjsua_transport_create( pjsip_transport_type_e type,
 	pjsip_tpmgr *tpmgr;
 
 	tpmgr = pjsip_endpt_get_tpmgr(pjsua_var.endpt);
-	tpcb = pjsip_tpmgr_get_status_cb(tpmgr);
+	tpcb = pjsip_tpmgr_get_state_cb(tpmgr);
 
 	if (tpcb != &on_tp_state_callback) {
 	    pjsua_var.old_tp_cb = tpcb;
-	    pjsip_tpmgr_set_status_cb(tpmgr, &on_tp_state_callback);
+	    pjsip_tpmgr_set_state_cb(tpmgr, &on_tp_state_callback);
 	}
     }