Revert "Revert "core_unreal / core_local: Add multistream and re-negotiation.""

This reverts commit ec8b3850d2.

Reason for revert: Time for this to return.

Change-Id: I5858f279bd523c609830c09333902b5d7d048aaf

include/asterisk/core_unreal.h

@@ -40,6 +40,7 @@ extern "C" {
 
 /* Forward declare some struct names */
 struct ast_format_cap;
+struct ast_stream_topology;
 
 /* ------------------------------------------------------------------- */
 
@@ -96,6 +97,7 @@ struct ast_unreal_pvt {
 	unsigned int flags;                         /*!< Private option flags */
 	/*! Base name of the unreal channels.  exten@context or other name. */
 	char name[AST_MAX_EXTENSION + AST_MAX_CONTEXT + 2];
+	struct ast_stream_topology *reqtopology;    /*!< Requested stream topology */
 };
 
 #define AST_UNREAL_IS_OUTBOUND(a, b) ((a) == (b)->chan ? 1 : 0)
@@ -146,6 +148,9 @@ struct ast_frame *ast_unreal_read(struct ast_channel *ast);
 /*! Unreal channel framework struct ast_channel_tech.write callback */
 int ast_unreal_write(struct ast_channel *ast, struct ast_frame *f);
 
+/*! Unreal channel framework struct ast_channel_tech.write_stream callback */
+int ast_unreal_write_stream(struct ast_channel *ast, int stream_num, struct ast_frame *f);
+
 /*! Unreal channel framework struct ast_channel_tech.indicate callback */
 int ast_unreal_indicate(struct ast_channel *ast, int condition, const void *data, size_t datalen);
 
@@ -187,6 +192,20 @@ void ast_unreal_destructor(void *vdoomed);
  */
 struct ast_unreal_pvt *ast_unreal_alloc(size_t size, ao2_destructor_fn destructor, struct ast_format_cap *cap);
 
+/*!
+ * \brief Allocate the base unreal struct for a derivative.
+ * \since 16.12.0
+ * \since 17.6.0
+ *
+ * \param size Size of the unreal struct to allocate.
+ * \param destructor Destructor callback.
+ * \param cap Format capabilities to give the unreal private struct.
+ *
+ * \retval pvt on success.
+ * \retval NULL on error.
+ */
+struct ast_unreal_pvt *ast_unreal_alloc_stream_topology(size_t size, ao2_destructor_fn destructor, struct ast_stream_topology *topology);
+
 /*!
  * \brief Create the semi1 and semi2 unreal channels.
  * \since 12.0.0

main/core_local.c

@@ -48,6 +48,8 @@
 #include "asterisk/stasis_channels.h"
 #include "asterisk/_private.h"
 #include "asterisk/stasis_channels.h"
+#include "asterisk/stream.h"
+#include "asterisk/translate.h"
 
 /*** DOCUMENTATION
 	<manager name="LocalOptimizeAway" language="en_US">
@@ -136,6 +138,7 @@ static const char tdesc[] = "Local Proxy Channel Driver";
 static struct ao2_container *locals;
 
 static struct ast_channel *local_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 *local_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 local_call(struct ast_channel *ast, const char *dest, int timeout);
 static int local_hangup(struct ast_channel *ast);
 static int local_devicestate(const char *data);
@@ -170,14 +173,15 @@ static struct ast_channel_tech local_tech = {
 	.type = "Local",
 	.description = tdesc,
 	.requester = local_request,
+	.requester_with_stream_topology = local_request_with_stream_topology,
 	.send_digit_begin = ast_unreal_digit_begin,
 	.send_digit_end = ast_unreal_digit_end,
 	.call = local_call,
 	.hangup = local_hangup,
 	.answer = ast_unreal_answer,
-	.read = ast_unreal_read,
+	.read_stream = ast_unreal_read,
 	.write = ast_unreal_write,
-	.write_video = ast_unreal_write,
+	.write_stream = ast_unreal_write_stream,
 	.exception = ast_unreal_read,
 	.indicate = ast_unreal_indicate,
 	.fixup = ast_unreal_fixup,
@@ -858,14 +862,14 @@ static void local_pvt_destructor(void *vdoomed)
 }
 
 /*! \brief Create a call structure */
-static struct local_pvt *local_alloc(const char *data, struct ast_format_cap *cap)
+static struct local_pvt *local_alloc(const char *data, struct ast_stream_topology *topology)
 {
 	struct local_pvt *pvt;
 	char *parse;
 	char *context;
 	char *opts;
 
-	pvt = (struct local_pvt *) ast_unreal_alloc(sizeof(*pvt), local_pvt_destructor, cap);
+	pvt = (struct local_pvt *) ast_unreal_alloc_stream_topology(sizeof(*pvt), local_pvt_destructor, topology);
 	if (!pvt) {
 		return NULL;
 	}
@@ -916,12 +920,95 @@ static struct local_pvt *local_alloc(const char *data, struct ast_format_cap *ca
 /*! \brief Part of PBX interface */
 static struct ast_channel *local_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 = local_request_with_stream_topology(type, topology, assignedids, requestor, data, cause);
+
+	ast_stream_topology_free(topology);
+
+	return chan;
+}
+
+/*! \brief Part of PBX interface */
+static struct ast_channel *local_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 ast_stream_topology *audio_filtered_topology;
+	int i;
 	struct local_pvt *p;
 	struct ast_channel *chan;
 	ast_callid callid;
 
+	/* Create a copy of the requested topology as we don't have ownership over
+	 * the one that is passed in.
+	 */
+	audio_filtered_topology = ast_stream_topology_clone(topology);
+	if (!audio_filtered_topology) {
+		return NULL;
+	}
+
+	/* Some users of Local channels request every known format in the
+	 * universe. The core itself automatically pruned this list down to a single
+	 * "best" format for audio in non-multistream. We replicate the logic here to
+	 * do the same thing.
+	 */
+	for (i = 0; i < ast_stream_topology_get_count(audio_filtered_topology); ++i) {
+		struct ast_stream *stream;
+		int res;
+		struct ast_format *tmp_fmt = NULL;
+		struct ast_format *best_audio_fmt = NULL;
+		struct ast_format_cap *caps;
+
+		stream = ast_stream_topology_get_stream(audio_filtered_topology, i);
+
+		if (ast_stream_get_type(stream) != AST_MEDIA_TYPE_AUDIO) {
+			continue;
+		}
+
+		/* Respect the immutable state of formats on the stream and create a new
+		 * format capabilities to replace the existing one.
+		 */
+		caps = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
+		if (!caps) {
+			ao2_ref(audio_filtered_topology, -1);
+			return NULL;
+		}
+
+		/* The ast_translator_best_choice function treats both caps as const
+		 * but does not declare it in the API.
+		 */
+		res = ast_translator_best_choice((struct ast_format_cap *)ast_stream_get_formats(stream), local_tech.capabilities,
+			&tmp_fmt, &best_audio_fmt);
+		if (res < 0) {
+			struct ast_str *tech_codecs = ast_str_alloca(AST_FORMAT_CAP_NAMES_LEN);
+			struct ast_str *request_codecs = ast_str_alloca(AST_FORMAT_CAP_NAMES_LEN);
+
+			ast_log(LOG_WARNING, "No translator path exists for channel type %s (native %s) to %s\n", type,
+				ast_format_cap_get_names(local_tech.capabilities, &tech_codecs),
+				ast_format_cap_get_names(ast_stream_get_formats(stream), &request_codecs));
+
+			/* If there are no formats then we abort */
+			ao2_ref(caps, -1);
+			ao2_ref(audio_filtered_topology, -1);
+			return NULL;
+		}
+
+		ast_format_cap_append(caps, best_audio_fmt, 0);
+		ast_stream_set_formats(stream, caps);
+
+		ao2_ref(caps, -1);
+		ao2_ref(tmp_fmt, -1);
+		ao2_ref(best_audio_fmt, -1);
+	}
+
 	/* Allocate a new private structure and then Asterisk channels */
-	p = local_alloc(data, cap);
+	p = local_alloc(data, audio_filtered_topology);
+	ao2_ref(audio_filtered_topology, -1);
 	if (!p) {
 		return NULL;
 	}
@@ -936,6 +1023,7 @@ static struct ast_channel *local_request(const char *type, struct ast_format_cap
 	return chan;
 }
 
+
 /*! \brief CLI command "local show channels" */
 static char *locals_show(struct ast_cli_entry *e, int cmd, struct ast_cli_args *a)
 {

main/core_unreal.c

@@ -40,6 +40,7 @@
 #include "asterisk/astobj2.h"
 #include "asterisk/bridge.h"
 #include "asterisk/core_unreal.h"
+#include "asterisk/stream.h"
 
 static unsigned int name_sequence = 0;
 
@@ -315,6 +316,11 @@ struct ast_frame  *ast_unreal_read(struct ast_channel *ast)
 }
 
 int ast_unreal_write(struct ast_channel *ast, struct ast_frame *f)
+{
+	return ast_unreal_write_stream(ast, -1, f);
+}
+
+int ast_unreal_write_stream(struct ast_channel *ast, int stream_num, struct ast_frame *f)
 {
 	struct ast_unreal_pvt *p = ast_channel_tech_pvt(ast);
 	int res = -1;
@@ -337,6 +343,9 @@ int ast_unreal_write(struct ast_channel *ast, struct ast_frame *f)
 		}
 	}
 
+	/* Update the frame to reflect the stream */
+	f->stream_num = stream_num;
+
 	/* Just queue for delivery to the other side */
 	ao2_ref(p, 1);
 	ao2_lock(p);
@@ -530,6 +539,86 @@ static int unreal_colp_redirect_indicate(struct ast_unreal_pvt *p, struct ast_ch
 	return res;
 }
 
+/*!
+ * \internal
+ * \brief Handle stream topology change request.
+ * \since 16.12.0
+ * \since 17.6.0
+ *
+ * \param p Unreal private structure.
+ * \param ast Channel indicating the condition.
+ * \param topology The requested topology.
+ *
+ * \retval 0 on success.
+ * \retval -1 on error.
+ */
+static int unreal_colp_stream_topology_request_change(struct ast_unreal_pvt *p, struct ast_channel *ast, const struct ast_stream_topology *topology)
+{
+	struct ast_stream_topology *this_channel_topology;
+	struct ast_stream_topology *the_other_channel_topology;
+	int i;
+	struct ast_stream *stream;
+	struct ast_channel *my_chan;
+	struct ast_channel *my_owner;
+	struct ast_channel *this_channel;
+	struct ast_channel *the_other_channel;
+	int res = 0;
+
+	this_channel_topology = ast_stream_topology_clone(topology);
+	if (!this_channel_topology) {
+		return -1;
+	}
+
+	the_other_channel_topology = ast_stream_topology_clone(topology);
+	if (!the_other_channel_topology) {
+		ast_stream_topology_free(this_channel_topology);
+		return -1;
+	}
+
+	/* We swap the stream state on the other channel because it is as if the channel is
+	 * connected to an external endpoint, so the perspective changes.
+	 */
+	for (i = 0; i < ast_stream_topology_get_count(the_other_channel_topology); ++i) {
+		stream = ast_stream_topology_get_stream(the_other_channel_topology, i);
+
+		if (ast_stream_get_state(stream) == AST_STREAM_STATE_RECVONLY) {
+			ast_stream_set_state(stream, AST_STREAM_STATE_SENDONLY);
+		} else if (ast_stream_get_state(stream) == AST_STREAM_STATE_SENDONLY) {
+			ast_stream_set_state(stream, AST_STREAM_STATE_RECVONLY);
+		}
+	}
+
+	ast_channel_unlock(ast);
+	ast_unreal_lock_all(p, &my_chan, &my_owner);
+	if (AST_UNREAL_IS_OUTBOUND(ast, p)) {
+		this_channel = p->chan;
+		the_other_channel = p->owner;
+	} else {
+		this_channel = p->owner;
+		the_other_channel = p->chan;
+	}
+	if (this_channel) {
+		ast_channel_set_stream_topology(this_channel, this_channel_topology);
+		ast_queue_control(this_channel, AST_CONTROL_STREAM_TOPOLOGY_CHANGED);
+	}
+	if (the_other_channel) {
+		ast_channel_set_stream_topology(the_other_channel, the_other_channel_topology);
+		ast_channel_stream_topology_changed_externally(the_other_channel);
+	}
+	if (my_chan) {
+		ast_channel_unlock(my_chan);
+		ast_channel_unref(my_chan);
+	}
+	if (my_owner) {
+		ast_channel_unlock(my_owner);
+		ast_channel_unref(my_owner);
+	}
+	ao2_unlock(p);
+	ast_channel_lock(ast);
+
+	return res;
+}
+
 int ast_unreal_indicate(struct ast_channel *ast, int condition, const void *data, size_t datalen)
 {
 	struct ast_unreal_pvt *p = ast_channel_tech_pvt(ast);
@@ -583,6 +672,11 @@ int ast_unreal_indicate(struct ast_channel *ast, int condition, const void *data
 		unreal_queue_indicate(p, ast, condition, data, datalen);
 		res = -1;
 		break;
+	case AST_CONTROL_STREAM_TOPOLOGY_REQUEST_CHANGE:
+		if (ast_channel_is_multistream(ast)) {
+			res = unreal_colp_stream_topology_request_change(p, ast, data);
+		}
+		break;
 	default:
 		res = unreal_queue_indicate(p, ast, condition, data, datalen);
 		break;
@@ -916,9 +1010,28 @@ void ast_unreal_destructor(void *vdoomed)
 
 	ao2_cleanup(doomed->reqcap);
 	doomed->reqcap = NULL;
+	ast_stream_topology_free(doomed->reqtopology);
+	doomed->reqtopology = NULL;
 }
 
 struct ast_unreal_pvt *ast_unreal_alloc(size_t size, ao2_destructor_fn destructor, struct ast_format_cap *cap)
+{
+	struct ast_stream_topology *topology;
+	struct ast_unreal_pvt *unreal;
+
+	topology = ast_stream_topology_create_from_format_cap(cap);
+	if (!topology) {
+		return NULL;
+	}
+
+	unreal = ast_unreal_alloc_stream_topology(size, destructor, topology);
+
+	ast_stream_topology_free(topology);
+
+	return unreal;
+}
+
+struct ast_unreal_pvt *ast_unreal_alloc_stream_topology(size_t size, ao2_destructor_fn destructor, struct ast_stream_topology *topology)
 {
 	struct ast_unreal_pvt *unreal;
 
@@ -935,12 +1048,17 @@ struct ast_unreal_pvt *ast_unreal_alloc(size_t size, ao2_destructor_fn destructo
 		return NULL;
 	}
 
-	unreal->reqcap = ast_format_cap_alloc(AST_FORMAT_CAP_FLAG_DEFAULT);
+	unreal->reqtopology = ast_stream_topology_clone(topology);
+	if (!unreal->reqtopology) {
+		ao2_ref(unreal, -1);
+		return NULL;
+	}
+
+	unreal->reqcap = ast_format_cap_from_stream_topology(topology);
 	if (!unreal->reqcap) {
 		ao2_ref(unreal, -1);
 		return NULL;
 	}
-	ast_format_cap_append_from_cap(unreal->reqcap, cap, AST_MEDIA_TYPE_UNKNOWN);
 
 	memcpy(&unreal->jb_conf, &jb_conf, sizeof(unreal->jb_conf));
 
@@ -958,6 +1076,7 @@ struct ast_channel *ast_unreal_new_channels(struct ast_unreal_pvt *p,
 	struct ast_assigned_ids id1 = {NULL, NULL};
 	struct ast_assigned_ids id2 = {NULL, NULL};
 	int generated_seqno = ast_atomic_fetchadd_int((int *) &name_sequence, +1);
+	struct ast_stream_topology *topology;
 
 	/* set unique ids for the two channels */
 	if (assignedids && !ast_strlen_zero(assignedids->uniqueid)) {
@@ -975,6 +1094,14 @@ struct ast_channel *ast_unreal_new_channels(struct ast_unreal_pvt *p,
 		id2.uniqueid = uniqueid2;
 	}
 
+	/* We need to create a topology to place on the first channel, as we can't
+	 * share a single one between both.
+	 */
+	topology = ast_stream_topology_clone(p->reqtopology);
+	if (!topology) {
+		return NULL;
+	}
+
 	/*
 	 * Allocate two new Asterisk channels
 	 *
@@ -987,6 +1114,7 @@ struct ast_channel *ast_unreal_new_channels(struct ast_unreal_pvt *p,
 		"%s/%s-%08x;1", tech->type, p->name, (unsigned)generated_seqno);
 	if (!owner) {
 		ast_log(LOG_WARNING, "Unable to allocate owner channel structure\n");
+		ast_stream_topology_free(topology);
 		return NULL;
 	}
 
@@ -1000,6 +1128,10 @@ struct ast_channel *ast_unreal_new_channels(struct ast_unreal_pvt *p,
 
 	ast_channel_nativeformats_set(owner, p->reqcap);
 
+	if (ast_channel_is_multistream(owner)) {
+		ast_channel_set_stream_topology(owner, topology);
+	}
+
 	/* Determine our read/write format and set it on each channel */
 	fmt = ast_format_cap_get_format(p->reqcap, 0);
 	if (!fmt) {
@@ -1054,6 +1186,11 @@ struct ast_channel *ast_unreal_new_channels(struct ast_unreal_pvt *p,
 
 	ast_channel_nativeformats_set(chan, p->reqcap);
 
+	if (ast_channel_is_multistream(chan)) {
+		ast_channel_set_stream_topology(chan, p->reqtopology);
+		p->reqtopology = NULL;
+	}
+
 	/* Format was already determined when setting up owner */
 	ast_channel_set_writeformat(chan, fmt);
 	ast_channel_set_rawwriteformat(chan, fmt);