res_pjsip_pubsub: Persist subscriptions in sorcery so they are recreated on startup.

This change makes res_pjsip_pubsub persist inbound subscriptions in sorcery. By default
this uses the local astdb but it can also be configured to store within an outside
database. When Asterisk is started these subscriptions are recreated if they have not
expired. Notifications are sent to the devices which have subscribed and they are none
the wiser that the system has restarted.

Review: https://reviewboard.asterisk.org/r/3598/
........

Merged revisions 415766 from http://svn.asterisk.org/svn/asterisk/branches/12


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@415767 65c4cc65-6c06-0410-ace0-fbb531ad65f3

contrib/ast-db-manage/config/versions/c6d929b23a8_create_pjsip_subscription_persistence_.py

@@ -0,0 +1,36 @@
+"""create pjsip subscription persistence table
+
+Revision ID: c6d929b23a8
+Revises: e96a0b8071c
+Create Date: 2014-06-06 02:17:38.660447
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = 'c6d929b23a8'
+down_revision = 'e96a0b8071c'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.create_table(
+        'ps_subscription_persistence',
+        sa.Column('id', sa.String(40), nullable=False, unique=True),
+	sa.Column('packet', sa.String(2048)),
+	sa.Column('src_name', sa.String(128)),
+	sa.Column('src_port', sa.Integer),
+	sa.Column('transport_key', sa.String(64)),
+	sa.Column('local_name', sa.String(128)),
+	sa.Column('local_port', sa.Integer),
+	sa.Column('cseq', sa.Integer),
+	sa.Column('tag', sa.String(128)),
+	sa.Column('endpoint', sa.String(40)),
+	sa.Column('expires', sa.Integer),
+    )
+
+    op.create_index('ps_subscription_persistence_id', 'ps_subscription_persistence', ['id'])
+
+def downgrade():
+	op.drop_table('ps_subscription_persistence')

include/asterisk/res_pjsip.h

@@ -1220,6 +1220,23 @@ pjsip_dialog *ast_sip_create_dialog_uac(const struct ast_sip_endpoint *endpoint,
  */
 pjsip_dialog *ast_sip_create_dialog_uas(const struct ast_sip_endpoint *endpoint, pjsip_rx_data *rdata);
 
+/*!
+ * \brief General purpose method for creating an rdata structure using specific information
+ *
+ * \param rdata[out] The rdata structure that will be populated
+ * \param packet A SIP message
+ * \param src_name The source IP address of the message
+ * \param src_port The source port of the message
+ * \param transport_type The type of transport the message was received on
+ * \param local_name The local IP address the message was received on
+ * \param local_port The local port the message was received on
+ *
+ * \retval 0 success
+ * \retval -1 failure
+ */
+int ast_sip_create_rdata(pjsip_rx_data *rdata, char *packet, const char *src_name, int src_port, char *transport_type,
+	const char *local_name, int local_port);
+
 /*!
  * \brief General purpose method for creating a SIP request
  *

include/asterisk/res_pjsip_pubsub.h

@@ -455,6 +455,18 @@ pjsip_evsub *ast_sip_subscription_get_evsub(struct ast_sip_subscription *sub);
  */
 pjsip_dialog *ast_sip_subscription_get_dlg(struct ast_sip_subscription *sub);
 
+/*!
+ * \brief Accept a subscription request
+ *
+ * \param sub The subscription to be accepted
+ * \param rdata The received subscription request
+ * \param response The response code to send
+ *
+ * \retval 0 Success
+ * \retval non-zero Failure
+ */
+int ast_sip_subscription_accept(struct ast_sip_subscription *sub, pjsip_rx_data *rdata, int response);
+
 /*!
  * \brief Send a request created via a PJSIP evsub method
  *

res/res_pjsip.c

@@ -1613,6 +1613,36 @@ pjsip_dialog *ast_sip_create_dialog_uas(const struct ast_sip_endpoint *endpoint,
 	return dlg;
 }
 
+int ast_sip_create_rdata(pjsip_rx_data *rdata, char *packet, const char *src_name, int src_port,
+	char *transport_type, const char *local_name, int local_port)
+{
+	pj_str_t tmp;
+
+	rdata->tp_info.transport = PJ_POOL_ZALLOC_T(rdata->tp_info.pool, pjsip_transport);
+	if (!rdata->tp_info.transport) {
+		return -1;
+	}
+
+	ast_copy_string(rdata->pkt_info.packet, packet, sizeof(rdata->pkt_info.packet));
+	ast_copy_string(rdata->pkt_info.src_name, src_name, sizeof(rdata->pkt_info.src_name));
+	rdata->pkt_info.src_port = src_port;
+
+	pjsip_parse_rdata(packet, strlen(packet), rdata);
+	if (!rdata->msg_info.msg) {
+		return -1;
+	}
+
+	pj_strdup2(rdata->tp_info.pool, &rdata->msg_info.via->recvd_param, rdata->pkt_info.src_name);
+	rdata->msg_info.via->rport_param = -1;
+
+	rdata->tp_info.transport->key.type = pjsip_transport_get_type_from_name(pj_cstr(&tmp, transport_type));
+	rdata->tp_info.transport->type_name = transport_type;
+	pj_strdup2(rdata->tp_info.pool, &rdata->tp_info.transport->local_name.host, local_name);
+	rdata->tp_info.transport->local_name.port = local_port;
+
+	return 0;
+}
+
 /* PJSIP doesn't know about the INFO method, so we have to define it ourselves */
 static const pjsip_method info_method = {PJSIP_OTHER_METHOD, {"INFO", 4} };
 static const pjsip_method message_method = {PJSIP_OTHER_METHOD, {"MESSAGE", 7} };

res/res_pjsip_exten_state.c

@@ -445,8 +445,7 @@ static struct ast_sip_subscription *new_subscribe(struct ast_sip_endpoint *endpo
 		return NULL;
 	}
 
-	if (pjsip_evsub_accept(ast_sip_subscription_get_evsub(exten_state_sub->sip_sub),
-			       rdata, 200, NULL) != PJ_SUCCESS) {
+	if (ast_sip_subscription_accept(exten_state_sub->sip_sub, rdata, 200)) {
 		ast_log(LOG_WARNING, "Unable to accept the incoming extension state subscription.\n");
 		pjsip_evsub_terminate(ast_sip_subscription_get_evsub(exten_state_sub->sip_sub), PJ_FALSE);
 		return NULL;

res/res_pjsip_mwi.c

@@ -699,7 +699,6 @@ static struct ast_sip_subscription *mwi_new_subscribe(struct ast_sip_endpoint *e
 	RAII_VAR(struct mwi_subscription *, sub, NULL, ao2_cleanup);
 	pjsip_uri *ruri = rdata->msg_info.msg->line.req.uri;
 	pjsip_sip_uri *sip_ruri;
-	pjsip_evsub *evsub;
 	char aor_name[80];
 
 	if (!PJSIP_URI_SCHEME_IS_SIP(ruri) && !PJSIP_URI_SCHEME_IS_SIPS(ruri)) {
@@ -715,8 +714,7 @@ static struct ast_sip_subscription *mwi_new_subscribe(struct ast_sip_endpoint *e
 		return NULL;
 	}
 
-	evsub = ast_sip_subscription_get_evsub(sub->sip_sub);
-	pjsip_evsub_accept(evsub, rdata, 200, NULL);
+	ast_sip_subscription_accept(sub->sip_sub, rdata, 200);
 	send_mwi_notify(sub, PJSIP_EVSUB_STATE_ACTIVE, NULL);
 
 	return sub->sip_sub;

res/res_pjsip_pubsub.c

@@ -72,6 +72,44 @@
                         </para>
 		</description>
 	</manager>
+	<configInfo name="res_pjsip_pubsub" language="en_US">
+		<synopsis>Module that implements publish and subscribe support.</synopsis>
+		<configFile name="pjsip.conf">
+			<configObject name="subscription_persistence">
+				<synopsis>Persists SIP subscriptions so they survive restarts.</synopsis>
+				<configOption name="packet">
+					<synopsis>Entire SIP SUBSCRIBE packet that created the subscription</synopsis>
+				</configOption>
+				<configOption name="src_name">
+					<synopsis>The source address of the subscription</synopsis>
+				</configOption>
+				<configOption name="src_port">
+					<synopsis>The source port of the subscription</synopsis>
+				</configOption>
+				<configOption name="transport_key">
+					<synopsis>The type of transport the subscription was received on</synopsis>
+				</configOption>
+				<configOption name="local_name">
+					<synopsis>The local address the subscription was received on</synopsis>
+				</configOption>
+				<configOption name="local_port">
+					<synopsis>The local port the subscription was received on</synopsis>
+				</configOption>
+				<configOption name="cseq">
+					<synopsis>The sequence number of the next NOTIFY to be sent</synopsis>
+				</configOption>
+				<configOption name="tag">
+					<synopsis>The local tag of the dialog for the subscription</synopsis>
+				</configOption>
+				<configOption name="endpoint">
+					<synopsis>The name of the endpoint that subscribed</synopsis>
+				</configOption>
+				<configOption name="expires">
+					<synopsis>The time at which the subscription expires</synopsis>
+				</configOption>
+			</configObject>
+		</configFile>
+	</configInfo>
  ***/
 
 static pj_bool_t pubsub_on_rx_request(pjsip_rx_data *rdata);
@@ -83,6 +121,7 @@ static struct pjsip_module pubsub_module = {
 };
 
 #define MOD_DATA_BODY_GENERATOR "sub_body_generator"
+#define MOD_DATA_PERSISTENCE "sub_persistence"
 
 static const pj_str_t str_event_name = { "Event", 5 };
 
@@ -180,6 +219,35 @@ struct ast_sip_publication {
 	int sched_id;
 };
 
+
+/*!
+ * \brief Structure used for persisting an inbound subscription
+ */
+struct subscription_persistence {
+	/*! Sorcery object details */
+	SORCERY_OBJECT(details);
+	/*! The name of the endpoint involved in the subscrption */
+	char *endpoint;
+	/*! SIP message that creates the subscription */
+	char packet[PJSIP_MAX_PKT_LEN];
+	/*! Source address of the message */
+	char src_name[PJ_INET6_ADDRSTRLEN];
+	/*! Source port of the message */
+	int src_port;
+	/*! Local transport key type */
+	char transport_key[32];
+	/*! Local transport address */
+	char local_name[PJ_INET6_ADDRSTRLEN];
+	/*! Local transport port */
+	int local_port;
+	/*! Next CSeq to use for message */
+	unsigned int cseq;
+	/*! Local tag of the dialog */
+	char *tag;
+	/*! When this subscription expires */
+	struct timeval expires;
+};
+
 /*!
  * \brief Structure representing a SIP subscription
  */
@@ -200,6 +268,8 @@ struct ast_sip_subscription {
 	pjsip_dialog *dlg;
 	/*! Body generaator for NOTIFYs */
 	struct ast_sip_pubsub_body_generator *body_generator;
+	/*! Persistence information */
+	struct subscription_persistence *persistence;
 	/*! Next item in the list */
 	AST_LIST_ENTRY(ast_sip_subscription) next;
 };
@@ -214,6 +284,265 @@ AST_RWLIST_HEAD_STATIC(subscriptions, ast_sip_subscription);
 AST_RWLIST_HEAD_STATIC(body_generators, ast_sip_pubsub_body_generator);
 AST_RWLIST_HEAD_STATIC(body_supplements, ast_sip_pubsub_body_supplement);
 
+/*! \brief Destructor for subscription persistence */
+static void subscription_persistence_destroy(void *obj)
+{
+	struct subscription_persistence *persistence = obj;
+
+	ast_free(persistence->endpoint);
+	ast_free(persistence->tag);
+}
+
+/*! \brief Allocator for subscription persistence */
+static void *subscription_persistence_alloc(const char *name)
+{
+	return ast_sorcery_generic_alloc(sizeof(struct subscription_persistence), subscription_persistence_destroy);
+}
+
+/*! \brief Function which creates initial persistence information of a subscription in sorcery */
+static struct subscription_persistence *subscription_persistence_create(struct ast_sip_subscription *sub)
+{
+	char tag[PJ_GUID_STRING_LENGTH + 1];
+
+	/* The id of this persistence object doesn't matter as we keep it on the subscription and don't need to
+	 * look it up by id at all.
+	 */
+	struct subscription_persistence *persistence = ast_sorcery_alloc(ast_sip_get_sorcery(),
+		"subscription_persistence", NULL);
+
+	if (!persistence) {
+		return NULL;
+	}
+
+	persistence->endpoint = ast_strdup(ast_sorcery_object_get_id(sub->endpoint));
+	ast_copy_pj_str(tag, &sub->dlg->local.info->tag, sizeof(tag));
+	persistence->tag = ast_strdup(tag);
+
+	ast_sorcery_create(ast_sip_get_sorcery(), persistence);
+	return persistence;
+}
+
+/*! \brief Function which updates persistence information of a subscription in sorcery */
+static void subscription_persistence_update(struct ast_sip_subscription *sub,
+	pjsip_rx_data *rdata)
+{
+	if (!sub->persistence) {
+		return;
+	}
+
+	sub->persistence->cseq = sub->dlg->local.cseq;
+
+	if (rdata) {
+		int expires;
+		pjsip_expires_hdr *expires_hdr = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_EXPIRES, NULL);
+
+		expires = expires_hdr ? expires_hdr->ivalue : DEFAULT_PUBLISH_EXPIRES;
+		sub->persistence->expires = ast_tvadd(ast_tvnow(), ast_samp2tv(expires, 1));
+
+		ast_copy_string(sub->persistence->packet, rdata->pkt_info.packet, sizeof(sub->persistence->packet));
+		ast_copy_string(sub->persistence->src_name, rdata->pkt_info.src_name, sizeof(sub->persistence->src_name));
+		sub->persistence->src_port = rdata->pkt_info.src_port;
+		ast_copy_string(sub->persistence->transport_key, rdata->tp_info.transport->type_name,
+			sizeof(sub->persistence->transport_key));
+		ast_copy_pj_str(sub->persistence->local_name, &rdata->tp_info.transport->local_name.host,
+			sizeof(sub->persistence->local_name));
+		sub->persistence->local_port = rdata->tp_info.transport->local_name.port;
+	}
+
+	ast_sorcery_update(ast_sip_get_sorcery(), sub->persistence);
+}
+
+/*! \brief Function which removes persistence of a subscription from sorcery */
+static void subscription_persistence_remove(struct ast_sip_subscription *sub)
+{
+	if (!sub->persistence) {
+		return;
+	}
+
+	ast_sorcery_delete(ast_sip_get_sorcery(), sub->persistence);
+	ao2_ref(sub->persistence, -1);
+}
+
+
+static struct ast_sip_subscription_handler *find_sub_handler_for_event_name(const char *event_name);
+static struct ast_sip_pubsub_body_generator *find_body_generator(char accept[AST_SIP_MAX_ACCEPT][64],
+		size_t num_accept);
+
+/*! \brief Retrieve a handler using the Event header of an rdata message */
+static struct ast_sip_subscription_handler *subscription_get_handler_from_rdata(pjsip_rx_data *rdata)
+{
+	pjsip_event_hdr *event_header;
+	char event[32];
+	struct ast_sip_subscription_handler *handler;
+
+	event_header = pjsip_msg_find_hdr_by_name(rdata->msg_info.msg, &str_event_name, rdata->msg_info.msg->hdr.next);
+	if (!event_header) {
+		ast_log(LOG_WARNING, "Incoming SUBSCRIBE request with no Event header\n");
+		return NULL;
+	}
+	ast_copy_pj_str(event, &event_header->event_type, sizeof(event));
+
+	handler = find_sub_handler_for_event_name(event);
+	if (!handler) {
+		ast_log(LOG_WARNING, "No registered subscribe handler for event %s\n", event);
+	}
+
+	return handler;
+}
+
+/*! \brief Retrieve a body generator using the Accept header of an rdata message */
+static struct ast_sip_pubsub_body_generator *subscription_get_generator_from_rdata(pjsip_rx_data *rdata,
+	const struct ast_sip_subscription_handler *handler)
+{
+	pjsip_accept_hdr *accept_header;
+	char accept[AST_SIP_MAX_ACCEPT][64];
+	size_t num_accept_headers;
+
+	accept_header = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_ACCEPT, rdata->msg_info.msg->hdr.next);
+	if (accept_header) {
+		int i;
+
+		for (i = 0; i < accept_header->count; ++i) {
+			ast_copy_pj_str(accept[i], &accept_header->values[i], sizeof(accept[i]));
+		}
+		num_accept_headers = accept_header->count;
+	} else {
+		/* If a SUBSCRIBE contains no Accept headers, then we must assume that
+		 * the default accept type for the event package is to be used.
+		 */
+		ast_copy_string(accept[0], handler->default_accept, sizeof(accept[0]));
+		num_accept_headers = 1;
+	}
+
+	return find_body_generator(accept, num_accept_headers);
+}
+
+/*! \brief Callback function to perform the actual recreation of a subscription */
+static int subscription_persistence_recreate(void *obj, void *arg, int flags)
+{
+	struct subscription_persistence *persistence = obj;
+	pj_pool_t *pool = arg;
+	pjsip_rx_data rdata = { { 0, }, };
+	pjsip_expires_hdr *expires_header;
+	struct ast_sip_subscription_handler *handler;
+	RAII_VAR(struct ast_sip_endpoint *, endpoint, NULL, ao2_cleanup);
+	struct ast_sip_subscription *sub;
+	struct ast_sip_pubsub_body_generator *generator;
+
+	/* If this subscription has already expired remove it */
+	if (ast_tvdiff_ms(persistence->expires, ast_tvnow()) <= 0) {
+		ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
+		return 0;
+	}
+
+	endpoint = ast_sorcery_retrieve_by_id(ast_sip_get_sorcery(), "endpoint", persistence->endpoint);
+	if (!endpoint) {
+		ast_log(LOG_WARNING, "A subscription for '%s' could not be recreated as the endpoint was not found\n",
+			persistence->endpoint);
+		ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
+		return 0;
+	}
+
+	pj_pool_reset(pool);
+	rdata.tp_info.pool = pool;
+
+	if (ast_sip_create_rdata(&rdata, persistence->packet, persistence->src_name, persistence->src_port,
+		persistence->transport_key, persistence->local_name, persistence->local_port)) {
+		ast_log(LOG_WARNING, "A subscription for '%s' could not be recreated as the message could not be parsed\n",
+			persistence->endpoint);
+		ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
+		return 0;
+	}
+
+	/* Update the expiration header with the new expiration */
+	expires_header = pjsip_msg_find_hdr(rdata.msg_info.msg, PJSIP_H_EXPIRES, rdata.msg_info.msg->hdr.next);
+	if (!expires_header) {
+		expires_header = pjsip_expires_hdr_create(pool, 0);
+		if (!expires_header) {
+			ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
+			return 0;
+		}
+		pjsip_msg_add_hdr(rdata.msg_info.msg, (pjsip_hdr*)expires_header);
+	}
+	expires_header->ivalue = (ast_tvdiff_ms(persistence->expires, ast_tvnow()) / 1000);
+
+	handler = subscription_get_handler_from_rdata(&rdata);
+	if (!handler) {
+		ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
+		return 0;
+	}
+
+	generator = subscription_get_generator_from_rdata(&rdata, handler);
+	if (!generator) {
+		ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
+		return 0;
+	}
+
+	ast_sip_mod_data_set(rdata.tp_info.pool, rdata.endpt_info.mod_data,
+			pubsub_module.id, MOD_DATA_BODY_GENERATOR, generator);
+	ast_sip_mod_data_set(rdata.tp_info.pool, rdata.endpt_info.mod_data,
+			pubsub_module.id, MOD_DATA_PERSISTENCE, persistence);
+
+	sub = handler->new_subscribe(endpoint, &rdata);
+	if (sub) {
+		sub->persistence = ao2_bump(persistence);
+		subscription_persistence_update(sub, &rdata);
+	} else {
+		ast_sorcery_delete(ast_sip_get_sorcery(), persistence);
+	}
+
+	return 0;
+}
+
+/*! \brief Function which loads and recreates persisted subscriptions upon startup when the system is fully booted */
+static int subscription_persistence_load(void *data)
+{
+	struct ao2_container *persisted_subscriptions = ast_sorcery_retrieve_by_fields(ast_sip_get_sorcery(),
+		"subscription_persistence", AST_RETRIEVE_FLAG_MULTIPLE | AST_RETRIEVE_FLAG_ALL, NULL);
+	pj_pool_t *pool;
+
+	pool = pjsip_endpt_create_pool(ast_sip_get_pjsip_endpoint(), "rtd%p", PJSIP_POOL_RDATA_LEN,
+		PJSIP_POOL_RDATA_INC);
+	if (!pool) {
+		ast_log(LOG_WARNING, "Could not create a memory pool for recreating SIP subscriptions\n");
+		return 0;
+	}
+
+	ao2_callback(persisted_subscriptions, OBJ_NODATA, subscription_persistence_recreate, pool);
+
+	pjsip_endpt_release_pool(ast_sip_get_pjsip_endpoint(), pool);
+
+	ao2_ref(persisted_subscriptions, -1);
+	return 0;
+}
+
+/*! \brief Event callback which fires subscription persistence recreation when the system is fully booted */
+static void subscription_persistence_event_cb(void *data, struct stasis_subscription *sub, struct stasis_message *message)
+{
+	struct ast_json_payload *payload;
+	const char *type;
+
+	if (stasis_message_type(message) != ast_manager_get_generic_type()) {
+		return;
+	}
+
+	payload = stasis_message_data(message);
+	type = ast_json_string_get(ast_json_object_get(payload->json, "type"));
+
+	/* This subscription only responds to the FullyBooted event so that all modules have been loaded when we
+	 * recreate SIP subscriptions.
+	 */
+	if (strcmp(type, "FullyBooted")) {
+		return;
+	}
+
+	/* This has to be here so the subscription is recreated when the body generator is available */
+	ast_sip_push_task(NULL, subscription_persistence_load, NULL);
+
+	/* Once the system is fully booted we don't care anymore */
+	stasis_unsubscribe(sub);
+}
+
 static void add_subscription(struct ast_sip_subscription *obj)
 {
 	SCOPED_LOCK(lock, &subscriptions, AST_RWLIST_WRLOCK, AST_RWLIST_UNLOCK);
@@ -335,6 +664,9 @@ static void subscription_destructor(void *obj)
 	struct ast_sip_subscription *sub = obj;
 
 	ast_debug(3, "Destroying SIP subscription\n");
+
+	subscription_persistence_remove(sub);
+
 	remove_subscription(sub);
 
 	ao2_cleanup(sub->datastores);
@@ -388,6 +720,7 @@ struct ast_sip_subscription *ast_sip_create_subscription(const struct ast_sip_su
 {
 	struct ast_sip_subscription *sub = ao2_alloc(sizeof(*sub), subscription_destructor);
 	pjsip_dialog *dlg;
+	struct subscription_persistence *persistence;
 
 	if (!sub) {
 		return NULL;
@@ -424,6 +757,17 @@ struct ast_sip_subscription *ast_sip_create_subscription(const struct ast_sip_su
 		ao2_ref(sub, -1);
 		return NULL;
 	}
+	persistence = ast_sip_mod_data_get(rdata->endpt_info.mod_data,
+			pubsub_module.id, MOD_DATA_PERSISTENCE);
+	if (persistence) {
+		/* Update the created dialog with the persisted information */
+		pjsip_ua_unregister_dlg(pjsip_ua_instance(), dlg);
+		pj_strdup2(dlg->pool, &dlg->local.info->tag, persistence->tag);
+		dlg->local.tag_hval = pj_hash_calc_tolower(0, NULL, &dlg->local.info->tag);
+		pjsip_ua_register_dlg(pjsip_ua_instance(), dlg);
+		dlg->local.cseq = persistence->cseq;
+		dlg->remote.cseq = persistence->cseq;
+	}
 	sub->evsub = allocate_evsub(handler->event_name, role, endpoint, rdata, dlg);
 	/* We keep a reference to the dialog until our subscription is destroyed. See
 	 * the subscription_destructor for more details
@@ -463,6 +807,16 @@ pjsip_dialog *ast_sip_subscription_get_dlg(struct ast_sip_subscription *sub)
 	return sub->dlg;
 }
 
+int ast_sip_subscription_accept(struct ast_sip_subscription *sub, pjsip_rx_data *rdata, int response)
+{
+	/* If this is a persistence recreation the subscription has already been accepted */
+	if (ast_sip_mod_data_get(rdata->endpt_info.mod_data, pubsub_module.id, MOD_DATA_PERSISTENCE)) {
+		return 0;
+	}
+
+	return pjsip_evsub_accept(ast_sip_subscription_get_evsub(sub), rdata, response, NULL) == PJ_SUCCESS ? 0 : -1;
+}
+
 int ast_sip_subscription_send_request(struct ast_sip_subscription *sub, pjsip_tx_data *tdata)
 {
 	struct ast_sip_endpoint *endpoint = ast_sip_subscription_get_endpoint(sub);
@@ -472,6 +826,8 @@ int ast_sip_subscription_send_request(struct ast_sip_subscription *sub, pjsip_tx
 	res = pjsip_evsub_send_request(ast_sip_subscription_get_evsub(sub),
 			tdata) == PJ_SUCCESS ? 0 : -1;
 
+	subscription_persistence_update(sub, NULL);
+
 	ast_test_suite_event_notify("SUBSCRIPTION_STATE_SET",
 		"StateText: %s\r\n"
 		"Endpoint: %s\r\n",
@@ -688,6 +1044,7 @@ int ast_sip_register_subscription_handler(struct ast_sip_subscription_handler *h
 	pjsip_evsub_register_pkg(&pubsub_module, &event, DEFAULT_EXPIRES, i, accept);
 
 	sub_add_handler(handler);
+
 	return 0;
 }
 
@@ -755,15 +1112,10 @@ static struct ast_sip_pubsub_body_generator *find_body_generator(char accept[AST
 
 static pj_bool_t pubsub_on_rx_subscribe_request(pjsip_rx_data *rdata)
 {
-	char event[32];
-	char accept[AST_SIP_MAX_ACCEPT][64];
-	pjsip_accept_hdr *accept_header;
-	pjsip_event_hdr *event_header;
 	pjsip_expires_hdr *expires_header;
 	struct ast_sip_subscription_handler *handler;
 	RAII_VAR(struct ast_sip_endpoint *, endpoint, NULL, ao2_cleanup);
 	struct ast_sip_subscription *sub;
-	size_t num_accept_headers;
 	struct ast_sip_pubsub_body_generator *generator;
 
 	endpoint = ast_pjsip_rdata_get_endpoint(rdata);
@@ -784,38 +1136,13 @@ static pj_bool_t pubsub_on_rx_subscribe_request(pjsip_rx_data *rdata)
 		return PJ_TRUE;
 	}
 
-	event_header = pjsip_msg_find_hdr_by_name(rdata->msg_info.msg, &str_event_name, rdata->msg_info.msg->hdr.next);
-	if (!event_header) {
-		ast_log(LOG_WARNING, "Incoming SUBSCRIBE request with no Event header\n");
-		pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 489, NULL, NULL, NULL);
-		return PJ_TRUE;
-	}
-	ast_copy_pj_str(event, &event_header->event_type, sizeof(event));
-
-	handler = find_sub_handler_for_event_name(event);
+	handler = subscription_get_handler_from_rdata(rdata);
 	if (!handler) {
-		ast_log(LOG_WARNING, "No registered subscribe handler for event %s\n", event);
 		pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 489, NULL, NULL, NULL);
 		return PJ_TRUE;
 	}
 
-	accept_header = pjsip_msg_find_hdr(rdata->msg_info.msg, PJSIP_H_ACCEPT, rdata->msg_info.msg->hdr.next);
-	if (accept_header) {
-		int i;
-
-		for (i = 0; i < accept_header->count; ++i) {
-			ast_copy_pj_str(accept[i], &accept_header->values[i], sizeof(accept[i]));
-		}
-		num_accept_headers = accept_header->count;
-	} else {
-		/* If a SUBSCRIBE contains no Accept headers, then we must assume that
-		 * the default accept type for the event package is to be used.
-		 */
-		ast_copy_string(accept[0], handler->default_accept, sizeof(accept[0]));
-		num_accept_headers = 1;
-	}
-
-	generator = find_body_generator(accept, num_accept_headers);
+	generator = subscription_get_generator_from_rdata(rdata, handler);
 	if (!generator) {
 		pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 489, NULL, NULL, NULL);
 		return PJ_TRUE;
@@ -839,7 +1166,11 @@ static pj_bool_t pubsub_on_rx_subscribe_request(pjsip_rx_data *rdata)
 		} else {
 			pjsip_endpt_respond_stateless(ast_sip_get_pjsip_endpoint(), rdata, 500, NULL, NULL, NULL);
 		}
+	} else {
+		sub->persistence = subscription_persistence_create(sub);
+		subscription_persistence_update(sub, rdata);
 	}
+
 	return PJ_TRUE;
 }
 
@@ -1471,9 +1802,54 @@ static int ami_show_subscriptions_outbound(struct mansession *s, const struct me
 #define AMI_SHOW_SUBSCRIPTIONS_INBOUND "PJSIPShowSubscriptionsInbound"
 #define AMI_SHOW_SUBSCRIPTIONS_OUTBOUND "PJSIPShowSubscriptionsOutbound"
 
+static int persistence_endpoint_str2struct(const struct aco_option *opt, struct ast_variable *var, void *obj)
+{
+	struct subscription_persistence *persistence = obj;
+
+	persistence->endpoint = ast_strdup(var->value);
+	return 0;
+}
+
+static int persistence_endpoint_struct2str(const void *obj, const intptr_t *args, char **buf)
+{
+	const struct subscription_persistence *persistence = obj;
+
+	*buf = ast_strdup(persistence->endpoint);
+	return 0;
+}
+
+static int persistence_tag_str2struct(const struct aco_option *opt, struct ast_variable *var, void *obj)
+{
+	struct subscription_persistence *persistence = obj;
+
+	persistence->tag = ast_strdup(var->value);
+	return 0;
+}
+
+static int persistence_tag_struct2str(const void *obj, const intptr_t *args, char **buf)
+{
+	const struct subscription_persistence *persistence = obj;
+
+	*buf = ast_strdup(persistence->tag);
+	return 0;
+}
+
+static int persistence_expires_str2struct(const struct aco_option *opt, struct ast_variable *var, void *obj)
+{
+	struct subscription_persistence *persistence = obj;
+	return ast_get_timeval(var->value, &persistence->expires, ast_tv(0, 0), NULL);
+}
+
+static int persistence_expires_struct2str(const void *obj, const intptr_t *args, char **buf)
+{
+	const struct subscription_persistence *persistence = obj;
+	return (ast_asprintf(buf, "%ld", persistence->expires.tv_sec) < 0) ? -1 : 0;
+}
+
 static int load_module(void)
 {
 	static const pj_str_t str_PUBLISH = { "PUBLISH", 7 };
+	struct ast_sorcery *sorcery = ast_sip_get_sorcery();
 
 	pjsip_evsub_init_module(ast_sip_get_pjsip_endpoint());
 
@@ -1496,6 +1872,42 @@ static int load_module(void)
 		return AST_MODULE_LOAD_FAILURE;
 	}
 
+	ast_sorcery_apply_config(sorcery, "res_pjsip_pubsub");
+	ast_sorcery_apply_default(sorcery, "subscription_persistence", "astdb", "subscription_persistence");
+	if (ast_sorcery_object_register(sorcery, "subscription_persistence", subscription_persistence_alloc,
+		NULL, NULL)) {
+		ast_log(LOG_ERROR, "Could not register subscription persistence object support\n");
+		ast_sip_unregister_service(&pubsub_module);
+		ast_sched_context_destroy(sched);
+		return AST_MODULE_LOAD_FAILURE;
+	}
+	ast_sorcery_object_field_register(sorcery, "subscription_persistence", "packet", "", OPT_CHAR_ARRAY_T, 0,
+		CHARFLDSET(struct subscription_persistence, packet));
+	ast_sorcery_object_field_register(sorcery, "subscription_persistence", "src_name", "", OPT_CHAR_ARRAY_T, 0,
+		CHARFLDSET(struct subscription_persistence, src_name));
+	ast_sorcery_object_field_register(sorcery, "subscription_persistence", "src_port", "0", OPT_UINT_T, 0,
+		FLDSET(struct subscription_persistence, src_port));
+	ast_sorcery_object_field_register(sorcery, "subscription_persistence", "transport_key", "0", OPT_CHAR_ARRAY_T, 0,
+		CHARFLDSET(struct subscription_persistence, transport_key));
+	ast_sorcery_object_field_register(sorcery, "subscription_persistence", "local_name", "", OPT_CHAR_ARRAY_T, 0,
+		CHARFLDSET(struct subscription_persistence, local_name));
+	ast_sorcery_object_field_register(sorcery, "subscription_persistence", "local_port", "0", OPT_UINT_T, 0,
+		FLDSET(struct subscription_persistence, local_port));
+	ast_sorcery_object_field_register(sorcery, "subscription_persistence", "cseq", "0", OPT_UINT_T, 0,
+		FLDSET(struct subscription_persistence, cseq));
+	ast_sorcery_object_field_register_custom(sorcery, "subscription_persistence", "endpoint", "",
+		persistence_endpoint_str2struct, persistence_endpoint_struct2str, NULL, 0, 0);
+	ast_sorcery_object_field_register_custom(sorcery, "subscription_persistence", "tag", "",
+		persistence_tag_str2struct, persistence_tag_struct2str, NULL, 0, 0);
+	ast_sorcery_object_field_register_custom(sorcery, "subscription_persistence", "expires", "",
+		persistence_expires_str2struct, persistence_expires_struct2str, NULL, 0, 0);
+
+	if (ast_test_flag(&ast_options, AST_OPT_FLAG_FULLY_BOOTED)) {
+		ast_sip_push_task(NULL, subscription_persistence_load, NULL);
+	} else {
+		stasis_subscribe(ast_manager_get_topic(), subscription_persistence_event_cb, NULL);
+	}
+
 	ast_manager_register_xml(AMI_SHOW_SUBSCRIPTIONS_INBOUND, EVENT_FLAG_SYSTEM,
 				 ami_show_subscriptions_inbound);
 	ast_manager_register_xml(AMI_SHOW_SUBSCRIPTIONS_OUTBOUND, EVENT_FLAG_SYSTEM,

res/res_pjsip_pubsub.exports.in

@@ -5,6 +5,7 @@
 		LINKER_SYMBOL_PREFIXast_sip_subscription_get_serializer;
 		LINKER_SYMBOL_PREFIXast_sip_subscription_get_evsub;
 		LINKER_SYMBOL_PREFIXast_sip_subscription_get_dlg;
+		LINKER_SYMBOL_PREFIXast_sip_subscription_accept;
 		LINKER_SYMBOL_PREFIXast_sip_subscription_send_request;
 		LINKER_SYMBOL_PREFIXast_sip_subscription_alloc_datastore;
 		LINKER_SYMBOL_PREFIXast_sip_subscription_add_datastore;