1 /* GStreamer
2  *
3  * Copyright (C) 2014 William Manley <will@williammanley.net>
4  *
5  * This library is free software; you can redistribute it and/or
6  * modify it under the terms of the GNU Library General Public
7  * License as published by the Free Software Foundation; either
8  * version 2 of the License, or (at your option) any later version.
9  *
10  * This library is distributed in the hope that it will be useful,
11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * Library General Public License for more details.
14  *
15  * You should have received a copy of the GNU Library General Public
16  * License along with this library; if not, write to the
17  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
18  * Boston, MA 02110-1301, USA.
19  */
20 #ifdef HAVE_CONFIG_H
21 #include "config.h"
22 #endif
23 
24 #include <gio/gio.h>
25 #include <gst/check/gstcheck.h>
26 #include <gst/app/gstappsink.h>
27 #include <gst/app/gstappsrc.h>
28 #include <gst/net/gstnetcontrolmessagemeta.h>
29 
30 #ifdef HAVE_GIO_UNIX_2_0
31 #include <gio/gunixfdmessage.h>
32 #endif /*  HAVE_GIO_UNIX_2_0 */
33 
34 #include <sys/types.h>
35 #include <sys/stat.h>
36 #include <sys/socket.h>
37 #include <unistd.h>
38 
39 static gboolean
40 g_socketpair (GSocketFamily family, GSocketType type, GSocketProtocol protocol,
41     GSocket * gsv[2], GError ** error);
42 
43 typedef struct
44 {
45   GstElement *sink;
46   GstElement *src;
47 
48   GstPipeline *sink_pipeline;
49   GstPipeline *src_pipeline;
50   GstAppSrc *sink_src;
51   GstAppSink *src_sink;
52 } SymmetryTest;
53 
54 static void
symmetry_test_setup(SymmetryTest * st,GstElement * sink,GstElement * src)55 symmetry_test_setup (SymmetryTest * st, GstElement * sink, GstElement * src)
56 {
57   GstCaps *caps;
58   st->sink = sink;
59   g_object_set (sink, "sync", FALSE, NULL);
60   st->src = src;
61 
62   st->sink_pipeline = GST_PIPELINE (gst_pipeline_new (NULL));
63   st->src_pipeline = GST_PIPELINE (gst_pipeline_new (NULL));
64 
65   st->sink_src = GST_APP_SRC (gst_element_factory_make ("appsrc", NULL));
66   fail_unless (st->sink_src != NULL);
67   caps = gst_caps_from_string ("application/x-gst-check");
68   gst_app_src_set_caps (st->sink_src, caps);
69   gst_caps_unref (caps);
70 
71   gst_bin_add_many (GST_BIN (st->sink_pipeline), GST_ELEMENT (st->sink_src),
72       st->sink, NULL);
73   fail_unless (gst_element_link_many (GST_ELEMENT (st->sink_src), st->sink,
74           NULL));
75 
76   st->src_sink = GST_APP_SINK (gst_element_factory_make ("appsink", NULL));
77   fail_unless (st->src_sink != NULL);
78   gst_bin_add_many (GST_BIN (st->src_pipeline), st->src,
79       GST_ELEMENT (st->src_sink), NULL);
80   fail_unless (gst_element_link_many (st->src, GST_ELEMENT (st->src_sink),
81           NULL));
82 
83   gst_element_set_state (GST_ELEMENT (st->sink_pipeline), GST_STATE_PLAYING);
84   gst_element_set_state (GST_ELEMENT (st->src_pipeline), GST_STATE_PLAYING);
85 }
86 
87 static void
symmetry_test_teardown(SymmetryTest * st)88 symmetry_test_teardown (SymmetryTest * st)
89 {
90   fail_unless (gst_element_set_state (GST_ELEMENT (st->sink_pipeline),
91           GST_STATE_NULL) != GST_STATE_CHANGE_FAILURE);
92   fail_unless (gst_element_set_state (GST_ELEMENT (st->src_pipeline),
93           GST_STATE_NULL) != GST_STATE_CHANGE_FAILURE);
94 
95   gst_object_unref (st->sink_pipeline);
96   gst_object_unref (st->src_pipeline);
97 
98   memset (st, 0, sizeof (*st));
99 }
100 
101 static void
symmetry_test_assert_passthrough(SymmetryTest * st,GstBuffer * in)102 symmetry_test_assert_passthrough (SymmetryTest * st, GstBuffer * in)
103 {
104   gpointer copy;
105   gsize data_size;
106   GstSample *out;
107 
108   gst_buffer_extract_dup (in, 0, -1, &copy, &data_size);
109 
110   fail_unless (gst_app_src_push_buffer (st->sink_src, in) == GST_FLOW_OK);
111   in = NULL;
112   out = gst_app_sink_pull_sample (st->src_sink);
113   fail_unless (out != NULL);
114 
115   fail_unless (gst_buffer_get_size (gst_sample_get_buffer (out)) == data_size);
116   fail_unless (gst_buffer_memcmp (gst_sample_get_buffer (out), 0, copy,
117           data_size) == 0);
118   g_free (copy);
119   gst_sample_unref (out);
120 }
121 
122 static gboolean
g_socketpair(GSocketFamily family,GSocketType type,GSocketProtocol protocol,GSocket * gsv[2],GError ** error)123 g_socketpair (GSocketFamily family, GSocketType type, GSocketProtocol protocol,
124     GSocket * gsv[2], GError ** error)
125 {
126   int ret;
127   int sv[2];
128 
129   ret = socketpair (family, type, protocol, sv);
130   if (ret != 0) {
131     g_set_error (error, G_IO_ERROR, G_IO_ERROR_FAILED, "socketpair failed: %s",
132         g_strerror (errno));
133     return FALSE;
134   }
135 
136   gsv[0] = g_socket_new_from_fd (sv[0], error);
137   if (gsv[0] == NULL) {
138     close (sv[0]);
139     close (sv[1]);
140     return FALSE;
141   }
142   gsv[1] = g_socket_new_from_fd (sv[1], error);
143   if (gsv[1] == NULL) {
144     g_object_unref (gsv[0]);
145     gsv[0] = NULL;
146     close (sv[1]);
147     return FALSE;
148   }
149   return TRUE;
150 }
151 
152 static void
setup_multisocketsink_and_socketsrc(SymmetryTest * st)153 setup_multisocketsink_and_socketsrc (SymmetryTest * st)
154 {
155   GSocket *sockets[2] = { NULL, NULL };
156   GError *err = NULL;
157 
158   st->sink = gst_check_setup_element ("multisocketsink");
159   st->src = gst_check_setup_element ("socketsrc");
160 
161   fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
162           G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, sockets, &err));
163 
164   g_object_set (st->src, "socket", sockets[0], NULL);
165   g_object_unref (sockets[0]);
166   sockets[0] = NULL;
167 
168   symmetry_test_setup (st, st->sink, st->src);
169 
170   g_signal_emit_by_name (st->sink, "add", sockets[1], NULL);
171   g_object_unref (sockets[1]);
172   sockets[1] = NULL;
173 }
174 
GST_START_TEST(test_that_socketsrc_and_multisocketsink_are_symmetrical)175 GST_START_TEST (test_that_socketsrc_and_multisocketsink_are_symmetrical)
176 {
177   SymmetryTest st = { 0 };
178   setup_multisocketsink_and_socketsrc (&st);
179   symmetry_test_assert_passthrough (&st,
180       gst_buffer_new_wrapped (g_strdup ("hello"), 5));
181   symmetry_test_teardown (&st);
182 }
183 
184 GST_END_TEST;
185 
186 
GST_START_TEST(test_that_tcpclientsink_and_tcpserversrc_are_symmetrical)187 GST_START_TEST (test_that_tcpclientsink_and_tcpserversrc_are_symmetrical)
188 {
189   SymmetryTest st = { 0 };
190   GstElement *serversrc = gst_check_setup_element ("tcpserversrc");
191 
192   gst_element_set_state (serversrc, GST_STATE_PAUSED);
193   symmetry_test_setup (&st, gst_check_setup_element ("tcpclientsink"),
194       serversrc);
195 
196   symmetry_test_assert_passthrough (&st,
197       gst_buffer_new_wrapped (g_strdup ("hello"), 5));
198 
199   symmetry_test_teardown (&st);
200 }
201 
202 GST_END_TEST;
203 
204 
GST_START_TEST(test_that_tcpserversink_and_tcpclientsrc_are_symmetrical)205 GST_START_TEST (test_that_tcpserversink_and_tcpclientsrc_are_symmetrical)
206 {
207   SymmetryTest st = { 0 };
208   GstElement *serversink = gst_check_setup_element ("tcpserversink");
209   guint timeout = 100;
210 
211   symmetry_test_setup (&st, serversink,
212       gst_check_setup_element ("tcpclientsrc"));
213 
214   /* Wait for the client to *actually* be connected before doing the
215    * test.  The socket connection from the client might very well
216    * succeed, but that doesn't mean the server has accepted it yet. If
217    * we don't wait for the server to have accepted the connection, we
218    * would end up dropping the buffer (because no one is "connected")
219    * and the receiving side would wait forever. */
220   while (timeout) {
221     guint handles;
222     g_object_get (serversink, "num-handles", &handles, NULL);
223     if (handles > 0)
224       break;
225     /* Wait for 10ms to see if client connected */
226     g_usleep (G_USEC_PER_SEC / 100);
227     timeout--;
228   }
229 
230   symmetry_test_assert_passthrough (&st,
231       gst_buffer_new_wrapped (g_strdup ("hello"), 5));
232   symmetry_test_teardown (&st);
233 }
234 
235 GST_END_TEST;
236 
237 static void
on_connection_closed(GstElement * socketsrc,gpointer user_data)238 on_connection_closed (GstElement * socketsrc, gpointer user_data)
239 {
240   GSocket *socket = (GSocket *) user_data;
241 
242   g_object_set (socketsrc, "socket", socket, NULL);
243 }
244 
GST_START_TEST(test_that_we_can_provide_new_socketsrc_sockets_during_signal)245 GST_START_TEST (test_that_we_can_provide_new_socketsrc_sockets_during_signal)
246 {
247   GSocket *sockets[4] = { NULL, NULL };
248 
249   GstPipeline *pipeline = NULL;
250   GstAppSink *appsink = NULL;
251   GstElement *socketsrc = NULL;
252   GstSample *sample = NULL;
253 
254   socketsrc = gst_check_setup_element ("socketsrc");
255 
256   fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
257           G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, &sockets[0], NULL));
258 
259   fail_unless (g_socket_send (sockets[0], "hello", 5, NULL, NULL) == 5);
260   fail_unless (g_socket_shutdown (sockets[0], FALSE, TRUE, NULL));
261 
262   fail_unless (g_socketpair (G_SOCKET_FAMILY_UNIX,
263           G_SOCKET_TYPE_STREAM, G_SOCKET_PROTOCOL_DEFAULT, &sockets[2], NULL));
264   fail_unless (g_socket_send (sockets[2], "goodbye", 7, NULL, NULL) == 7);
265   fail_unless (g_socket_shutdown (sockets[2], FALSE, TRUE, NULL));
266 
267   g_object_set (socketsrc, "socket", sockets[1], NULL);
268 
269   g_signal_connect (socketsrc, "connection-closed-by-peer",
270       G_CALLBACK (on_connection_closed), sockets[3]);
271 
272   pipeline = (GstPipeline *) gst_pipeline_new (NULL);
273   appsink = GST_APP_SINK (gst_check_setup_element ("appsink"));
274   gst_bin_add_many (GST_BIN (pipeline), socketsrc, GST_ELEMENT (appsink), NULL);
275   fail_unless (gst_element_link_many (socketsrc, GST_ELEMENT (appsink), NULL));
276 
277   gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_PLAYING);
278 
279   fail_unless ((sample = gst_app_sink_pull_sample (appsink)) != NULL);
280   gst_buffer_memcmp (gst_sample_get_buffer (sample), 0, "hello", 5);
281   gst_sample_unref (sample);
282 
283   fail_unless ((sample = gst_app_sink_pull_sample (appsink)) != NULL);
284   gst_buffer_memcmp (gst_sample_get_buffer (sample), 0, "goodbye", 7);
285   gst_sample_unref (sample);
286 
287   fail_unless (NULL == gst_app_sink_pull_sample (appsink));
288   fail_unless (gst_app_sink_is_eos (appsink));
289 
290   gst_element_set_state (GST_ELEMENT (pipeline), GST_STATE_NULL);
291   g_clear_object (&sockets[0]);
292   g_clear_object (&sockets[1]);
293   g_clear_object (&sockets[2]);
294   g_clear_object (&sockets[3]);
295   gst_object_unref (pipeline);
296 }
297 
298 GST_END_TEST
299 #ifdef HAVE_GIO_UNIX_2_0
300 static GSocketControlMessage *
get_control_message_meta(GstBuffer * buf)301 get_control_message_meta (GstBuffer * buf)
302 {
303   GstMeta *meta;
304   gpointer iter_state = NULL;
305 
306   while ((meta = gst_buffer_iterate_meta (buf, &iter_state)) != NULL) {
307     if (meta->info->api == GST_NET_CONTROL_MESSAGE_META_API_TYPE)
308       return ((GstNetControlMessageMeta *) meta)->message;
309   }
310   fail ("Expected GSocketControlMessage attached to buffer");
311   return NULL;
312 }
313 
314 
GST_START_TEST(test_that_multisocketsink_and_socketsrc_preserve_meta)315 GST_START_TEST (test_that_multisocketsink_and_socketsrc_preserve_meta)
316 {
317   GstBuffer *buf;
318   GSocketControlMessage *msg;
319   SymmetryTest st = { 0 };
320   char tmpfilename[] = "/tmp/tcp-test.XXXXXX";
321   GstSample *out;
322   int orig_fd, *new_fds, new_fds_len;
323   struct stat orig_stat, new_stat;
324 
325   setup_multisocketsink_and_socketsrc (&st);
326 
327   orig_fd = mkstemp (tmpfilename);
328   fail_unless (orig_fd > 0);
329   fail_unless (unlink (tmpfilename) == 0);
330   fstat (orig_fd, &orig_stat);
331 
332   msg = g_unix_fd_message_new ();
333   fail_unless (g_unix_fd_message_append_fd ((GUnixFDMessage *) msg, orig_fd,
334           NULL));
335   close (orig_fd);
336   orig_fd = -1;
337 
338   buf = gst_buffer_new_wrapped (g_strdup ("hello"), 5);
339   gst_buffer_add_net_control_message_meta (buf, msg);
340   g_clear_object (&msg);
341 
342   fail_unless (gst_app_src_push_buffer (st.sink_src, buf) == GST_FLOW_OK);
343   buf = NULL;
344 
345   out = gst_app_sink_pull_sample (st.src_sink);
346   fail_unless (out != NULL);
347 
348   fail_unless (gst_buffer_get_size (gst_sample_get_buffer (out)) == 5);
349   fail_unless (gst_buffer_memcmp (gst_sample_get_buffer (out), 0, "hello",
350           5) == 0);
351 
352   msg = get_control_message_meta (gst_sample_get_buffer (out));
353   fail_unless (g_socket_control_message_get_msg_type (msg) == SCM_RIGHTS);
354   new_fds = g_unix_fd_message_steal_fds ((GUnixFDMessage *) msg, &new_fds_len);
355   fail_unless (new_fds_len == 1);
356 
357   fstat (new_fds[0], &new_stat);
358   fail_unless (orig_stat.st_ino, new_stat.st_ino);
359 
360   close (new_fds[0]);
361   g_free (new_fds);
362 
363   gst_sample_unref (out);
364 
365   symmetry_test_teardown (&st);
366 }
367 
368 GST_END_TEST;
369 #endif /* HAVE_GIO_UNIX_2_0 */
370 
371 static Suite *
socketintegrationtest_suite(void)372 socketintegrationtest_suite (void)
373 {
374   Suite *s = suite_create ("socketintegrationtest");
375   TCase *tc_chain = tcase_create ("general");
376 
377   suite_add_tcase (s, tc_chain);
378   tcase_add_test (tc_chain,
379       test_that_socketsrc_and_multisocketsink_are_symmetrical);
380   tcase_add_test (tc_chain,
381       test_that_tcpclientsink_and_tcpserversrc_are_symmetrical);
382   tcase_add_test (tc_chain,
383       test_that_tcpserversink_and_tcpclientsrc_are_symmetrical);
384   tcase_add_test (tc_chain,
385       test_that_we_can_provide_new_socketsrc_sockets_during_signal);
386 #ifdef HAVE_GIO_UNIX_2_0
387   tcase_add_test (tc_chain,
388       test_that_multisocketsink_and_socketsrc_preserve_meta);
389 #endif /* HAVE_GIO_UNIX_2_0 */
390 
391   return s;
392 }
393 
394 GST_CHECK_MAIN (socketintegrationtest);
395