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

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

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

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

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

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

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

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

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

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

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

ASTERISK-27076

Change-Id: I8afd8dd2eb538806a39b887af0abd046266e14c7

3 files changed, 197 insertions, 36 deletions

diff --git a/include/asterisk/res_pjsip.h b/include/asterisk/res_pjsip.h
index b9c50adda..f907effcf 100644
--- a/include/asterisk/res_pjsip.h
+++ b/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;
 };
 
 /*!
diff --git a/include/asterisk/res_pjsip_session.h b/include/asterisk/res_pjsip_session.h
index e2a90662e..e298e1f32 100644
--- a/include/asterisk/res_pjsip_session.h
+++ b/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 {								\
diff --git a/include/asterisk/stream.h b/include/asterisk/stream.h
index 00169a3f1..4027231ed 100644
--- a/include/asterisk/stream.h
+++ b/include/asterisk/stream.h
@@ -290,6 +290,20 @@ 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
  *
  * \param topology The topology of streams
@@ -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.
  *