tests/check/pipelines/tcp.c - imx-gst-plugins-base - Git at Google

 /* GStreamer
  *
  * Copyright (C) 2014 William Manley <will@williammanley.net>
  *
  * 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 <gio/gio.h>
 #include <gst/check/gstcheck.h>
 #include <gst/app/gstappsink.h>
 #include <gst/app/gstappsrc.h>
 #include <gst/net/gstnetcontrolmessagemeta.h>

 #ifdef HAVE_GIO_UNIX_2_0
 #include <gio/gunixfdmessage.h>
 #endif /*  HAVE_GIO_UNIX_2_0 */

 #include <sys/types.h>
 #include <sys/stat.h>
 #include <sys/socket.h>
 #include <unistd.h>

 static gboolean
 g_socketpair (GSocketFamily family, GSocketType type, GSocketProtocol protocol,
     GSocket * gsv[2], GError ** error);

 typedef struct
 {
   GstElement *sink;
   GstElement *src;

   GstPipeline *sink_pipeline;
   GstPipeline *src_pipeline;
   GstAppSrc *sink_src;
   GstAppSink *src_sink;
 } SymmetryTest;

 static void
 symmetry_test_setup (SymmetryTest * st, GstElement * sink, GstElement * src)
 {
   GstCaps *caps;
   st->sink = sink;
   g_object_set (sink, "sync", FALSE, NULL);
   st->src = src;

   st->sink_pipeline = GST_PIPELINE (gst_pipeline_new (NULL));
   st->src_pipeline = GST_PIPELINE (gst_pipeline_new (NULL));

   st->sink_src = GST_APP_SRC (gst_element_factory_make ("appsrc", NULL));
   fail_unless (st->sink_src != NULL);
   caps = gst_caps_from_string ("application/x-gst-check");
   gst_app_src_set_caps (st->sink_src, caps);
   gst_caps_unref (caps);

   gst_bin_add_many (GST_BIN (st->sink_pipeline), GST_ELEMENT (st->sink_src),
       st->sink, NULL);
   fail_unless (gst_element_link_many (GST_ELEMENT (st->sink_src), st->sink,
           NULL));

   st->src_sink = GST_APP_SINK (gst_element_factory_make ("appsink", NULL));
   fail_unless (st->src_sink != NULL);
   gst_bin_add_many (GST_BIN (st->src_pipeline), st->src,
       GST_ELEMENT (st->src_sink), NULL);
   fail_unless (gst_element_link_many (st->src, GST_ELEMENT (st->src_sink),
           NULL));

   gst_element_set_state (GST_ELEMENT (st->sink_pipeline), GST_STATE_PLAYING);
   gst_element_set_state (GST_ELEMENT (st->src_pipeline), GST_STATE_PLAYING);
 }

 static void
 symmetry_test_teardown (SymmetryTest * st)
 {
   fail_unless (gst_element_set_state (GST_ELEMENT (st->sink_pipeline),
           GST_STATE_NULL) != GST_STATE_CHANGE_FAILURE);
   fail_unless (gst_element_set_state (GST_ELEMENT (st->src_pipeline),
           GST_STATE_NULL) != GST_STATE_CHANGE_FAILURE);

   gst_object_unref (st->sink_pipeline);
   gst_object_unref (st->src_pipeline);

   memset (st, 0, sizeof (*st));
 }

 static void
 symmetry_test_assert_passthrough (SymmetryTest * st, GstBuffer * in)
 {
   gpointer copy;
   gsize data_size;
   GstSample *out;

   gst_buffer_extract_dup (in, 0, -1, &copy, &data_size);

   fail_unless (gst_app_src_push_buffer (st->sink_src, in) == GST_FLOW_OK);
   in = NULL;
   out = gst_app_sink_pull_sample (st->src_sink);
   fail_unless (out != NULL);

   fail_unless (gst_buffer_get_size (gst_sample_get_buffer (out)) == data_size);
   fail_unless (gst_buffer_memcmp (gst_sample_get_buffer (out), 0, copy,
           data_size) == 0);
   g_free (copy);
   gst_sample_unref (out);
 }

 static gboolean
 g_socketpair (GSocketFamily family, GSocketType type, GSocketProtocol protocol,
     GSocket * gsv[2], GError ** error)
 {
   int ret;
   int sv[2];

   ret = socketpair (family, type, protocol, sv);
   if (ret != 0) {
     g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED, "socketpair failed: %s",
         g_strerror (errno));
     return FALSE;
   }

   gsv[0] = g_socket_new_from_fd (sv[0], error);
   if (gsv[0] == NULL) {
     close (sv[0]);
     close (sv[1]);
     return FALSE;
   }
   gsv[1] = g_socket_new_from_fd (sv[1], error);
   if (gsv[1] == NULL) {
     g_object_unref (gsv[0]);
     gsv[0] = NULL;
     close (sv[1]);
     return FALSE;
   }
   return TRUE;
 }

 static void
 setup_multisocketsink_and_socketsrc (SymmetryTest * st)
 {
   GSocket *sockets[2] = { NULL, NULL };
   GError *err = NULL;

   st->sink = gst_check_setup_element ("multisocketsink");
   st->src = gst_check_setup_element ("socketsrc");

   fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
           G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, sockets, &err));

   g_object_set (st->src, "socket", sockets[0], NULL);
   g_object_unref (sockets[0]);
   sockets[0] = NULL;

   symmetry_test_setup (st, st->sink, st->src);

   g_signal_emit_by_name (st->sink, "add", sockets[1], NULL);
   g_object_unref (sockets[1]);
   sockets[1] = NULL;
 }

 GST_START_TEST (test_that_socketsrc_and_multisocketsink_are_symmetrical)
 {
   SymmetryTest st = { 0 };
   setup_multisocketsink_and_socketsrc (&st);
   symmetry_test_assert_passthrough (&st,
       gst_buffer_new_wrapped (g_strdup ("hello"), 5));
   symmetry_test_teardown (&st);
 }

 GST_END_TEST;


 GST_START_TEST (test_that_tcpclientsink_and_tcpserversrc_are_symmetrical)
 {
   SymmetryTest st = { 0 };
   GstElement *serversrc = gst_check_setup_element ("tcpserversrc");

   gst_element_set_state (serversrc, GST_STATE_PAUSED);
   symmetry_test_setup (&st, gst_check_setup_element ("tcpclientsink"),
       serversrc);

   symmetry_test_assert_passthrough (&st,
       gst_buffer_new_wrapped (g_strdup ("hello"), 5));

   symmetry_test_teardown (&st);
 }

 GST_END_TEST;


 GST_START_TEST (test_that_tcpserversink_and_tcpclientsrc_are_symmetrical)
 {
   SymmetryTest st = { 0 };

   symmetry_test_setup (&st, gst_check_setup_element ("tcpserversink"),
       gst_check_setup_element ("tcpclientsrc"));

   symmetry_test_assert_passthrough (&st,
       gst_buffer_new_wrapped (g_strdup ("hello"), 5));
   symmetry_test_teardown (&st);
 }

 GST_END_TEST;

 static void
 on_connection_closed (GstElement * socketsrc, gpointer user_data)
 {
   GSocket *socket = (GSocket *) user_data;

   g_object_set (socketsrc, "socket", socket, NULL);
 }

 GST_START_TEST (test_that_we_can_provide_new_socketsrc_sockets_during_signal)
 {
   GSocket *sockets[4] = { NULL, NULL };

   GstPipeline *pipeline = NULL;
   GstAppSink *appsink = NULL;
   GstElement *socketsrc = NULL;
   GstSample *sample = NULL;

   socketsrc = gst_check_setup_element ("socketsrc");

   fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
           G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, &sockets[0], NULL));

   fail_unless (g_socket_send (sockets[0], "hello", 5, NULL, NULL) == 5);
   fail_unless (g_socket_shutdown (sockets[0], FALSE, TRUE, NULL));

   fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
           G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, &sockets[2], NULL));
   fail_unless (g_socket_send (sockets[2], "goodbye", 7, NULL, NULL) == 7);
   fail_unless (g_socket_shutdown (sockets[2], FALSE, TRUE, NULL));

   g_object_set (socketsrc, "socket", sockets[1], NULL);

   g_signal_connect (socketsrc, "connection-closed-by-peer",
       G_CALLBACK (on_connection_closed), sockets[3]);

   pipeline = (GstPipeline *) gst_pipeline_new (NULL);
   appsink = GST_APP_SINK (gst_check_setup_element ("appsink"));
   gst_bin_add_many (GST_BIN (pipeline), socketsrc, GST_ELEMENT (appsink), NULL);
   fail_unless (gst_element_link_many (socketsrc, GST_ELEMENT (appsink), NULL));

   gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_PLAYING);

   fail_unless ((sample = gst_app_sink_pull_sample (appsink)) != NULL);
   gst_buffer_memcmp (gst_sample_get_buffer (sample), 0, "hello", 5);
   gst_sample_unref (sample);

   fail_unless ((sample = gst_app_sink_pull_sample (appsink)) != NULL);
   gst_buffer_memcmp (gst_sample_get_buffer (sample), 0, "goodbye", 7);
   gst_sample_unref (sample);

   fail_unless (NULL == gst_app_sink_pull_sample (appsink));
   fail_unless (gst_app_sink_is_eos (appsink));

   gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_NULL);
   g_clear_object (&sockets[0]);
   g_clear_object (&sockets[1]);
   g_clear_object (&sockets[2]);
   g_clear_object (&sockets[3]);
   gst_object_unref (pipeline);
 }

 GST_END_TEST
 #ifdef HAVE_GIO_UNIX_2_0
 static GSocketControlMessage *
 get_control_message_meta (GstBuffer * buf)
 {
   GstMeta *meta;
   gpointer iter_state = NULL;

   while ((meta = gst_buffer_iterate_meta (buf, &iter_state)) != NULL) {
     if (meta->info->api == GST_NET_CONTROL_MESSAGE_META_API_TYPE)
       return ((GstNetControlMessageMeta *) meta)->message;
   }
   fail ("Expected GSocketControlMessage attached to buffer");
   return NULL;
 }


 GST_START_TEST (test_that_multisocketsink_and_socketsrc_preserve_meta)
 {
   GstBuffer *buf;
   GSocketControlMessage *msg;
   SymmetryTest st = { 0 };
   char tmpfilename[] = "/tmp/tcp-test.XXXXXX";
   GstSample *out;
   int orig_fd, *new_fds, new_fds_len;
   struct stat orig_stat, new_stat;

   setup_multisocketsink_and_socketsrc (&st);

   orig_fd = mkstemp (tmpfilename);
   fail_unless (orig_fd > 0);
   fail_unless (unlink (tmpfilename) == 0);
   fstat (orig_fd, &orig_stat);

   msg = g_unix_fd_message_new ();
   fail_unless (g_unix_fd_message_append_fd ((GUnixFDMessage *) msg, orig_fd,
           NULL));
   close (orig_fd);
   orig_fd = -1;

   buf = gst_buffer_new_wrapped (g_strdup ("hello"), 5);
   gst_buffer_add_net_control_message_meta (buf, msg);
   g_clear_object (&msg);

   fail_unless (gst_app_src_push_buffer (st.sink_src, buf) == GST_FLOW_OK);
   buf = NULL;

   out = gst_app_sink_pull_sample (st.src_sink);
   fail_unless (out != NULL);

   fail_unless (gst_buffer_get_size (gst_sample_get_buffer (out)) == 5);
   fail_unless (gst_buffer_memcmp (gst_sample_get_buffer (out), 0, "hello",
           5) == 0);

   msg = get_control_message_meta (gst_sample_get_buffer (out));
   fail_unless (g_socket_control_message_get_msg_type (msg) == SCM_RIGHTS);
   new_fds = g_unix_fd_message_steal_fds ((GUnixFDMessage *) msg, &new_fds_len);
   fail_unless (new_fds_len == 1);

   fstat (new_fds[0], &new_stat);
   fail_unless (orig_stat.st_ino, new_stat.st_ino);

   close (new_fds[0]);
   g_free (new_fds);

   gst_sample_unref (out);

   symmetry_test_teardown (&st);
 }

 GST_END_TEST;
 #endif /* HAVE_GIO_UNIX_2_0 */

 static Suite *
 socketintegrationtest_suite (void)
 {
   Suite *s = suite_create ("socketintegrationtest");
   TCase *tc_chain = tcase_create ("general");

   suite_add_tcase (s, tc_chain);
   tcase_add_test (tc_chain,
       test_that_socketsrc_and_multisocketsink_are_symmetrical);
   tcase_add_test (tc_chain,
       test_that_tcpclientsink_and_tcpserversrc_are_symmetrical);
   tcase_add_test (tc_chain,
       test_that_tcpserversink_and_tcpclientsrc_are_symmetrical);
   tcase_add_test (tc_chain,
       test_that_we_can_provide_new_socketsrc_sockets_during_signal);
 #ifdef HAVE_GIO_UNIX_2_0
   tcase_add_test (tc_chain,
       test_that_multisocketsink_and_socketsrc_preserve_meta);
 #endif /* HAVE_GIO_UNIX_2_0 */

   return s;
 }

 GST_CHECK_MAIN (socketintegrationtest);
	/* GStreamer
	*
	* Copyright (C) 2014 William Manley <will@williammanley.net>
	*
	* 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 <gio/gio.h>
	#include <gst/check/gstcheck.h>
	#include <gst/app/gstappsink.h>
	#include <gst/app/gstappsrc.h>
	#include <gst/net/gstnetcontrolmessagemeta.h>

	#ifdef HAVE_GIO_UNIX_2_0
	#include <gio/gunixfdmessage.h>
	#endif /* HAVE_GIO_UNIX_2_0 */

	#include <sys/types.h>
	#include <sys/stat.h>
	#include <sys/socket.h>
	#include <unistd.h>

	static gboolean
	g_socketpair (GSocketFamily family, GSocketType type, GSocketProtocol protocol,
	GSocket * gsv[2], GError ** error);

	typedef struct
	{
	GstElement *sink;
	GstElement *src;

	GstPipeline *sink_pipeline;
	GstPipeline *src_pipeline;
	GstAppSrc *sink_src;
	GstAppSink *src_sink;
	} SymmetryTest;

	static void
	symmetry_test_setup (SymmetryTest * st, GstElement * sink, GstElement * src)
	{
	GstCaps *caps;
	st->sink = sink;
	g_object_set (sink, "sync", FALSE, NULL);
	st->src = src;

	st->sink_pipeline = GST_PIPELINE (gst_pipeline_new (NULL));
	st->src_pipeline = GST_PIPELINE (gst_pipeline_new (NULL));

	st->sink_src = GST_APP_SRC (gst_element_factory_make ("appsrc", NULL));
	fail_unless (st->sink_src != NULL);
	caps = gst_caps_from_string ("application/x-gst-check");
	gst_app_src_set_caps (st->sink_src, caps);
	gst_caps_unref (caps);

	gst_bin_add_many (GST_BIN (st->sink_pipeline), GST_ELEMENT (st->sink_src),
	st->sink, NULL);
	fail_unless (gst_element_link_many (GST_ELEMENT (st->sink_src), st->sink,
	NULL));

	st->src_sink = GST_APP_SINK (gst_element_factory_make ("appsink", NULL));
	fail_unless (st->src_sink != NULL);
	gst_bin_add_many (GST_BIN (st->src_pipeline), st->src,
	GST_ELEMENT (st->src_sink), NULL);
	fail_unless (gst_element_link_many (st->src, GST_ELEMENT (st->src_sink),
	NULL));

	gst_element_set_state (GST_ELEMENT (st->sink_pipeline), GST_STATE_PLAYING);
	gst_element_set_state (GST_ELEMENT (st->src_pipeline), GST_STATE_PLAYING);
	}

	static void
	symmetry_test_teardown (SymmetryTest * st)
	{
	fail_unless (gst_element_set_state (GST_ELEMENT (st->sink_pipeline),
	GST_STATE_NULL) != GST_STATE_CHANGE_FAILURE);
	fail_unless (gst_element_set_state (GST_ELEMENT (st->src_pipeline),
	GST_STATE_NULL) != GST_STATE_CHANGE_FAILURE);

	gst_object_unref (st->sink_pipeline);
	gst_object_unref (st->src_pipeline);

	memset (st, 0, sizeof (*st));
	}

	static void
	symmetry_test_assert_passthrough (SymmetryTest * st, GstBuffer * in)
	{
	gpointer copy;
	gsize data_size;
	GstSample *out;

	gst_buffer_extract_dup (in, 0, -1, &copy, &data_size);

	fail_unless (gst_app_src_push_buffer (st->sink_src, in) == GST_FLOW_OK);
	in = NULL;
	out = gst_app_sink_pull_sample (st->src_sink);
	fail_unless (out != NULL);

	fail_unless (gst_buffer_get_size (gst_sample_get_buffer (out)) == data_size);
	fail_unless (gst_buffer_memcmp (gst_sample_get_buffer (out), 0, copy,
	data_size) == 0);
	g_free (copy);
	gst_sample_unref (out);
	}

	static gboolean
	g_socketpair (GSocketFamily family, GSocketType type, GSocketProtocol protocol,
	GSocket * gsv[2], GError ** error)
	{
	int ret;
	int sv[2];

	ret = socketpair (family, type, protocol, sv);
	if (ret != 0) {
	g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED, "socketpair failed: %s",
	g_strerror (errno));
	return FALSE;
	}

	gsv[0] = g_socket_new_from_fd (sv[0], error);
	if (gsv[0] == NULL) {
	close (sv[0]);
	close (sv[1]);
	return FALSE;
	}
	gsv[1] = g_socket_new_from_fd (sv[1], error);
	if (gsv[1] == NULL) {
	g_object_unref (gsv[0]);
	gsv[0] = NULL;
	close (sv[1]);
	return FALSE;
	}
	return TRUE;
	}

	static void
	setup_multisocketsink_and_socketsrc (SymmetryTest * st)
	{
	GSocket *sockets[2] = { NULL, NULL };
	GError *err = NULL;

	st->sink = gst_check_setup_element ("multisocketsink");
	st->src = gst_check_setup_element ("socketsrc");

	fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
	G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, sockets, &err));

	g_object_set (st->src, "socket", sockets[0], NULL);
	g_object_unref (sockets[0]);
	sockets[0] = NULL;

	symmetry_test_setup (st, st->sink, st->src);

	g_signal_emit_by_name (st->sink, "add", sockets[1], NULL);
	g_object_unref (sockets[1]);
	sockets[1] = NULL;
	}

	GST_START_TEST (test_that_socketsrc_and_multisocketsink_are_symmetrical)
	{
	SymmetryTest st = { 0 };
	setup_multisocketsink_and_socketsrc (&st);
	symmetry_test_assert_passthrough (&st,
	gst_buffer_new_wrapped (g_strdup ("hello"), 5));
	symmetry_test_teardown (&st);
	}

	GST_END_TEST;


	GST_START_TEST (test_that_tcpclientsink_and_tcpserversrc_are_symmetrical)
	{
	SymmetryTest st = { 0 };
	GstElement *serversrc = gst_check_setup_element ("tcpserversrc");

	gst_element_set_state (serversrc, GST_STATE_PAUSED);
	symmetry_test_setup (&st, gst_check_setup_element ("tcpclientsink"),
	serversrc);

	symmetry_test_assert_passthrough (&st,
	gst_buffer_new_wrapped (g_strdup ("hello"), 5));

	symmetry_test_teardown (&st);
	}

	GST_END_TEST;


	GST_START_TEST (test_that_tcpserversink_and_tcpclientsrc_are_symmetrical)
	{
	SymmetryTest st = { 0 };

	symmetry_test_setup (&st, gst_check_setup_element ("tcpserversink"),
	gst_check_setup_element ("tcpclientsrc"));

	symmetry_test_assert_passthrough (&st,
	gst_buffer_new_wrapped (g_strdup ("hello"), 5));
	symmetry_test_teardown (&st);
	}

	GST_END_TEST;

	static void
	on_connection_closed (GstElement * socketsrc, gpointer user_data)
	{
	GSocket socket = (GSocket ) user_data;

	g_object_set (socketsrc, "socket", socket, NULL);
	}

	GST_START_TEST (test_that_we_can_provide_new_socketsrc_sockets_during_signal)
	{
	GSocket *sockets[4] = { NULL, NULL };

	GstPipeline *pipeline = NULL;
	GstAppSink *appsink = NULL;
	GstElement *socketsrc = NULL;
	GstSample *sample = NULL;

	socketsrc = gst_check_setup_element ("socketsrc");

	fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
	G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, &sockets[0], NULL));

	fail_unless (g_socket_send (sockets[0], "hello", 5, NULL, NULL) == 5);
	fail_unless (g_socket_shutdown (sockets[0], FALSE, TRUE, NULL));

	fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
	G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, &sockets[2], NULL));
	fail_unless (g_socket_send (sockets[2], "goodbye", 7, NULL, NULL) == 7);
	fail_unless (g_socket_shutdown (sockets[2], FALSE, TRUE, NULL));

	g_object_set (socketsrc, "socket", sockets[1], NULL);

	g_signal_connect (socketsrc, "connection-closed-by-peer",
	G_CALLBACK (on_connection_closed), sockets[3]);

	pipeline = (GstPipeline *) gst_pipeline_new (NULL);
	appsink = GST_APP_SINK (gst_check_setup_element ("appsink"));
	gst_bin_add_many (GST_BIN (pipeline), socketsrc, GST_ELEMENT (appsink), NULL);
	fail_unless (gst_element_link_many (socketsrc, GST_ELEMENT (appsink), NULL));

	gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_PLAYING);

	fail_unless ((sample = gst_app_sink_pull_sample (appsink)) != NULL);
	gst_buffer_memcmp (gst_sample_get_buffer (sample), 0, "hello", 5);
	gst_sample_unref (sample);

	fail_unless ((sample = gst_app_sink_pull_sample (appsink)) != NULL);
	gst_buffer_memcmp (gst_sample_get_buffer (sample), 0, "goodbye", 7);
	gst_sample_unref (sample);

	fail_unless (NULL == gst_app_sink_pull_sample (appsink));
	fail_unless (gst_app_sink_is_eos (appsink));

	gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_NULL);
	g_clear_object (&sockets[0]);
	g_clear_object (&sockets[1]);
	g_clear_object (&sockets[2]);
	g_clear_object (&sockets[3]);
	gst_object_unref (pipeline);
	}

	GST_END_TEST
	#ifdef HAVE_GIO_UNIX_2_0
	static GSocketControlMessage *
	get_control_message_meta (GstBuffer * buf)
	{
	GstMeta *meta;
	gpointer iter_state = NULL;

	while ((meta = gst_buffer_iterate_meta (buf, &iter_state)) != NULL) {
	if (meta->info->api == GST_NET_CONTROL_MESSAGE_META_API_TYPE)
	return ((GstNetControlMessageMeta *) meta)->message;
	}
	fail ("Expected GSocketControlMessage attached to buffer");
	return NULL;
	}


	GST_START_TEST (test_that_multisocketsink_and_socketsrc_preserve_meta)
	{
	GstBuffer *buf;
	GSocketControlMessage *msg;
	SymmetryTest st = { 0 };
	char tmpfilename[] = "/tmp/tcp-test.XXXXXX";
	GstSample *out;
	int orig_fd, *new_fds, new_fds_len;
	struct stat orig_stat, new_stat;

	setup_multisocketsink_and_socketsrc (&st);

	orig_fd = mkstemp (tmpfilename);
	fail_unless (orig_fd > 0);
	fail_unless (unlink (tmpfilename) == 0);
	fstat (orig_fd, &orig_stat);

	msg = g_unix_fd_message_new ();
	fail_unless (g_unix_fd_message_append_fd ((GUnixFDMessage *) msg, orig_fd,
	NULL));
	close (orig_fd);
	orig_fd = -1;

	buf = gst_buffer_new_wrapped (g_strdup ("hello"), 5);
	gst_buffer_add_net_control_message_meta (buf, msg);
	g_clear_object (&msg);

	fail_unless (gst_app_src_push_buffer (st.sink_src, buf) == GST_FLOW_OK);
	buf = NULL;

	out = gst_app_sink_pull_sample (st.src_sink);
	fail_unless (out != NULL);

	fail_unless (gst_buffer_get_size (gst_sample_get_buffer (out)) == 5);
	fail_unless (gst_buffer_memcmp (gst_sample_get_buffer (out), 0, "hello",
	5) == 0);

	msg = get_control_message_meta (gst_sample_get_buffer (out));
	fail_unless (g_socket_control_message_get_msg_type (msg) == SCM_RIGHTS);
	new_fds = g_unix_fd_message_steal_fds ((GUnixFDMessage *) msg, &new_fds_len);
	fail_unless (new_fds_len == 1);

	fstat (new_fds[0], &new_stat);
	fail_unless (orig_stat.st_ino, new_stat.st_ino);

	close (new_fds[0]);
	g_free (new_fds);

	gst_sample_unref (out);

	symmetry_test_teardown (&st);
	}

	GST_END_TEST;
	#endif /* HAVE_GIO_UNIX_2_0 */

	static Suite *
	socketintegrationtest_suite (void)
	{
	Suite *s = suite_create ("socketintegrationtest");
	TCase *tc_chain = tcase_create ("general");

	suite_add_tcase (s, tc_chain);
	tcase_add_test (tc_chain,
	test_that_socketsrc_and_multisocketsink_are_symmetrical);
	tcase_add_test (tc_chain,
	test_that_tcpclientsink_and_tcpserversrc_are_symmetrical);
	tcase_add_test (tc_chain,
	test_that_tcpserversink_and_tcpclientsrc_are_symmetrical);
	tcase_add_test (tc_chain,
	test_that_we_can_provide_new_socketsrc_sockets_during_signal);
	#ifdef HAVE_GIO_UNIX_2_0
	tcase_add_test (tc_chain,
	test_that_multisocketsink_and_socketsrc_preserve_meta);
	#endif /* HAVE_GIO_UNIX_2_0 */

	return s;
	}

	GST_CHECK_MAIN (socketintegrationtest);