/* GStreamer unix file-descriptor source/sink tests * * Copyright (C) 2023 Netflix Inc. * Author: Xavier Claessens * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Library General Public * License as published by the Free Software Foundation; either * version 2 of the License, or (at your option) any later version. * * This library is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU * Library General Public License for more details. * * You should have received a copy of the GNU Library General Public * License along with this library; if not, write to the * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor, * Boston, MA 02110-1301, USA. */ #ifdef HAVE_CONFIG_H #include "config.h" #endif #include #include #include #include static void wait_preroll (GstElement * element) { GstStateChangeReturn state_res = gst_element_set_state (element, GST_STATE_PLAYING); fail_unless (state_res != GST_STATE_CHANGE_FAILURE); state_res = gst_element_get_state (element, NULL, NULL, GST_CLOCK_TIME_NONE); fail_unless (state_res == GST_STATE_CHANGE_SUCCESS); } static GstPadProbeReturn buffer_pad_probe_cb (GstPad * pad, GstPadProbeInfo * info, gpointer user_data) { GstBuffer *buffer = GST_PAD_PROBE_INFO_BUFFER (info); if (buffer != NULL) { GstCustomMeta *cmeta = gst_buffer_add_custom_meta (buffer, "unix-fd-custom-meta"); GstStructure *s = gst_custom_meta_get_structure (cmeta); gst_structure_set (s, "field", G_TYPE_INT, 42, NULL); } return GST_PAD_PROBE_OK; } GST_START_TEST (test_unixfd_videotestsrc) { GError *error = NULL; const gchar *tags[] = { NULL }; gst_meta_register_custom ("unix-fd-custom-meta", tags, NULL, NULL, NULL); /* Ensure we don't have socket from previous failed test */ gchar *tempdir = g_dir_make_tmp ("unixfd-test-XXXXXX", &error); g_assert_no_error (error); gchar *socket_path = g_strdup_printf ("%s/socket", tempdir); /* Setup source */ gchar *pipeline_str = g_strdup_printf ("videotestsrc name=src ! unixfdsink socket-path=%s", socket_path); GstElement *pipeline_service = gst_parse_launch (pipeline_str, &error); g_assert_no_error (error); g_free (pipeline_str); /* Add a custom meta on each buffer */ GstElement *src = gst_bin_get_by_name (GST_BIN (pipeline_service), "src"); GstPad *pad = gst_element_get_static_pad (src, "src"); gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BUFFER, buffer_pad_probe_cb, NULL, NULL); gst_object_unref (src); gst_object_unref (pad); wait_preroll (pipeline_service); /* Setup sink */ pipeline_str = g_strdup_printf ("unixfdsrc socket-path=%s ! fakesink name=sink", socket_path); GstElement *pipeline_client_1 = gst_parse_launch (pipeline_str, &error); g_assert_no_error (error); wait_preroll (pipeline_client_1); /* disconnect, reconnect */ fail_unless (gst_element_set_state (pipeline_client_1, GST_STATE_READY) == GST_STATE_CHANGE_SUCCESS); wait_preroll (pipeline_client_1); /* Connect 2nd sink */ GstElement *pipeline_client_2 = gst_parse_launch (pipeline_str, &error); g_assert_no_error (error); wait_preroll (pipeline_client_2); /* Check we received our custom meta */ GstSample *sample; GstElement *sink = gst_bin_get_by_name (GST_BIN (pipeline_client_2), "sink"); g_object_get (sink, "last-sample", &sample, NULL); fail_unless (sample); GstBuffer *buffer = gst_sample_get_buffer (sample); GstCustomMeta *cmeta = gst_buffer_get_custom_meta (buffer, "unix-fd-custom-meta"); fail_unless (cmeta); GstStructure *s = gst_custom_meta_get_structure (cmeta); gint value; fail_unless (gst_structure_get_int (s, "field", &value)); fail_unless_equals_int (value, 42); gst_object_unref (sink); gst_sample_unref (sample); /* Teardown */ fail_unless (gst_element_set_state (pipeline_client_1, GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS); fail_unless (gst_element_set_state (pipeline_client_2, GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS); fail_unless (gst_element_set_state (pipeline_service, GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS); fail_if (g_file_test (socket_path, G_FILE_TEST_EXISTS)); g_rmdir (tempdir); g_free (tempdir); gst_object_unref (pipeline_service); gst_object_unref (pipeline_client_1); gst_object_unref (pipeline_client_2); g_free (socket_path); g_free (pipeline_str); } GST_END_TEST; GST_START_TEST (test_unixfd_segment) { GError *error = NULL; /* Ensure we don't have socket from previous failed test */ gchar *tempdir = g_dir_make_tmp ("unixfd-test-XXXXXX", &error); g_assert_no_error (error); gchar *socket_path = g_strdup_printf ("%s/socket", tempdir); GstCaps *caps = gst_caps_new_empty_simple ("video/x-raw"); /* Setup service */ gchar *pipeline_str = g_strdup_printf ("appsrc name=src format=time handle-segment-change=true ! unixfdsink socket-path=%s sync=false async=false wait-for-connection=true", socket_path); GstElement *pipeline_service = gst_parse_launch (pipeline_str, &error); g_assert_no_error (error); fail_unless (gst_element_set_state (pipeline_service, GST_STATE_PLAYING) == GST_STATE_CHANGE_SUCCESS); GstElement *appsrc = gst_bin_get_by_name (GST_BIN (pipeline_service), "src"); gst_object_unref (appsrc); g_free (pipeline_str); /* Setup client */ pipeline_str = g_strdup_printf ("unixfdsrc socket-path=%s ! appsink name=sink sync=false async=false", socket_path); GstElement *pipeline_client = gst_parse_launch (pipeline_str, &error); g_assert_no_error (error); fail_unless (gst_element_set_state (pipeline_client, GST_STATE_PLAYING) == GST_STATE_CHANGE_SUCCESS); GstElement *appsink = gst_bin_get_by_name (GST_BIN (pipeline_client), "sink"); gst_object_unref (appsink); g_free (pipeline_str); /* Send a buffer with PTS=30s */ GstSegment segment; gst_segment_init (&segment, GST_FORMAT_TIME); GstBuffer *buf = gst_buffer_new (); GST_BUFFER_PTS (buf) = 30 * GST_SECOND; GstSample *sample = gst_sample_new (buf, caps, &segment, NULL); gst_app_src_push_sample (GST_APP_SRC (appsrc), sample); gst_sample_unref (sample); gst_buffer_unref (buf); /* Wait for it */ sample = gst_app_sink_pull_sample (GST_APP_SINK (appsink)); buf = gst_sample_get_buffer (sample); GstClockTime first_pts = GST_BUFFER_PTS (buf); gst_sample_unref (sample); /* Send a buffer with PTS=1s but with 30s offset in the segment */ segment.base = 30 * GST_SECOND; buf = gst_buffer_new (); GST_BUFFER_PTS (buf) = 1 * GST_SECOND; sample = gst_sample_new (buf, caps, &segment, NULL); gst_app_src_push_sample (GST_APP_SRC (appsrc), sample); gst_sample_unref (sample); gst_buffer_unref (buf); /* Wait for it */ sample = gst_app_sink_pull_sample (GST_APP_SINK (appsink)); buf = gst_sample_get_buffer (sample); GstClockTime second_pts = GST_BUFFER_PTS (buf); gst_sample_unref (sample); /* They should be 1s appart */ fail_unless_equals_uint64 (second_pts - first_pts, GST_SECOND); /* Teardown */ fail_unless (gst_element_set_state (pipeline_client, GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS); fail_unless (gst_element_set_state (pipeline_service, GST_STATE_NULL) == GST_STATE_CHANGE_SUCCESS); g_rmdir (tempdir); g_free (tempdir); gst_object_unref (pipeline_service); gst_object_unref (pipeline_client); g_free (socket_path); gst_caps_unref (caps); } GST_END_TEST; static Suite * unixfd_suite (void) { Suite *s = suite_create ("unixfd"); TCase *tc = tcase_create ("unixfd"); suite_add_tcase (s, tc); tcase_add_test (tc, test_unixfd_videotestsrc); tcase_add_test (tc, test_unixfd_segment); return s; } GST_CHECK_MAIN (unixfd);