chan_pjsip: Add support for multiple streams of the same type.

The stream topology (list of streams and order) is now stored with the
configured PJSIP endpoints and used during the negotiation process.

Media negotiation state information has been changed to be stored
in a separate object. Two of these objects exist at any one time
on a session. The active media state information is what was previously
negotiated and the pending media state information is what the
media state will become if negotiation succeeds. Streams and other
state information is stored in this object using the index (or
position) of each individual stream for easy lookup.

The ability for a media type handler to specify a callback for
writing has been added as well as the ability to add file
descriptors with a callback which is invoked when data is available
to be read on them. This allows media logic to live outside of
the chan_pjsip module.

Direct media has been changed so that only the first audio and
video stream are directly connected. In the future once the RTP
engine glue API has been updated to know about streams each individual
stream can be directly connected as appropriate.

Media negotiation itself will currently answer all the provided streams
on an offer within configured limits and on an offer will use the
topology created as a result of the disallow/allow codec lines.

If a stream has been removed or declined we will now mark it as such
within the resulting SDP.

Applications can now also request that the stream topology change.
If we are told to do so we will limit any provided formats to the ones
configured on the endpoint and send a re-invite with the new topology.

Two new configuration options have also been added to PJSIP endpoints:

max_audio_streams: determines the maximum number of audio streams to
offer/accept from an endpoint. Defaults to 1.

max_video_streams: determines the maximum number of video streams to
offer/accept from an endpoint. Defaults to 1.

ASTERISK-27076

Change-Id: I8afd8dd2eb538806a39b887af0abd046266e14c7

3 files changed, 190 insertions, 45 deletions

diff --git a/channels/pjsip/cli_commands.c b/channels/pjsip/cli_commands.c
index fc14b25a8..33d0e02c1 100644
--- a/channels/pjsip/cli_commands.c
+++ b/channels/pjsip/cli_commands.c
@@ -342,8 +342,9 @@ static int cli_channelstats_print_body(void *obj, void *arg, int flags)
 	const struct ast_channel_snapshot *snapshot = obj;
 	struct ast_channel *channel = ast_channel_get_by_name(snapshot->name);
 	struct ast_sip_channel_pvt *cpvt = channel ? ast_channel_tech_pvt(channel) : NULL;
-	struct chan_pjsip_pvt *pvt = cpvt ? cpvt->pvt : NULL;
-	struct ast_sip_session_media *media = pvt ? pvt->media[SIP_MEDIA_AUDIO] : NULL;
+	struct ast_sip_session *session;
+	struct ast_sip_session_media *media;
+	struct ast_rtp_instance *rtp;
 	struct ast_rtp_instance_stats stats;
 	char *print_name = NULL;
 	char *print_time = alloca(32);
@@ -351,29 +352,46 @@ static int cli_channelstats_print_body(void *obj, void *arg, int flags)
 
 	ast_assert(context->output_buffer != NULL);
 
+	if (!channel) {
+		ast_str_append(&context->output_buffer, 0, " %s not valid\n", snapshot->name);
+		return -1;
+	}
+
+	ast_channel_lock(channel);
+
+	session = cpvt->session;
+	if (!session) {
+		ast_str_append(&context->output_buffer, 0, " %s not valid\n", snapshot->name);
+		ast_channel_unlock(channel);
+		ao2_cleanup(channel);
+		return -1;
+	}
+
+	media = session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO];
 	if (!media || !media->rtp) {
 		ast_str_append(&context->output_buffer, 0, " %s not valid\n", snapshot->name);
+		ast_channel_unlock(channel);
 		ao2_cleanup(channel);
 		return -1;
 	}
 
+	rtp = ao2_bump(media->rtp);
+
 	codec_in_use[0] = '\0';
 
-	if (channel) {
-		ast_channel_lock(channel);
-		if (ast_channel_rawreadformat(channel)) {
-			ast_copy_string(codec_in_use, ast_format_get_name(ast_channel_rawreadformat(channel)), sizeof(codec_in_use));
-		}
-		ast_channel_unlock(channel);
+	if (ast_channel_rawreadformat(channel)) {
+		ast_copy_string(codec_in_use, ast_format_get_name(ast_channel_rawreadformat(channel)), sizeof(codec_in_use));
 	}
 
+	ast_channel_unlock(channel);
+
 	print_name = ast_strdupa(snapshot->name);
 	/* Skip the PJSIP/.  We know what channel type it is and we need the space. */
 	print_name += 6;
 
 	ast_format_duration_hh_mm_ss(ast_tvnow().tv_sec - snapshot->creationtime.tv_sec, print_time, 32);
 
-	if (ast_rtp_instance_get_stats(media->rtp, &stats, AST_RTP_INSTANCE_STAT_ALL)) {
+	if (ast_rtp_instance_get_stats(rtp, &stats, AST_RTP_INSTANCE_STAT_ALL)) {
 		ast_str_append(&context->output_buffer, 0, "%s direct media\n", snapshot->name);
 	} else {
 		ast_str_append(&context->output_buffer, 0,
@@ -398,6 +416,7 @@ static int cli_channelstats_print_body(void *obj, void *arg, int flags)
 		);
 	}
 
+	ao2_cleanup(rtp);
 	ao2_cleanup(channel);
 
 	return 0;
diff --git a/channels/pjsip/dialplan_functions.c b/channels/pjsip/dialplan_functions.c
index e2c78cd87..59ca9d791 100644
--- a/channels/pjsip/dialplan_functions.c
+++ b/channels/pjsip/dialplan_functions.c
@@ -437,6 +437,7 @@
 #include "asterisk/acl.h"
 #include "asterisk/app.h"
 #include "asterisk/channel.h"
+#include "asterisk/stream.h"
 #include "asterisk/format.h"
 #include "asterisk/pbx.h"
 #include "asterisk/res_pjsip.h"
@@ -461,8 +462,8 @@ static const char *t38state_to_string[T38_MAX_ENUM] = {
 static int channel_read_rtp(struct ast_channel *chan, const char *type, const char *field, char *buf, size_t buflen)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(chan);
-	struct chan_pjsip_pvt *pvt;
-	struct ast_sip_session_media *media = NULL;
+	struct ast_sip_session *session;
+	struct ast_sip_session_media *media;
 	struct ast_sockaddr addr;
 
 	if (!channel) {
@@ -470,9 +471,9 @@ static int channel_read_rtp(struct ast_channel *chan, const char *type, const ch
 		return -1;
 	}
 
-	pvt = channel->pvt;
-	if (!pvt) {
-		ast_log(AST_LOG_WARNING, "Channel %s has no chan_pjsip pvt!\n", ast_channel_name(chan));
+	session = channel->session;
+	if (!session) {
+		ast_log(AST_LOG_WARNING, "Channel %s has no session!\n", ast_channel_name(chan));
 		return -1;
 	}
 
@@ -482,9 +483,9 @@ static int channel_read_rtp(struct ast_channel *chan, const char *type, const ch
 	}
 
 	if (ast_strlen_zero(field) || !strcmp(field, "audio")) {
-		media = pvt->media[SIP_MEDIA_AUDIO];
+		media = session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO];
 	} else if (!strcmp(field, "video")) {
-		media = pvt->media[SIP_MEDIA_VIDEO];
+		media = session->active_media_state->default_session[AST_MEDIA_TYPE_VIDEO];
 	} else {
 		ast_log(AST_LOG_WARNING, "Unknown media type field '%s' for 'rtp' information\n", field);
 		return -1;
@@ -522,17 +523,17 @@ static int channel_read_rtp(struct ast_channel *chan, const char *type, const ch
 static int channel_read_rtcp(struct ast_channel *chan, const char *type, const char *field, char *buf, size_t buflen)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(chan);
-	struct chan_pjsip_pvt *pvt;
-	struct ast_sip_session_media *media = NULL;
+	struct ast_sip_session *session;
+	struct ast_sip_session_media *media;
 
 	if (!channel) {
 		ast_log(AST_LOG_WARNING, "Channel %s has no pvt!\n", ast_channel_name(chan));
 		return -1;
 	}
 
-	pvt = channel->pvt;
-	if (!pvt) {
-		ast_log(AST_LOG_WARNING, "Channel %s has no chan_pjsip pvt!\n", ast_channel_name(chan));
+	session = channel->session;
+	if (!session) {
+		ast_log(AST_LOG_WARNING, "Channel %s has no session!\n", ast_channel_name(chan));
 		return -1;
 	}
 
@@ -542,9 +543,9 @@ static int channel_read_rtcp(struct ast_channel *chan, const char *type, const c
 	}
 
 	if (ast_strlen_zero(field) || !strcmp(field, "audio")) {
-		media = pvt->media[SIP_MEDIA_AUDIO];
+		media = session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO];
 	} else if (!strcmp(field, "video")) {
-		media = pvt->media[SIP_MEDIA_VIDEO];
+		media = session->active_media_state->default_session[AST_MEDIA_TYPE_VIDEO];
 	} else {
 		ast_log(AST_LOG_WARNING, "Unknown media type field '%s' for 'rtcp' information\n", field);
 		return -1;
@@ -924,22 +925,117 @@ int pjsip_acf_dial_contacts_read(struct ast_channel *chan, const char *cmd, char
 	return 0;
 }
 
+/*! \brief Session refresh state information */
+struct session_refresh_state {
+	/*! \brief Created proposed media state */
+	struct ast_sip_session_media_state *media_state;
+};
+
+/*! \brief Destructor for session refresh information */
+static void session_refresh_state_destroy(void *obj)
+{
+	struct session_refresh_state *state = obj;
+
+	ast_sip_session_media_state_free(state->media_state);
+	ast_free(obj);
+}
+
+/*! \brief Datastore for attaching session refresh state information */
+static const struct ast_datastore_info session_refresh_datastore = {
+	.type = "pjsip_session_refresh",
+	.destroy = session_refresh_state_destroy,
+};
+
+/*! \brief Helper function which retrieves or allocates a session refresh state information datastore */
+static struct session_refresh_state *session_refresh_state_get_or_alloc(struct ast_sip_session *session)
+{
+	RAII_VAR(struct ast_datastore *, datastore, ast_sip_session_get_datastore(session, "pjsip_session_refresh"), ao2_cleanup);
+	struct session_refresh_state *state;
+
+	/* While the datastore refcount is decremented this is operating in the serializer so it will remain valid regardless */
+	if (datastore) {
+		return datastore->data;
+	}
+
+	if (!(datastore = ast_sip_session_alloc_datastore(&session_refresh_datastore, "pjsip_session_refresh"))
+		|| !(datastore->data = ast_calloc(1, sizeof(struct session_refresh_state)))
+		|| ast_sip_session_add_datastore(session, datastore)) {
+		return NULL;
+	}
+
+	state = datastore->data;
+	state->media_state = ast_sip_session_media_state_alloc();
+	if (!state->media_state) {
+		ast_sip_session_remove_datastore(session, "pjsip_session_refresh");
+		return NULL;
+	}
+	state->media_state->topology = ast_stream_topology_clone(session->endpoint->media.topology);
+	if (!state->media_state->topology) {
+		ast_sip_session_remove_datastore(session, "pjsip_session_refresh");
+		return NULL;
+	}
+
+	datastore->data = state;
+
+	return state;
+}
+
 static int media_offer_read_av(struct ast_sip_session *session, char *buf,
 			       size_t len, enum ast_media_type media_type)
 {
+	struct ast_stream_topology *topology;
 	int idx;
+	struct ast_stream *stream = NULL;
+	struct ast_format_cap *caps;
 	size_t accum = 0;
 
+	if (session->inv_session->dlg->state == PJSIP_DIALOG_STATE_ESTABLISHED) {
+		struct session_refresh_state *state;
+
+		/* As we've already answered we need to store our media state until we are ready to send it */
+		state = session_refresh_state_get_or_alloc(session);
+		if (!state) {
+			return -1;
+		}
+		topology = state->media_state->topology;
+	} else {
+		/* The session is not yet up so we are initially answering or offering */
+		if (!session->pending_media_state->topology) {
+			session->pending_media_state->topology = ast_stream_topology_clone(session->endpoint->media.topology);
+			if (!session->pending_media_state->topology) {
+				return -1;
+			}
+		}
+		topology = session->pending_media_state->topology;
+	}
+
+	/* Find the first suitable stream */
+	for (idx = 0; idx < ast_stream_topology_get_count(topology); ++idx) {
+		stream = ast_stream_topology_get_stream(topology, idx);
+
+		if (ast_stream_get_type(stream) != media_type ||
+			ast_stream_get_state(stream) == AST_STREAM_STATE_REMOVED) {
+			stream = NULL;
+			continue;
+		}
+
+		break;
+	}
+
+	/* If no suitable stream then exit early */
+	if (!stream) {
+		buf[0] = '\0';
+		return 0;
+	}
+
+	caps = ast_stream_get_formats(stream);
+
 	/* Note: buf is not terminated while the string is being built. */
-	for (idx = 0; idx < ast_format_cap_count(session->req_caps); ++idx) {
+	for (idx = 0; idx < ast_format_cap_count(caps); ++idx) {
 		struct ast_format *fmt;
 		size_t size;
 
-		fmt = ast_format_cap_get_format(session->req_caps, idx);
-		if (ast_format_get_type(fmt) != media_type) {
-			ao2_ref(fmt, -1);
-			continue;
-		}
+		fmt = ast_format_cap_get_format(caps, idx);
 
 		/* Add one for a comma or terminator */
 		size = strlen(ast_format_get_name(fmt)) + 1;
@@ -973,9 +1069,43 @@ struct media_offer_data {
 static int media_offer_write_av(void *obj)
 {
 	struct media_offer_data *data = obj;
+	struct ast_stream_topology *topology;
+	struct ast_stream *stream;
+	struct ast_format_cap *caps;
 
-	ast_format_cap_remove_by_type(data->session->req_caps, data->media_type);
-	ast_format_cap_update_by_allow_disallow(data->session->req_caps, data->value, 1);
+	if (data->session->inv_session->dlg->state == PJSIP_DIALOG_STATE_ESTABLISHED) {
+		struct session_refresh_state *state;
+
+		/* As we've already answered we need to store our media state until we are ready to send it */
+		state = session_refresh_state_get_or_alloc(data->session);
+		if (!state) {
+			return -1;
+		}
+		topology = state->media_state->topology;
+	} else {
+		/* The session is not yet up so we are initially answering or offering */
+		if (!data->session->pending_media_state->topology) {
+			data->session->pending_media_state->topology = ast_stream_topology_clone(data->session->endpoint->media.topology);
+			if (!data->session->pending_media_state->topology) {
+				return -1;
+			}
+		}
+		topology = data->session->pending_media_state->topology;
+	}
+
+	/* XXX This method won't work when it comes time to do multistream support. The proper way to do this
+	 * will either be to
+	 * a) Alter all media streams of a particular type.
+	 * b) Change the dialplan function to be able to specify which stream to alter and alter only that
+	 * one stream
+	 */
+	stream = ast_stream_topology_get_first_stream_by_type(topology, data->media_type);
+	if (!stream) {
+		return 0;
+	}
+	caps = ast_stream_get_formats(stream);
+	ast_format_cap_remove_by_type(caps, data->media_type);
+	ast_format_cap_update_by_allow_disallow(caps, data->value, 1);
 
 	return 0;
 }
@@ -1068,9 +1198,18 @@ static int sip_session_response_cb(struct ast_sip_session *session, pjsip_rx_dat
 static int refresh_write_cb(void *obj)
 {
 	struct refresh_data *data = obj;
+	struct session_refresh_state *state;
+
+	state = session_refresh_state_get_or_alloc(data->session);
+	if (!state) {
+		return -1;
+	}
 
 	ast_sip_session_refresh(data->session, NULL, NULL,
-		sip_session_response_cb, data->method, 1);
+		sip_session_response_cb, data->method, 1, state->media_state);
+
+	state->media_state = NULL;
+	ast_sip_session_remove_datastore(data->session, "pjsip_session_refresh");
 
 	return 0;
 }
diff --git a/channels/pjsip/include/chan_pjsip.h b/channels/pjsip/include/chan_pjsip.h
index b229a0487..1fee86419 100644
--- a/channels/pjsip/include/chan_pjsip.h
+++ b/channels/pjsip/include/chan_pjsip.h
@@ -34,25 +34,12 @@ struct transport_info_data {
 	pj_sockaddr local_addr;
 };
 
-/*!
- * \brief Positions of various media
- */
-enum sip_session_media_position {
-	/*! \brief First is audio */
-	SIP_MEDIA_AUDIO = 0,
-	/*! \brief Second is video */
-	SIP_MEDIA_VIDEO,
-	/*! \brief Last is the size for media details */
-	SIP_MEDIA_SIZE,
-};
 
 /*!
  * \brief The PJSIP channel driver pvt, stored in the \ref ast_sip_channel_pvt
  * data structure
  */
 struct chan_pjsip_pvt {
-	/*! \brief The available media sessions */
-	struct ast_sip_session_media *media[SIP_MEDIA_SIZE];
 };
 
 #endif /* _CHAN_PJSIP_HEADER */