1 /* GStreamer
2 * Copyright (C) <2005> Philippe Khalaf <burger@speedy.org>
3 * Copyright (C) <2005> Nokia Corporation <kai.vehmanen@nokia.com>
4 * Copyright (C) <2006> Joni Valtanen <joni.valtanen@movial.fi>
5 * Copyright (C) <2012> Collabora Ltd.
6 * Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
7 *
8 * This library is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Library General Public
10 * License as published by the Free Software Foundation; either
11 * version 2 of the License, or (at your option) any later version.
12 *
13 * This library is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * Library General Public License for more details.
17 *
18 * You should have received a copy of the GNU Library General Public
19 * License along with this library; if not, write to the
20 * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
21 * Boston, MA 02110-1301, USA.
22 */
23
24 #ifdef HAVE_CONFIG_H
25 #include "config.h"
26 #endif
27 #include "gstdynudpsink.h"
28
29 #include <gst/net/gstnetaddressmeta.h>
30
31 GST_DEBUG_CATEGORY_STATIC (dynudpsink_debug);
32 #define GST_CAT_DEFAULT (dynudpsink_debug)
33
34 static GstStaticPadTemplate sink_template = GST_STATIC_PAD_TEMPLATE ("sink",
35 GST_PAD_SINK,
36 GST_PAD_ALWAYS,
37 GST_STATIC_CAPS_ANY);
38
39 /* DynUDPSink signals and args */
40 enum
41 {
42 /* methods */
43 SIGNAL_GET_STATS,
44
45 /* signals */
46
47 /* FILL ME */
48 LAST_SIGNAL
49 };
50
51 #define UDP_DEFAULT_SOCKET NULL
52 #define UDP_DEFAULT_CLOSE_SOCKET TRUE
53 #define UDP_DEFAULT_BIND_ADDRESS NULL
54 #define UDP_DEFAULT_BIND_PORT 0
55
56 enum
57 {
58 PROP_0,
59 PROP_SOCKET,
60 PROP_SOCKET_V6,
61 PROP_CLOSE_SOCKET,
62 PROP_BIND_ADDRESS,
63 PROP_BIND_PORT
64 };
65
66 static void gst_dynudpsink_finalize (GObject * object);
67
68 static GstFlowReturn gst_dynudpsink_render (GstBaseSink * sink,
69 GstBuffer * buffer);
70 static gboolean gst_dynudpsink_stop (GstBaseSink * bsink);
71 static gboolean gst_dynudpsink_start (GstBaseSink * bsink);
72 static gboolean gst_dynudpsink_unlock (GstBaseSink * bsink);
73 static gboolean gst_dynudpsink_unlock_stop (GstBaseSink * bsink);
74
75 static void gst_dynudpsink_set_property (GObject * object, guint prop_id,
76 const GValue * value, GParamSpec * pspec);
77 static void gst_dynudpsink_get_property (GObject * object, guint prop_id,
78 GValue * value, GParamSpec * pspec);
79 static GstStructure *gst_dynudpsink_get_stats (GstDynUDPSink * sink,
80 const gchar * host, gint port);
81
82 static guint gst_dynudpsink_signals[LAST_SIGNAL] = { 0 };
83
84 #define gst_dynudpsink_parent_class parent_class
85 G_DEFINE_TYPE (GstDynUDPSink, gst_dynudpsink, GST_TYPE_BASE_SINK);
86
87 static void
gst_dynudpsink_class_init(GstDynUDPSinkClass * klass)88 gst_dynudpsink_class_init (GstDynUDPSinkClass * klass)
89 {
90 GObjectClass *gobject_class;
91 GstElementClass *gstelement_class;
92 GstBaseSinkClass *gstbasesink_class;
93
94 gobject_class = (GObjectClass *) klass;
95 gstelement_class = (GstElementClass *) klass;
96 gstbasesink_class = (GstBaseSinkClass *) klass;
97
98 parent_class = g_type_class_peek_parent (klass);
99
100 gobject_class->set_property = gst_dynudpsink_set_property;
101 gobject_class->get_property = gst_dynudpsink_get_property;
102 gobject_class->finalize = gst_dynudpsink_finalize;
103
104 gst_dynudpsink_signals[SIGNAL_GET_STATS] =
105 g_signal_new ("get-stats", G_TYPE_FROM_CLASS (klass),
106 G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
107 G_STRUCT_OFFSET (GstDynUDPSinkClass, get_stats),
108 NULL, NULL, g_cclosure_marshal_generic, GST_TYPE_STRUCTURE, 2,
109 G_TYPE_STRING, G_TYPE_INT);
110
111 g_object_class_install_property (gobject_class, PROP_SOCKET,
112 g_param_spec_object ("socket", "Socket",
113 "Socket to use for UDP sending. (NULL == allocate)",
114 G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
115 g_object_class_install_property (gobject_class, PROP_SOCKET_V6,
116 g_param_spec_object ("socket-v6", "Socket IPv6",
117 "Socket to use for UDPv6 sending. (NULL == allocate)",
118 G_TYPE_SOCKET, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
119 g_object_class_install_property (gobject_class, PROP_CLOSE_SOCKET,
120 g_param_spec_boolean ("close-socket", "Close socket",
121 "Close socket if passed as property on state change",
122 UDP_DEFAULT_CLOSE_SOCKET,
123 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
124 g_object_class_install_property (gobject_class, PROP_BIND_ADDRESS,
125 g_param_spec_string ("bind-address", "Bind Address",
126 "Address to bind the socket to", UDP_DEFAULT_BIND_ADDRESS,
127 G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
128 g_object_class_install_property (gobject_class, PROP_BIND_PORT,
129 g_param_spec_int ("bind-port", "Bind Port",
130 "Port to bind the socket to", 0, G_MAXUINT16,
131 UDP_DEFAULT_BIND_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
132
133 gst_element_class_add_static_pad_template (gstelement_class, &sink_template);
134
135 gst_element_class_set_static_metadata (gstelement_class, "UDP packet sender",
136 "Sink/Network",
137 "Send data over the network via UDP with packet destinations picked up "
138 "dynamically from meta on the buffers passed",
139 "Philippe Khalaf <burger@speedy.org>");
140
141 gstbasesink_class->render = gst_dynudpsink_render;
142 gstbasesink_class->start = gst_dynudpsink_start;
143 gstbasesink_class->stop = gst_dynudpsink_stop;
144 gstbasesink_class->unlock = gst_dynudpsink_unlock;
145 gstbasesink_class->unlock_stop = gst_dynudpsink_unlock_stop;
146
147 klass->get_stats = gst_dynudpsink_get_stats;
148
149 GST_DEBUG_CATEGORY_INIT (dynudpsink_debug, "dynudpsink", 0, "UDP sink");
150 }
151
152 static void
gst_dynudpsink_init(GstDynUDPSink * sink)153 gst_dynudpsink_init (GstDynUDPSink * sink)
154 {
155 sink->socket = UDP_DEFAULT_SOCKET;
156 sink->socket_v6 = UDP_DEFAULT_SOCKET;
157 sink->close_socket = UDP_DEFAULT_CLOSE_SOCKET;
158 sink->external_socket = FALSE;
159 sink->bind_address = UDP_DEFAULT_BIND_ADDRESS;
160 sink->bind_port = UDP_DEFAULT_BIND_PORT;
161
162 sink->used_socket = NULL;
163 sink->used_socket_v6 = NULL;
164 }
165
166 static void
gst_dynudpsink_finalize(GObject * object)167 gst_dynudpsink_finalize (GObject * object)
168 {
169 GstDynUDPSink *sink;
170
171 sink = GST_DYNUDPSINK (object);
172
173 if (sink->socket)
174 g_object_unref (sink->socket);
175 sink->socket = NULL;
176
177 if (sink->socket_v6)
178 g_object_unref (sink->socket_v6);
179 sink->socket_v6 = NULL;
180
181 if (sink->used_socket)
182 g_object_unref (sink->used_socket);
183 sink->used_socket = NULL;
184
185 if (sink->used_socket_v6)
186 g_object_unref (sink->used_socket_v6);
187 sink->used_socket_v6 = NULL;
188
189 g_free (sink->bind_address);
190 sink->bind_address = NULL;
191
192 G_OBJECT_CLASS (parent_class)->finalize (object);
193 }
194
195 static GstFlowReturn
gst_dynudpsink_render(GstBaseSink * bsink,GstBuffer * buffer)196 gst_dynudpsink_render (GstBaseSink * bsink, GstBuffer * buffer)
197 {
198 GstDynUDPSink *sink;
199 gssize ret;
200 GstMapInfo map;
201 GstNetAddressMeta *meta;
202 GSocketAddress *addr;
203 GError *err = NULL;
204 GSocketFamily family;
205 GSocket *socket;
206
207 meta = gst_buffer_get_net_address_meta (buffer);
208
209 if (meta == NULL) {
210 GST_DEBUG ("Received buffer without GstNetAddressMeta, skipping");
211 return GST_FLOW_OK;
212 }
213
214 sink = GST_DYNUDPSINK (bsink);
215
216 /* let's get the address from the metadata */
217 addr = meta->addr;
218
219 family = g_socket_address_get_family (addr);
220 if (family == G_SOCKET_FAMILY_IPV6 && !sink->used_socket_v6)
221 goto invalid_family;
222
223 gst_buffer_map (buffer, &map, GST_MAP_READ);
224
225 GST_DEBUG ("about to send %" G_GSIZE_FORMAT " bytes", map.size);
226
227 #ifndef GST_DISABLE_GST_DEBUG
228 {
229 gchar *host;
230
231 host =
232 g_inet_address_to_string (g_inet_socket_address_get_address
233 (G_INET_SOCKET_ADDRESS (addr)));
234 GST_DEBUG ("sending %" G_GSIZE_FORMAT " bytes to client %s port %d",
235 map.size, host,
236 g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (addr)));
237 g_free (host);
238 }
239 #endif
240
241 /* Select socket to send from for this address */
242 if (family == G_SOCKET_FAMILY_IPV6 || !sink->used_socket)
243 socket = sink->used_socket_v6;
244 else
245 socket = sink->used_socket;
246
247 ret =
248 g_socket_send_to (socket, addr, (gchar *) map.data, map.size,
249 sink->cancellable, &err);
250 gst_buffer_unmap (buffer, &map);
251
252 if (ret < 0)
253 goto send_error;
254
255 GST_DEBUG ("sent %" G_GSSIZE_FORMAT " bytes", ret);
256
257 return GST_FLOW_OK;
258
259 send_error:
260 {
261 GstFlowReturn flow_ret;
262
263 if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
264 GST_DEBUG_OBJECT (sink, "send cancelled");
265 flow_ret = GST_FLOW_FLUSHING;
266 } else {
267 GST_ELEMENT_ERROR (sink, RESOURCE, WRITE, (NULL),
268 ("send error: %s", err->message));
269 flow_ret = GST_FLOW_ERROR;
270 }
271 g_clear_error (&err);
272 return flow_ret;
273 }
274 invalid_family:
275 {
276 GST_DEBUG ("invalid address family (got %d)", family);
277 return GST_FLOW_ERROR;
278 }
279 }
280
281 static void
gst_dynudpsink_set_property(GObject * object,guint prop_id,const GValue * value,GParamSpec * pspec)282 gst_dynudpsink_set_property (GObject * object, guint prop_id,
283 const GValue * value, GParamSpec * pspec)
284 {
285 GstDynUDPSink *udpsink;
286
287 udpsink = GST_DYNUDPSINK (object);
288
289 switch (prop_id) {
290 case PROP_SOCKET:
291 if (udpsink->socket != NULL && udpsink->socket != udpsink->used_socket &&
292 udpsink->close_socket) {
293 GError *err = NULL;
294
295 if (!g_socket_close (udpsink->socket, &err)) {
296 GST_ERROR ("failed to close socket %p: %s", udpsink->socket,
297 err->message);
298 g_clear_error (&err);
299 }
300 }
301 if (udpsink->socket)
302 g_object_unref (udpsink->socket);
303 udpsink->socket = g_value_dup_object (value);
304 GST_DEBUG ("setting socket to %p", udpsink->socket);
305 break;
306 case PROP_SOCKET_V6:
307 if (udpsink->socket_v6 != NULL
308 && udpsink->socket_v6 != udpsink->used_socket_v6
309 && udpsink->close_socket) {
310 GError *err = NULL;
311
312 if (!g_socket_close (udpsink->socket_v6, &err)) {
313 GST_ERROR ("failed to close socket %p: %s", udpsink->socket_v6,
314 err->message);
315 g_clear_error (&err);
316 }
317 }
318 if (udpsink->socket_v6)
319 g_object_unref (udpsink->socket_v6);
320 udpsink->socket_v6 = g_value_dup_object (value);
321 GST_DEBUG ("setting socket v6 to %p", udpsink->socket_v6);
322 break;
323 case PROP_CLOSE_SOCKET:
324 udpsink->close_socket = g_value_get_boolean (value);
325 break;
326 case PROP_BIND_ADDRESS:
327 g_free (udpsink->bind_address);
328 udpsink->bind_address = g_value_dup_string (value);
329 break;
330 case PROP_BIND_PORT:
331 udpsink->bind_port = g_value_get_int (value);
332 break;
333 default:
334 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
335 break;
336 }
337 }
338
339 static void
gst_dynudpsink_get_property(GObject * object,guint prop_id,GValue * value,GParamSpec * pspec)340 gst_dynudpsink_get_property (GObject * object, guint prop_id, GValue * value,
341 GParamSpec * pspec)
342 {
343 GstDynUDPSink *udpsink;
344
345 udpsink = GST_DYNUDPSINK (object);
346
347 switch (prop_id) {
348 case PROP_SOCKET:
349 g_value_set_object (value, udpsink->socket);
350 break;
351 case PROP_SOCKET_V6:
352 g_value_set_object (value, udpsink->socket_v6);
353 break;
354 case PROP_CLOSE_SOCKET:
355 g_value_set_boolean (value, udpsink->close_socket);
356 break;
357 case PROP_BIND_ADDRESS:
358 g_value_set_string (value, udpsink->bind_address);
359 break;
360 case PROP_BIND_PORT:
361 g_value_set_int (value, udpsink->bind_port);
362 break;
363 default:
364 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
365 break;
366 }
367 }
368
369 static void
gst_dynudpsink_create_cancellable(GstDynUDPSink * sink)370 gst_dynudpsink_create_cancellable (GstDynUDPSink * sink)
371 {
372 GPollFD pollfd;
373
374 sink->cancellable = g_cancellable_new ();
375 sink->made_cancel_fd = g_cancellable_make_pollfd (sink->cancellable, &pollfd);
376 }
377
378 static void
gst_dynudpsink_free_cancellable(GstDynUDPSink * sink)379 gst_dynudpsink_free_cancellable (GstDynUDPSink * sink)
380 {
381 if (sink->made_cancel_fd) {
382 g_cancellable_release_fd (sink->cancellable);
383 sink->made_cancel_fd = FALSE;
384 }
385 g_object_unref (sink->cancellable);
386 sink->cancellable = NULL;
387 }
388
389 /* create a socket for sending to remote machine */
390 static gboolean
gst_dynudpsink_start(GstBaseSink * bsink)391 gst_dynudpsink_start (GstBaseSink * bsink)
392 {
393 GstDynUDPSink *udpsink;
394 GError *err = NULL;
395
396 udpsink = GST_DYNUDPSINK (bsink);
397
398 gst_dynudpsink_create_cancellable (udpsink);
399
400 udpsink->external_socket = FALSE;
401
402 if (udpsink->socket) {
403 if (g_socket_get_family (udpsink->socket) == G_SOCKET_FAMILY_IPV6) {
404 udpsink->used_socket_v6 = G_SOCKET (g_object_ref (udpsink->socket));
405 udpsink->external_socket = TRUE;
406 } else {
407 udpsink->used_socket = G_SOCKET (g_object_ref (udpsink->socket));
408 udpsink->external_socket = TRUE;
409 }
410 }
411
412 if (udpsink->socket_v6) {
413 g_return_val_if_fail (g_socket_get_family (udpsink->socket) !=
414 G_SOCKET_FAMILY_IPV6, FALSE);
415
416 if (udpsink->used_socket_v6
417 && udpsink->used_socket_v6 != udpsink->socket_v6) {
418 GST_ERROR_OBJECT (udpsink,
419 "Provided different IPv6 sockets in socket and socket-v6 properties");
420 return FALSE;
421 }
422
423 udpsink->used_socket_v6 = G_SOCKET (g_object_ref (udpsink->socket_v6));
424 udpsink->external_socket = TRUE;
425 }
426
427 if (!udpsink->used_socket && !udpsink->used_socket_v6) {
428 GSocketAddress *bind_addr;
429 GInetAddress *bind_iaddr;
430
431 if (udpsink->bind_address) {
432 GSocketFamily family;
433
434 bind_iaddr = g_inet_address_new_from_string (udpsink->bind_address);
435 if (!bind_iaddr) {
436 GList *results;
437 GResolver *resolver;
438
439 resolver = g_resolver_get_default ();
440 results =
441 g_resolver_lookup_by_name (resolver, udpsink->bind_address,
442 udpsink->cancellable, &err);
443 if (!results) {
444 g_object_unref (resolver);
445 goto name_resolve;
446 }
447 bind_iaddr = G_INET_ADDRESS (g_object_ref (results->data));
448 g_resolver_free_addresses (results);
449 g_object_unref (resolver);
450 }
451
452 bind_addr = g_inet_socket_address_new (bind_iaddr, udpsink->bind_port);
453 g_object_unref (bind_iaddr);
454 family = g_socket_address_get_family (G_SOCKET_ADDRESS (bind_addr));
455
456 if ((udpsink->used_socket =
457 g_socket_new (family, G_SOCKET_TYPE_DATAGRAM,
458 G_SOCKET_PROTOCOL_UDP, &err)) == NULL) {
459 g_object_unref (bind_addr);
460 goto no_socket;
461 }
462
463 g_socket_bind (udpsink->used_socket, bind_addr, TRUE, &err);
464 if (err != NULL)
465 goto bind_error;
466 } else {
467 /* create sender sockets if none available */
468 if ((udpsink->used_socket = g_socket_new (G_SOCKET_FAMILY_IPV4,
469 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err)) == NULL)
470 goto no_socket;
471
472 bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV4);
473 bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
474 g_socket_bind (udpsink->used_socket, bind_addr, TRUE, &err);
475 g_object_unref (bind_addr);
476 g_object_unref (bind_iaddr);
477 if (err != NULL)
478 goto bind_error;
479
480 if ((udpsink->used_socket_v6 = g_socket_new (G_SOCKET_FAMILY_IPV6,
481 G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP,
482 &err)) == NULL) {
483 GST_INFO_OBJECT (udpsink, "Failed to create IPv6 socket: %s",
484 err->message);
485 g_clear_error (&err);
486 } else {
487 bind_iaddr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV6);
488 bind_addr = g_inet_socket_address_new (bind_iaddr, 0);
489 g_socket_bind (udpsink->used_socket_v6, bind_addr, TRUE, &err);
490 g_object_unref (bind_addr);
491 g_object_unref (bind_iaddr);
492 if (err != NULL)
493 goto bind_error;
494 }
495 }
496 }
497
498 if (udpsink->used_socket)
499 g_socket_set_broadcast (udpsink->used_socket, TRUE);
500 if (udpsink->used_socket_v6)
501 g_socket_set_broadcast (udpsink->used_socket_v6, TRUE);
502
503 return TRUE;
504
505 /* ERRORS */
506 no_socket:
507 {
508 GST_ERROR_OBJECT (udpsink, "Failed to create IPv4 socket: %s",
509 err->message);
510 g_clear_error (&err);
511 return FALSE;
512 }
513 bind_error:
514 {
515 GST_ELEMENT_ERROR (udpsink, RESOURCE, FAILED, (NULL),
516 ("Failed to bind socket: %s", err->message));
517 g_clear_error (&err);
518 return FALSE;
519 }
520 name_resolve:
521 {
522 GST_ELEMENT_ERROR (udpsink, RESOURCE, FAILED, (NULL),
523 ("Failed to resolve bind address %s: %s", udpsink->bind_address,
524 err->message));
525 g_clear_error (&err);
526 return FALSE;
527 }
528 }
529
530 static GstStructure *
gst_dynudpsink_get_stats(GstDynUDPSink * sink,const gchar * host,gint port)531 gst_dynudpsink_get_stats (GstDynUDPSink * sink, const gchar * host, gint port)
532 {
533 return NULL;
534 }
535
536 static gboolean
gst_dynudpsink_stop(GstBaseSink * bsink)537 gst_dynudpsink_stop (GstBaseSink * bsink)
538 {
539 GstDynUDPSink *udpsink;
540
541 udpsink = GST_DYNUDPSINK (bsink);
542
543 if (udpsink->used_socket) {
544 if (udpsink->close_socket || !udpsink->external_socket) {
545 GError *err = NULL;
546
547 if (!g_socket_close (udpsink->used_socket, &err)) {
548 GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
549 g_clear_error (&err);
550 }
551 }
552
553 g_object_unref (udpsink->used_socket);
554 udpsink->used_socket = NULL;
555 }
556
557 if (udpsink->used_socket_v6) {
558 if (udpsink->close_socket || !udpsink->external_socket) {
559 GError *err = NULL;
560
561 if (!g_socket_close (udpsink->used_socket_v6, &err)) {
562 GST_ERROR_OBJECT (udpsink, "Failed to close socket: %s", err->message);
563 g_clear_error (&err);
564 }
565 }
566
567 g_object_unref (udpsink->used_socket_v6);
568 udpsink->used_socket_v6 = NULL;
569 }
570
571 gst_dynudpsink_free_cancellable (udpsink);
572
573 return TRUE;
574 }
575
576 static gboolean
gst_dynudpsink_unlock(GstBaseSink * bsink)577 gst_dynudpsink_unlock (GstBaseSink * bsink)
578 {
579 GstDynUDPSink *udpsink;
580
581 udpsink = GST_DYNUDPSINK (bsink);
582
583 g_cancellable_cancel (udpsink->cancellable);
584
585 return TRUE;
586 }
587
588 static gboolean
gst_dynudpsink_unlock_stop(GstBaseSink * bsink)589 gst_dynudpsink_unlock_stop (GstBaseSink * bsink)
590 {
591 GstDynUDPSink *udpsink;
592
593 udpsink = GST_DYNUDPSINK (bsink);
594
595 gst_dynudpsink_free_cancellable (udpsink);
596 gst_dynudpsink_create_cancellable (udpsink);
597
598 return TRUE;
599 }
600