Add 'mix' flag for subscribe request to produce single mixed audio stream

When subscribing to a call with multiple from-tags, adding the 'mix'
flag produces a single destination media with audio_player activated
to mix all audio sources into one stream, instead of creating separate
destination medias per source.

Handles bidirectional transitions between mix and non-mix modes on the
same to-tag: retiring stale destination medias, deactivating
audio_player, and compacting the media array to prevent index holes
and unbounded growth across re-subscribe cycles.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: fergusean

daemon/call.c

@@ -4404,6 +4404,13 @@ static int monologue_subscribe_request1(struct call_media *src_media, struct cal
 	__auto_type ms = t_hash_table_lookup(ht, src_media);
 	if (ms)
 		dst_media = ms->media;
+	// Don't reuse a dest that was already claimed by another source in this
+	// request. This can happen when transitioning from mix mode (one shared
+	// dest) to non-mix mode (one dest per source): all ht entries point to
+	// the same old dest, but we need separate dests. Check head instead of
+	// length because i_queue_delete doesn't update length.
+	if (dst_media && dst_media->media_subscriptions.head)
+		dst_media = NULL;
 	if (!dst_media) {
 		// new media needed
 		dst_media = call_get_media(dst_ml, &src_media->type, src_media->type_id,
@@ -4486,14 +4493,205 @@ static int monologue_subscribe_request1(struct call_media *src_media, struct cal
 	return 0;
 }
 
+static void inject_reconfigure_destination_media(struct call_media *, bool, const sdp_ng_flags *);
+
+/* called with call->master_lock held in W */
+__attribute__((nonnull(1, 2, 3)))
+static int monologue_subscribe_request_mix(const subscription_q *srms, struct call_monologue *dst_ml,
+		sdp_ng_flags *flags)
+{
+	g_auto(str_ht) mid_tracker_dst = str_ht_new();
+	g_auto(subscription_store_ht) ht = subscription_store_ht_new();
+
+	__unsubscribe_medias_from_all(dst_ml, ht);
+	__call_monologue_init_from_flags(dst_ml, NULL, flags);
+
+	// For audio sources, create ONE destination media and subscribe all sources to it.
+	// For non-audio, fall through to normal per-source handling.
+
+	struct call_media *dst_audio_media = NULL;
+	bool dst_audio_initialized = false;
+
+	IQUEUE_FOREACH(srms, ms) {
+		struct call_media *src_media = ms->media;
+		if (!src_media)
+			continue;
+
+		if (src_media->type_id != MT_AUDIO) {
+			// non-audio: create separate destination media per source (normal behavior)
+			int ret = monologue_subscribe_request1(src_media, dst_ml, flags, ht);
+			if (ret)
+				return -1;
+			continue;
+		}
+
+		// Audio source: all go to one shared destination media
+		if (!dst_audio_media) {
+			// check all audio sources in ht for an existing dest media
+			IQUEUE_FOREACH(srms, check_ms) {
+				if (!check_ms->media || check_ms->media->type_id != MT_AUDIO)
+					continue;
+				__auto_type old_ms = t_hash_table_lookup(ht, check_ms->media);
+				if (old_ms) {
+					dst_audio_media = old_ms->media;
+					break;
+				}
+			}
+			if (!dst_audio_media)
+				dst_audio_media = call_get_media(dst_ml, &src_media->type,
+						src_media->type_id, NULL, false,
+						dst_ml->medias->len + 1, mid_tracker_dst);
+		}
+
+		__add_media_subscription(dst_audio_media, src_media,
+				&(struct sink_attrs) { .egress = !!flags->egress });
+
+		if (flags->rtcp_mirror)
+			__add_media_subscription(src_media, dst_audio_media,
+				&(struct sink_attrs) { .egress = !!flags->egress, .rtcp_only = true });
+
+		// Initialize destination media properties from first audio source
+		if (!dst_audio_initialized) {
+			struct stream_params *sp = &src_media->sp;
+
+			media_init_from_flags(src_media, flags);
+			media_init_from_flags(dst_audio_media, flags);
+			media_set_echo(src_media, flags);
+			media_set_echo_reverse(dst_audio_media, flags);
+			media_set_siprec_label(dst_audio_media, flags, src_media->unique_id);
+			media_update_label(dst_audio_media, flags, &src_media->label);
+			media_update_type(dst_audio_media, sp);
+			media_set_protocol(dst_audio_media, src_media, sp, flags);
+			media_gen_media_id(dst_audio_media, flags);
+			media_update_flags(dst_audio_media, sp);
+			media_update_crypto(dst_audio_media, sp, flags);
+			media_copy_format(dst_audio_media, src_media);
+			media_set_address_family(dst_audio_media, src_media, flags);
+			media_set_ptime(src_media, sp, flags->rev_ptime, flags->ptime);
+			media_set_ptime(dst_audio_media, sp, flags->ptime, flags->rev_ptime);
+			media_set_extmap(dst_audio_media, &src_media->extmap, media_extmap_strip, flags);
+
+			codec_store_populate(&dst_audio_media->codecs, &src_media->codecs,
+					.allow_asymmetric = !!flags->allow_asymmetric_codecs);
+			codec_store_strip(&dst_audio_media->codecs, &flags->codec_strip, flags->codec_except);
+			codec_store_strip(&dst_audio_media->codecs, &flags->codec_consume, flags->codec_except);
+			codec_store_strip(&dst_audio_media->codecs, &flags->codec_mask, flags->codec_except);
+			codec_store_offer(&dst_audio_media->codecs, &flags->codec_offer, &sp->codecs);
+			codec_store_transcode(&dst_audio_media->codecs, &flags->codec_transcode, &sp->codecs);
+			codec_store_synthesise(&dst_audio_media->codecs, &src_media->codecs);
+
+			if (!flags->inactive)
+				bf_copy(&dst_audio_media->media_flags, MEDIA_FLAG_SEND,
+						&src_media->media_flags, SP_FLAG_RECV);
+			else
+				MEDIA_CLEAR(dst_audio_media, SEND);
+			MEDIA_CLEAR(dst_audio_media, RECV);
+
+			__rtcp_mux_set(flags, dst_audio_media);
+			__generate_crypto(flags, dst_audio_media, src_media);
+
+			unsigned int num_ports = proto_num_ports(sp->num_ports, dst_audio_media, flags, false);
+
+			__init_interface(dst_audio_media, &flags->interface, num_ports);
+			if (dst_audio_media->logical_intf == NULL)
+				return -1;
+
+			__ice_offer(flags, dst_audio_media, src_media,
+					ice_is_restart(src_media->ice_agent, sp));
+
+			struct endpoint_map *em = __get_endpoint_map(dst_audio_media, num_ports,
+					NULL, flags, true);
+			if (!em)
+				return -1;
+
+			__num_media_streams(dst_audio_media, num_ports);
+
+			if (!__init_streams(dst_audio_media, NULL, flags))
+				return -1;
+
+			dst_audio_initialized = true;
+		} else {
+			// Additional audio sources: init the source side
+			media_init_from_flags(src_media, flags);
+			media_set_echo(src_media, flags);
+			media_set_ptime(src_media, &src_media->sp, flags->rev_ptime, flags->ptime);
+		}
+
+		update_init_subscribers(src_media, NULL, NULL, flags->opmode);
+	}
+
+	// Activate audio_player for the mixed destination media
+	if (dst_audio_media) {
+		MEDIA_SET(dst_audio_media, MIX);
+		inject_reconfigure_destination_media(dst_audio_media, true, flags);
+	}
+
+	// Retire stale audio destination medias left over from a previous
+	// non-mix topology (e.g. non-mix had 2 audio dests, mix reuses 1).
+	// Without this, sdp_create() would emit extra m=audio lines.
+	for (unsigned int i = 0; i < dst_ml->medias->len; i++) {
+		struct call_media *media = dst_ml->medias->pdata[i];
+		if (!media)
+			continue;
+		if (media == dst_audio_media)
+			continue;
+		if (media->type_id != MT_AUDIO)
+			continue;
+		if (media->media_subscriptions.head)
+			continue;
+		__disable_streams(media, 0);
+		dst_ml->medias->pdata[i] = NULL;
+	}
+
+	// Compact: collapse NULL holes so the mixed destination always
+	// occupies the lowest index. Without this, monologue_subscribe_answer
+	// resolves streams by SDP index (1-based) via call_get_media, so a
+	// leading NULL hole causes it to allocate a new empty media instead
+	// of finding the active mixed destination.
+	unsigned int dst_idx = 0;
+	for (unsigned int src_idx = 0; src_idx < dst_ml->medias->len; src_idx++) {
+		struct call_media *media = dst_ml->medias->pdata[src_idx];
+		if (!media)
+			continue;
+		if (dst_idx != src_idx) {
+			dst_ml->medias->pdata[dst_idx] = media;
+			dst_ml->medias->pdata[src_idx] = NULL;
+			media->index = dst_idx + 1;
+		}
+		dst_idx++;
+	}
+	if (dst_idx < dst_ml->medias->len)
+		t_ptr_array_set_size(dst_ml->medias, dst_idx);
+
+	monologue_open_ports(dst_ml);
+	monologue_media_start(dst_ml);
+
+	return 0;
+}
+
 /* called with call->master_lock held in W */
 __attribute__((nonnull(1, 2, 3)))
 int monologue_subscribe_request(const subscription_q *srms, struct call_monologue *dst_ml, sdp_ng_flags *flags) {
+	if (flags->mix)
+		return monologue_subscribe_request_mix(srms, dst_ml, flags);
+
 	g_auto(subscription_store_ht) ht = subscription_store_ht_new();
 
 	__unsubscribe_medias_from_all(dst_ml, ht);
 	__call_monologue_init_from_flags(dst_ml, NULL, flags);
 
+	// Deactivate audio_player on previously mixed medias before reuse
+	for (unsigned int i = 0; i < dst_ml->medias->len; i++) {
+		struct call_media *media = dst_ml->medias->pdata[i];
+		if (!media)
+			continue;
+		if (MEDIA_ISSET(media, MIX)) {
+			MEDIA_CLEAR(media, AUDIO_PLAYER);
+			audio_player_stop(media);
+			MEDIA_CLEAR(media, MIX);
+		}
+	}
+
 	IQUEUE_FOREACH(srms, ms) {
 		struct call_media *src_media = ms->media;
 		if (!src_media)
@@ -4526,10 +4724,12 @@ int monologue_subscribe_answer(struct call_monologue *dst_ml, sdp_ng_flags *flag
 		/* set src_media based on subscription (assuming it is one-to-one)
 		 * TODO: this should probably be reworked to support one-to-multi subscriptions.
 		 */
-		__auto_type ms = dst_media->media_subscriptions.head;
-		if (!ms)
+		__auto_type head_ms = dst_media->media_subscriptions.head;
+		if (!head_ms)
 			continue;
-		struct call_media *src_media = ms->media;
+		struct call_media *src_media = head_ms->media;
+
+		bool is_mix = MEDIA_ISSET(dst_media, MIX);
 
 		rev_ms = call_get_media_subscription(src_media->media_subscribers_ht, dst_media);
 		if (rev_ms)
@@ -4545,7 +4745,7 @@ int monologue_subscribe_answer(struct call_monologue *dst_ml, sdp_ng_flags *flag
 		media_set_ptime(dst_media, sp, flags->ptime, 0);
 		media_update_extmap(dst_media, sp, NULL, flags);
 
-		if (flags->allow_transcoding) {
+		if (flags->allow_transcoding || is_mix) {
 			codec_store_populate(&dst_media->codecs, &sp->codecs,
 					.codec_set = flags->codec_set,
 					.answer_cs = &src_media->codecs,
@@ -4560,11 +4760,16 @@ int monologue_subscribe_answer(struct call_monologue *dst_ml, sdp_ng_flags *flag
 				return -1;
 		}
 
-		codec_handlers_update(src_media, dst_media, .flags = flags,
-				.allow_asymmetric = !!flags->allow_asymmetric_codecs);
-		codec_handlers_update(dst_media, src_media, .flags = flags, .sp = sp,
-				.allow_asymmetric = !!flags->allow_asymmetric_codecs,
-				.reset_transcoding = true);
+		if (is_mix) {
+			/* Mix mode: reconfigure audio_player with answer codecs for all subscriptions */
+			inject_reconfigure_destination_media(dst_media, true, flags);
+		} else {
+			codec_handlers_update(src_media, dst_media, .flags = flags,
+					.allow_asymmetric = !!flags->allow_asymmetric_codecs);
+			codec_handlers_update(dst_media, src_media, .flags = flags, .sp = sp,
+					.allow_asymmetric = !!flags->allow_asymmetric_codecs,
+					.reset_transcoding = true);
+		}
 
 		__dtls_logic(flags, dst_media, sp);
 
@@ -4578,15 +4783,26 @@ int monologue_subscribe_answer(struct call_monologue *dst_ml, sdp_ng_flags *flag
 		MEDIA_SET(dst_media, INITIALIZED);
 
 		update_init_subscribers(dst_media, sp, flags, flags->opmode);
-		update_init_subscribers(src_media, NULL, NULL, flags->opmode);
+
+		/* For mix mode, update all source subscriptions */
+		if (is_mix) {
+			IQUEUE_FOREACH(&dst_media->media_subscriptions, sub_ms) {
+				struct call_media *sub_src = sub_ms->media;
+				update_init_subscribers(sub_src, NULL, NULL, flags->opmode);
+				__media_unconfirm(sub_src, "subscribe answer event");
+				media_update_transcoding_flag(sub_src);
+			}
+		} else {
+			update_init_subscribers(src_media, NULL, NULL, flags->opmode);
+			__media_unconfirm(src_media, "subscribe answer event");
+			media_update_transcoding_flag(src_media);
+		}
 
 		__media_unconfirm(dst_media, "subscribe answer event");
-		__media_unconfirm(src_media, "subscribe answer event");
 
 		sdp_sp_move(&dst_media->sp, sp);
 
 		media_update_transcoding_flag(dst_media);
-		media_update_transcoding_flag(src_media);
 	}
 
 	monologue_media_start(dst_ml);

daemon/call_interfaces.c

@@ -1208,6 +1208,9 @@ void call_ng_flags_flags(str *s, unsigned int idx, helper_arg arg) {
 		case CSH_LOOKUP("media handover"):
 			out->media_handover = true;
 			break;
+		case CSH_LOOKUP("mix"):
+			out->mix = true;
+			break;
 		case CSH_LOOKUP("mirror-RTCP"):
 		case CSH_LOOKUP("mirror-rtcp"):
 		case CSH_LOOKUP("RTCP-mirror"):
@@ -4252,41 +4255,42 @@ const char *call_subscribe_request_ng(ng_command_ctx_t *ctx) {
 		if (!dest_media)
 			continue;
 
-		// each media should be subscribed to just one other media
 		if (!dest_media->media_subscriptions.length)
 			continue;
 
-		struct call_media *source_media = dest_media->media_subscriptions.head->media;
-		struct call_monologue *source_ml = source_media->monologue;
+		/* iterate all subscriptions per dest media (supports mix mode
+		 * where one dest media has multiple source subscriptions) */
+		IQUEUE_FOREACH(&dest_media->media_subscriptions, ms) {
+			struct call_media *source_media = ms->media;
+			struct call_monologue *source_ml = source_media->monologue;
+
+			parser->list_add_str_dup(from_list, &source_ml->tag);
+
+			if (media_labels.gen && dest_media->label.len) {
+				parser_arg label =
+					parser->dict_add_dict(media_labels, dest_media->label.s);
+				parser->dict_add_str(label, "tag", &source_ml->tag);
+				parser->dict_add_int(label, "index", source_media->index);
+				parser->dict_add_str(label, "type", &dest_media->type);
+				if (source_ml->label.len)
+					parser->dict_add_str(label, "label", &source_ml->label);
+				parser->dict_add_string(label, "mode", sdp_get_sendrecv(source_media));
+			}
 
-		parser->list_add_str_dup(from_list, &source_ml->tag);
+			if (tag_medias.gen) {
+				parser_arg tag_label = parser->list_add_dict(tag_medias);
+				parser->dict_add_str(tag_label, "tag", &source_ml->tag);
+				if (source_ml->label.len)
+					parser->dict_add_str(tag_label, "label", &source_ml->label);
 
-		if (media_labels.gen && dest_media->label.len) {
-			parser_arg label =
-				parser->dict_add_dict(media_labels, dest_media->label.s);
-			parser->dict_add_str(label, "tag", &source_ml->tag);
-			parser->dict_add_int(label, "index", source_media->index);
-			parser->dict_add_str(label, "type", &dest_media->type);
-			if (source_ml->label.len)
-				parser->dict_add_str(label, "label", &source_ml->label);
-			parser->dict_add_string(label, "mode", sdp_get_sendrecv(source_media));
-		}
+				parser_arg medias = parser->dict_add_list(tag_label, "medias");
 
-		if (tag_medias.gen) {
-			parser_arg tag_label = parser->list_add_dict(tag_medias);
-			parser->dict_add_str(tag_label, "tag", &source_ml->tag);
-			if (source_ml->label.len)
-				parser->dict_add_str(tag_label, "label", &source_ml->label);
-
-			parser_arg medias = parser->dict_add_list(tag_label, "medias");
-
-			// this is a bit strange because in this mode, each list can only
-			// ever get one entry...
-			parser_arg med_ent = parser->list_add_dict(medias);
-			parser->dict_add_int(med_ent, "index", source_media->index);
-			parser->dict_add_str(med_ent, "type", &dest_media->type);
-			parser->dict_add_str(med_ent, "label", &dest_media->label);
-			parser->dict_add_string(med_ent, "mode", sdp_get_sendrecv(source_media));
+				parser_arg med_ent = parser->list_add_dict(medias);
+				parser->dict_add_int(med_ent, "index", source_media->index);
+				parser->dict_add_str(med_ent, "type", &dest_media->type);
+				parser->dict_add_str(med_ent, "label", &dest_media->label);
+				parser->dict_add_string(med_ent, "mode", sdp_get_sendrecv(source_media));
+			}
 		}
 	}
 

include/call.h

@@ -219,6 +219,7 @@ enum {
 #define MEDIA_FLAG_SELECT_PT			(1LL << 36)
 #define MEDIA_FLAG_RECRYPT			(1LL << 37)
 #define MEDIA_FLAG_PUBLIC			(1LL << 38)
+#define MEDIA_FLAG_MIX				(1LL << 41)
 #define MEDIA_FLAG_EXTMAP_SHORT			SHARED_FLAG_EXTMAP_SHORT
 #define MEDIA_FLAG_BUNDLE_ONLY			(1LL << 40)
 

include/call_interfaces.h

@@ -240,6 +240,7 @@ RTPE_NG_FLAGS_STR_CASE_HT_PARAMS
 	             reset:1,
 		     egress:1,
 		     siprec:1,
+		     mix:1,
 	             fragment:1,
 	             record_call:1,
 		     discard_recording:1,