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

channels/chan_pjsip.c

@@ -64,6 +64,7 @@
 
 #include "asterisk/res_pjsip.h"
 #include "asterisk/res_pjsip_session.h"
+#include "asterisk/stream.h"
 
 #include "pjsip/include/chan_pjsip.h"
 #include "pjsip/include/dialplan_functions.h"
@@ -78,25 +79,22 @@ static unsigned int chan_idx;
 
 static void chan_pjsip_pvt_dtor(void *obj)
 {
-	struct chan_pjsip_pvt *pvt = obj;
-	int i;
-
-	for (i = 0; i < SIP_MEDIA_SIZE; ++i) {
-		ao2_cleanup(pvt->media[i]);
-		pvt->media[i] = NULL;
-	}
 }
 
 /* \brief Asterisk core interaction functions */
 static struct ast_channel *chan_pjsip_request(const char *type, struct ast_format_cap *cap, const struct ast_assigned_ids *assignedids, const struct ast_channel *requestor, const char *data, int *cause);
+static struct ast_channel *chan_pjsip_request_with_stream_topology(const char *type,
+	struct ast_stream_topology *topology, const struct ast_assigned_ids *assignedids,
+	const struct ast_channel *requestor, const char *data, int *cause);
 static int chan_pjsip_sendtext(struct ast_channel *ast, const char *text);
 static int chan_pjsip_digit_begin(struct ast_channel *ast, char digit);
 static int chan_pjsip_digit_end(struct ast_channel *ast, char digit, unsigned int duration);
 static int chan_pjsip_call(struct ast_channel *ast, const char *dest, int timeout);
 static int chan_pjsip_hangup(struct ast_channel *ast);
 static int chan_pjsip_answer(struct ast_channel *ast);
-static struct ast_frame *chan_pjsip_read(struct ast_channel *ast);
+static struct ast_frame *chan_pjsip_read_stream(struct ast_channel *ast);
 static int chan_pjsip_write(struct ast_channel *ast, struct ast_frame *f);
+static int chan_pjsip_write_stream(struct ast_channel *ast, int stream_num, struct ast_frame *f);
 static int chan_pjsip_indicate(struct ast_channel *ast, int condition, const void *data, size_t datalen);
 static int chan_pjsip_transfer(struct ast_channel *ast, const char *target);
 static int chan_pjsip_fixup(struct ast_channel *oldchan, struct ast_channel *newchan);
@@ -109,16 +107,17 @@ struct ast_channel_tech chan_pjsip_tech = {
 	.type = channel_type,
 	.description = "PJSIP Channel Driver",
 	.requester = chan_pjsip_request,
+	.requester_with_stream_topology = chan_pjsip_request_with_stream_topology,
 	.send_text = chan_pjsip_sendtext,
 	.send_digit_begin = chan_pjsip_digit_begin,
 	.send_digit_end = chan_pjsip_digit_end,
 	.call = chan_pjsip_call,
 	.hangup = chan_pjsip_hangup,
 	.answer = chan_pjsip_answer,
-	.read = chan_pjsip_read,
+	.read_stream = chan_pjsip_read_stream,
 	.write = chan_pjsip_write,
-	.write_video = chan_pjsip_write,
-	.exception = chan_pjsip_read,
+	.write_stream = chan_pjsip_write_stream,
+	.exception = chan_pjsip_read_stream,
 	.indicate = chan_pjsip_indicate,
 	.transfer = chan_pjsip_transfer,
 	.fixup = chan_pjsip_fixup,
@@ -159,11 +158,20 @@ static struct ast_sip_session_supplement chan_pjsip_ack_supplement = {
 static enum ast_rtp_glue_result chan_pjsip_get_rtp_peer(struct ast_channel *chan, struct ast_rtp_instance **instance)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(chan);
-	struct chan_pjsip_pvt *pvt;
 	struct ast_sip_endpoint *endpoint;
 	struct ast_datastore *datastore;
+	struct ast_sip_session_media *media;
 
-	if (!channel || !channel->session || !(pvt = channel->pvt) || !pvt->media[SIP_MEDIA_AUDIO]->rtp) {
+	if (!channel || !channel->session) {
+		return AST_RTP_GLUE_RESULT_FORBID;
+	}
+
+	/* XXX Getting the first RTP instance for direct media related stuff seems just
+	 * absolutely wrong. But the native RTP bridge knows no other method than single-stream
+	 * for direct media. So this is the best we can do.
+	 */
+	media = channel->session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO];
+	if (!media || !media->rtp) {
 		return AST_RTP_GLUE_RESULT_FORBID;
 	}
 
@@ -175,7 +183,7 @@ static enum ast_rtp_glue_result chan_pjsip_get_rtp_peer(struct ast_channel *chan
 
 	endpoint = channel->session->endpoint;
 
-	*instance = pvt->media[SIP_MEDIA_AUDIO]->rtp;
+	*instance = media->rtp;
 	ao2_ref(*instance, +1);
 
 	ast_assert(endpoint != NULL);
@@ -194,16 +202,21 @@ static enum ast_rtp_glue_result chan_pjsip_get_rtp_peer(struct ast_channel *chan
 static enum ast_rtp_glue_result chan_pjsip_get_vrtp_peer(struct ast_channel *chan, struct ast_rtp_instance **instance)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(chan);
-	struct chan_pjsip_pvt *pvt = channel->pvt;
 	struct ast_sip_endpoint *endpoint;
+	struct ast_sip_session_media *media;
 
-	if (!pvt || !channel->session || !pvt->media[SIP_MEDIA_VIDEO]->rtp) {
+	if (!channel || !channel->session) {
+		return AST_RTP_GLUE_RESULT_FORBID;
+	}
+
+	media = channel->session->active_media_state->default_session[AST_MEDIA_TYPE_VIDEO];
+	if (!media || !media->rtp) {
 		return AST_RTP_GLUE_RESULT_FORBID;
 	}
 
 	endpoint = channel->session->endpoint;
 
-	*instance = pvt->media[SIP_MEDIA_VIDEO]->rtp;
+	*instance = media->rtp;
 	ao2_ref(*instance, +1);
 
 	ast_assert(endpoint != NULL);
@@ -265,18 +278,43 @@ static int direct_media_mitigate_glare(struct ast_sip_session *session)
 	return 0;
 }
 
+/*! \brief Helper function to find the position for RTCP */
+static int rtp_find_rtcp_fd_position(struct ast_sip_session *session, struct ast_rtp_instance *rtp)
+{
+	int index;
+
+	for (index = 0; index < AST_VECTOR_SIZE(&session->active_media_state->read_callbacks); ++index) {
+		struct ast_sip_session_media_read_callback_state *callback_state =
+			AST_VECTOR_GET_ADDR(&session->active_media_state->read_callbacks, index);
+
+		if (callback_state->fd != ast_rtp_instance_fd(rtp, 1)) {
+			continue;
+		}
+
+		return index;
+	}
+
+	return -1;
+}
+
 /*!
  * \pre chan is locked
  */
 static int check_for_rtp_changes(struct ast_channel *chan, struct ast_rtp_instance *rtp,
-		struct ast_sip_session_media *media, int rtcp_fd)
+		struct ast_sip_session_media *media, struct ast_sip_session *session)
 {
-	int changed = 0;
+	int changed = 0, position = -1;
+
+	if (media->rtp) {
+		position = rtp_find_rtcp_fd_position(session, media->rtp);
+	}
 
 	if (rtp) {
 		changed = ast_rtp_instance_get_and_cmp_remote_address(rtp, &media->direct_media_addr);
 		if (media->rtp) {
-			ast_channel_set_fd(chan, rtcp_fd, -1);
+			if (position != -1) {
+				ast_channel_set_fd(chan, position + AST_EXTENDED_FDS, -1);
+			}
 			ast_rtp_instance_set_prop(media->rtp, AST_RTP_PROPERTY_RTCP, 0);
 		}
 	} else if (!ast_sockaddr_isnull(&media->direct_media_addr)){
@@ -284,7 +322,9 @@ static int check_for_rtp_changes(struct ast_channel *chan, struct ast_rtp_instan
 		changed = 1;
 		if (media->rtp) {
 			ast_rtp_instance_set_prop(media->rtp, AST_RTP_PROPERTY_RTCP, 1);
-			ast_channel_set_fd(chan, rtcp_fd, ast_rtp_instance_fd(media->rtp, 1));
+			if (position != -1) {
+				ast_channel_set_fd(chan, position + AST_EXTENDED_FDS, ast_rtp_instance_fd(media->rtp, 1));
+			}
 		}
 	}
 
@@ -333,22 +373,27 @@ static int send_direct_media_request(void *data)
 {
 	struct rtp_direct_media_data *cdata = data;
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(cdata->chan);
-	struct chan_pjsip_pvt *pvt = channel->pvt;
+	struct ast_sip_session *session;
 	int changed = 0;
 	int res = 0;
 
+	/* XXX In an ideal world each media stream would be direct, but for now preserve behavior
+	 * and connect only the default media sessions for audio and video.
+	 */
+
 	/* The channel needs to be locked when checking for RTP changes.
 	 * Otherwise, we could end up destroying an underlying RTCP structure
 	 * at the same time that the channel thread is attempting to read RTCP
 	 */
 	ast_channel_lock(cdata->chan);
-	if (pvt->media[SIP_MEDIA_AUDIO]) {
+	session = channel->session;
+	if (session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO]) {
 		changed |= check_for_rtp_changes(
-			cdata->chan, cdata->rtp, pvt->media[SIP_MEDIA_AUDIO], 1);
+			cdata->chan, cdata->rtp, session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO], session);
 	}
-	if (pvt->media[SIP_MEDIA_VIDEO]) {
+	if (session->active_media_state->default_session[AST_MEDIA_TYPE_VIDEO]) {
 		changed |= check_for_rtp_changes(
-			cdata->chan, cdata->vrtp, pvt->media[SIP_MEDIA_VIDEO], 3);
+			cdata->chan, cdata->vrtp, session->active_media_state->default_session[AST_MEDIA_TYPE_VIDEO], session);
 	}
 	ast_channel_unlock(cdata->chan);
 
@@ -368,7 +413,7 @@ static int send_direct_media_request(void *data)
 	if (changed) {
 		ast_debug(4, "RTP changed on %s; initiating direct media update\n", ast_channel_name(cdata->chan));
 		res = ast_sip_session_refresh(cdata->session, NULL, NULL, NULL,
-			cdata->session->endpoint->media.direct_media.method, 1);
+			cdata->session->endpoint->media.direct_media.method, 1, NULL);
 	}
 
 	ao2_ref(cdata, -1);
@@ -420,14 +465,53 @@ static struct ast_rtp_glue chan_pjsip_rtp_glue = {
 	.update_peer = chan_pjsip_set_rtp_peer,
 };
 
-static void set_channel_on_rtp_instance(struct chan_pjsip_pvt *pvt, const char *channel_id)
+static void set_channel_on_rtp_instance(const struct ast_sip_session *session,
+	const char *channel_id)
 {
-	if (pvt->media[SIP_MEDIA_AUDIO] && pvt->media[SIP_MEDIA_AUDIO]->rtp) {
-		ast_rtp_instance_set_channel_id(pvt->media[SIP_MEDIA_AUDIO]->rtp, channel_id);
+	int i;
+
+	for (i = 0; i < AST_VECTOR_SIZE(&session->active_media_state->sessions); ++i) {
+		struct ast_sip_session_media *session_media;
+
+		session_media = AST_VECTOR_GET(&session->active_media_state->sessions, i);
+		if (!session_media || !session_media->rtp) {
+			continue;
+		}
+
+		ast_rtp_instance_set_channel_id(session_media->rtp, channel_id);
 	}
-	if (pvt->media[SIP_MEDIA_VIDEO] && pvt->media[SIP_MEDIA_VIDEO]->rtp) {
-		ast_rtp_instance_set_channel_id(pvt->media[SIP_MEDIA_VIDEO]->rtp, channel_id);
+}
+
+/*!
+ * \brief Determine if a topology is compatible with format capabilities
+ *
+ * This will return true if ANY formats in the topology are compatible with the format
+ * capabilities.
+ *
+ * XXX When supporting true multistream, we will need to be sure to mark which streams from
+ * top1 are compatible with which streams from top2. Then the ones that are not compatible
+ * will need to be marked as "removed" so that they are negotiated as expected.
+ *
+ * \param top Topology
+ * \param cap Format capabilities
+ * \retval 1 The topology has at least one compatible format
+ * \retval 0 The topology has no compatible formats or an error occurred.
+ */
+static int compatible_formats_exist(struct ast_stream_topology *top, struct ast_format_cap *cap)
+{
+	struct ast_format_cap *cap_from_top;
+	int res;
+
+	cap_from_top = ast_format_cap_from_stream_topology(top);
+
+	if (!cap_from_top) {
+		return 0;
 	}
+
+	res = ast_format_cap_iscompatible(cap_from_top, cap);
+	ao2_ref(cap_from_top, -1);
+
+	return res;
 }
 
 /*! \brief Function called to create a new PJSIP Asterisk channel */
@@ -438,12 +522,9 @@ static struct ast_channel *chan_pjsip_new(struct ast_sip_session *session, int s
 	RAII_VAR(struct chan_pjsip_pvt *, pvt, NULL, ao2_cleanup);
 	struct ast_sip_channel_pvt *channel;
 	struct ast_variable *var;
+	struct ast_stream_topology *topology;
 
-	if (!(pvt = ao2_alloc(sizeof(*pvt), chan_pjsip_pvt_dtor))) {
-		return NULL;
-	}
-	caps = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
-	if (!caps) {
+	if (!(pvt = ao2_alloc_options(sizeof(*pvt), chan_pjsip_pvt_dtor, AO2_ALLOC_OPT_LOCK_NOLOCK))) {
 		return NULL;
 	}
 
@@ -457,31 +538,46 @@ static struct ast_channel *chan_pjsip_new(struct ast_sip_session *session, int s
 		ast_sorcery_object_get_id(session->endpoint),
 		(unsigned) ast_atomic_fetchadd_int((int *) &chan_idx, +1));
 	if (!chan) {
-		ao2_ref(caps, -1);
 		return NULL;
 	}
 
 	ast_channel_tech_set(chan, &chan_pjsip_tech);
 
 	if (!(channel = ast_sip_channel_pvt_alloc(pvt, session))) {
-		ao2_ref(caps, -1);
 		ast_channel_unlock(chan);
 		ast_hangup(chan);
 		return NULL;
 	}
 
-	ast_channel_stage_snapshot(chan);
-
 	ast_channel_tech_pvt_set(chan, channel);
 
-	if (!ast_format_cap_count(session->req_caps) ||
-		!ast_format_cap_iscompatible(session->req_caps, session->endpoint->media.codecs)) {
+	if (!ast_stream_topology_get_count(session->pending_media_state->topology) ||
+		!compatible_formats_exist(session->pending_media_state->topology, session->endpoint->media.codecs)) {
+		caps = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
+		if (!caps) {
+			ast_channel_unlock(chan);
+			ast_hangup(chan);
+			return NULL;
+		}
 		ast_format_cap_append_from_cap(caps, session->endpoint->media.codecs, AST_MEDIA_TYPE_UNKNOWN);
+		topology = ast_stream_topology_clone(session->endpoint->media.topology);
 	} else {
-		ast_format_cap_append_from_cap(caps, session->req_caps, AST_MEDIA_TYPE_UNKNOWN);
+		caps = ast_format_cap_from_stream_topology(session->pending_media_state->topology);
+		topology = ast_stream_topology_clone(session->pending_media_state->topology);
 	}
 
+	if (!topology || !caps) {
+		ao2_cleanup(caps);
+		ast_stream_topology_free(topology);
+		ast_channel_unlock(chan);
+		ast_hangup(chan);
+		return NULL;
+	}
+
+	ast_channel_stage_snapshot(chan);
+
 	ast_channel_nativeformats_set(chan, caps);
+	ast_channel_set_stream_topology(chan, topology);
 
 	if (!ast_format_cap_empty(caps)) {
 		struct ast_format *fmt;
@@ -538,12 +634,7 @@ static struct ast_channel *chan_pjsip_new(struct ast_sip_session *session, int s
 	ast_channel_stage_snapshot_done(chan);
 	ast_channel_unlock(chan);
 
-	/* If res_pjsip_session is ever updated to create/destroy ast_sip_session_media
-	 * during a call such as if multiple same-type stream support is introduced,
-	 * these will need to be recaptured as well */
-	pvt->media[SIP_MEDIA_AUDIO] = ao2_find(session->media, "audio", OBJ_KEY);
-	pvt->media[SIP_MEDIA_VIDEO] = ao2_find(session->media, "video", OBJ_KEY);
-	set_channel_on_rtp_instance(pvt, ast_channel_uniqueid(chan));
+	set_channel_on_rtp_instance(session, ast_channel_uniqueid(chan));
 
 	return chan;
 }
@@ -682,49 +773,32 @@ static struct ast_frame *chan_pjsip_cng_tone_detected(struct ast_sip_session *se
  *
  * \note The channel is already locked.
  */
-static struct ast_frame *chan_pjsip_read(struct ast_channel *ast)
+static struct ast_frame *chan_pjsip_read_stream(struct ast_channel *ast)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(ast);
-	struct ast_sip_session *session;
-	struct chan_pjsip_pvt *pvt = channel->pvt;
+	struct ast_sip_session *session = channel->session;
+	struct ast_sip_session_media_read_callback_state *callback_state;
 	struct ast_frame *f;
-	struct ast_sip_session_media *media = NULL;
-	int rtcp = 0;
-	int fdno = ast_channel_fdno(ast);
+	int fdno = ast_channel_fdno(ast) - AST_EXTENDED_FDS;
 
-	switch (fdno) {
-	case 0:
-		media = pvt->media[SIP_MEDIA_AUDIO];
-		break;
-	case 1:
-		media = pvt->media[SIP_MEDIA_AUDIO];
-		rtcp = 1;
-		break;
-	case 2:
-		media = pvt->media[SIP_MEDIA_VIDEO];
-		break;
-	case 3:
-		media = pvt->media[SIP_MEDIA_VIDEO];
-		rtcp = 1;
-		break;
-	}
-
-	if (!media || !media->rtp) {
+	if (fdno >= AST_VECTOR_SIZE(&session->active_media_state->read_callbacks)) {
 		return &ast_null_frame;
 	}
 
-	if (!(f = ast_rtp_instance_read(media->rtp, rtcp))) {
+	callback_state = AST_VECTOR_GET_ADDR(&session->active_media_state->read_callbacks, fdno);
+	f = callback_state->read_callback(session, callback_state->session);
+
+	if (!f) {
 		return f;
 	}
 
-	ast_rtp_instance_set_last_rx(media->rtp, time(NULL));
+	f->stream_num = callback_state->session->stream_num;
 
-	if (f->frametype != AST_FRAME_VOICE) {
+	if (f->frametype != AST_FRAME_VOICE ||
+		callback_state->session != session->active_media_state->default_session[callback_state->session->type]) {
 		return f;
 	}
 
-	session = channel->session;
-
 	if (ast_format_cap_iscompatible_format(ast_channel_nativeformats(ast), f->subclass.format) == AST_FORMAT_CMP_NOT_EQUAL) {
 		ast_debug(1, "Oooh, got a frame with format of %s on channel '%s' when it has not been negotiated\n",
 			ast_format_get_name(f->subclass.format), ast_channel_name(ast));
@@ -794,22 +868,31 @@ static struct ast_frame *chan_pjsip_read(struct ast_channel *ast)
 	return f;
 }
 
-/*! \brief Function called by core to write frames */
-static int chan_pjsip_write(struct ast_channel *ast, struct ast_frame *frame)
+static int chan_pjsip_write_stream(struct ast_channel *ast, int stream_num, struct ast_frame *frame)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(ast);
-	struct chan_pjsip_pvt *pvt = channel->pvt;
-	struct ast_sip_session_media *media;
+	struct ast_sip_session *session = channel->session;
+	struct ast_sip_session_media *media = NULL;
 	int res = 0;
 
+	/* The core provides a guarantee that the stream will exist when we are called if stream_num is provided */
+	if (stream_num >= 0) {
+		/* What is not guaranteed is that a media session will exist */
+		if (stream_num < AST_VECTOR_SIZE(&channel->session->active_media_state->sessions)) {
+			media = AST_VECTOR_GET(&channel->session->active_media_state->sessions, stream_num);
+		}
+	}
+
 	switch (frame->frametype) {
 	case AST_FRAME_VOICE:
-		media = pvt->media[SIP_MEDIA_AUDIO];
-
 		if (!media) {
 			return 0;
-		}
-		if (ast_format_cap_iscompatible_format(ast_channel_nativeformats(ast), frame->subclass.format) == AST_FORMAT_CMP_NOT_EQUAL) {
+		} else if (media->type != AST_MEDIA_TYPE_AUDIO) {
+			ast_debug(3, "Channel %s stream %d is of type '%s', not audio!\n",
+				ast_channel_name(ast), stream_num, ast_codec_media_type2str(media->type));
+			return 0;
+		} else if (media == channel->session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO] &&
+			ast_format_cap_iscompatible_format(ast_channel_nativeformats(ast), frame->subclass.format) == AST_FORMAT_CMP_NOT_EQUAL) {
 			struct ast_str *cap_buf = ast_str_alloca(AST_FORMAT_CAP_NAMES_LEN);
 			struct ast_str *write_transpath = ast_str_alloca(256);
 			struct ast_str *read_transpath = ast_str_alloca(256);
@@ -826,17 +909,32 @@ static int chan_pjsip_write(struct ast_channel *ast, struct ast_frame *frame)
 				ast_format_get_name(ast_channel_rawwriteformat(ast)),
 				ast_translate_path_to_str(ast_channel_writetrans(ast), &write_transpath));
 			return 0;
-		}
-		if (media->rtp) {
-			res = ast_rtp_instance_write(media->rtp, frame);
+		} else if (media->write_callback) {
+			res = media->write_callback(session, media, frame);
+
 		}
 		break;
 	case AST_FRAME_VIDEO:
-		if ((media = pvt->media[SIP_MEDIA_VIDEO]) && media->rtp) {
-			res = ast_rtp_instance_write(media->rtp, frame);
+		if (!media) {
+			return 0;
+		} else if (media->type != AST_MEDIA_TYPE_VIDEO) {
+			ast_debug(3, "Channel %s stream %d is of type '%s', not video!\n",
+				ast_channel_name(ast), stream_num, ast_codec_media_type2str(media->type));
+			return 0;
+		} else if (media->write_callback) {
+			res = media->write_callback(session, media, frame);
 		}
 		break;
 	case AST_FRAME_MODEM:
+		if (!media) {
+			return 0;
+		} else if (media->type != AST_MEDIA_TYPE_IMAGE) {
+			ast_debug(3, "Channel %s stream %d is of type '%s', not image!\n",
+				ast_channel_name(ast), stream_num, ast_codec_media_type2str(media->type));
+			return 0;
+		} else if (media->write_callback) {
+			res = media->write_callback(session, media, frame);
+		}
 		break;
 	default:
 		ast_log(LOG_WARNING, "Can't send %u type frames with PJSIP\n", frame->frametype);
@@ -846,11 +944,15 @@ static int chan_pjsip_write(struct ast_channel *ast, struct ast_frame *frame)
 	return res;
 }
 
+static int chan_pjsip_write(struct ast_channel *ast, struct ast_frame *frame)
+{
+	return chan_pjsip_write_stream(ast, -1, frame);
+}
+
 /*! \brief Function called by core to change the underlying owner channel */
 static int chan_pjsip_fixup(struct ast_channel *oldchan, struct ast_channel *newchan)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(newchan);
-	struct chan_pjsip_pvt *pvt = channel->pvt;
 
 	if (channel->session->channel != oldchan) {
 		return -1;
@@ -863,7 +965,7 @@ static int chan_pjsip_fixup(struct ast_channel *oldchan, struct ast_channel *new
 	 */
 	channel->session->channel = newchan;
 
-	set_channel_on_rtp_instance(pvt, ast_channel_uniqueid(newchan));
+	set_channel_on_rtp_instance(channel->session, ast_channel_uniqueid(newchan));
 
 	return 0;
 }
@@ -1278,7 +1380,7 @@ static int update_connected_line_information(void *data)
 			/* Only the INVITE method actually needs SDP, UPDATE can do without */
 			generate_new_sdp = (method == AST_SIP_SESSION_REFRESH_METHOD_INVITE);
 
-			ast_sip_session_refresh(session, NULL, NULL, NULL, method, generate_new_sdp);
+			ast_sip_session_refresh(session, NULL, NULL, NULL, method, generate_new_sdp, NULL);
 		}
 	} else if (session->endpoint->id.rpid_immediate
 		&& session->inv_session->state != PJSIP_INV_STATE_DISCONNECTED
@@ -1309,21 +1411,18 @@ static int update_connected_line_information(void *data)
 }
 
 /*! \brief Callback which changes the value of locally held on the media stream */
-static int local_hold_set_state(void *obj, void *arg, int flags)
+static void local_hold_set_state(struct ast_sip_session_media *session_media, unsigned int held)
 {
-	struct ast_sip_session_media *session_media = obj;
-	unsigned int *held = arg;
-
-	session_media->locally_held = *held;
-
-	return 0;
+	if (session_media) {
+		session_media->locally_held = held;
+	}
 }
 
 /*! \brief Update local hold state and send a re-INVITE with the new SDP */
 static int remote_send_hold_refresh(struct ast_sip_session *session, unsigned int held)
 {
-	ao2_callback(session->media, OBJ_NODATA, local_hold_set_state, &held);
-	ast_sip_session_refresh(session, NULL, NULL, NULL, AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1);
+	AST_VECTOR_CALLBACK_VOID(&session->active_media_state->sessions, local_hold_set_state, held);
+	ast_sip_session_refresh(session, NULL, NULL, NULL, AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1, NULL);
 	ao2_ref(session, -1);
 
 	return 0;
@@ -1341,16 +1440,103 @@ static int remote_send_unhold(void *data)
 	return remote_send_hold_refresh(data, 0);
 }
 
+struct topology_change_refresh_data {
+	struct ast_sip_session *session;
+	struct ast_sip_session_media_state *media_state;
+};
+
+static void topology_change_refresh_data_free(struct topology_change_refresh_data *refresh_data)
+{
+	ao2_cleanup(refresh_data->session);
+
+	ast_sip_session_media_state_free(refresh_data->media_state);
+	ast_free(refresh_data);
+}
+
+static struct topology_change_refresh_data *topology_change_refresh_data_alloc(
+	struct ast_sip_session *session, const struct ast_stream_topology *topology)
+{
+	struct topology_change_refresh_data *refresh_data;
+
+	refresh_data = ast_calloc(1, sizeof(*refresh_data));
+	if (!refresh_data) {
+		return NULL;
+	}
+
+	refresh_data->session = ao2_bump(session);
+	refresh_data->media_state = ast_sip_session_media_state_alloc();
+	if (!refresh_data->media_state) {
+		topology_change_refresh_data_free(refresh_data);
+		return NULL;
+	}
+	refresh_data->media_state->topology = ast_stream_topology_clone(topology);
+	if (!refresh_data->media_state->topology) {
+		topology_change_refresh_data_free(refresh_data);
+		return NULL;
+	}
+
+	return refresh_data;
+}
+
+static int on_topology_change_response(struct ast_sip_session *session, pjsip_rx_data *rdata)
+{
+	if (rdata->msg_info.msg->line.status.code == 200) {
+		/* The topology was changed to something new so give notice to what requested
+		 * it so it queries the channel and updates accordingly.
+		 */
+		if (session->channel) {
+			ast_queue_control(session->channel, AST_CONTROL_STREAM_TOPOLOGY_CHANGED);
+		}
+	} else if (rdata->msg_info.msg->line.status.code != 100) {
+		/* The topology change failed, so drop the current pending media state */
+		ast_sip_session_media_state_reset(session->pending_media_state);
+	}
+
+	return 0;
+}
+
+static int send_topology_change_refresh(void *data)
+{
+	struct topology_change_refresh_data *refresh_data = data;
+	int ret;
+
+	ret = ast_sip_session_refresh(refresh_data->session, NULL, NULL, on_topology_change_response,
+		AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1, refresh_data->media_state);
+	refresh_data->media_state = NULL;
+	topology_change_refresh_data_free(refresh_data);
+
+	return ret;
+}
+
+static int handle_topology_request_change(struct ast_sip_session *session,
+	const struct ast_stream_topology *proposed)
+{
+	struct topology_change_refresh_data *refresh_data;
+	int res;
+
+	refresh_data = topology_change_refresh_data_alloc(session, proposed);
+	if (!refresh_data) {
+		return -1;
+	}
+
+	res = ast_sip_push_task(session->serializer, send_topology_change_refresh, refresh_data);
+	if (res) {
+		topology_change_refresh_data_free(refresh_data);
+	}
+	return res;
+}
+
 /*! \brief Function called by core to ask the channel to indicate some sort of condition */
 static int chan_pjsip_indicate(struct ast_channel *ast, int condition, const void *data, size_t datalen)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(ast);
-	struct chan_pjsip_pvt *pvt = channel->pvt;
 	struct ast_sip_session_media *media;
 	int response_code = 0;
 	int res = 0;
 	char *device_buf;
 	size_t device_buf_size;
+	int i;
+	const struct ast_stream_topology *topology;
 
 	switch (condition) {
 	case AST_CONTROL_RINGING:
@@ -1403,39 +1589,47 @@ static int chan_pjsip_indicate(struct ast_channel *ast, int condition, const voi
 		ast_devstate_changed(AST_DEVICE_UNKNOWN, AST_DEVSTATE_CACHABLE, "PJSIP/%s", ast_sorcery_object_get_id(channel->session->endpoint));
 		break;
 	case AST_CONTROL_VIDUPDATE:
-		media = pvt->media[SIP_MEDIA_VIDEO];
-		if (media && media->rtp) {
-			/* FIXME: Only use this for VP8. Additional work would have to be done to
-			 * fully support other video codecs */
-
-			if (ast_format_cap_iscompatible_format(ast_channel_nativeformats(ast), ast_format_vp8) != AST_FORMAT_CMP_NOT_EQUAL) {
-				/* FIXME Fake RTP write, this will be sent as an RTCP packet. Ideally the
-				 * RTP engine would provide a way to externally write/schedule RTCP
-				 * packets */
-				struct ast_frame fr;
-				fr.frametype = AST_FRAME_CONTROL;
-				fr.subclass.integer = AST_CONTROL_VIDUPDATE;
-				res = ast_rtp_instance_write(media->rtp, &fr);
-			} else {
-				ao2_ref(channel->session, +1);
-#ifdef HAVE_PJSIP_INV_SESSION_REF
-				if (pjsip_inv_add_ref(channel->session->inv_session) != PJ_SUCCESS) {
-					ast_log(LOG_ERROR, "Can't increase the session reference counter\n");
-					ao2_cleanup(channel->session);
+		for (i = 0; i < AST_VECTOR_SIZE(&channel->session->active_media_state->sessions); ++i) {
+			media = AST_VECTOR_GET(&channel->session->active_media_state->sessions, i);
+			if (!media || media->type != AST_MEDIA_TYPE_VIDEO) {
+				continue;
+			}
+			if (media->rtp) {
+				/* FIXME: Only use this for VP8. Additional work would have to be done to
+				 * fully support other video codecs */
+
+				if (ast_format_cap_iscompatible_format(ast_channel_nativeformats(ast), ast_format_vp8) != AST_FORMAT_CMP_NOT_EQUAL) {
+					/* FIXME Fake RTP write, this will be sent as an RTCP packet. Ideally the
+					 * RTP engine would provide a way to externally write/schedule RTCP
+					 * packets */
+					struct ast_frame fr;
+					fr.frametype = AST_FRAME_CONTROL;
+					fr.subclass.integer = AST_CONTROL_VIDUPDATE;
+					res = ast_rtp_instance_write(media->rtp, &fr);
 				} else {
-#endif
-					if (ast_sip_push_task(channel->session->serializer, transmit_info_with_vidupdate, channel->session)) {
+					ao2_ref(channel->session, +1);
+#ifdef HAVE_PJSIP_INV_SESSION_REF
+					if (pjsip_inv_add_ref(channel->session->inv_session) != PJ_SUCCESS) {
+						ast_log(LOG_ERROR, "Can't increase the session reference counter\n");
 						ao2_cleanup(channel->session);
-					}
+					} else {
+#endif
+						if (ast_sip_push_task(channel->session->serializer, transmit_info_with_vidupdate, channel->session)) {
+							ao2_cleanup(channel->session);
+						}
 #ifdef HAVE_PJSIP_INV_SESSION_REF
-				}
+					}
 #endif
+				}
+				ast_test_suite_event_notify("AST_CONTROL_VIDUPDATE", "Result: Success");
+			} else {
+				ast_test_suite_event_notify("AST_CONTROL_VIDUPDATE", "Result: Failure");
+				res = -1;
 			}
-			ast_test_suite_event_notify("AST_CONTROL_VIDUPDATE", "Result: Success");
-		} else {
-			ast_test_suite_event_notify("AST_CONTROL_VIDUPDATE", "Result: Failure");
-			res = -1;
 		}
+		/* XXX If there were no video streams, then this should set
+		 * res to -1
+		 */
 		break;
 	case AST_CONTROL_CONNECTED_LINE:
 		ao2_ref(channel->session, +1);
@@ -1530,6 +1724,10 @@ static int chan_pjsip_indicate(struct ast_channel *ast, int condition, const voi
 			}
 		}
 
+		break;
+	case AST_CONTROL_STREAM_TOPOLOGY_REQUEST_CHANGE:
+		topology = data;
+		res = handle_topology_request_change(channel->session, topology);
 		break;
 	case -1:
 		res = -1;
@@ -1744,10 +1942,11 @@ static int chan_pjsip_transfer(struct ast_channel *chan, const char *target)
 static int chan_pjsip_digit_begin(struct ast_channel *chan, char digit)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(chan);
-	struct chan_pjsip_pvt *pvt = channel->pvt;
-	struct ast_sip_session_media *media = pvt->media[SIP_MEDIA_AUDIO];
+	struct ast_sip_session_media *media;
 	int res = 0;
 
+	media = channel->session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO];
+
 	switch (channel->session->endpoint->dtmf) {
 	case AST_SIP_DTMF_RFC_4733:
 		if (!media || !media->rtp) {
@@ -1755,14 +1954,14 @@ static int chan_pjsip_digit_begin(struct ast_channel *chan, char digit)
 		}
 
 		ast_rtp_instance_dtmf_begin(media->rtp, digit);
-                break;
+		break;
 	case AST_SIP_DTMF_AUTO:
-                       if (!media || !media->rtp || (ast_rtp_instance_dtmf_mode_get(media->rtp) == AST_RTP_DTMF_MODE_INBAND)) {
-                        return -1;
-                }
+		if (!media || !media->rtp || (ast_rtp_instance_dtmf_mode_get(media->rtp) == AST_RTP_DTMF_MODE_INBAND)) {
+			return -1;
+		}
 
-                ast_rtp_instance_dtmf_begin(media->rtp, digit);
-                break;
+		ast_rtp_instance_dtmf_begin(media->rtp, digit);
+		break;
 	case AST_SIP_DTMF_NONE:
 		break;
 	case AST_SIP_DTMF_INBAND:
@@ -1858,10 +2057,11 @@ failure:
 static int chan_pjsip_digit_end(struct ast_channel *ast, char digit, unsigned int duration)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(ast);
-	struct chan_pjsip_pvt *pvt = channel->pvt;
-	struct ast_sip_session_media *media = pvt->media[SIP_MEDIA_AUDIO];
+	struct ast_sip_session_media *media;
 	int res = 0;
 
+	media = channel->session->active_media_state->default_session[AST_MEDIA_TYPE_AUDIO];
+
 	switch (channel->session->endpoint->dtmf) {
 	case AST_SIP_DTMF_INFO:
 	{
@@ -1943,7 +2143,6 @@ static int call(void *data)
 {
 	struct ast_sip_channel_pvt *channel = data;
 	struct ast_sip_session *session = channel->session;
-	struct chan_pjsip_pvt *pvt = channel->pvt;
 	pjsip_tx_data *tdata;
 
 	int res = ast_sip_session_create_invite(session, &tdata);
@@ -1952,7 +2151,7 @@ static int call(void *data)
 		ast_set_hangupsource(session->channel, ast_channel_name(session->channel), 0);
 		ast_queue_hangup(session->channel);
 	} else {
-		set_channel_on_rtp_instance(pvt, ast_channel_uniqueid(session->channel));
+		set_channel_on_rtp_instance(session, ast_channel_uniqueid(session->channel));
 		update_initial_connected_line(session);
 		ast_sip_session_send_request(session, tdata);
 	}
@@ -2050,10 +2249,10 @@ static struct hangup_data *hangup_data_alloc(int cause, struct ast_channel *chan
 }
 
 /*! \brief Clear a channel from a session along with its PVT */
-static void clear_session_and_channel(struct ast_sip_session *session, struct ast_channel *ast, struct chan_pjsip_pvt *pvt)
+static void clear_session_and_channel(struct ast_sip_session *session, struct ast_channel *ast)
 {
 	session->channel = NULL;
-	set_channel_on_rtp_instance(pvt, "");
+	set_channel_on_rtp_instance(session, "");
 	ast_channel_tech_pvt_set(ast, NULL);
 }
 
@@ -2062,7 +2261,6 @@ static int hangup(void *data)
 	struct hangup_data *h_data = data;
 	struct ast_channel *ast = h_data->chan;
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(ast);
-	struct chan_pjsip_pvt *pvt = channel->pvt;
 	struct ast_sip_session *session = channel->session;
 	int cause = h_data->cause;
 
@@ -2072,7 +2270,7 @@ static int hangup(void *data)
 	 * afterwards.
 	 */
 	ast_sip_session_terminate(ao2_bump(session), cause);
-	clear_session_and_channel(session, ast, pvt);
+	clear_session_and_channel(session, ast);
 	ao2_cleanup(session);
 	ao2_cleanup(channel);
 	ao2_cleanup(h_data);
@@ -2083,7 +2281,6 @@ static int hangup(void *data)
 static int chan_pjsip_hangup(struct ast_channel *ast)
 {
 	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(ast);
-	struct chan_pjsip_pvt *pvt;
 	int cause;
 	struct hangup_data *h_data;
 
@@ -2091,7 +2288,6 @@ static int chan_pjsip_hangup(struct ast_channel *ast)
 		return -1;
 	}
 
-	pvt = channel->pvt;
 	cause = hangup_cause2sip(ast_channel_hangupcause(channel->session->channel));
 	h_data = hangup_data_alloc(cause, ast);
 
@@ -2110,7 +2306,7 @@ failure:
 	/* Go ahead and do our cleanup of the session and channel even if we're not going
 	 * to be able to send our SIP request/response
 	 */
-	clear_session_and_channel(channel->session, ast, pvt);
+	clear_session_and_channel(channel->session, ast);
 	ao2_cleanup(channel);
 	ao2_cleanup(h_data);
 
@@ -2119,7 +2315,7 @@ failure:
 
 struct request_data {
 	struct ast_sip_session *session;
-	struct ast_format_cap *caps;
+	struct ast_stream_topology *topology;
 	const char *dest;
 	int cause;
 };
@@ -2193,7 +2389,7 @@ static int request(void *obj)
 		}
 	}
 
-	if (!(session = ast_sip_session_create_outgoing(endpoint, NULL, args.aor, request_user, req_data->caps))) {
+	if (!(session = ast_sip_session_create_outgoing(endpoint, NULL, args.aor, request_user, req_data->topology))) {
 		ast_log(LOG_ERROR, "Failed to create outgoing session to endpoint '%s'\n", endpoint_name);
 		req_data->cause = AST_CAUSE_NO_ROUTE_DESTINATION;
 		return -1;
@@ -2205,12 +2401,12 @@ static int request(void *obj)
 }
 
 /*! \brief Function called by core to create a new outgoing PJSIP session */
-static struct ast_channel *chan_pjsip_request(const char *type, struct ast_format_cap *cap, const struct ast_assigned_ids *assignedids, const struct ast_channel *requestor, const char *data, int *cause)
+static struct ast_channel *chan_pjsip_request_with_stream_topology(const char *type, struct ast_stream_topology *topology, const struct ast_assigned_ids *assignedids, const struct ast_channel *requestor, const char *data, int *cause)
 {
 	struct request_data req_data;
 	RAII_VAR(struct ast_sip_session *, session, NULL, ao2_cleanup);
 
-	req_data.caps = cap;
+	req_data.topology = topology;
 	req_data.dest = data;
 
 	if (ast_sip_push_task_synchronous(NULL, request, &req_data)) {
@@ -2228,6 +2424,23 @@ static struct ast_channel *chan_pjsip_request(const char *type, struct ast_forma
 	return session->channel;
 }
 
+static struct ast_channel *chan_pjsip_request(const char *type, struct ast_format_cap *cap, const struct ast_assigned_ids *assignedids, const struct ast_channel *requestor, const char *data, int *cause)
+{
+	struct ast_stream_topology *topology;
+	struct ast_channel *chan;
+
+	topology = ast_stream_topology_create_from_format_cap(cap);
+	if (!topology) {
+		return NULL;
+	}
+
+	chan = chan_pjsip_request_with_stream_topology(type, topology, assignedids, requestor, data, cause);
+
+	ast_stream_topology_free(topology);
+
+	return chan;
+}
+
 struct sendtext_data {
 	struct ast_sip_session *session;
 	char text[0];

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;

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;
 }

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 */

configs/samples/pjsip.conf.sample

@@ -779,6 +779,10 @@
                               ; The value "yes" is useful for some SIP phones
                               ; (Cisco SPA) to be able to indicate and pick up
                               ; ringing devices.
+;max_audio_streams= ; The maximum number of allowed negotiated audio streams
+                    ; (default: 1)
+;max_video_streams= ; The maximum number of allowed negotiated video streams
+                    ; (default: 1)
 
 ;==========================AUTH SECTION OPTIONS=========================
 ;[auth]

contrib/ast-db-manage/config/versions/39959b9c2566_pjsip_stream_maximum.py

@@ -0,0 +1,24 @@
+"""pjsip_stream_maximum
+
+Revision ID: 39959b9c2566
+Revises: d7983954dd96
+Create Date: 2017-06-15 13:18:12.372333
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = '39959b9c2566'
+down_revision = 'd7983954dd96'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+    op.add_column('ps_endpoints', sa.Column('max_audio_streams', sa.Integer))
+    op.add_column('ps_endpoints', sa.Column('max_video_streams', sa.Integer))
+
+
+def downgrade():
+    op.drop_column('ps_endpoints', 'max_audio_streams')
+    op.drop_column('ps_endpoints', 'max_video_streams')

include/asterisk/res_pjsip.h

@@ -666,6 +666,8 @@ struct ast_sip_endpoint_media_configuration {
 	struct ast_sip_t38_configuration t38;
 	/*! Configured codecs */
 	struct ast_format_cap *codecs;
+	/*! Capabilities in topology form */
+	struct ast_stream_topology *topology;
 	/*! DSCP TOS bits for audio streams */
 	unsigned int tos_audio;
 	/*! Priority for audio streams */
@@ -680,6 +682,10 @@ struct ast_sip_endpoint_media_configuration {
 	unsigned int bind_rtp_to_media_address;
 	/*! Use RTCP-MUX */
 	unsigned int rtcp_mux;
+	/*! Maximum number of audio streams to offer/accept */
+	unsigned int max_audio_streams;
+	/*! Maximum number of video streams to offer/accept */
+	unsigned int max_video_streams;
 };
 
 /*!

include/asterisk/res_pjsip_session.h

@@ -28,6 +28,8 @@
 #include "asterisk/netsock2.h"
 /* Needed for ast_sdp_srtp struct */
 #include "asterisk/sdp_srtp.h"
+/* Needed for ast_media_type */
+#include "asterisk/codec.h"
 
 /* Forward declarations */
 struct ast_sip_endpoint;
@@ -56,17 +58,21 @@ enum ast_sip_session_t38state {
 };
 
 struct ast_sip_session_sdp_handler;
+struct ast_sip_session;
+struct ast_sip_session_media;
+
+typedef struct ast_frame *(*ast_sip_session_media_read_cb)(struct ast_sip_session *session, struct ast_sip_session_media *session_media);
+typedef int (*ast_sip_session_media_write_cb)(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
+	struct ast_frame *frame);
 
 /*!
  * \brief A structure containing SIP session media information
  */
 struct ast_sip_session_media {
-	union {
-		/*! \brief RTP instance itself */
-		struct ast_rtp_instance *rtp;
-		/*! \brief UDPTL instance itself */
-		struct ast_udptl *udptl;
-	};
+	/*! \brief RTP instance itself */
+	struct ast_rtp_instance *rtp;
+	/*! \brief UDPTL instance itself */
+	struct ast_udptl *udptl;
 	/*! \brief Direct media address */
 	struct ast_sockaddr direct_media_addr;
 	/*! \brief SDP handler that setup the RTP */
@@ -87,8 +93,38 @@ struct ast_sip_session_media {
 	unsigned int locally_held:1;
 	/*! \brief Does remote support rtcp_mux */
 	unsigned int remote_rtcp_mux:1;
-	/*! \brief Stream type this session media handles */
-	char stream_type[1];
+	/*! \brief Media type of this session media */
+	enum ast_media_type type;
+	/*! \brief The write callback when writing frames */
+	ast_sip_session_media_write_cb write_callback;
+	/*! \brief The stream number to place into any resulting frames */
+	int stream_num;
+};
+
+/*!
+ * \brief Structure which contains read callback information
+ */
+struct ast_sip_session_media_read_callback_state {
+	/*! \brief The file descriptor itself */
+	int fd;
+	/*! \brief The callback to invoke */
+	ast_sip_session_media_read_cb read_callback;
+	/*! \brief The media session */
+	struct ast_sip_session_media *session;
+};
+
+/*!
+ * \brief Structure which contains media state information (streams, sessions)
+ */
+struct ast_sip_session_media_state {
+	/*! \brief Mapping of stream to media sessions */
+	AST_VECTOR(, struct ast_sip_session_media *) sessions;
+	/*! \brief Added read callbacks - these are whole structs and not pointers */
+	AST_VECTOR(, struct ast_sip_session_media_read_callback_state) read_callbacks;
+	/*! \brief Default media sessions for each type */
+	struct ast_sip_session_media *default_session[AST_MEDIA_TYPE_END];
+	/*! \brief The media stream topology */
+	struct ast_stream_topology *topology;
 };
 
 /*!
@@ -123,8 +159,6 @@ struct ast_sip_session {
 	AST_LIST_HEAD(, ast_sip_session_supplement) supplements;
 	/*! Datastores added to the session by supplements to the session */
 	struct ao2_container *datastores;
-	/*! Media streams */
-	struct ao2_container *media;
 	/*! Serializer for tasks relating to this SIP session */
 	struct ast_taskprocessor *serializer;
 	/*! Non-null if the session serializer is suspended or being suspended. */
@@ -139,8 +173,10 @@ struct ast_sip_session {
 	pj_timer_entry scheduled_termination;
 	/*! Identity of endpoint this session deals with */
 	struct ast_party_id id;
-	/*! Requested capabilities */
-	struct ast_format_cap *req_caps;
+	/*! Active media state (sessions + streams) - contents are guaranteed not to change */
+	struct ast_sip_session_media_state *active_media_state;
+	/*! Pending media state (sessions + streams) */
+	struct ast_sip_session_media_state *pending_media_state;
 	/*! Optional DSP, used only for inband DTMF/Fax-CNG detection if configured */
 	struct ast_dsp *dsp;
 	/*! Whether the termination of the session should be deferred */
@@ -315,34 +351,29 @@ struct ast_sip_session_sdp_handler {
 	/*!
 	 * \brief Set session details based on a stream in an incoming SDP offer or answer
 	 * \param session The session for which the media is being negotiated
-	 * \param session_media The media to be setup for this session
+	 * \param session_media The media session
 	 * \param sdp The entire SDP. Useful for getting "global" information, such as connections or attributes
-	 * \param stream The stream on which to operate
-	 * \retval 0 The stream was not handled by this handler. If there are other registered handlers for this stream type, they will be called.
-	 * \retval <0 There was an error encountered. No further operation will take place and the current negotiation will be abandoned.
-	 * \retval >0 The stream was handled by this handler. No further handler of this stream type will be called.
-	 */
-	int (*negotiate_incoming_sdp_stream)(struct ast_sip_session *session, struct ast_sip_session_media *session_media, const struct pjmedia_sdp_session *sdp, const struct pjmedia_sdp_media *stream);
-	/*!
-	 * \brief Create an SDP media stream and add it to the outgoing SDP offer or answer
-	 * \param session The session for which media is being added
-	 * \param session_media The media to be setup for this session
-	 * \param stream The stream on which to operate
+	 * \param index The index for the session media, Asterisk stream, and PJMEDIA stream being negotiated
+	 * \param asterisk_stream The Asterisk stream representation
 	 * \retval 0 The stream was not handled by this handler. If there are other registered handlers for this stream type, they will be called.
 	 * \retval <0 There was an error encountered. No further operation will take place and the current negotiation will be abandoned.
 	 * \retval >0 The stream was handled by this handler. No further handler of this stream type will be called.
 	 */
-	int (*handle_incoming_sdp_stream)(struct ast_sip_session *session, struct ast_sip_session_media *session_media, const struct pjmedia_sdp_session *sdp, struct pjmedia_sdp_media *stream);
+	int (*negotiate_incoming_sdp_stream)(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
+		const struct pjmedia_sdp_session *sdp, int index, struct ast_stream *asterisk_stream);
 	/*!
 	 * \brief Create an SDP media stream and add it to the outgoing SDP offer or answer
 	 * \param session The session for which media is being added
 	 * \param session_media The media to be setup for this session
 	 * \param sdp The entire SDP as currently built
+	 * \param remote Optional remote SDP if this is an answer
+	 * \param stream The stream that is to be added to the outgoing SDP
 	 * \retval 0 This handler has no stream to add. If there are other registered handlers for this stream type, they will be called.
 	 * \retval <0 There was an error encountered. No further operation will take place and the current SDP negotiation will be abandoned.
 	 * \retval >0 The handler has a stream to be added to the SDP. No further handler of this stream type will be called.
 	 */
-	int (*create_outgoing_sdp_stream)(struct ast_sip_session *session, struct ast_sip_session_media *session_media, struct pjmedia_sdp_session *sdp);
+	int (*create_outgoing_sdp_stream)(struct ast_sip_session *session, struct ast_sip_session_media *session_media, struct pjmedia_sdp_session *sdp,
+		const struct pjmedia_sdp_session *remote, struct ast_stream *stream);
 	/*!
 	 * \brief Update media stream with external address if applicable
 	 * \param tdata The outgoing message itself
@@ -353,17 +384,18 @@ struct ast_sip_session_sdp_handler {
 	/*!
 	 * \brief Apply a negotiated SDP media stream
 	 * \param session The session for which media is being applied
-	 * \param session_media The media to be setup for this session
+	 * \param session_media The media session
 	 * \param local The entire local negotiated SDP
-	 * \param local_stream The local stream which to apply
 	 * \param remote The entire remote negotiated SDP
-	 * \param remote_stream The remote stream which to apply
+	 * \param index The index of the session media, SDP streams, and Asterisk streams
+	 * \param asterisk_stream The Asterisk stream representation
 	 * \retval 0 The stream was not applied by this handler. If there are other registered handlers for this stream type, they will be called.
 	 * \retval <0 There was an error encountered. No further operation will take place and the current application will be abandoned.
 	 * \retval >0 The stream was handled by this handler. No further handler of this stream type will be called.
 	 */
-	int (*apply_negotiated_sdp_stream)(struct ast_sip_session *session, struct ast_sip_session_media *session_media, const struct pjmedia_sdp_session *local, const struct pjmedia_sdp_media *local_stream,
-		const struct pjmedia_sdp_session *remote, const struct pjmedia_sdp_media *remote_stream);
+	int (*apply_negotiated_sdp_stream)(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
+		const struct pjmedia_sdp_session *local, const struct pjmedia_sdp_session *remote, int index,
+		struct ast_stream *asterisk_stream);
 	/*!
 	 * \brief Stop a session_media created by this handler but do not destroy resources
 	 * \param session The session for which media is being stopped
@@ -393,7 +425,7 @@ struct ast_sip_channel_pvt {
 /*!
  * \brief Allocate a new SIP channel pvt structure
  *
- * \param pvt Pointer to channel specific implementation
+ * \param pvt Pointer to channel specific information
  * \param session Pointer to SIP session
  *
  * \retval non-NULL success
@@ -452,11 +484,11 @@ void ast_sip_session_unsuspend(struct ast_sip_session *session);
  * \param contact The contact that this session will communicate with
  * \param location Name of the location to call, be it named location or explicit URI. Overrides contact if present.
  * \param request_user Optional request user to place in the request URI if permitted
- * \param req_caps The requested capabilities
+ * \param req_topology The requested capabilities
  */
 struct ast_sip_session *ast_sip_session_create_outgoing(struct ast_sip_endpoint *endpoint,
 	struct ast_sip_contact *contact, const char *location, const char *request_user,
-	struct ast_format_cap *req_caps);
+	struct ast_stream_topology *req_topology);
 
 /*!
  * \brief Terminate a session and, if possible, send the provided response code
@@ -613,15 +645,20 @@ void ast_sip_session_remove_datastore(struct ast_sip_session *session, const cha
  * \param on_response Callback called when response for request is received
  * \param method The method that should be used when constructing the session refresh
  * \param generate_new_sdp Boolean to indicate if a new SDP should be created
+ * \param media_state Optional requested media state for the SDP
+ *
  * \retval 0 Successfully sent refresh
  * \retval -1 Failure to send refresh
+ *
+ * \note If a media_state is passed in ownership will be taken in all cases
  */
 int ast_sip_session_refresh(struct ast_sip_session *session,
 		ast_sip_session_request_creation_cb on_request_creation,
 		ast_sip_session_sdp_creation_cb on_sdp_creation,
 		ast_sip_session_response_cb on_response,
 		enum ast_sip_session_refresh_method method,
-		int generate_new_sdp);
+		int generate_new_sdp,
+		struct ast_sip_session_media_state *media_state);
 
 /*!
  * \brief Send a SIP response
@@ -692,6 +729,110 @@ struct ast_sip_session *ast_sip_dialog_get_session(pjsip_dialog *dlg);
  */
 void ast_sip_session_resume_reinvite(struct ast_sip_session *session);
 
+/*!
+ * \brief Determines if a provided pending stream will be the default stream or not
+ * \since 15.0.0
+ *
+ * \param session The session to check against
+ * \param stream The pending stream
+ *
+ * \retval 1 if stream will be default
+ * \retval 0 if stream will NOT be the default
+ */
+int ast_sip_session_is_pending_stream_default(const struct ast_sip_session *session, const struct ast_stream *stream);
+
+/*!
+ * \brief Allocate a session media state structure
+ * \since 15.0.0
+ *
+ * \retval non-NULL success
+ * \retval NULL failure
+ */
+struct ast_sip_session_media_state *ast_sip_session_media_state_alloc(void);
+
+/*!
+ * \brief Allocate an ast_session_media and add it to the media state's vector.
+ * \since 15.0.0
+ *
+ * This allocates a session media of the specified type. The position argument
+ * determines where in the vector that the new session media will be inserted.
+ *
+ * \note The returned ast_session_media is the reference held by the vector. Callers
+ * of this function must NOT decrement the refcount of the session media.
+ *
+ * \param session Session on which to query active media state for
+ * \param media_state Media state to place the session media into
+ * \param type The type of the session media
+ * \param position Position at which to insert the new session media.
+ *
+ * \note The active media state will be queried and if a media session already
+ * exists at the given position for the same type it will be reused instead of
+ * allocating a new one.
+ *
+ * \retval non-NULL success
+ * \retval NULL failure
+ */
+struct ast_sip_session_media *ast_sip_session_media_state_add(struct ast_sip_session *session,
+	struct ast_sip_session_media_state *media_state, enum ast_media_type type, int position);
+
+/*!
+ * \brief Reset a media state to a clean state
+ * \since 15.0.0
+ *
+ * \param media_state The media state to reset
+ */
+void ast_sip_session_media_state_reset(struct ast_sip_session_media_state *media_state);
+
+/*!
+ * \brief Clone a media state
+ * \since 15.0.0
+ *
+ * \param media_state The media state to clone
+ *
+ * \retval non-NULL success
+ * \retval NULL failure
+ */
+struct ast_sip_session_media_state *ast_sip_session_media_state_clone(const struct ast_sip_session_media_state *media_state);
+
+/*!
+ * \brief Free a session media state structure
+ * \since 15.0.0
+ */
+void ast_sip_session_media_state_free(struct ast_sip_session_media_state *media_state);
+
+/*!
+ * \brief Set a read callback for a media session with a specific file descriptor
+ * \since 15.0.0
+ *
+ * \param session The session
+ * \param session_media The media session
+ * \param fd The file descriptor
+ * \param callback The read callback
+ *
+ * \retval 0 the read callback was successfully added
+ * \retval -1 the read callback could not be added
+ *
+ * \note This operations on the pending media state
+ */
+int ast_sip_session_media_add_read_callback(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
+	int fd, ast_sip_session_media_read_cb callback);
+
+/*!
+ * \brief Set a write callback for a media session
+ * \since 15.0.0
+ *
+ * \param session The session
+ * \param session_media The media session
+ * \param callback The write callback
+ *
+ * \retval 0 the write callback was successfully add
+ * \retval -1 the write callback is already set to something different
+ *
+ * \note This operates on the pending media state
+ */
+int ast_sip_session_media_set_write_callback(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
+	ast_sip_session_media_write_cb callback);
+
 /*! \brief Determines whether the res_pjsip_session module is loaded */
 #define CHECK_PJSIP_SESSION_MODULE_LOADED()				\
 	do {								\

include/asterisk/stream.h

@@ -289,6 +289,20 @@ struct ast_stream_topology *ast_stream_topology_alloc(void);
 struct ast_stream_topology *ast_stream_topology_clone(
 	const struct ast_stream_topology *topology);
 
+/*!
+ * \brief Compare two stream topologies to see if they are equal
+ *
+ * \param left The left topology
+ * \param right The right topology
+ *
+ * \retval 1 topologies are equivalent
+ * \retval 0 topologies differ
+ *
+ * \since 15
+ */
+int ast_stream_topology_equal(const struct ast_stream_topology *left,
+	const struct ast_stream_topology *right);
+
 /*!
  * \brief Destroy a stream topology
  *
@@ -391,7 +405,7 @@ int ast_stream_topology_del_stream(struct ast_stream_topology *topology,
  * since a new format capabilities structure is created for each media type.
  *
  * \note Each stream will have its name set to the corresponding media type.
- * For example: "AST_MEDIA_TYPE_AUDIO".
+ * For example: "audio".
  *
  * \note Each stream will be set to the sendrecv state.
  *

main/channel.c

@@ -4928,17 +4928,28 @@ int ast_write_stream(struct ast_channel *chan, int stream_num, struct ast_frame
 		goto done;
 	}
 
-	/* If this frame is writing an audio or video frame get the stream information */
-	if (fr->frametype == AST_FRAME_VOICE || fr->frametype == AST_FRAME_VIDEO) {
-		/* Initially use the default stream unless an explicit stream is provided */
-		stream = default_stream = ast_channel_get_default_stream(chan, ast_format_get_type(fr->subclass.format));
+	if (stream_num >= 0) {
+		/* If we were told to write to an explicit stream then allow this frame through, no matter
+		 * if the type is expected or not (a framehook could change)
+		 */
+		if (stream_num >= ast_stream_topology_get_count(ast_channel_get_stream_topology(chan))) {
+			goto done;
+		}
+		stream = ast_stream_topology_get_stream(ast_channel_get_stream_topology(chan), stream_num);
+		default_stream = ast_channel_get_default_stream(chan, ast_stream_get_type(stream));
+	} else if (fr->frametype == AST_FRAME_VOICE || fr->frametype == AST_FRAME_VIDEO || fr->frametype == AST_FRAME_MODEM) {
+		/* If we haven't been told of a stream then we need to figure out which once we need */
+		enum ast_media_type type = AST_MEDIA_TYPE_UNKNOWN;
 
-		if (stream_num >= 0) {
-			if (stream_num >= ast_stream_topology_get_count(ast_channel_get_stream_topology(chan))) {
-				goto done;
-			}
-			stream = ast_stream_topology_get_stream(ast_channel_get_stream_topology(chan), stream_num);
+		/* Some frame types have a fixed media type */
+		if (fr->frametype == AST_FRAME_VOICE || fr->frametype == AST_FRAME_VIDEO) {
+			type = ast_format_get_type(fr->subclass.format);
+		} else if (fr->frametype == AST_FRAME_MODEM) {
+			type = AST_MEDIA_TYPE_IMAGE;
 		}
+
+		/* No stream was specified, so use the default one */
+		stream = default_stream = ast_channel_get_default_stream(chan, type);
 	}
 
 	/* Perform the framehook write event here. After the frame enters the framehook list
@@ -5035,12 +5046,16 @@ int ast_write_stream(struct ast_channel *chan, int stream_num, struct ast_frame
 			res = ast_channel_tech(chan)->write_video(chan, fr);
 		} else {
 			res = 0;
-
 		}
 		break;
 	case AST_FRAME_MODEM:
-		res = (ast_channel_tech(chan)->write == NULL) ? 0 :
-			ast_channel_tech(chan)->write(chan, fr);
+		if (ast_channel_tech(chan)->write_stream) {
+			res = ast_channel_tech(chan)->write_stream(chan, ast_stream_get_position(stream), fr);
+		} else if ((stream == default_stream) && ast_channel_tech(chan)->write) {
+			res = ast_channel_tech(chan)->write(chan, fr);
+		} else {
+			res = 0;
+		}
 		break;
 	case AST_FRAME_VOICE:
 		if (ast_opt_generic_plc && ast_format_cmp(fr->subclass.format, ast_format_slin) == AST_FORMAT_CMP_EQUAL) {
@@ -10948,6 +10963,12 @@ int ast_channel_request_stream_topology_change(struct ast_channel *chan,
 		return -1;
 	}
 
+	if (ast_stream_topology_equal(ast_channel_get_stream_topology(chan), topology)) {
+		ast_debug(3, "Topology of %s already matches what is requested so ignoring topology change request\n",
+				ast_channel_name(chan));
+		return 0;
+	}
+
 	ast_channel_internal_set_stream_topology_change_source(chan, change_source);
 
 	return ast_channel_tech(chan)->indicate(chan, AST_CONTROL_STREAM_TOPOLOGY_REQUEST_CHANGE, topology, sizeof(topology));

main/stream.c

@@ -284,6 +284,53 @@ struct ast_stream_topology *ast_stream_topology_clone(
 	return new_topology;
 }
 
+int ast_stream_topology_equal(const struct ast_stream_topology *left,
+	const struct ast_stream_topology *right)
+{
+	int index;
+
+	ast_assert(left != NULL);
+	ast_assert(right != NULL);
+
+	if (ast_stream_topology_get_count(left) != ast_stream_topology_get_count(right)) {
+		return 0;
+	}
+
+	for (index = 0; index < ast_stream_topology_get_count(left); ++index) {
+		const struct ast_stream *left_stream = ast_stream_topology_get_stream(left, index);
+		const struct ast_stream *right_stream = ast_stream_topology_get_stream(right, index);
+
+		if (ast_stream_get_type(left_stream) != ast_stream_get_type(right_stream)) {
+			return 0;
+		}
+
+		if (ast_stream_get_state(left_stream) != ast_stream_get_state(right_stream)) {
+			return 0;
+		}
+
+		if (!ast_stream_get_formats(left_stream) && ast_stream_get_formats(right_stream) &&
+			ast_format_cap_count(ast_stream_get_formats(right_stream))) {
+			/* A NULL format capabilities and an empty format capabilities are the same, as they have
+			 * no formats inside. If one does though... they are not equal.
+			 */
+			return 0;
+		} else if (!ast_stream_get_formats(right_stream) && ast_stream_get_formats(left_stream) &&
+			ast_format_cap_count(ast_stream_get_formats(left_stream))) {
+			return 0;
+		} else if (ast_stream_get_formats(left_stream) && ast_stream_get_formats(right_stream) &&
+			!ast_format_cap_identical(ast_stream_get_formats(left_stream), ast_stream_get_formats(right_stream))) {
+			/* But if both are actually present we need to do an actual identical check. */
+			return 0;
+		}
+
+		if (strcmp(ast_stream_get_name(left_stream), ast_stream_get_name(right_stream))) {
+			return 0;
+		}
+	}
+
+	return 1;
+}
+
 void ast_stream_topology_free(struct ast_stream_topology *topology)
 {
 	if (!topology) {

res/res_pjsip.c

@@ -978,6 +978,20 @@
 						on Ringing when already INUSE.
 					</para></description>
 				</configOption>
+				<configOption name="max_audio_streams" default="1">
+					<synopsis>The maximum number of allowed audio streams for the endpoint</synopsis>
+					<description><para>
+						This option enforces a limit on the maximum simultaneous negotiated audio
+						streams allowed for the endpoint.
+					</para></description>
+				</configOption>
+				<configOption name="max_video_streams" default="1">
+					<synopsis>The maximum number of allowed video streams for the endpoint</synopsis>
+					<description><para>
+						This option enforces a limit on the maximum simultaneous negotiated video
+						streams allowed for the endpoint.
+					</para></description>
+				</configOption>
 			</configObject>
 			<configObject name="auth">
 				<synopsis>Authentication type</synopsis>

res/res_pjsip/pjsip_configuration.c

@@ -22,6 +22,7 @@
 #include "asterisk/test.h"
 #include "asterisk/statsd.h"
 #include "asterisk/pbx.h"
+#include "asterisk/stream.h"
 
 /*! \brief Number of buckets for persistent endpoint information */
 #define PERSISTENT_BUCKETS 53
@@ -1321,6 +1322,11 @@ static int sip_endpoint_apply_handler(const struct ast_sorcery *sorcery, void *o
 		return -1;
 	}
 
+	endpoint->media.topology = ast_stream_topology_create_from_format_cap(endpoint->media.codecs);
+	if (!endpoint->media.topology) {
+		return -1;
+	}
+
 	return 0;
 }
 
@@ -1941,6 +1947,8 @@ int ast_res_pjsip_initialize_configuration(void)
 	ast_sorcery_object_field_register(sip_sorcery, "endpoint", "allow_overlap", "yes", OPT_BOOL_T, 1, FLDSET(struct ast_sip_endpoint, allow_overlap));
 	ast_sorcery_object_field_register(sip_sorcery, "endpoint", "refer_blind_progress", "yes", OPT_BOOL_T, 1, FLDSET(struct ast_sip_endpoint, refer_blind_progress));
 	ast_sorcery_object_field_register(sip_sorcery, "endpoint", "notify_early_inuse_ringing", "no", OPT_BOOL_T, 1, FLDSET(struct ast_sip_endpoint, notify_early_inuse_ringing));
+	ast_sorcery_object_field_register(sip_sorcery, "endpoint", "max_audio_streams", "1", OPT_UINT_T, 0, FLDSET(struct ast_sip_endpoint, media.max_audio_streams));
+	ast_sorcery_object_field_register(sip_sorcery, "endpoint", "max_video_streams", "1", OPT_UINT_T, 0, FLDSET(struct ast_sip_endpoint, media.max_video_streams));
 
 	if (ast_sip_initialize_sorcery_transport()) {
 		ast_log(LOG_ERROR, "Failed to register SIP transport support with sorcery\n");
@@ -2060,7 +2068,8 @@ static void endpoint_destructor(void* obj)
 
 	ast_string_field_free_memory(endpoint);
 
-	ao2_ref(endpoint->media.codecs, -1);
+	ao2_cleanup(endpoint->media.codecs);
+	ast_stream_topology_free(endpoint->media.topology);
 	subscription_configuration_destroy(&endpoint->subscription);
 	info_configuration_destroy(&endpoint->info);
 	media_configuration_destroy(&endpoint->media);

res/res_pjsip_sdp_rtp.c

@@ -51,6 +51,8 @@
 #include "asterisk/sdp_srtp.h"
 #include "asterisk/dsp.h"
 #include "asterisk/linkedlists.h"       /* for AST_LIST_NEXT */
+#include "asterisk/stream.h"
+#include "asterisk/format_cache.h"
 
 #include "asterisk/res_pjsip.h"
 #include "asterisk/res_pjsip_session.h"
@@ -62,48 +64,7 @@ static struct ast_sched_context *sched;
 static struct ast_sockaddr address_rtp;
 
 static const char STR_AUDIO[] = "audio";
-static const int FD_AUDIO = 0;
-
 static const char STR_VIDEO[] = "video";
-static const int FD_VIDEO = 2;
-
-/*! \brief Retrieves an ast_format_type based on the given stream_type */
-static enum ast_media_type stream_to_media_type(const char *stream_type)
-{
-	if (!strcasecmp(stream_type, STR_AUDIO)) {
-		return AST_MEDIA_TYPE_AUDIO;
-	} else if (!strcasecmp(stream_type, STR_VIDEO)) {
-		return AST_MEDIA_TYPE_VIDEO;
-	}
-
-	return 0;
-}
-
-/*! \brief Get the starting descriptor for a media type */
-static int media_type_to_fdno(enum ast_media_type media_type)
-{
-	switch (media_type) {
-	case AST_MEDIA_TYPE_AUDIO: return FD_AUDIO;
-	case AST_MEDIA_TYPE_VIDEO: return FD_VIDEO;
-	case AST_MEDIA_TYPE_TEXT:
-	case AST_MEDIA_TYPE_UNKNOWN:
-	case AST_MEDIA_TYPE_IMAGE:
-	case AST_MEDIA_TYPE_END: break;
-	}
-	return -1;
-}
-
-/*! \brief Remove all other cap types but the one given */
-static void format_cap_only_type(struct ast_format_cap *caps, enum ast_media_type media_type)
-{
-	int i = 0;
-	while (i <= AST_MEDIA_TYPE_TEXT) {
-		if (i != media_type && i != AST_MEDIA_TYPE_UNKNOWN) {
-			ast_format_cap_remove_by_type(caps, i);
-		}
-		i += 1;
-	}
-}
 
 static int send_keepalive(const void *data)
 {
@@ -253,11 +214,11 @@ static int create_rtp(struct ast_sip_session *session, struct ast_sip_session_me
 		ast_rtp_instance_dtmf_mode_set(session_media->rtp, AST_RTP_DTMF_MODE_INBAND);
 	}
 
-	if (!strcmp(session_media->stream_type, STR_AUDIO) &&
+	if (session_media->type == AST_MEDIA_TYPE_AUDIO &&
 			(session->endpoint->media.tos_audio || session->endpoint->media.cos_audio)) {
 		ast_rtp_instance_set_qos(session_media->rtp, session->endpoint->media.tos_audio,
 				session->endpoint->media.cos_audio, "SIP RTP Audio");
-	} else if (!strcmp(session_media->stream_type, STR_VIDEO) &&
+	} else if (session_media->type == AST_MEDIA_TYPE_VIDEO &&
 			(session->endpoint->media.tos_video || session->endpoint->media.cos_video)) {
 		ast_rtp_instance_set_qos(session_media->rtp, session->endpoint->media.tos_video,
 				session->endpoint->media.cos_video, "SIP RTP Video");
@@ -347,12 +308,13 @@ static void get_codecs(struct ast_sip_session *session, const struct pjmedia_sdp
 static int set_caps(struct ast_sip_session *session,
 	struct ast_sip_session_media *session_media,
 	const struct pjmedia_sdp_media *stream,
-	int is_offer)
+	int is_offer, struct ast_stream *asterisk_stream)
 {
 	RAII_VAR(struct ast_format_cap *, caps, NULL, ao2_cleanup);
 	RAII_VAR(struct ast_format_cap *, peer, NULL, ao2_cleanup);
 	RAII_VAR(struct ast_format_cap *, joint, NULL, ao2_cleanup);
-	enum ast_media_type media_type = stream_to_media_type(session_media->stream_type);
+	RAII_VAR(struct ast_format_cap *, endpoint_caps, NULL, ao2_cleanup);
+	enum ast_media_type media_type = session_media->type;
 	struct ast_rtp_codecs codecs = AST_RTP_CODECS_NULL_INIT;
 	int fmts = 0;
 	int direct_media_enabled = !ast_sockaddr_isnull(&session_media->direct_media_addr) &&
@@ -362,14 +324,14 @@ static int set_caps(struct ast_sip_session *session,
 	if (!(caps = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT)) ||
 	    !(peer = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT)) ||
 	    !(joint = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT))) {
-		ast_log(LOG_ERROR, "Failed to allocate %s capabilities\n", session_media->stream_type);
+		ast_log(LOG_ERROR, "Failed to allocate %s capabilities\n",
+			ast_codec_media_type2str(session_media->type));
 		return -1;
 	}
 
 	/* get the endpoint capabilities */
 	if (direct_media_enabled) {
 		ast_format_cap_get_compatible(session->endpoint->media.codecs, session->direct_media_cap, caps);
-		format_cap_only_type(caps, media_type);
 	} else {
 		ast_format_cap_append_from_cap(caps, session->endpoint->media.codecs, media_type);
 	}
@@ -386,7 +348,7 @@ static int set_caps(struct ast_sip_session *session,
 
 		ast_rtp_codecs_payloads_destroy(&codecs);
 		ast_log(LOG_NOTICE, "No joint capabilities for '%s' media stream between our configuration(%s) and incoming SDP(%s)\n",
-			session_media->stream_type,
+			ast_codec_media_type2str(session_media->type),
 			ast_format_cap_get_names(caps, &usbuf),
 			ast_format_cap_get_names(peer, &thembuf));
 		return -1;
@@ -402,9 +364,9 @@ static int set_caps(struct ast_sip_session *session,
 	ast_rtp_codecs_payloads_copy(&codecs, ast_rtp_instance_get_codecs(session_media->rtp),
 		session_media->rtp);
 
-	ast_format_cap_append_from_cap(session->req_caps, joint, AST_MEDIA_TYPE_UNKNOWN);
+	ast_stream_set_formats(asterisk_stream, joint);
 
-	if (session->channel) {
+	if (session->channel && ast_sip_session_is_pending_stream_default(session, asterisk_stream)) {
 		ast_channel_lock(session->channel);
 		ast_format_cap_remove_by_type(caps, AST_MEDIA_TYPE_UNKNOWN);
 		ast_format_cap_append_from_cap(caps, ast_channel_nativeformats(session->channel),
@@ -968,24 +930,21 @@ static void set_ice_components(struct ast_sip_session *session, struct ast_sip_s
 }
 
 /*! \brief Function which negotiates an incoming media stream */
-static int negotiate_incoming_sdp_stream(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
-					 const struct pjmedia_sdp_session *sdp, const struct pjmedia_sdp_media *stream)
+static int negotiate_incoming_sdp_stream(struct ast_sip_session *session,
+	struct ast_sip_session_media *session_media, const pjmedia_sdp_session *sdp,
+	int index, struct ast_stream *asterisk_stream)
 {
 	char host[NI_MAXHOST];
 	RAII_VAR(struct ast_sockaddr *, addrs, NULL, ast_free);
-	enum ast_media_type media_type = stream_to_media_type(session_media->stream_type);
+	pjmedia_sdp_media *stream = sdp->media[index];
+	enum ast_media_type media_type = session_media->type;
 	enum ast_sip_session_media_encryption encryption = AST_SIP_MEDIA_ENCRYPT_NONE;
 	int res;
 
-	/* If port is 0, ignore this media stream */
-	if (!stream->desc.port) {
-		ast_debug(3, "Media stream '%s' is already declined\n", session_media->stream_type);
-		return 0;
-	}
-
 	/* If no type formats have been configured reject this stream */
 	if (!ast_format_cap_has_type(session->endpoint->media.codecs, media_type)) {
-		ast_debug(3, "Endpoint has no codecs for media type '%s', declining stream\n", session_media->stream_type);
+		ast_debug(3, "Endpoint has no codecs for media type '%s', declining stream\n",
+			ast_codec_media_type2str(session_media->type));
 		return 0;
 	}
 
@@ -1040,7 +999,7 @@ static int negotiate_incoming_sdp_stream(struct ast_sip_session *session, struct
 		pj_strdup(session->inv_session->pool, &session_media->transport, &stream->desc.transport);
  	}
 
-	if (set_caps(session, session_media, stream, 1)) {
+	if (set_caps(session, session_media, stream, 1, asterisk_stream)) {
 		return 0;
 	}
 	return 1;
@@ -1161,9 +1120,10 @@ static int add_crypto_to_stream(struct ast_sip_session *session,
 
 /*! \brief Function which creates an outgoing stream */
 static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
-				      struct pjmedia_sdp_session *sdp)
+				      struct pjmedia_sdp_session *sdp, const struct pjmedia_sdp_session *remote, struct ast_stream *stream)
 {
 	pj_pool_t *pool = session->inv_session->pool_prov;
+	static const pj_str_t STR_RTP_AVP = { "RTP/AVP", 7 };
 	static const pj_str_t STR_IN = { "IN", 2 };
 	static const pj_str_t STR_IP4 = { "IP4", 3};
 	static const pj_str_t STR_IP6 = { "IP6", 3};
@@ -1180,33 +1140,60 @@ static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct as
 	int min_packet_size = 0, max_packet_size = 0;
 	int rtp_code;
 	RAII_VAR(struct ast_format_cap *, caps, NULL, ao2_cleanup);
-	enum ast_media_type media_type = stream_to_media_type(session_media->stream_type);
-	int use_override_prefs = ast_format_cap_count(session->req_caps);
+	enum ast_media_type media_type = session_media->type;
 
 	int direct_media_enabled = !ast_sockaddr_isnull(&session_media->direct_media_addr) &&
 		ast_format_cap_count(session->direct_media_cap);
 
-	if ((use_override_prefs && !ast_format_cap_has_type(session->req_caps, media_type)) ||
-	    (!use_override_prefs && !ast_format_cap_has_type(session->endpoint->media.codecs, media_type))) {
-		/* If no type formats are configured don't add a stream */
-		return 0;
-	} else if (!session_media->rtp && create_rtp(session, session_media)) {
+	media = pj_pool_zalloc(pool, sizeof(struct pjmedia_sdp_media));
+	if (!media) {
 		return -1;
 	}
+	pj_strdup2(pool, &media->desc.media, ast_codec_media_type2str(session_media->type));
 
-	set_ice_components(session, session_media);
-	enable_rtcp(session, session_media, NULL);
+	/* If this is a removed (or declined) stream OR if no formats exist then construct a minimal stream in SDP */
+	if (ast_stream_get_state(stream) == AST_STREAM_STATE_REMOVED || !ast_stream_get_formats(stream) ||
+		!ast_format_cap_count(ast_stream_get_formats(stream))) {
+		media->desc.port = 0;
+		media->desc.port_count = 1;
+
+		if (remote) {
+			pjmedia_sdp_media *remote_media = remote->media[ast_stream_get_position(stream)];
+			int index;
+
+			media->desc.transport = remote_media->desc.transport;
+
+			/* Preserve existing behavior by copying the formats provided from the offer */
+			for (index = 0; index < remote_media->desc.fmt_count; ++index) {
+				media->desc.fmt[index] = remote_media->desc.fmt[index];
+			}
+			media->desc.fmt_count = remote_media->desc.fmt_count;
+		} else {
+			/* This is actually an offer so put a dummy payload in that is ignored and sane transport */
+			media->desc.transport = STR_RTP_AVP;
+			pj_strdup2(pool, &media->desc.fmt[media->desc.fmt_count++], "32");
+		}
+
+		sdp->media[sdp->media_count++] = media;
+		ast_stream_set_state(stream, AST_STREAM_STATE_REMOVED);
+
+		return 1;
+	}
 
-	if (!(media = pj_pool_zalloc(pool, sizeof(struct pjmedia_sdp_media))) ||
-		!(media->conn = pj_pool_zalloc(pool, sizeof(struct pjmedia_sdp_conn)))) {
+	if (!session_media->rtp && create_rtp(session, session_media)) {
 		return -1;
 	}
 
+	set_ice_components(session, session_media);
+	enable_rtcp(session, session_media, NULL);
+
+	/* Crypto has to be added before setting the media transport so that SRTP is properly
+	 * set up according to the configuration. This ends up changing the media transport.
+	 */
 	if (add_crypto_to_stream(session, session_media, pool, media)) {
 		return -1;
 	}
 
-	media->desc.media = pj_str(session_media->stream_type);
 	if (pj_strlen(&session_media->transport)) {
 		/* If a transport has already been specified use it */
 		media->desc.transport = session_media->transport;
@@ -1219,6 +1206,11 @@ static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct as
 			session->endpoint->media.rtp.force_avp));
 	}
 
+	media->conn = pj_pool_zalloc(pool, sizeof(struct pjmedia_sdp_conn));
+	if (!media->conn) {
+		return -1;
+	}
+
 	/* Add connection level details */
 	if (direct_media_enabled) {
 		hostip = ast_sockaddr_stringify_fmt(&session_media->direct_media_addr, AST_SOCKADDR_STR_ADDR);
@@ -1229,7 +1221,8 @@ static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct as
 	}
 
 	if (ast_strlen_zero(hostip)) {
-		ast_log(LOG_ERROR, "No local host IP available for stream %s\n", session_media->stream_type);
+		ast_log(LOG_ERROR, "No local host IP available for stream %s\n",
+			ast_codec_media_type2str(session_media->type));
 		return -1;
 	}
 
@@ -1247,25 +1240,23 @@ static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct as
 		}
 	}
 
+	/* Add ICE attributes and candidates */
+	add_ice_to_stream(session, session_media, pool, media);
+
 	ast_rtp_instance_get_local_address(session_media->rtp, &addr);
 	media->desc.port = direct_media_enabled ? ast_sockaddr_port(&session_media->direct_media_addr) : (pj_uint16_t) ast_sockaddr_port(&addr);
 	media->desc.port_count = 1;
 
-	/* Add ICE attributes and candidates */
-	add_ice_to_stream(session, session_media, pool, media);
-
 	if (!(caps = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT))) {
-		ast_log(LOG_ERROR, "Failed to allocate %s capabilities\n", session_media->stream_type);
+		ast_log(LOG_ERROR, "Failed to allocate %s capabilities\n",
+			ast_codec_media_type2str(session_media->type));
 		return -1;
 	}
 
 	if (direct_media_enabled) {
 		ast_format_cap_get_compatible(session->endpoint->media.codecs, session->direct_media_cap, caps);
-	} else if (!ast_format_cap_count(session->req_caps) ||
-		!ast_format_cap_iscompatible(session->req_caps, session->endpoint->media.codecs)) {
-		ast_format_cap_append_from_cap(caps, session->endpoint->media.codecs, media_type);
 	} else {
-		ast_format_cap_append_from_cap(caps, session->req_caps, media_type);
+		ast_format_cap_append_from_cap(caps, ast_stream_get_formats(stream), media_type);
 	}
 
 	for (index = 0; index < ast_format_cap_count(caps); ++index) {
@@ -1302,7 +1293,8 @@ static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct as
 	}
 
 	/* Add non-codec formats */
-	if (media_type != AST_MEDIA_TYPE_VIDEO && media->desc.fmt_count < PJMEDIA_MAX_SDP_FMT) {
+	if (ast_sip_session_is_pending_stream_default(session, stream) && media_type != AST_MEDIA_TYPE_VIDEO
+		&& media->desc.fmt_count < PJMEDIA_MAX_SDP_FMT) {
 		for (index = 1LL; index <= AST_RTP_MAX; index <<= 1) {
 			if (!(noncodec & index)) {
 				continue;
@@ -1368,23 +1360,65 @@ static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct as
 	return 1;
 }
 
-static int apply_negotiated_sdp_stream(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
-				       const struct pjmedia_sdp_session *local, const struct pjmedia_sdp_media *local_stream,
-				       const struct pjmedia_sdp_session *remote, const struct pjmedia_sdp_media *remote_stream)
+static struct ast_frame *media_session_rtp_read_callback(struct ast_sip_session *session, struct ast_sip_session_media *session_media)
+{
+	struct ast_frame *f;
+
+	if (!session_media->rtp) {
+		return &ast_null_frame;
+	}
+
+	f = ast_rtp_instance_read(session_media->rtp, 0);
+	if (!f) {
+		return NULL;
+	}
+
+	ast_rtp_instance_set_last_rx(session_media->rtp, time(NULL));
+
+	return f;
+}
+
+static struct ast_frame *media_session_rtcp_read_callback(struct ast_sip_session *session, struct ast_sip_session_media *session_media)
+{
+	struct ast_frame *f;
+
+	if (!session_media->rtp) {
+		return &ast_null_frame;
+	}
+
+	f = ast_rtp_instance_read(session_media->rtp, 1);
+	if (!f) {
+		return NULL;
+	}
+
+	ast_rtp_instance_set_last_rx(session_media->rtp, time(NULL));
+
+	return f;
+}
+
+static int media_session_rtp_write_callback(struct ast_sip_session *session, struct ast_sip_session_media *session_media, struct ast_frame *frame)
+{
+	if (!session_media->rtp) {
+		return 0;
+	}
+
+	return ast_rtp_instance_write(session_media->rtp, frame);
+}
+
+static int apply_negotiated_sdp_stream(struct ast_sip_session *session,
+	struct ast_sip_session_media *session_media, const struct pjmedia_sdp_session *local,
+	const struct pjmedia_sdp_session *remote, int index, struct ast_stream *asterisk_stream)
 {
 	RAII_VAR(struct ast_sockaddr *, addrs, NULL, ast_free);
-	enum ast_media_type media_type = stream_to_media_type(session_media->stream_type);
+	struct pjmedia_sdp_media *remote_stream = remote->media[index];
+	enum ast_media_type media_type = session_media->type;
 	char host[NI_MAXHOST];
-	int fdno, res;
+	int res;
 
 	if (!session->channel) {
 		return 1;
 	}
 
-	if (!local_stream->desc.port || !remote_stream->desc.port) {
-		return 1;
-	}
-
 	/* Ensure incoming transport is compatible with the endpoint's configuration */
 	if (!session->endpoint->media.rtp.use_received_transport &&
 		check_endpoint_media_transport(session->endpoint, remote_stream) == AST_SIP_MEDIA_TRANSPORT_INVALID) {
@@ -1424,21 +1458,26 @@ static int apply_negotiated_sdp_stream(struct ast_sip_session *session, struct a
 	/* Apply connection information to the RTP instance */
 	ast_sockaddr_set_port(addrs, remote_stream->desc.port);
 	ast_rtp_instance_set_remote_address(session_media->rtp, addrs);
-	if (set_caps(session, session_media, remote_stream, 0)) {
+	if (set_caps(session, session_media, remote_stream, 0, asterisk_stream)) {
 		return 1;
 	}
 
-	if ((fdno = media_type_to_fdno(media_type)) < 0) {
-		return -1;
-	}
-	ast_channel_set_fd(session->channel, fdno, ast_rtp_instance_fd(session_media->rtp, 0));
+	ast_sip_session_media_set_write_callback(session, session_media, media_session_rtp_write_callback);
+	ast_sip_session_media_add_read_callback(session, session_media, ast_rtp_instance_fd(session_media->rtp, 0),
+		media_session_rtp_read_callback);
 	if (!session->endpoint->media.rtcp_mux || !session_media->remote_rtcp_mux) {
-		ast_channel_set_fd(session->channel, fdno + 1, ast_rtp_instance_fd(session_media->rtp, 1));
+		ast_sip_session_media_add_read_callback(session, session_media, ast_rtp_instance_fd(session_media->rtp, 1),
+			media_session_rtcp_read_callback);
 	}
 
 	/* If ICE support is enabled find all the needed attributes */
 	process_ice_attributes(session, session_media, remote, remote_stream);
 
+	/* Set the channel uniqueid on the RTP instance now that it is becoming active */
+	ast_channel_lock(session->channel);
+	ast_rtp_instance_set_channel_id(session_media->rtp, ast_channel_uniqueid(session->channel));
+	ast_channel_unlock(session->channel);
+
 	/* Ensure the RTP instance is active */
 	ast_rtp_instance_activate(session_media->rtp);
 
@@ -1476,7 +1515,7 @@ static int apply_negotiated_sdp_stream(struct ast_sip_session *session, struct a
 	session_media->encryption = session->endpoint->media.rtp.encryption;
 
 	if (session->endpoint->media.rtp.keepalive > 0 &&
-			stream_to_media_type(session_media->stream_type) == AST_MEDIA_TYPE_AUDIO) {
+			session_media->type == AST_MEDIA_TYPE_AUDIO) {
 		ast_rtp_instance_set_keepalive(session_media->rtp, session->endpoint->media.rtp.keepalive);
 		/* Schedule the initial keepalive early in case this is being used to punch holes through
 		 * a NAT. This way there won't be an awkward delay before media starts flowing in some

res/res_pjsip_session.exports.in

@@ -1,27 +1,7 @@
 {
 	global:
-		LINKER_SYMBOL_PREFIXast_sip_session_terminate;
-		LINKER_SYMBOL_PREFIXast_sip_session_defer_termination;
-		LINKER_SYMBOL_PREFIXast_sip_session_defer_termination_cancel;
-		LINKER_SYMBOL_PREFIXast_sip_session_end_if_deferred;
-		LINKER_SYMBOL_PREFIXast_sip_session_register_sdp_handler;
-		LINKER_SYMBOL_PREFIXast_sip_session_unregister_sdp_handler;
-		LINKER_SYMBOL_PREFIXast_sip_session_register_supplement;
-		LINKER_SYMBOL_PREFIXast_sip_session_unregister_supplement;
-		LINKER_SYMBOL_PREFIXast_sip_session_alloc_datastore;
-		LINKER_SYMBOL_PREFIXast_sip_session_add_datastore;
-		LINKER_SYMBOL_PREFIXast_sip_session_get_datastore;
-		LINKER_SYMBOL_PREFIXast_sip_session_remove_datastore;
-		LINKER_SYMBOL_PREFIXast_sip_session_get_identity;
-		LINKER_SYMBOL_PREFIXast_sip_session_refresh;
-		LINKER_SYMBOL_PREFIXast_sip_session_send_response;
-		LINKER_SYMBOL_PREFIXast_sip_session_send_request;
-		LINKER_SYMBOL_PREFIXast_sip_session_create_invite;
-		LINKER_SYMBOL_PREFIXast_sip_session_create_outgoing;
-		LINKER_SYMBOL_PREFIXast_sip_session_suspend;
-		LINKER_SYMBOL_PREFIXast_sip_session_unsuspend;
+		LINKER_SYMBOL_PREFIXast_sip_session_*;
 		LINKER_SYMBOL_PREFIXast_sip_dialog_get_session;
-		LINKER_SYMBOL_PREFIXast_sip_session_resume_reinvite;
 		LINKER_SYMBOL_PREFIXast_sip_channel_pvt_alloc;
 	local:
 		*;

res/res_pjsip_t38.c

@@ -43,6 +43,8 @@
 #include "asterisk/netsock2.h"
 #include "asterisk/channel.h"
 #include "asterisk/acl.h"
+#include "asterisk/stream.h"
+#include "asterisk/format_cache.h"
 
 #include "asterisk/res_pjsip.h"
 #include "asterisk/res_pjsip_session.h"
@@ -63,11 +65,16 @@ struct t38_state {
 	struct ast_control_t38_parameters their_parms;
 	/*! \brief Timer entry for automatically rejecting an inbound re-invite */
 	pj_timer_entry timer;
+	/*! Preserved media state for when T.38 ends */
+	struct ast_sip_session_media_state *media_state;
 };
 
 /*! \brief Destructor for T.38 state information */
 static void t38_state_destroy(void *obj)
 {
+	struct t38_state *state = obj;
+
+	ast_sip_session_media_state_free(state->media_state);
 	ast_free(obj);
 }
 
@@ -195,7 +202,7 @@ static int t38_automatic_reject(void *obj)
 {
 	RAII_VAR(struct ast_sip_session *, session, obj, ao2_cleanup);
 	RAII_VAR(struct ast_datastore *, datastore, ast_sip_session_get_datastore(session, "t38"), ao2_cleanup);
-	RAII_VAR(struct ast_sip_session_media *, session_media, ao2_find(session->media, "image", OBJ_KEY), ao2_cleanup);
+	struct ast_sip_session_media *session_media;
 
 	if (!datastore) {
 		return 0;
@@ -204,6 +211,7 @@ static int t38_automatic_reject(void *obj)
 	ast_debug(2, "Automatically rejecting T.38 request on channel '%s'\n",
 		session->channel ? ast_channel_name(session->channel) : "<gone>");
 
+	session_media = session->pending_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
 	t38_change_state(session, session_media, datastore->data, T38_REJECTED);
 	ast_sip_session_resume_reinvite(session);
 
@@ -259,7 +267,6 @@ static int t38_initialize_session(struct ast_sip_session *session, struct ast_si
 		return -1;
 	}
 
-	ast_channel_set_fd(session->channel, 5, ast_udptl_fd(session_media->udptl));
 	ast_udptl_set_error_correction_scheme(session_media->udptl, session->endpoint->media.t38.error_correction);
 	ast_udptl_setnat(session_media->udptl, session->endpoint->media.t38.nat);
 	ast_udptl_set_far_max_datagram(session_media->udptl, session->endpoint->media.t38.maxdatagram);
@@ -271,19 +278,15 @@ static int t38_initialize_session(struct ast_sip_session *session, struct ast_si
 /*! \brief Callback for when T.38 reinvite SDP is created */
 static int t38_reinvite_sdp_cb(struct ast_sip_session *session, pjmedia_sdp_session *sdp)
 {
-	int stream;
-
-	/* Move the image media stream to the front and have it as the only stream, pjmedia will fill in
-	 * dummy streams for the rest
-	 */
-	for (stream = 0; stream < sdp->media_count; ++stream) {
-		if (!pj_strcmp2(&sdp->media[stream]->desc.media, "image")) {
-			sdp->media[0] = sdp->media[stream];
-			sdp->media_count = 1;
-			break;
-		}
+	struct t38_state *state;
+
+	state = t38_state_get_or_alloc(session);
+	if (!state) {
+		return -1;
 	}
 
+	state->media_state = ast_sip_session_media_state_clone(session->active_media_state);
+
 	return 0;
 }
 
@@ -292,34 +295,109 @@ static int t38_reinvite_response_cb(struct ast_sip_session *session, pjsip_rx_da
 {
 	struct pjsip_status_line status = rdata->msg_info.msg->line.status;
 	struct t38_state *state;
-	RAII_VAR(struct ast_sip_session_media *, session_media, NULL, ao2_cleanup);
+	struct ast_sip_session_media *session_media = NULL;
 
 	if (status.code == 100) {
 		return 0;
 	}
 
-	if (!(state = t38_state_get_or_alloc(session)) ||
-		!(session_media = ao2_find(session->media, "image", OBJ_KEY))) {
+	state = t38_state_get_or_alloc(session);
+	if (!state) {
 		ast_log(LOG_WARNING, "Received response to T.38 re-invite on '%s' but state unavailable\n",
 			ast_channel_name(session->channel));
 		return 0;
 	}
 
-	t38_change_state(session, session_media, state, (status.code == 200) ? T38_ENABLED : T38_REJECTED);
+	if (status.code == 200) {
+		int index;
+
+		session_media = session->active_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
+		t38_change_state(session, session_media, state, T38_ENABLED);
+
+		/* Stop all the streams in the stored away active state, they'll go back to being active once
+		 * we reinvite back.
+		 */
+		for (index = 0; index < AST_VECTOR_SIZE(&state->media_state->sessions); ++index) {
+			struct ast_sip_session_media *session_media = AST_VECTOR_GET(&state->media_state->sessions, index);
+
+			if (session_media && session_media->handler && session_media->handler->stream_stop) {
+				session_media->handler->stream_stop(session_media);
+			}
+		}
+	} else {
+		session_media = session->pending_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
+		t38_change_state(session, session_media, state, T38_REJECTED);
+
+		/* Abort this attempt at switching to T.38 by resetting the pending state and freeing our stored away active state */
+		ast_sip_session_media_state_free(state->media_state);
+		state->media_state = NULL;
+		ast_sip_session_media_state_reset(session->pending_media_state);
+	}
 
 	return 0;
 }
 
+/*! \brief Helper function which creates a media state for strictly T.38 */
+static struct ast_sip_session_media_state *t38_create_media_state(struct ast_sip_session *session)
+{
+	struct ast_sip_session_media_state *media_state;
+	struct ast_stream *stream;
+	struct ast_format_cap *caps;
+	struct ast_sip_session_media *session_media;
+
+	media_state = ast_sip_session_media_state_alloc();
+	if (!media_state) {
+		return NULL;
+	}
+
+	media_state->topology = ast_stream_topology_alloc();
+	if (!media_state->topology) {
+		ast_sip_session_media_state_free(media_state);
+		return NULL;
+	}
+
+	stream = ast_stream_alloc("t38", AST_MEDIA_TYPE_IMAGE);
+	if (!stream) {
+		ast_sip_session_media_state_free(media_state);
+		return NULL;
+	}
+
+	ast_stream_set_state(stream, AST_STREAM_STATE_SENDRECV);
+	ast_stream_topology_set_stream(media_state->topology, 0, stream);
+
+	caps = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
+	if (!caps) {
+		ast_sip_session_media_state_free(media_state);
+		return NULL;
+	}
+
+	ast_format_cap_append(caps, ast_format_t38, 0);
+	ast_stream_set_formats(stream, caps);
+	ao2_ref(caps, -1);
+
+	session_media = ast_sip_session_media_state_add(session, media_state, AST_MEDIA_TYPE_IMAGE, 0);
+	if (!session_media) {
+		ast_sip_session_media_state_free(media_state);
+		return NULL;
+	}
+
+	if (t38_initialize_session(session, session_media)) {
+		ast_sip_session_media_state_free(media_state);
+		return NULL;
+	}
+
+	return media_state;
+}
+
 /*! \brief Task for reacting to T.38 control frame */
 static int t38_interpret_parameters(void *obj)
 {
 	RAII_VAR(struct t38_parameters_task_data *, data, obj, ao2_cleanup);
 	const struct ast_control_t38_parameters *parameters = data->frame->data.ptr;
 	struct t38_state *state = t38_state_get_or_alloc(data->session);
-	RAII_VAR(struct ast_sip_session_media *, session_media, ao2_find(data->session->media, "image", OBJ_KEY), ao2_cleanup);
+	struct ast_sip_session_media *session_media = NULL;
 
-	/* Without session media or state we can't interpret parameters */
-	if (!session_media || !state) {
+	if (!state) {
 		return 0;
 	}
 
@@ -329,12 +407,15 @@ static int t38_interpret_parameters(void *obj)
 		/* Negotiation can not take place without a valid max_ifp value. */
 		if (!parameters->max_ifp) {
 			if (data->session->t38state == T38_PEER_REINVITE) {
+				session_media = data->session->pending_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
 				t38_change_state(data->session, session_media, state, T38_REJECTED);
 				ast_sip_session_resume_reinvite(data->session);
 			} else if (data->session->t38state == T38_ENABLED) {
+				session_media = data->session->active_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
 				t38_change_state(data->session, session_media, state, T38_DISABLED);
 				ast_sip_session_refresh(data->session, NULL, NULL, NULL,
-					AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1);
+					AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1, state->media_state);
+				state->media_state = NULL;
 			}
 			break;
 		} else if (data->session->t38state == T38_PEER_REINVITE) {
@@ -353,37 +434,46 @@ static int t38_interpret_parameters(void *obj)
 			}
 			state->our_parms.version = MIN(state->our_parms.version, state->their_parms.version);
 			state->our_parms.rate_management = state->their_parms.rate_management;
+			session_media = data->session->pending_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
 			ast_udptl_set_local_max_ifp(session_media->udptl, state->our_parms.max_ifp);
 			t38_change_state(data->session, session_media, state, T38_ENABLED);
 			ast_sip_session_resume_reinvite(data->session);
 		} else if ((data->session->t38state != T38_ENABLED) ||
 				((data->session->t38state == T38_ENABLED) &&
                                 (parameters->request_response == AST_T38_REQUEST_NEGOTIATE))) {
-			if (t38_initialize_session(data->session, session_media)) {
+			struct ast_sip_session_media_state *media_state;
+
+			media_state = t38_create_media_state(data->session);
+			if (!media_state) {
 				break;
 			}
 			state->our_parms = *parameters;
+			session_media = media_state->default_session[AST_MEDIA_TYPE_IMAGE];
 			ast_udptl_set_local_max_ifp(session_media->udptl, state->our_parms.max_ifp);
 			t38_change_state(data->session, session_media, state, T38_LOCAL_REINVITE);
 			ast_sip_session_refresh(data->session, NULL, t38_reinvite_sdp_cb, t38_reinvite_response_cb,
-				AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1);
+				AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1, media_state);
 		}
 		break;
 	case AST_T38_TERMINATED:
 	case AST_T38_REFUSED:
 	case AST_T38_REQUEST_TERMINATE:         /* Shutdown T38 */
 		if (data->session->t38state == T38_PEER_REINVITE) {
+			session_media = data->session->pending_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
 			t38_change_state(data->session, session_media, state, T38_REJECTED);
 			ast_sip_session_resume_reinvite(data->session);
 		} else if (data->session->t38state == T38_ENABLED) {
+			session_media = data->session->active_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
 			t38_change_state(data->session, session_media, state, T38_DISABLED);
-			ast_sip_session_refresh(data->session, NULL, NULL, NULL, AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1);
+			ast_sip_session_refresh(data->session, NULL, NULL, NULL, AST_SIP_SESSION_REFRESH_METHOD_INVITE, 1, state->media_state);
+			state->media_state = NULL;
 		}
 		break;
 	case AST_T38_REQUEST_PARMS: {		/* Application wants remote's parameters re-sent */
 		struct ast_control_t38_parameters parameters = state->their_parms;
 
 		if (data->session->t38state == T38_PEER_REINVITE) {
+			session_media = data->session->pending_media_state->default_session[AST_MEDIA_TYPE_IMAGE];
 			parameters.max_ifp = ast_udptl_get_far_max_ifp(session_media->udptl);
 			parameters.request_response = AST_T38_REQUEST_NEGOTIATE;
 			ast_queue_control_data(data->session->channel, AST_CONTROL_T38_PARAMETERS, &parameters, sizeof(parameters));
@@ -397,67 +487,27 @@ static int t38_interpret_parameters(void *obj)
 	return 0;
 }
 
-/*! \brief Frame hook callback for writing */
-static struct ast_frame *t38_framehook_write(struct ast_channel *chan,
-	struct ast_sip_session *session, struct ast_frame *f)
+/*! \brief Frame hook callback for T.38 related stuff */
+static struct ast_frame *t38_framehook(struct ast_channel *chan, struct ast_frame *f,
+	enum ast_framehook_event event, void *data)
 {
+	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(chan);
+
+	if (event != AST_FRAMEHOOK_EVENT_WRITE) {
+		return f;
+	}
+
 	if (f->frametype == AST_FRAME_CONTROL && f->subclass.integer == AST_CONTROL_T38_PARAMETERS &&
-		session->endpoint->media.t38.enabled) {
-		struct t38_parameters_task_data *data = t38_parameters_task_data_alloc(session, f);
+		channel->session->endpoint->media.t38.enabled) {
+		struct t38_parameters_task_data *data = t38_parameters_task_data_alloc(channel->session, f);
 
 		if (!data) {
 			return f;
 		}
 
-		if (ast_sip_push_task(session->serializer, t38_interpret_parameters, data)) {
+		if (ast_sip_push_task(channel->session->serializer, t38_interpret_parameters, data)) {
 			ao2_ref(data, -1);
 		}
-	} else if (f->frametype == AST_FRAME_MODEM) {
-		struct ast_sip_session_media *session_media;
-
-		/* Avoid deadlock between chan and the session->media container lock */
-		ast_channel_unlock(chan);
-		session_media = ao2_find(session->media, "image", OBJ_SEARCH_KEY);
-		ast_channel_lock(chan);
-		if (session_media && session_media->udptl) {
-			ast_udptl_write(session_media->udptl, f);
-		}
-		ao2_cleanup(session_media);
-	}
-
-	return f;
-}
-
-/*! \brief Frame hook callback for reading */
-static struct ast_frame *t38_framehook_read(struct ast_channel *chan,
-	struct ast_sip_session *session, struct ast_frame *f)
-{
-	if (ast_channel_fdno(session->channel) == 5) {
-		struct ast_sip_session_media *session_media;
-
-		/* Avoid deadlock between chan and the session->media container lock */
-		ast_channel_unlock(chan);
-		session_media = ao2_find(session->media, "image", OBJ_SEARCH_KEY);
-		ast_channel_lock(chan);
-		if (session_media && session_media->udptl) {
-			f = ast_udptl_read(session_media->udptl);
-		}
-		ao2_cleanup(session_media);
-	}
-
-	return f;
-}
-
-/*! \brief Frame hook callback for T.38 related stuff */
-static struct ast_frame *t38_framehook(struct ast_channel *chan, struct ast_frame *f,
-	enum ast_framehook_event event, void *data)
-{
-	struct ast_sip_channel_pvt *channel = ast_channel_tech_pvt(chan);
-
-	if (event == AST_FRAMEHOOK_EVENT_READ) {
-		f = t38_framehook_read(chan, channel->session, f);
-	} else if (event == AST_FRAMEHOOK_EVENT_WRITE) {
-		f = t38_framehook_write(chan, channel->session, f);
 	}
 
 	return f;
@@ -476,7 +526,7 @@ static void t38_masq(void *data, int framehook_id,
 
 static int t38_consume(void *data, enum ast_frame_type type)
 {
-	return 0;
+	return (type == AST_FRAME_CONTROL) ? 1 : 0;
 }
 
 static const struct ast_datastore_info t38_framehook_datastore = {
@@ -676,11 +726,13 @@ static enum ast_sip_session_sdp_stream_defer defer_incoming_sdp_stream(
 }
 
 /*! \brief Function which negotiates an incoming media stream */
-static int negotiate_incoming_sdp_stream(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
-					 const struct pjmedia_sdp_session *sdp, const struct pjmedia_sdp_media *stream)
+static int negotiate_incoming_sdp_stream(struct ast_sip_session *session,
+	struct ast_sip_session_media *session_media, const struct pjmedia_sdp_session *sdp,
+	int index, struct ast_stream *asterisk_stream)
 {
 	struct t38_state *state;
 	char host[NI_MAXHOST];
+	pjmedia_sdp_media *stream = sdp->media[index];
 	RAII_VAR(struct ast_sockaddr *, addrs, NULL, ast_free);
 
 	if (!session->endpoint->media.t38.enabled) {
@@ -720,7 +772,7 @@ static int negotiate_incoming_sdp_stream(struct ast_sip_session *session, struct
 
 /*! \brief Function which creates an outgoing stream */
 static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
-				      struct pjmedia_sdp_session *sdp)
+				      struct pjmedia_sdp_session *sdp, const struct pjmedia_sdp_session *remote, struct ast_stream *stream)
 {
 	pj_pool_t *pool = session->inv_session->pool_prov;
 	static const pj_str_t STR_IN = { "IN", 2 };
@@ -758,7 +810,7 @@ static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct as
 		return -1;
 	}
 
-	media->desc.media = pj_str(session_media->stream_type);
+	pj_strdup2(pool, &media->desc.media, ast_codec_media_type2str(session_media->type));
 	media->desc.transport = STR_UDPTL;
 
 	if (ast_strlen_zero(session->endpoint->media.address)) {
@@ -826,12 +878,31 @@ static int create_outgoing_sdp_stream(struct ast_sip_session *session, struct as
 	return 1;
 }
 
+static struct ast_frame *media_session_udptl_read_callback(struct ast_sip_session *session, struct ast_sip_session_media *session_media)
+{
+	if (!session_media->udptl) {
+		return &ast_null_frame;
+	}
+
+	return ast_udptl_read(session_media->udptl);
+}
+
+static int media_session_udptl_write_callback(struct ast_sip_session *session, struct ast_sip_session_media *session_media, struct ast_frame *frame)
+{
+	if (!session_media->udptl) {
+		return 0;
+	}
+
+	return ast_udptl_write(session_media->udptl, frame);
+}
+
 /*! \brief Function which applies a negotiated stream */
-static int apply_negotiated_sdp_stream(struct ast_sip_session *session, struct ast_sip_session_media *session_media,
-				       const struct pjmedia_sdp_session *local, const struct pjmedia_sdp_media *local_stream,
-				       const struct pjmedia_sdp_session *remote, const struct pjmedia_sdp_media *remote_stream)
+static int apply_negotiated_sdp_stream(struct ast_sip_session *session,
+	struct ast_sip_session_media *session_media, const struct pjmedia_sdp_session *local,
+	const struct pjmedia_sdp_session *remote, int index, struct ast_stream *asterisk_stream)
 {
 	RAII_VAR(struct ast_sockaddr *, addrs, NULL, ast_free);
+	pjmedia_sdp_media *remote_stream = remote->media[index];
 	char host[NI_MAXHOST];
 	struct t38_state *state;
 
@@ -858,6 +929,10 @@ static int apply_negotiated_sdp_stream(struct ast_sip_session *session, struct a
 
 	t38_interpret_sdp(state, session, session_media, remote_stream);
 
+	ast_sip_session_media_set_write_callback(session, session_media, media_session_udptl_write_callback);
+	ast_sip_session_media_add_read_callback(session, session_media, ast_udptl_fd(session_media->udptl),
+		media_session_udptl_read_callback);
+
 	return 0;
 }