/* GStreamer unix file-descriptor source/sink tests
 *
 * Copyright (C) 2023 Netflix Inc.
 *  Author: Xavier Claessens <xavier.claessens@collabora.com>
 *
 * 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 <gst/gst.h>
#include <gst/check/gstcheck.h>
#include <gst/app/app.h>
#include <glib/gstdio.h>

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;

GST_START_TEST (test_unixfd_copy)
{
  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 ! 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 system memory */
  GstSegment segment;
  gst_segment_init (&segment, GST_FORMAT_TIME);
  const char content[] = "Hello world!";
  GstBuffer *buf = gst_buffer_new_memdup (content, strlen (content));
  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);
  fail_unless (gst_buffer_memcmp (buf, 0, content, strlen (content)) == 0);
  gst_sample_unref (sample);

  /* 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;

GST_START_TEST (test_unixfd_big_payload)
{
  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 ! 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 1MB meta payload */
  gsize data_size = 1000000;
  GstBuffer *indata = gst_buffer_new_allocate (NULL, data_size, NULL);
  gst_buffer_memset (indata, 0, 0x42, data_size);

  GstSegment segment;
  gst_segment_init (&segment, GST_FORMAT_TIME);
  gst_meta_register_custom_simple ("test_unixfd_big_payload");
  const char content[] = "Hello world!";
  GstBuffer *buf = gst_buffer_new_memdup (content, strlen (content));
  GstCustomMeta *meta =
      gst_buffer_add_custom_meta (buf, "test_unixfd_big_payload");
  gst_structure_set (meta->structure, "data", GST_TYPE_BUFFER, indata, NULL);
  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);
  fail_unless (gst_buffer_memcmp (buf, 0, content, strlen (content)) == 0);
  meta = gst_buffer_get_custom_meta (buf, "test_unixfd_big_payload");
  fail_unless (meta != NULL);
  GstBuffer *outdata = NULL;
  gst_structure_get (meta->structure, "data", GST_TYPE_BUFFER, &outdata, NULL);
  gst_sample_unref (sample);

  /* Check the buffer we received is identical */
  GstMapInfo inmap;
  GstMapInfo outmap;
  gst_buffer_map (indata, &inmap, GST_MAP_READ);
  gst_buffer_map (outdata, &outmap, GST_MAP_READ);
  fail_unless (inmap.size == outmap.size);
  fail_unless (memcmp (inmap.data, outmap.data, inmap.size) == 0);
  gst_buffer_unmap (indata, &inmap);
  gst_buffer_unmap (outdata, &outmap);
  gst_buffer_unref (indata);
  gst_buffer_unref (outdata);

  /* 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);
  tcase_add_test (tc, test_unixfd_copy);
  tcase_add_test (tc, test_unixfd_big_payload);

  return s;
}

GST_CHECK_MAIN (unixfd);