res_pjsip_pubsub: Prune subs with reliable transports at startup

In an earlier release, inbound registrations on a reliable transport
were pruned on Asterisk restart since the TCP connection would have
been torn down and become unusable when Asterisk stopped.  This same
process is now also applied to inbound subscriptions.

Also fixed issues in res_pjsip_registrar where it wasn't handling the
monitoring correctly when multiple registrations came in over the same
transport.

To accomplish this, the pjsip_transport_event feature needed to
be refactored to allow multiple monitors (multiple subcriptions or
registrations from the same endpoint) to exist on the same transport.
Since this changed the API, any external modules that may have used the
transport monitor feature (highly unlikey) will need to be changed.

ASTERISK-27612
Reported by: Ross Beer

Change-Id: Iee87cf4eb9b7b2b93d5739a72af52d6ca8fbbe36

2 files changed, 62 insertions, 35 deletions

diff --git a/res/res_pjsip/include/res_pjsip_private.h b/res/res_pjsip/include/res_pjsip_private.h
index 7fafd8007..7d434aa95 100644
--- a/res/res_pjsip/include/res_pjsip_private.h
+++ b/res/res_pjsip/include/res_pjsip_private.h
@@ -347,4 +347,18 @@ int ast_sip_initialize_scheduler(void);
  */
 int ast_sip_destroy_scheduler(void);
 
+/*!
+ * \internal
+ * \brief Determines if a uri will still be valid after an asterisk restart
+ * \since 13.20.0
+ *
+ * \param uri uri to test
+ * \param endpoint The associated endpoint
+ * \param rdata The rdata to get transport information from
+ *
+ * \retval 1 Yes, 0 No
+ */
+int ast_sip_will_uri_survive_restart(pjsip_sip_uri *uri, struct ast_sip_endpoint *endpoint,
+	pjsip_rx_data *rdata);
+
 #endif /* RES_PJSIP_PRIVATE_H_ */
diff --git a/res/res_pjsip/pjsip_transport_events.c b/res/res_pjsip/pjsip_transport_events.c
index 0f57303ba..c701b8411 100644
--- a/res/res_pjsip/pjsip_transport_events.c
+++ b/res/res_pjsip/pjsip_transport_events.c
@@ -135,7 +135,7 @@ static void transport_state_callback(pjsip_transport *transport,
 				break;
 			}
 			monitored->transport = transport;
-			if (AST_VECTOR_INIT(&monitored->monitors, 2)) {
+			if (AST_VECTOR_INIT(&monitored->monitors, 5)) {
 				ao2_ref(monitored, -1);
 				break;
 			}
@@ -166,6 +166,8 @@ static void transport_state_callback(pjsip_transport *transport,
 					struct transport_monitor_notifier *notifier;
 
 					notifier = AST_VECTOR_GET_ADDR(&monitored->monitors, idx);
+					ast_debug(3, "running callback %p(%p) for transport %s\n",
+						notifier->cb, notifier->data, transport->obj_name);
 					notifier->cb(notifier->data);
 				}
 				ao2_ref(monitored, -1);
@@ -195,43 +197,66 @@ static void transport_state_callback(pjsip_transport *transport,
 	}
 }
 
-static int transport_monitor_unregister_all(void *obj, void *arg, int flags)
+struct callback_data {
+	ast_transport_monitor_shutdown_cb cb;
+	void *data;
+	ast_transport_monitor_data_matcher matches;
+};
+
+static int transport_monitor_unregister_cb(void *obj, void *arg, int flags)
 {
 	struct transport_monitor *monitored = obj;
-	ast_transport_monitor_shutdown_cb cb = arg;
+	struct callback_data *cb_data = arg;
 	int idx;
 
 	for (idx = AST_VECTOR_SIZE(&monitored->monitors); idx--;) {
 		struct transport_monitor_notifier *notifier;
 
 		notifier = AST_VECTOR_GET_ADDR(&monitored->monitors, idx);
-		if (notifier->cb == cb) {
+		if (notifier->cb == cb_data->cb && (!cb_data->data
+			|| cb_data->matches(cb_data->data, notifier->data))) {
 			ao2_cleanup(notifier->data);
 			AST_VECTOR_REMOVE_UNORDERED(&monitored->monitors, idx);
-			break;
+			ast_debug(3, "Unregistered monitor %p(%p) from transport %s\n",
+				notifier->cb, notifier->data, monitored->transport->obj_name);
 		}
 	}
 	return 0;
 }
 
-void ast_sip_transport_monitor_unregister_all(ast_transport_monitor_shutdown_cb cb)
+static int ptr_matcher(void *a, void *b)
+{
+	return a == b;
+}
+
+void ast_sip_transport_monitor_unregister_all(ast_transport_monitor_shutdown_cb cb,
+	void *data, ast_transport_monitor_data_matcher matches)
 {
 	struct ao2_container *transports;
+	struct callback_data cb_data = {
+		.cb = cb,
+		.data = data,
+		.matches = matches ?: ptr_matcher,
+	};
+
+	ast_assert(cb != NULL);
 
 	transports = ao2_global_obj_ref(active_transports);
 	if (!transports) {
 		return;
 	}
-	ao2_callback(transports, OBJ_MULTIPLE | OBJ_NODATA, transport_monitor_unregister_all,
-		cb);
+	ao2_callback(transports, OBJ_MULTIPLE | OBJ_NODATA, transport_monitor_unregister_cb, &cb_data);
 	ao2_ref(transports, -1);
 }
 
-void ast_sip_transport_monitor_unregister(pjsip_transport *transport, ast_transport_monitor_shutdown_cb cb)
+void ast_sip_transport_monitor_unregister(pjsip_transport *transport,
+	ast_transport_monitor_shutdown_cb cb, void *data, ast_transport_monitor_data_matcher matches)
 {
 	struct ao2_container *transports;
 	struct transport_monitor *monitored;
 
+	ast_assert(transport != NULL && cb != NULL);
+
 	transports = ao2_global_obj_ref(active_transports);
 	if (!transports) {
 		return;
@@ -240,18 +265,13 @@ void ast_sip_transport_monitor_unregister(pjsip_transport *transport, ast_transp
 	ao2_lock(transports);
 	monitored = ao2_find(transports, transport->obj_name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
 	if (monitored) {
-		int idx;
+		struct callback_data cb_data = {
+			.cb = cb,
+			.data = data,
+			.matches = matches ?: ptr_matcher,
+		};
 
-		for (idx = AST_VECTOR_SIZE(&monitored->monitors); idx--;) {
-			struct transport_monitor_notifier *notifier;
-
-			notifier = AST_VECTOR_GET_ADDR(&monitored->monitors, idx);
-			if (notifier->cb == cb) {
-				ao2_cleanup(notifier->data);
-				AST_VECTOR_REMOVE_UNORDERED(&monitored->monitors, idx);
-				break;
-			}
-		}
+		transport_monitor_unregister_cb(monitored, &cb_data, 0);
 		ao2_ref(monitored, -1);
 	}
 	ao2_unlock(transports);
@@ -265,6 +285,8 @@ enum ast_transport_monitor_reg ast_sip_transport_monitor_register(pjsip_transpor
 	struct transport_monitor *monitored;
 	enum ast_transport_monitor_reg res = AST_TRANSPORT_MONITOR_REG_NOT_FOUND;
 
+	ast_assert(transport != NULL && cb != NULL);
+
 	transports = ao2_global_obj_ref(active_transports);
 	if (!transports) {
 		return res;
@@ -273,31 +295,22 @@ enum ast_transport_monitor_reg ast_sip_transport_monitor_register(pjsip_transpor
 	ao2_lock(transports);
 	monitored = ao2_find(transports, transport->obj_name, OBJ_SEARCH_KEY | OBJ_NOLOCK);
 	if (monitored) {
-		int idx;
 		struct transport_monitor_notifier new_monitor;
 
-		/* Check if the callback monitor already exists */
-		for (idx = AST_VECTOR_SIZE(&monitored->monitors); idx--;) {
-			struct transport_monitor_notifier *notifier;
-
-			notifier = AST_VECTOR_GET_ADDR(&monitored->monitors, idx);
-			if (notifier->cb == cb) {
-				/* The monitor is already in the vector replace with new ao2_data. */
-				ao2_replace(notifier->data, ao2_data);
-				res = AST_TRANSPORT_MONITOR_REG_REPLACED;
-				goto register_done;
-			}
-		}
-
 		/* Add new monitor to vector */
 		new_monitor.cb = cb;
 		new_monitor.data = ao2_bump(ao2_data);
 		if (AST_VECTOR_APPEND(&monitored->monitors, new_monitor)) {
 			ao2_cleanup(ao2_data);
 			res = AST_TRANSPORT_MONITOR_REG_FAILED;
+			ast_debug(3, "Register monitor %p(%p) to transport %s FAILED\n",
+				cb, ao2_data, transport->obj_name);
+		} else {
+			res = AST_TRANSPORT_MONITOR_REG_SUCCESS;
+			ast_debug(3, "Registered monitor %p(%p) to transport %s\n",
+				cb, ao2_data, transport->obj_name);
 		}
 
-register_done:
 		ao2_ref(monitored, -1);
 	}
 	ao2_unlock(transports);