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, ©, &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