libobs: Add a packet callback mechanism

Packet callbacks are invoked in `send_interleaved()` and
are useful for any plugin to extend functionality at the
packet processing level without needing to modify code in
libobs. Closed caption support is one candidate that is
suitable for migration to a packet callback.

The packet callback also supports the new encoder packet
timing feature. This means a registered callback will have
access to both the compressed encoder packet and the associated
encoder packet timing information.

docs/sphinx/reference-outputs.rst

@@ -734,6 +734,44 @@ General Output Functions
 
 ---------------------
 
+.. function:: void obs_output_add_packet_callback(obs_output_t *output, void (*packet_cb)(obs_output_t *output,
+              struct encoder_packet *pkt, struct encoder_packet_time *pkt_time, void *param), void *param)
+
+   Register a packet callback function for the output. The callback is invoked for each compressed
+   packet just before sending to the service. This packet callback mechanism is the preferred method
+   for all packet-level processing that is not required to be implemented in libobs. Any reallocation
+   of the packet buffer, if necessary, must be done with functions in `libobs\util\bmem.h`, otherwise
+   a memory leak may occur. Never use `memset()` to clear the packet buffer, as the buffer data is
+   needed for subsequent callback processing.
+
+   :param output:     The output to register the packet_cb() function against
+   :param packet_cb:  Function pointer to the callback function
+   :param param:      Data passed to the callback
+   :return:           When the callback is added
+
+   packet_cb() arguments:
+   :param output:     The output associated with the invoked callback function
+   :param pkt:        Compressed data packet (audio or video)
+   :param pkt_time:   encoder_packet_time structure associated with the data packet
+   :param param:      Data passed to the callback
+
+   .. versionadded:: 31.0
+
+---------------------
+
+.. function:: void obs_output_remove_packet_callback(obs_output_t *output, void (*packet_cb)(obs_output_t *output,
+              struct encoder_packet *pkt, struct encoder_packet_time *pkt_time, void *param), void *param)
+
+   Remove a packet callback function for the output, that had been previously registered with
+   `obs_output_add_packet_callback()`.
+
+   :param output:     The output to remove the packet_cb() function against
+   :param packet_cb:  Function pointer to the callback function
+   :param param:      Data passed to the callback
+   :return:           When the callback is removed
+
+   .. versionadded:: 31.0
+
 Functions used by outputs
 -------------------------
 

libobs/obs-internal.h

@@ -74,6 +74,12 @@ struct rendered_callback {
 	void *param;
 };
 
+struct packet_callback {
+	void (*packet_cb)(obs_output_t *output, struct encoder_packet *pkt,
+			  struct encoder_packet_time *pkt_time, void *param);
+	void *param;
+};
+
 /* ------------------------------------------------------------------------- */
 /* validity checks */
 
@@ -1190,7 +1196,12 @@ struct obs_output {
 	// captions are output per track
 	struct caption_track_data *caption_tracks[MAX_OUTPUT_VIDEO_ENCODERS];
 
-	DARRAY(struct encoder_packet_time) encoder_packet_times[MAX_OUTPUT_VIDEO_ENCODERS];
+	DARRAY(struct encoder_packet_time)
+	encoder_packet_times[MAX_OUTPUT_VIDEO_ENCODERS];
+
+	/* Packet callbacks */
+	pthread_mutex_t pkt_callbacks_mutex;
+	DARRAY(struct packet_callback) pkt_callbacks;
 
 	bool valid;
 

libobs/obs-output.c

@@ -178,6 +178,7 @@ obs_output_t *obs_output_create(const char *id, const char *name,
 	pthread_mutex_init_value(&output->interleaved_mutex);
 	pthread_mutex_init_value(&output->delay_mutex);
 	pthread_mutex_init_value(&output->pause.mutex);
+	pthread_mutex_init_value(&output->pkt_callbacks_mutex);
 
 	if (pthread_mutex_init(&output->interleaved_mutex, NULL) != 0)
 		goto fail;
@@ -185,6 +186,8 @@ obs_output_t *obs_output_create(const char *id, const char *name,
 		goto fail;
 	if (pthread_mutex_init(&output->pause.mutex, NULL) != 0)
 		goto fail;
+	if (pthread_mutex_init(&output->pkt_callbacks_mutex, NULL) != 0)
+		goto fail;
 	if (os_event_init(&output->stopping_event, OS_EVENT_TYPE_MANUAL) != 0)
 		goto fail;
 	if (!init_output_handlers(output, name, settings, hotkey_data))
@@ -312,12 +315,15 @@ void obs_output_destroy(obs_output_t *output)
 		for (size_t i = 0; i < MAX_OUTPUT_VIDEO_ENCODERS; i++)
 			da_free(output->encoder_packet_times[i]);
 
+		da_free(output->pkt_callbacks);
+
 		clear_raw_audio_buffers(output);
 
 		os_event_destroy(output->stopping_event);
 		pthread_mutex_destroy(&output->pause.mutex);
 		pthread_mutex_destroy(&output->interleaved_mutex);
 		pthread_mutex_destroy(&output->delay_mutex);
+		pthread_mutex_destroy(&output->pkt_callbacks_mutex);
 		os_event_destroy(output->reconnect_stop_event);
 		obs_context_data_free(&output->context);
 		deque_free(&output->delay_data);
@@ -1825,6 +1831,21 @@ static inline void send_interleaved(struct obs_output *output)
 		}
 	}
 
+	/* Iterate the registered packet callback(s) and invoke
+	 * each one. The caption track logic further above should
+	 * eventually migrate to the packet callback mechanism.
+	 */
+	pthread_mutex_lock(&output->pkt_callbacks_mutex);
+	for (size_t i = 0; i < output->pkt_callbacks.num; ++i) {
+		struct packet_callback *const callback =
+			&output->pkt_callbacks.array[i];
+		// Packet interleave request timestamp
+		ept_local.pir = os_gettime_ns();
+		callback->packet_cb(output, &out, found_ept ? &ept_local : NULL,
+				    callback->param);
+	}
+	pthread_mutex_unlock(&output->pkt_callbacks_mutex);
+
 	output->info.encoded_packet(output->context.data, &out);
 	obs_encoder_packet_release(&out);
 }
@@ -3384,3 +3405,29 @@ const char *obs_get_output_supported_audio_codecs(const char *id)
 	const struct obs_output_info *info = find_output(id);
 	return info ? info->encoded_audio_codecs : NULL;
 }
+
+void obs_output_add_packet_callback(
+	obs_output_t *output,
+	void (*packet_cb)(obs_output_t *output, struct encoder_packet *pkt,
+			  struct encoder_packet_time *pkt_time, void *param),
+	void *param)
+{
+	struct packet_callback data = {packet_cb, param};
+
+	pthread_mutex_lock(&output->pkt_callbacks_mutex);
+	da_insert(output->pkt_callbacks, 0, &data);
+	pthread_mutex_unlock(&output->pkt_callbacks_mutex);
+}
+
+void obs_output_remove_packet_callback(
+	obs_output_t *output,
+	void (*packet_cb)(obs_output_t *output, struct encoder_packet *pkt,
+			  struct encoder_packet_time *pkt_time, void *param),
+	void *param)
+{
+	struct packet_callback data = {packet_cb, param};
+
+	pthread_mutex_lock(&output->pkt_callbacks_mutex);
+	da_erase_item(output->pkt_callbacks, &data);
+	pthread_mutex_unlock(&output->pkt_callbacks_mutex);
+}

libobs/obs.h

@@ -2299,6 +2299,24 @@ EXPORT const char *obs_get_output_supported_video_codecs(const char *id);
 
 EXPORT const char *obs_get_output_supported_audio_codecs(const char *id);
 
+/* Add/remove packet-processing callbacks that are invoked in
+ * send_interleaved(), before forwarding packets to the output service.
+ * This provides a mechanism to perform packet processing outside of
+ * libobs, however any callback function registering with this API should keep
+ * keep code to a minimum and understand it is running synchronously with the
+ * calling thread.
+ */
+EXPORT void obs_output_add_packet_callback(
+	obs_output_t *output,
+	void (*packet_cb)(obs_output_t *output, struct encoder_packet *pkt,
+			  struct encoder_packet_time *pkt_time, void *param),
+	void *param);
+EXPORT void obs_output_remove_packet_callback(
+	obs_output_t *output,
+	void (*packet_cb)(obs_output_t *output, struct encoder_packet *pkt,
+			  struct encoder_packet_time *pkt_time, void *param),
+	void *param);
+
 /* ------------------------------------------------------------------------- */
 /* Functions used by outputs */