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