a/gst: Add GStreamer helper code

Co-authored-by: Aaron Boxer <aaron.boxer@collabora.com>

src/xrt/auxiliary/CMakeLists.txt

@@ -69,6 +69,14 @@ if(XRT_HAVE_DBUS)
 		)
 endif()
 
+set(GSTREAMER_SOURCE_FILES
+	gstreamer/gst_internal.h
+	gstreamer/gst_sink.h
+	gstreamer/gst_sink.c
+	gstreamer/gst_pipeline.h
+	gstreamer/gst_pipeline.c
+	)
+
 set(TRACKING_SOURCE_FILES
 	tracking/t_data_utils.c
 	tracking/t_imu_fusion.hpp
@@ -221,6 +229,23 @@ if(ANDROID)
 	target_link_libraries(aux_util PUBLIC ${ANDROID_LOG_LIBRARY})
 endif()
 
+# GStreamer library.
+if(XRT_HAVE_GST)
+	add_library(aux_gstreamer STATIC ${GSTREAMER_SOURCE_FILES})
+	target_link_libraries(aux_gstreamer PUBLIC
+		aux-includes
+		)
+	target_link_libraries(aux_gstreamer PRIVATE
+		xrt-interfaces
+		aux_math
+		aux_os
+		${GST_LIBRARIES}
+		)
+	target_include_directories(aux_gstreamer PRIVATE
+		${GST_INCLUDE_DIRS}
+		)
+endif()
+
 # Tracking library.
 add_library(aux_tracking STATIC ${TRACKING_SOURCE_FILES})
 target_link_libraries(aux_tracking PUBLIC aux-includes PRIVATE aux_math)

src/xrt/auxiliary/gstreamer/gst_internal.h

@@ -0,0 +1,80 @@
+// Copyright 2019-2021, Collabora, Ltd.
+// SPDX-License-Identifier: BSL-1.0
+/*!
+ * @file
+ * @brief  Semi internal structs for gstreamer code.
+ * @author Jakob Bornecrantz <jakob@collabora.com>
+ * @ingroup aux_util
+ */
+
+#pragma once
+
+#include "xrt/xrt_frame.h"
+
+#include <gst/app/gstappsink.h>
+#include <gst/app/gstappsrc.h>
+
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+
+/*
+ *
+ * Pipeline
+ *
+ */
+
+/*!
+ * A pipeline from which you can create one or more @ref gstreamer_sink from.
+ *
+ * @implements xrt_frame_node
+ */
+struct gstreamer_pipeline
+{
+	struct xrt_frame_node node;
+
+	struct xrt_frame_context *xfctx;
+
+	GstElement *pipeline;
+};
+
+
+/*
+ *
+ * Sink
+ *
+ */
+
+/*!
+ * An @ref xrt_frame_sink that uses appsrc.
+ *
+ * @implements xrt_frame_sink
+ * @implements xrt_frame_node
+ */
+struct gstreamer_sink
+{
+	//! The base structure exposing the sink interface.
+	struct xrt_frame_sink base;
+
+	//! A sink can expose multie @ref xrt_frame_sink but only one node.
+	struct xrt_frame_node node;
+
+	//! Pipeline this sink is producing frames into.
+	struct gstreamer_pipeline *gp;
+
+	//! Offset applied to timestamps given to GStreamer.
+	uint64_t offset_ns;
+
+	//! Last sent timestamp, used to calculate duration.
+	uint64_t timestamp_ns;
+
+	//! Cached appsrc element.
+	GstElement *appsrc;
+};
+
+
+#ifdef __cplusplus
+}
+#endif

src/xrt/auxiliary/gstreamer/gst_pipeline.c

@@ -0,0 +1,152 @@
+// Copyright 2019-2021, Collabora, Ltd.
+// SPDX-License-Identifier: BSL-1.0
+/*!
+ * @file
+ * @brief  An @ref xrt_frame_sink that does gst things.
+ * @author Jakob Bornecrantz <jakob@collabora.com>
+ * @author Aaron Boxer <aaron.boxer@collabora.com>
+ * @ingroup aux_util
+ */
+
+#include "util/u_misc.h"
+#include "util/u_debug.h"
+
+#include "gstreamer/gst_internal.h"
+#include "gstreamer/gst_pipeline.h"
+
+
+/*
+ *
+ * Internal pipeline functions.
+ *
+ */
+
+static void
+break_apart(struct xrt_frame_node *node)
+{
+	struct gstreamer_pipeline *gp = container_of(node, struct gstreamer_pipeline, node);
+
+	/*
+	 * This function is called when we are shutting down, after returning
+	 * from this function you are not allowed to call any other nodes in the
+	 * graph. But it must be safe for other nodes to call any normal
+	 * functions on us. Once the context is done calling break_aprt on all
+	 * objects it will call destroy on them.
+	 */
+
+	(void)gp;
+}
+
+static void
+destroy(struct xrt_frame_node *node)
+{
+	struct gstreamer_pipeline *gp = container_of(node, struct gstreamer_pipeline, node);
+
+	/*
+	 * All of the nodes has been broken apart and none of our functions will
+	 * be called, it's now safe to destroy and free ourselves.
+	 */
+
+	free(gp);
+}
+
+
+/*
+ *
+ * Exported functions.
+ *
+ */
+
+void
+gstreamer_pipeline_play(struct gstreamer_pipeline *gp)
+{
+	U_LOG_D("Starting pipeline");
+
+	gst_element_set_state(gp->pipeline, GST_STATE_PLAYING);
+}
+
+void
+gstreamer_pipeline_stop(struct gstreamer_pipeline *gp)
+{
+	U_LOG_D("Stopping pipeline");
+
+	// Settle the pipeline.
+	U_LOG_T("Sending EOS");
+	gst_element_send_event(gp->pipeline, gst_event_new_eos());
+
+	// Wait for EOS message on the pipeline bus.
+	U_LOG_T("Waiting for EOS");
+	GstMessage *msg = NULL;
+	msg = gst_bus_timed_pop_filtered(GST_ELEMENT_BUS(gp->pipeline), GST_CLOCK_TIME_NONE,
+	                                 GST_MESSAGE_EOS | GST_MESSAGE_ERROR);
+	//! @todo Should check if we got an error message here or an eos.
+	(void)msg;
+
+	// Completely stop the pipeline.
+	U_LOG_T("Setting to NULL");
+	gst_element_set_state(gp->pipeline, GST_STATE_NULL);
+}
+
+void
+gstreamer_pipeline_create_from_string(struct xrt_frame_context *xfctx,
+                                      const char *pipeline_string,
+                                      struct gstreamer_pipeline **out_gp)
+{
+	gst_init(NULL, NULL);
+
+	struct gstreamer_pipeline *gp = U_TYPED_CALLOC(struct gstreamer_pipeline);
+	gp->node.break_apart = break_apart;
+	gp->node.destroy = destroy;
+	gp->xfctx = xfctx;
+
+	// Setup pipeline.
+	gp->pipeline = gst_parse_launch(pipeline_string, NULL);
+
+	/*
+	 * Add ourselves to the context so we are destroyed.
+	 * This is done once we know everything is completed.
+	 */
+	xrt_frame_context_add(xfctx, &gp->node);
+
+	*out_gp = gp;
+}
+
+void
+gstreamer_pipeline_create_autovideo_sink(struct xrt_frame_context *xfctx,
+                                         const char *appsrc_name,
+                                         struct gstreamer_pipeline **out_gp)
+{
+	gst_init(NULL, NULL);
+
+	struct gstreamer_pipeline *gp = U_TYPED_CALLOC(struct gstreamer_pipeline);
+	gp->node.break_apart = break_apart;
+	gp->node.destroy = destroy;
+	gp->xfctx = xfctx;
+
+	// Setup pipeline.
+	gp->pipeline = gst_pipeline_new("pipeline");
+	GstElement *appsrc = gst_element_factory_make("appsrc", appsrc_name);
+	GstElement *conv = gst_element_factory_make("videoconvert", "conv");
+	GstElement *scale = gst_element_factory_make("videoscale", "scale");
+	GstElement *videosink = gst_element_factory_make("autovideosink", "videosink");
+
+	gst_bin_add_many(GST_BIN(gp->pipeline), //
+	                 appsrc,                //
+	                 conv,                  //
+	                 scale,                 //
+	                 videosink,             //
+	                 NULL);
+	gst_element_link_many(appsrc,    //
+	                      conv,      //
+	                      scale,     //
+	                      videosink, //
+	                      NULL);
+
+	/*
+	 * Add ourselves to the context so we are destroyed.
+	 * This is done once we know everything is completed.
+	 */
+	xrt_frame_context_add(xfctx, &gp->node);
+
+	*out_gp = gp;
+}

src/xrt/auxiliary/gstreamer/gst_pipeline.h

@@ -0,0 +1,40 @@
+// Copyright 2019-2021, Collabora, Ltd.
+// SPDX-License-Identifier: BSL-1.0
+/*!
+ * @file
+ * @brief  Functions for creating @ref gstreamer_pipeline objects.
+ * @author Jakob Bornecrantz <jakob@collabora.com>
+ * @ingroup aux_util
+ */
+
+#pragma once
+
+#include "xrt/xrt_frame.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+
+struct gstreamer_pipeline;
+
+void
+gstreamer_pipeline_create_from_string(struct xrt_frame_context *xfctx,
+                                      const char *pipeline_string,
+                                      struct gstreamer_pipeline **out_gp);
+
+void
+gstreamer_pipeline_create_autovideo_sink(struct xrt_frame_context *xfctx,
+                                         const char *appsrc_name,
+                                         struct gstreamer_pipeline **out_gp);
+
+void
+gstreamer_pipeline_play(struct gstreamer_pipeline *gp);
+
+void
+gstreamer_pipeline_stop(struct gstreamer_pipeline *gp);
+
+
+#ifdef __cplusplus
+}
+#endif

src/xrt/auxiliary/gstreamer/gst_sink.c

@@ -0,0 +1,187 @@
+// Copyright 2019-2021, Collabora, Ltd.
+// SPDX-License-Identifier: BSL-1.0
+/*!
+ * @file
+ * @brief  An @ref xrt_frame_sink that does gst things.
+ * @author Jakob Bornecrantz <jakob@collabora.com>
+ * @author Aaron Boxer <aaron.boxer@collabora.com>
+ * @ingroup aux_util
+ */
+
+#include "util/u_misc.h"
+#include "util/u_debug.h"
+#include "util/u_format.h"
+
+#include "gstreamer/gst_sink.h"
+#include "gstreamer/gst_pipeline.h"
+#include "gstreamer/gst_internal.h"
+
+#include <assert.h>
+
+
+/*
+ *
+ * Internal sink functions.
+ *
+ */
+
+static void
+wrapped_buffer_destroy(gpointer data)
+{
+	struct xrt_frame *xf = (struct xrt_frame *)data;
+
+	U_LOG_T("Called");
+
+	xrt_frame_reference(&xf, NULL);
+}
+
+static void
+push_frame(struct xrt_frame_sink *xfs, struct xrt_frame *xf)
+{
+	struct gstreamer_sink *gs = (struct gstreamer_sink *)xfs;
+	GstBuffer *buffer;
+	GstFlowReturn ret;
+
+	U_LOG_T(
+	    "Called"
+	    "\n\tformat: %s"
+	    "\n\twidth: %u"
+	    "\n\theight: %u",
+	    u_format_str(xf->format), xf->width, xf->height);
+
+	/* We need to take a reference on the frame to keep it alive. */
+	struct xrt_frame *taken = NULL;
+	xrt_frame_reference(&taken, xf);
+
+	/* Wrap the frame that we now hold a reference to. */
+	buffer = gst_buffer_new_wrapped_full( //
+	    0,                                // GstMemoryFlags flags
+	    (gpointer)xf->data,               // gpointer data
+	    taken->size,                      // gsize maxsize
+	    0,                                // gsize offset
+	    taken->size,                      // gsize size
+	    taken,                            // gpointer user_data
+	    wrapped_buffer_destroy);          // GDestroyNotify notify
+
+	//! Get the timestampe from the frame.
+	uint64_t xtimestamp_ns = xf->timestamp;
+
+	// Use the first frame as offset.
+	if (gs->offset_ns == 0) {
+		gs->offset_ns = xtimestamp_ns;
+	}
+
+	// Need to be offset or gstreamer becomes sad.
+	GST_BUFFER_PTS(buffer) = xtimestamp_ns - gs->offset_ns;
+
+	// Duration is measured from last time stamp.
+	GST_BUFFER_DURATION(buffer) = xtimestamp_ns - gs->timestamp_ns;
+	gs->timestamp_ns = xtimestamp_ns;
+
+	// All done, send it to the gstreamer pipeline.
+	ret = gst_app_src_push_buffer((GstAppSrc *)gs->appsrc, buffer);
+	if (ret != GST_FLOW_OK) {
+		U_LOG_E("Got GST error '%i'", ret);
+	}
+}
+
+static void
+enough_data(GstElement *appsrc, gpointer udata)
+{
+	// Debugging code.
+	U_LOG_T("Called");
+}
+
+static void
+break_apart(struct xrt_frame_node *node)
+{
+	struct gstreamer_sink *gs = container_of(node, struct gstreamer_sink, node);
+
+	/*
+	 * This function is called when we are shutting down, after returning
+	 * from this function you are not allowed to call any other nodes in the
+	 * graph. But it must be safe for other nodes to call any normal
+	 * functions on us. Once the context is done calling break_aprt on all
+	 * objects it will call destroy on them.
+	 */
+
+	(void)gs;
+}
+
+static void
+destroy(struct xrt_frame_node *node)
+{
+	struct gstreamer_sink *gs = container_of(node, struct gstreamer_sink, node);
+
+	/*
+	 * All of the nodes has been broken apart and none of our functions will
+	 * be called, it's now safe to destroy and free ourselves.
+	 */
+
+	free(gs);
+}
+
+
+/*
+ *
+ * Exported functions.
+ *
+ */
+
+void
+gstreamer_sink_send_eos(struct gstreamer_sink *gs)
+{
+	gst_element_send_event(gs->appsrc, gst_event_new_eos());
+}
+
+void
+gstreamer_sink_create_with_pipeline(struct gstreamer_pipeline *gp,
+                                    uint32_t width,
+                                    uint32_t height,
+                                    enum xrt_format format,
+                                    const char *appsrc_name,
+                                    struct gstreamer_sink **out_gs,
+                                    struct xrt_frame_sink **out_xfs)
+{
+	const char *format_str = NULL;
+	switch (format) {
+	case XRT_FORMAT_R8G8B8: format_str = "RGB"; break;
+	case XRT_FORMAT_YUYV422: format_str = "YUY2"; break;
+	case XRT_FORMAT_L8: format_str = "GRAY8"; break;
+	default: assert(false); break;
+	}
+
+	struct gstreamer_sink *gs = U_TYPED_CALLOC(struct gstreamer_sink);
+	gs->base.push_frame = push_frame;
+	gs->node.break_apart = break_apart;
+	gs->node.destroy = destroy;
+	gs->gp = gp;
+	gs->appsrc = gst_bin_get_by_name(GST_BIN(gp->pipeline), appsrc_name);
+
+
+	GstCaps *caps = gst_caps_new_simple(      //
+	    "video/x-raw",                        //
+	    "format", G_TYPE_STRING, format_str,  //
+	    "width", G_TYPE_INT, width,           //
+	    "height", G_TYPE_INT, height,         //
+	    "framerate", GST_TYPE_FRACTION, 0, 1, //
+	    NULL);
+
+	g_object_set(G_OBJECT(gs->appsrc),                      //
+	             "caps", caps,                              //
+	             "stream-type", GST_APP_STREAM_TYPE_STREAM, //
+	             "format", GST_FORMAT_TIME,                 //
+	             "is-live", TRUE,                           //
+	             NULL);
+
+	g_signal_connect(G_OBJECT(gs->appsrc), "enough-data", G_CALLBACK(enough_data), gs);
+
+	/*
+	 * Add ourselves to the context so we are destroyed.
+	 * This is done once we know everything is completed.
+	 */
+	xrt_frame_context_add(gp->xfctx, &gs->node);
+
+	*out_gs = gs;
+	*out_xfs = &gs->base;
+}

src/xrt/auxiliary/gstreamer/gst_sink.h

@@ -0,0 +1,37 @@
+// Copyright 2019-2021, Collabora, Ltd.
+// SPDX-License-Identifier: BSL-1.0
+/*!
+ * @file
+ * @brief  @ref xrt_frame_sink that does gst things.
+ * @author Jakob Bornecrantz <jakob@collabora.com>
+ * @ingroup aux_util
+ */
+
+#pragma once
+
+#include "xrt/xrt_frame.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+struct gstreamer_sink;
+struct gstreamer_pipeline;
+
+
+void
+gstreamer_sink_send_eos(struct gstreamer_sink *gp);
+
+void
+gstreamer_sink_create_with_pipeline(struct gstreamer_pipeline *gp,
+                                    uint32_t width,
+                                    uint32_t height,
+                                    enum xrt_format format,
+                                    const char *appsrc_name,
+                                    struct gstreamer_sink **out_gs,
+                                    struct xrt_frame_sink **out_xfs);
+
+
+#ifdef __cplusplus
+}
+#endif