]> git.ipfire.org Git - thirdparty/asterisk.git/commitdiff
res_pjsip_pubsub: Prune subs with reliable transports at startup
authorGeorge Joseph <gjoseph@digium.com>
Sun, 28 Jan 2018 16:10:00 +0000 (09:10 -0700)
committerGeorge Joseph <gjoseph@digium.com>
Thu, 1 Feb 2018 17:32:26 +0000 (10:32 -0700)
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

CHANGES
UPGRADE.txt
contrib/ast-db-manage/config/versions/d3e4284f8707_add_prune_on_boot_to_ps_subscription_.py [new file with mode: 0644]
include/asterisk/res_pjsip.h
res/res_pjsip.c
res/res_pjsip/include/res_pjsip_private.h
res/res_pjsip/pjsip_transport_events.c
res/res_pjsip_outbound_registration.c
res/res_pjsip_pubsub.c
res/res_pjsip_registrar.c

diff --git a/CHANGES b/CHANGES
index fc52916311b4b7a14752647fffc107a0df2ff6f6..04c4346f33690868ca2ecf8f90de9812d865fc3a 100644 (file)
--- a/CHANGES
+++ b/CHANGES
@@ -30,6 +30,10 @@ res_pjsip
    identifier method split into the "ip" and "header" endpoint identifier
    methods.
 
+ * The pjsip_transport_event feature introduced in 13.18.0 has been refactored.
+   Any external modules that may have used that feature (highly unlikey) will
+   need to be changed as the API has been altered slightly.
+
 res_pjsip_endpoint_identifier_ip
 ------------------
  * The endpoint identifier "ip" method previously recognized endpoints either
@@ -50,6 +54,17 @@ res_pjsip_registrar_expire
    from Asterisk 16.  The existing functionality was moved into
    res_pjsip_registrar.
 
+res_pjsip_pubsub
+------------------
+ * 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.  Since this
+   required the addition of a new column to the ps_subscription_persistence
+   realtime table, users who store their subscriptions in a database will
+   need to run the "alembic upgrade head" process to add the column to
+   the schema.
+
 ------------------------------------------------------------------------------
 --- Functionality changes from Asterisk 13.18.0 to Asterisk 13.19.0 ----------
 ------------------------------------------------------------------------------
index ff246ddffa20c57085db2b1c9de9aa1634d5e5f0..5d5cf8d0b34bd0a6a9f64da7fd1fc8794993b432 100644 (file)
@@ -30,6 +30,10 @@ res_pjsip
    identifier method split into the "ip" and "header" endpoint identifier
    methods.
 
+ * The pjsip_transport_event feature introduced in 13.18.0 has been refactored.
+   Any external modules that may have used that feature (highly unlikey) will
+   need to be changed as the API has been altered slightly.
+
 res_pjsip_endpoint_identifier_ip
 ------------------
  * The endpoint identifier "ip" method previously recognized endpoints either
diff --git a/contrib/ast-db-manage/config/versions/d3e4284f8707_add_prune_on_boot_to_ps_subscription_.py b/contrib/ast-db-manage/config/versions/d3e4284f8707_add_prune_on_boot_to_ps_subscription_.py
new file mode 100644 (file)
index 0000000..aa780be
--- /dev/null
@@ -0,0 +1,33 @@
+"""add prune_on_boot to ps_subscription_persistence
+
+Revision ID: d3e4284f8707
+Revises: 52798ad97bdf
+Create Date: 2018-01-28 17:45:36.218123
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = 'd3e4284f8707'
+down_revision = '52798ad97bdf'
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import ENUM
+
+YESNO_NAME = 'yesno_values'
+YESNO_VALUES = ['yes', 'no']
+
+
+def upgrade():
+    ############################# Enums ##############################
+
+    # yesno_values have already been created, so use postgres enum object
+    # type to get around "already created" issue - works okay with mysql
+    yesno_values = ENUM(*YESNO_VALUES, name=YESNO_NAME, create_type=False)
+
+    op.add_column('ps_subscription_persistence', sa.Column('prune_on_boot', yesno_values))
+
+def downgrade():
+    if op.get_context().bind.dialect.name == 'mssql':
+        op.drop_constraint('ck_ps_subscription_persistence_prune_on_boot_yesno_values','ps_subscription_persistence')
+    op.drop_column('ps_subscription_persistence', 'prune_on_boot')
index 2d3609b5ee3923d14c19351a7c5cd7804db10202..1a9076b442579ab4742a0dd1eed4e877e12a694c 100644 (file)
@@ -3009,6 +3009,18 @@ int ast_sip_str_to_dtmf(const char *dtmf_mode);
  */
 typedef void (*ast_transport_monitor_shutdown_cb)(void *data);
 
+/*!
+ * \brief Transport shutdown monitor data matcher
+ * \since 13.20.0
+ *
+ * \param a User data to compare.
+ * \param b User data to compare.
+ *
+ * \retval 1 The data objects match
+ * \retval 0 The data objects don't match
+ */
+typedef int (*ast_transport_monitor_data_matcher)(void *a, void *b);
+
 enum ast_transport_monitor_reg {
        /*! \brief Successfully registered the transport monitor */
        AST_TRANSPORT_MONITOR_REG_SUCCESS,
@@ -3025,37 +3037,59 @@ enum ast_transport_monitor_reg {
 
 /*!
  * \brief Register a reliable transport shutdown monitor callback.
- * \since 13.18.0
+ * \since 13.20.0
  *
  * \param transport Transport to monitor for shutdown.
  * \param cb Who to call when transport is shutdown.
  * \param ao2_data Data to pass with the callback.
  *
+ * \note The data object passed will have its reference count automatically
+ * incremented by this call and automatically decremented after the callback
+ * runs or when the callback is unregistered.
+ *
+ * There is no checking for duplicate registrations.
+ *
  * \return enum ast_transport_monitor_reg
  */
 enum ast_transport_monitor_reg ast_sip_transport_monitor_register(pjsip_transport *transport,
        ast_transport_monitor_shutdown_cb cb, void *ao2_data);
 
 /*!
- * \brief Unregister a reliable transport shutdown monitor callback.
- * \since 13.18.0
+ * \brief Unregister a reliable transport shutdown monitor
+ * \since 13.20.0
  *
  * \param transport Transport to monitor for shutdown.
- * \param cb Who to call when transport is shutdown.
+ * \param cb The callback that was used for the original register.
+ * \param data Data to pass to the matcher. May be NULL and does NOT need to be an ao2 object.
+ *             If NULL, all monitors with the provided callbck are unregistered.
+ * \param matches Matcher function that returns true if data matches the previously
+ *                registered data object.  If NULL, a simple pointer comparison is done.
+ *
+ * \note The data object passed into the original register will have its reference count
+ * automatically decremeneted.
  *
  * \return Nothing
  */
-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);
 
 /*!
- * \brief Unregister monitor callback from all reliable transports.
- * \since 13.18.0
+ * \brief Unregister a transport shutdown monitor from all reliable transports
+ * \since 13.20.0
+ *
+ * \param cb The callback that was used for the original register.
+ * \param data Data to pass to the matcher. May be NULL and does NOT need to be an ao2 object.
+ *             If NULL, all monitors with the provided callbck are unregistered.
+ * \param matches Matcher function that returns true if ao2_data matches the previously
+ *                registered data object.  If NULL, a simple pointer comparison is done.
  *
- * \param cb Who to call when a transport is shutdown.
+ * \note The data object passed into the original register will have its reference count
+ * automatically decremeneted.
  *
  * \return Nothing
  */
-void ast_sip_transport_monitor_unregister_all(ast_transport_monitor_shutdown_cb cb);
+void ast_sip_transport_monitor_unregister_all(ast_transport_monitor_shutdown_cb cb,
+       void *data, ast_transport_monitor_data_matcher matches);
 
 /*! Transport state notification registration element.  */
 struct ast_sip_tpmgr_state_callback {
index 05e10d31edae88a26bd532ad0fa08d3f61d51a23..8ec89c6586954f9cc3a95057a84064f039cf65df 100644 (file)
@@ -2894,6 +2894,45 @@ pjsip_endpoint *ast_sip_get_pjsip_endpoint(void)
        return ast_pjsip_endpoint;
 }
 
+int ast_sip_will_uri_survive_restart(pjsip_sip_uri *uri, struct ast_sip_endpoint *endpoint,
+       pjsip_rx_data *rdata)
+{
+       pj_str_t host_name;
+       int result = 1;
+
+       /* Determine if the contact cannot survive a restart/boot. */
+       if (uri->port == rdata->pkt_info.src_port
+               && !pj_strcmp(&uri->host,
+                       pj_cstr(&host_name, rdata->pkt_info.src_name))
+               /* We have already checked if the URI scheme is sip: or sips: */
+               && PJSIP_TRANSPORT_IS_RELIABLE(rdata->tp_info.transport)) {
+               pj_str_t type_name;
+
+               /* Determine the transport parameter value */
+               if (!strcasecmp("WSS", rdata->tp_info.transport->type_name)) {
+                       /* WSS is special, as it needs to be ws. */
+                       pj_cstr(&type_name, "ws");
+               } else {
+                       pj_cstr(&type_name, rdata->tp_info.transport->type_name);
+               }
+
+               if (!pj_stricmp(&uri->transport_param, &type_name)
+                       && (endpoint->nat.rewrite_contact
+                               /* Websockets are always rewritten */
+                               || !pj_stricmp(&uri->transport_param,
+                                       pj_cstr(&type_name, "ws")))) {
+                       /*
+                        * The contact was rewritten to the reliable transport's
+                        * source address.  Disconnecting the transport for any
+                        * reason invalidates the contact.
+                        */
+                       result = 0;
+               }
+       }
+
+       return result;
+}
+
 int ast_sip_get_transport_name(const struct ast_sip_endpoint *endpoint,
        pjsip_sip_uri *sip_uri, char *buf, size_t buf_len)
 {
index 151f59821a8c66af02f0a574dc00e4c3da76e3ff..7bffd2b6f58a1bf06ff0acc02d7f2973f6432b6d 100644 (file)
@@ -419,4 +419,18 @@ void internal_res_pjsip_ref(void);
  */
 void internal_res_pjsip_unref(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_ */
index 0f57303ba22fc385d65ded6a8d214fee2dec81d8..c701b841115dd439ebd4d23673414dbff7742037 100644 (file)
@@ -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);
index 0b177ae3223853bd28860c355deb2ee24557cd66..e3799fb2fe6e6d898ed270470c84e240aa487675 100644 (file)
@@ -851,6 +851,14 @@ static void registration_transport_shutdown_cb(void *obj)
        }
 }
 
+static int monitor_matcher(void *a, void *b)
+{
+       char *ma = a;
+       char *mb = b;
+
+       return strcmp(ma, mb) == 0;
+}
+
 static void registration_transport_monitor_setup(pjsip_transport *transport, const char *registration_name)
 {
        char *monitor;
@@ -951,7 +959,8 @@ static int handle_registration_response(void *data)
                        ast_debug(1, "Outbound unregistration to '%s' with client '%s' successful\n", server_uri, client_uri);
                        update_client_state_status(response->client_state, SIP_REGISTRATION_UNREGISTERED);
                        ast_sip_transport_monitor_unregister(response->rdata->tp_info.transport,
-                               registration_transport_shutdown_cb);
+                               registration_transport_shutdown_cb, response->client_state->registration_name,
+                               monitor_matcher);
                }
        } else if (response->client_state->destroy) {
                /* We need to deal with the pending destruction instead. */
@@ -2150,7 +2159,7 @@ static int unload_module(void)
 
        ao2_global_obj_release(current_states);
 
-       ast_sip_transport_monitor_unregister_all(registration_transport_shutdown_cb);
+       ast_sip_transport_monitor_unregister_all(registration_transport_shutdown_cb, NULL, NULL);
 
        /* Wait for registration serializers to get destroyed. */
        ast_debug(2, "Waiting for registration transactions to complete for unload.\n");
index 88005b85c83357a84bbfed3e7295af1edcf84c10..c2edb80d64cc013e51bb40371a9cddecbca176c2 100644 (file)
                                <configOption name="contact_uri">
                                        <synopsis>The Contact URI of the dialog for the subscription</synopsis>
                                </configOption>
+                               <configOption name="prune_on_boot">
+                                       <synopsis>If set, indicates that the contact used a reliable transport
+                                       and therefore the subscription must be deleted after an asterisk restart.
+                                       </synopsis>
+                               </configOption>
                        </configObject>
                        <configObject name="resource_list">
                                <synopsis>Resource list configuration parameters.</synopsis>
@@ -382,6 +387,8 @@ struct subscription_persistence {
        struct timeval expires;
        /*! Contact URI */
        char contact_uri[PJSIP_MAX_URL_SIZE];
+       /*! Prune subscription on restart */
+       int prune_on_boot;
 };
 
 /*!
@@ -446,6 +453,10 @@ struct sip_subscription_tree {
         * capable of restarting the timer.
         */
        struct ast_sip_sched_task *expiration_task;
+       /*! The transport the subscription was received on.
+        * Only used for reliable transports.
+        */
+       pjsip_transport *transport;
 };
 
 /*!
@@ -549,6 +560,17 @@ static void *publication_resource_alloc(const char *name)
        return ast_sorcery_generic_alloc(sizeof(struct ast_sip_publication_resource), publication_resource_destroy);
 }
 
+static void sub_tree_transport_cb(void *data) {
+       struct sip_subscription_tree *sub_tree = data;
+
+       ast_debug(3, "Transport destroyed.  Removing subscription '%s->%s'  prune on restart: %d\n",
+               sub_tree->persistence->endpoint, sub_tree->root->resource,
+               sub_tree->persistence->prune_on_boot);
+
+       sub_tree->state = SIP_SUB_TREE_TERMINATE_IN_PROGRESS;
+       pjsip_evsub_terminate(sub_tree->evsub, PJ_TRUE);
+}
+
 /*! \brief Destructor for subscription persistence */
 static void subscription_persistence_destroy(void *obj)
 {
@@ -599,8 +621,9 @@ static void subscription_persistence_update(struct sip_subscription_tree *sub_tr
                return;
        }
 
-       ast_debug(3, "Updating persistence for '%s->%s'\n", sub_tree->persistence->endpoint,
-               sub_tree->root->resource);
+       ast_debug(3, "Updating persistence for '%s->%s'  prune on restart: %s\n",
+               sub_tree->persistence->endpoint, sub_tree->root->resource,
+               sub_tree->persistence->prune_on_boot ? "yes" : "no");
 
        dlg = sub_tree->dlg;
        sub_tree->persistence->cseq = dlg->local.cseq;
@@ -614,6 +637,28 @@ static void subscription_persistence_update(struct sip_subscription_tree *sub_tr
                sub_tree->persistence->expires = ast_tvadd(ast_tvnow(), ast_samp2tv(expires, 1));
 
                if (contact_hdr) {
+                       if (contact_hdr) {
+                               if (type == SUBSCRIPTION_PERSISTENCE_CREATED) {
+                                       sub_tree->persistence->prune_on_boot =
+                                               !ast_sip_will_uri_survive_restart(
+                                                       (pjsip_sip_uri *)pjsip_uri_get_uri(contact_hdr->uri),
+                                                       sub_tree->endpoint, rdata);
+
+                                       if (sub_tree->persistence->prune_on_boot) {
+                                               ast_debug(3, "adding transport monitor on %s for '%s->%s'  prune on restart: %d\n",
+                                                       rdata->tp_info.transport->obj_name,
+                                                       sub_tree->persistence->endpoint, sub_tree->root->resource,
+                                                       sub_tree->persistence->prune_on_boot);
+                                               sub_tree->transport = rdata->tp_info.transport;
+                                               ast_sip_transport_monitor_register(rdata->tp_info.transport,
+                                                       sub_tree_transport_cb, sub_tree);
+                                               /*
+                                                * FYI: ast_sip_transport_monitor_register holds a reference to the sub_tree
+                                                */
+                                       }
+                               }
+                       }
+
                        pjsip_uri_print(PJSIP_URI_IN_CONTACT_HDR, contact_hdr->uri,
                                        sub_tree->persistence->contact_uri, sizeof(sub_tree->persistence->contact_uri));
                } else {
@@ -656,6 +701,15 @@ static void subscription_persistence_remove(struct sip_subscription_tree *sub_tr
                return;
        }
 
+       if (sub_tree->persistence->prune_on_boot && sub_tree->transport) {
+               ast_debug(3, "Unregistering transport monitor on %s '%s->%s'\n",
+                       sub_tree->transport->obj_name,
+                       sub_tree->endpoint ? ast_sorcery_object_get_id(sub_tree->endpoint) : "Unknown",
+                       sub_tree->root ? sub_tree->root->resource : "Unknown");
+               ast_sip_transport_monitor_unregister(sub_tree->transport,
+                       sub_tree_transport_cb, sub_tree, NULL);
+       }
+
        ast_sorcery_delete(ast_sip_get_sorcery(), sub_tree->persistence);
        ao2_ref(sub_tree->persistence, -1);
        sub_tree->persistence = NULL;
@@ -1586,6 +1640,14 @@ static int subscription_persistence_recreate(void *obj, void *arg, int flags)
        pjsip_rx_data rdata;
        struct persistence_recreate_data recreate_data;
 
+       /* If this subscription used a reliable transport it can't be reestablished so remove it */
+       if (persistence->prune_on_boot) {
+               ast_debug(3, "Deleting subscription marked as 'prune' from persistent store '%s' %s\n",
+                       persistence->endpoint, persistence->tag);
+               ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
+               return 0;
+       }
+
        /* If this subscription has already expired remove it */
        if (ast_tvdiff_ms(persistence->expires, ast_tvnow()) <= 0) {
                ast_debug(3, "Expired subscription retrived from persistent store '%s' %s\n",
@@ -5483,6 +5545,8 @@ static int load_module(void)
                persistence_expires_str2struct, persistence_expires_struct2str, NULL, 0, 0);
        ast_sorcery_object_field_register(sorcery, "subscription_persistence", "contact_uri", "", OPT_CHAR_ARRAY_T, 0,
                CHARFLDSET(struct subscription_persistence, contact_uri));
+       ast_sorcery_object_field_register(sorcery, "subscription_persistence", "prune_on_boot", "0", OPT_UINT_T, 0,
+               FLDSET(struct subscription_persistence, prune_on_boot));
 
        if (apply_list_configuration(sorcery)) {
                ast_sched_context_destroy(sched);
@@ -5559,6 +5623,8 @@ static int unload_module(void)
        AST_TEST_UNREGISTER(loop);
        AST_TEST_UNREGISTER(bad_event);
 
+       ast_sip_transport_monitor_unregister_all(sub_tree_transport_cb, NULL, NULL);
+
        ast_cli_unregister_multiple(cli_commands, ARRAY_LEN(cli_commands));
 
        ast_manager_unregister(AMI_SHOW_SUBSCRIPTIONS_OUTBOUND);
index a5d9b7421a4db4a986cc47a2e427817649218cc1..bddea27671eac94893775819159ef3107ba3daf4 100644 (file)
@@ -328,6 +328,15 @@ struct contact_transport_monitor {
        char aor_name[0];
 };
 
+static int contact_transport_monitor_matcher(void *a, void *b)
+{
+       struct contact_transport_monitor *ma = a;
+       struct contact_transport_monitor *mb = b;
+
+       return strcmp(ma->aor_name, mb->aor_name) == 0
+               && strcmp(ma->contact_name, mb->contact_name) == 0;
+}
+
 static void register_contact_transport_shutdown_cb(void *data)
 {
        struct contact_transport_monitor *monitor = data;
@@ -579,8 +588,7 @@ static void register_aor_core(pjsip_rx_data *rdata,
 
                contact = ao2_callback(contacts, OBJ_UNLINK, registrar_find_contact, &details);
                if (!contact) {
-                       int prune_on_boot = 0;
-                       pj_str_t host_name;
+                       int prune_on_boot;
 
                        /* If they are actually trying to delete a contact that does not exist... be forgiving */
                        if (!expiration) {
@@ -589,35 +597,7 @@ static void register_aor_core(pjsip_rx_data *rdata,
                                continue;
                        }
 
-                       /* Determine if the contact cannot survive a restart/boot. */
-                       if (details.uri->port == rdata->pkt_info.src_port
-                               && !pj_strcmp(&details.uri->host,
-                                       pj_cstr(&host_name, rdata->pkt_info.src_name))
-                               /* We have already checked if the URI scheme is sip: or sips: */
-                               && PJSIP_TRANSPORT_IS_RELIABLE(rdata->tp_info.transport)) {
-                               pj_str_t type_name;
-
-                               /* Determine the transport parameter value */
-                               if (!strcasecmp("WSS", rdata->tp_info.transport->type_name)) {
-                                       /* WSS is special, as it needs to be ws. */
-                                       pj_cstr(&type_name, "ws");
-                               } else {
-                                       pj_cstr(&type_name, rdata->tp_info.transport->type_name);
-                               }
-
-                               if (!pj_stricmp(&details.uri->transport_param, &type_name)
-                                       && (endpoint->nat.rewrite_contact
-                                               /* Websockets are always rewritten */
-                                               || !pj_stricmp(&details.uri->transport_param,
-                                                       pj_cstr(&type_name, "ws")))) {
-                                       /*
-                                        * The contact was rewritten to the reliable transport's
-                                        * source address.  Disconnecting the transport for any
-                                        * reason invalidates the contact.
-                                        */
-                                       prune_on_boot = 1;
-                               }
-                       }
+                       prune_on_boot = !ast_sip_will_uri_survive_restart(details.uri, endpoint, rdata);
 
                        contact = ast_sip_location_create_contact(aor, contact_uri,
                                ast_tvadd(ast_tvnow(), ast_samp2tv(expiration, 1)),
@@ -704,6 +684,21 @@ static void register_aor_core(pjsip_rx_data *rdata,
                                        contact_update->user_agent);
                        ao2_cleanup(contact_update);
                } else {
+                       if (contact->prune_on_boot) {
+                               struct contact_transport_monitor *monitor;
+                               const char *contact_name =
+                                       ast_sorcery_object_get_id(contact);
+
+                               monitor = ast_alloca(sizeof(*monitor) + 2 + strlen(aor_name)
+                                       + strlen(contact_name));
+                               strcpy(monitor->aor_name, aor_name);/* Safe */
+                               monitor->contact_name = monitor->aor_name + strlen(aor_name) + 1;
+                               strcpy(monitor->contact_name, contact_name);/* Safe */
+
+                               ast_sip_transport_monitor_unregister(rdata->tp_info.transport,
+                                       register_contact_transport_shutdown_cb, monitor, contact_transport_monitor_matcher);
+                       }
+
                        /* We want to report the user agent that was actually in the removed contact */
                        ast_sip_location_delete_contact(contact);
                        ast_verb(3, "Removed contact '%s' from AOR '%s' due to request\n", contact_uri, aor_name);
@@ -1125,6 +1120,19 @@ static int expire_contact(void *obj, void *arg, int flags)
         */
        ao2_lock(lock);
        if (ast_tvdiff_ms(ast_tvnow(), contact->expiration_time) > 0) {
+               if (contact->prune_on_boot) {
+                       struct contact_transport_monitor *monitor;
+                       const char *contact_name = ast_sorcery_object_get_id(contact);
+
+                       monitor = ast_alloca(sizeof(*monitor) + 2 + strlen(contact->aor)
+                               + strlen(contact_name));
+                       strcpy(monitor->aor_name, contact->aor);/* Safe */
+                       monitor->contact_name = monitor->aor_name + strlen(contact->aor) + 1;
+                       strcpy(monitor->contact_name, contact_name);/* Safe */
+
+                       ast_sip_transport_monitor_unregister_all(register_contact_transport_shutdown_cb,
+                               monitor, contact_transport_monitor_matcher);
+               }
                ast_sip_location_delete_contact(contact);
        }
        ao2_unlock(lock);
@@ -1236,7 +1244,7 @@ static int unload_module(void)
        ast_manager_unregister(AMI_SHOW_REGISTRATIONS);
        ast_manager_unregister(AMI_SHOW_REGISTRATION_CONTACT_STATUSES);
        ast_sip_unregister_service(&registrar_module);
-       ast_sip_transport_monitor_unregister_all(register_contact_transport_shutdown_cb);
+       ast_sip_transport_monitor_unregister_all(register_contact_transport_shutdown_cb, NULL, NULL);
        return 0;
 }