gst/udp/gstudpsrc.c

   1 /* GStreamer
   2  * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
   3  *
   4  * This library is free software; you can redistribute it and/or
   5  * modify it under the terms of the GNU Library General Public
   6  * License as published by the Free Software Foundation; either
   7  * version 2 of the License, or (at your option) any later version.
   8  *
   9  * This library is distributed in the hope that it will be useful,
  10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12  * Library General Public License for more details.
  13  *
  14  * You should have received a copy of the GNU Library General Public
  15  * License along with this library; if not, write to the
  16  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  17  * Boston, MA 02111-1307, USA.
  18  */
  19
  20
  21 #ifdef HAVE_CONFIG_H
  22 #include "config.h"
  23 #endif
  24
  25 #include "gstudpsrc.h"
  26 #include <unistd.h>
  27
  28 #define UDP_DEFAULT_PORT                4951
  29 #define UDP_DEFAULT_MULTICAST_GROUP     "0.0.0.0"
  30
  31 /* elementfactory information */
  32 static GstElementDetails gst_udpsrc_details = GST_ELEMENT_DETAILS (
  33   "UDP packet receiver",
  34   "Source/Network",
  35   "Receive data over the network via UDP",
  36   "Wim Taymans <wim.taymans@chello.be>"
  37 );
  38
  39 /* UDPSrc signals and args */
  40 enum {
  41   /* FILL ME */
  42   LAST_SIGNAL
  43 };
  44
  45 enum {
  46   ARG_0,
  47   ARG_PORT,
  48   ARG_CONTROL,
  49   ARG_MULTICAST_GROUP
  50   /* FILL ME */
  51 };
  52
  53 #define GST_TYPE_UDPSRC_CONTROL (gst_udpsrc_control_get_type())
  54 static GType
  55 gst_udpsrc_control_get_type(void) {
  56   static GType udpsrc_control_type = 0;
  57   static GEnumValue udpsrc_control[] = {
  58     {CONTROL_NONE, "1", "none"},
  59     {CONTROL_UDP, "2", "udp"},
  60     {CONTROL_TCP, "3", "tcp"},
  61     {CONTROL_ZERO, NULL, NULL},
  62   };
  63   if (!udpsrc_control_type) {
  64     udpsrc_control_type = g_enum_register_static("GstUDPSrcControl", udpsrc_control);
  65   }
  66   return udpsrc_control_type;
  67 }
  68
  69 static void             gst_udpsrc_base_init            (gpointer g_class);
  70 static void             gst_udpsrc_class_init           (GstUDPSrc *klass);
  71 static void             gst_udpsrc_init                 (GstUDPSrc *udpsrc);
  72
  73 static GstData* gst_udpsrc_get                  (GstPad *pad);
  74 static GstElementStateReturn
  75                         gst_udpsrc_change_state         (GstElement *element);
  76
  77 static void             gst_udpsrc_set_property         (GObject *object, guint prop_id,
  78                                                          const GValue *value, GParamSpec *pspec);
  79 static void             gst_udpsrc_get_property         (GObject *object, guint prop_id,
  80                                                          GValue *value, GParamSpec *pspec);
  81 static void             gst_udpsrc_set_clock            (GstElement *element, GstClock *clock);
  82
  83 static GstElementClass *parent_class = NULL;
  84 /*static guint gst_udpsrc_signals[LAST_SIGNAL] = { 0 }; */
  85
  86 GType
  87 gst_udpsrc_get_type (void)
  88 {
  89   static GType udpsrc_type = 0;
  90
  91   if (!udpsrc_type) {
  92     static const GTypeInfo udpsrc_info = {
  93       sizeof(GstUDPSrcClass),
  94       gst_udpsrc_base_init,
  95       NULL,
  96       (GClassInitFunc)gst_udpsrc_class_init,
  97       NULL,
  98       NULL,
  99       sizeof(GstUDPSrc),
 100       0,
 101       (GInstanceInitFunc)gst_udpsrc_init,
 102       NULL
 103     };
 104     udpsrc_type = g_type_register_static (GST_TYPE_ELEMENT, "GstUDPSrc", &udpsrc_info, 0);
 105   }
 106   return udpsrc_type;
 107 }
 108
 109 static void
 110 gst_udpsrc_base_init (gpointer g_class)
 111 {
 112   GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);
 113
 114   gst_element_class_set_details (element_class, &gst_udpsrc_details);
 115 }
 116
 117 static void
 118 gst_udpsrc_class_init (GstUDPSrc *klass)
 119 {
 120   GObjectClass *gobject_class;
 121   GstElementClass *gstelement_class;
 122
 123   gobject_class = (GObjectClass*) klass;
 124   gstelement_class = (GstElementClass*) klass;
 125
 126   parent_class = g_type_class_ref (GST_TYPE_ELEMENT);
 127
 128   g_object_class_install_property (G_OBJECT_CLASS (klass), ARG_PORT,
 129     g_param_spec_int ("port", "port", "The port to receive the packets from",
 130                        0, 32768, UDP_DEFAULT_PORT, G_PARAM_READWRITE));
 131   g_object_class_install_property (gobject_class, ARG_CONTROL,
 132     g_param_spec_enum ("control", "control", "The type of control",
 133                        GST_TYPE_UDPSRC_CONTROL, CONTROL_UDP, G_PARAM_READWRITE));
 134   g_object_class_install_property (gobject_class, ARG_MULTICAST_GROUP,
 135     g_param_spec_string ("multicast_group", "multicast_group",
 136                          "The Address of multicast group to join",
 137                          UDP_DEFAULT_MULTICAST_GROUP, G_PARAM_READWRITE));
 138
 139   gobject_class->set_property = gst_udpsrc_set_property;
 140   gobject_class->get_property = gst_udpsrc_get_property;
 141
 142   gstelement_class->change_state = gst_udpsrc_change_state;
 143   gstelement_class->set_clock = gst_udpsrc_set_clock;
 144 }
 145
 146 static void
 147 gst_udpsrc_set_clock (GstElement *element, GstClock *clock)
 148 {
 149   GstUDPSrc *udpsrc;
 150
 151   udpsrc = GST_UDPSRC (element);
 152
 153   udpsrc->clock = clock;
 154 }
 155
 156 static void
 157 gst_udpsrc_init (GstUDPSrc *udpsrc)
 158 {
 159   /* create the src and src pads */
 160   udpsrc->srcpad = gst_pad_new ("src", GST_PAD_SRC);
 161   gst_element_add_pad (GST_ELEMENT (udpsrc), udpsrc->srcpad);
 162   gst_pad_set_get_function (udpsrc->srcpad, gst_udpsrc_get);
 163
 164   udpsrc->port = UDP_DEFAULT_PORT;
 165   udpsrc->control = CONTROL_UDP;
 166   udpsrc->clock = NULL;
 167   udpsrc->sock = -1;
 168   udpsrc->control_sock = -1;
 169   udpsrc->multi_group = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 170
 171   udpsrc->first_buf = TRUE;
 172 }
 173
 174 static GstData*
 175 gst_udpsrc_get (GstPad *pad)
 176 {
 177   GstUDPSrc *udpsrc;
 178   GstBuffer *outbuf;
 179   struct sockaddr_in tmpaddr;
 180   socklen_t len;
 181   gint numbytes;
 182   fd_set read_fds;
 183   guint max_sock;
 184
 185   g_return_val_if_fail (pad != NULL, NULL);
 186   g_return_val_if_fail (GST_IS_PAD (pad), NULL);
 187
 188   udpsrc = GST_UDPSRC (GST_OBJECT_PARENT (pad));
 189
 190   FD_ZERO (&read_fds);
 191   FD_SET (udpsrc->sock, &read_fds);
 192   if (udpsrc->control != CONTROL_NONE) {
 193      FD_SET (udpsrc->control_sock, &read_fds);
 194   }
 195   max_sock = MAX(udpsrc->sock, udpsrc->control_sock);
 196
 197   if (select (max_sock+1, &read_fds, NULL, NULL, NULL) > 0) {
 198     if ((udpsrc->control_sock != -1) &&
 199         FD_ISSET (udpsrc->control_sock, &read_fds)) {
 200 #ifndef GST_DISABLE_LOADSAVE
 201       guchar *buf;
 202       int ret;
 203       int fdread;
 204       struct sockaddr addr;
 205       xmlDocPtr doc;
 206       GstCaps *caps;
 207
 208       buf = g_malloc (1024*10);
 209
 210       switch (udpsrc->control) {
 211         case CONTROL_TCP:
 212             len = sizeof (struct sockaddr);
 213             fdread = accept (udpsrc->control_sock, &addr, &len);
 214             if (fdread < 0) {
 215                 perror ("accept");
 216             }
 217
 218             ret = read (fdread, buf, 1024*10);
 219             break;
 220         case CONTROL_UDP:
 221             len = sizeof (struct sockaddr);
 222             ret = recvfrom (udpsrc->control_sock, buf, 1024*10, 0, (struct sockaddr *)&tmpaddr, &len);
 223             if (ret < 0) {
 224                 perror ("recvfrom");
 225             }
 226             break;
 227         case CONTROL_NONE:
 228         default:
 229             g_free (buf);
 230             return NULL;
 231             break;
 232       }
 233
 234       buf[ret] = '\0';
 235       doc = xmlParseMemory(buf, ret);
 236       caps = gst_caps_load_thyself(doc->xmlRootNode);
 237       if (caps == NULL) {
 238         return NULL;
 239       }
 240
 241       /* foward the connect, we don't signal back the result here... */
 242       if (gst_caps_is_fixed (caps)) {
 243         gst_pad_try_set_caps (udpsrc->srcpad, caps);
 244       } else {
 245         GST_ERROR ("caps %" GST_PTR_FORMAT, caps);
 246         GST_ELEMENT_ERROR (udpsrc, CORE, NEGOTIATION, (NULL), ("Got unfixed caps from peer"));
 247       }
 248
 249 #endif
 250       g_free (buf);
 251       outbuf = NULL;
 252     }
 253     else {
 254       outbuf = gst_buffer_new ();
 255       GST_BUFFER_DATA (outbuf) = g_malloc (24000);
 256       GST_BUFFER_SIZE (outbuf) = 24000;
 257
 258       if (udpsrc->first_buf) {
 259         if (udpsrc->clock) {
 260            GstClockTime current_time;
 261            GstEvent *discont;
 262
 263            current_time = gst_clock_get_time (udpsrc->clock);
 264
 265            GST_BUFFER_TIMESTAMP (outbuf) = current_time;
 266
 267            discont = gst_event_new_discontinuous (FALSE, GST_FORMAT_TIME,
 268                                 current_time, NULL);
 269
 270            gst_pad_push (udpsrc->srcpad, GST_DATA (discont));
 271         }
 272
 273         udpsrc->first_buf = FALSE;
 274       }
 275
 276       else {
 277         GST_BUFFER_TIMESTAMP (outbuf) = GST_CLOCK_TIME_NONE;
 278       }
 279
 280       len = sizeof (struct sockaddr);
 281       numbytes = recvfrom (udpsrc->sock, GST_BUFFER_DATA (outbuf),
 282                   GST_BUFFER_SIZE (outbuf), 0, (struct sockaddr *)&tmpaddr, &len);
 283
 284       if (numbytes != -1) {
 285         GST_BUFFER_SIZE (outbuf) = numbytes;
 286       }
 287       else {
 288         perror ("recvfrom");
 289         gst_buffer_unref (outbuf);
 290         outbuf = NULL;
 291       }
 292
 293     }
 294   }
 295   else {
 296     perror ("select");
 297     outbuf = NULL;
 298   }
 299   return GST_DATA (outbuf);
 300 }
 301
 302
 303 static void
 304 gst_udpsrc_set_property (GObject *object, guint prop_id, const GValue *value, GParamSpec *pspec)
 305 {
 306   GstUDPSrc *udpsrc;
 307
 308   /* it's not null if we got it, but it might not be ours */
 309   g_return_if_fail(GST_IS_UDPSRC(object));
 310   udpsrc = GST_UDPSRC(object);
 311
 312   switch (prop_id) {
 313     case ARG_PORT:
 314         udpsrc->port = g_value_get_int (value);
 315       break;
 316     case ARG_MULTICAST_GROUP:
 317       g_free(udpsrc->multi_group);
 318
 319       if (g_value_get_string (value) == NULL)
 320         udpsrc->multi_group = g_strdup (UDP_DEFAULT_MULTICAST_GROUP);
 321       else
 322         udpsrc->multi_group = g_strdup (g_value_get_string (value));
 323
 324       break;
 325     case ARG_CONTROL:
 326         udpsrc->control = g_value_get_enum (value);
 327       break;
 328     default:
 329       break;
 330   }
 331 }
 332
 333 static void
 334 gst_udpsrc_get_property (GObject *object, guint prop_id, GValue *value, GParamSpec *pspec)
 335 {
 336   GstUDPSrc *udpsrc;
 337
 338   /* it's not null if we got it, but it might not be ours */
 339   g_return_if_fail(GST_IS_UDPSRC(object));
 340   udpsrc = GST_UDPSRC(object);
 341
 342   switch (prop_id) {
 343     case ARG_PORT:
 344       g_value_set_int (value, udpsrc->port);
 345       break;
 346     case ARG_MULTICAST_GROUP:
 347       g_value_set_string (value, udpsrc->multi_group);
 348       break;
 349     case ARG_CONTROL:
 350       g_value_set_enum (value, udpsrc->control);
 351       break;
 352     default:
 353       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 354       break;
 355   }
 356 }
 357
 358 /* create a socket for sending to remote machine */
 359 static gboolean
 360 gst_udpsrc_init_receive (GstUDPSrc *src)
 361 {
 362   guint bc_val;
 363   gint reuse=1;
 364   memset (&src->myaddr, 0, sizeof (src->myaddr));
 365   src->myaddr.sin_family = AF_INET;           /* host byte order */
 366   src->myaddr.sin_port = htons (src->port);   /* short, network byte order */
 367   src->myaddr.sin_addr.s_addr = INADDR_ANY;
 368
 369   if ((src->sock = socket (AF_INET, SOCK_DGRAM, 0)) == -1) {
 370     perror("socket");
 371     return FALSE;
 372   }
 373
 374   if (setsockopt(src->sock, SOL_SOCKET, SO_REUSEADDR, &reuse, sizeof(reuse)) == -1) {
 375    perror("setsockopt");
 376    return FALSE;
 377   }
 378
 379   if (bind (src->sock, (struct sockaddr *) &src->myaddr, sizeof (src->myaddr)) == -1)  {
 380     perror("bind");
 381     return FALSE;
 382   }
 383
 384   if (inet_aton (src->multi_group, &(src->multi_addr.imr_multiaddr))) {
 385     if (src->multi_addr.imr_multiaddr.s_addr) {
 386       src->multi_addr.imr_interface.s_addr = INADDR_ANY;
 387       setsockopt (src->sock, IPPROTO_IP, IP_ADD_MEMBERSHIP, &src->multi_addr, sizeof(src->multi_addr));
 388     }
 389   }
 390
 391   bc_val = 1;
 392   setsockopt (src->sock, SOL_SOCKET, SO_BROADCAST, &bc_val, sizeof (bc_val));
 393   src->myaddr.sin_port = htons (src->port+1);
 394
 395   switch (src->control) {
 396     case CONTROL_TCP:
 397         if ((src->control_sock = socket (AF_INET, SOCK_STREAM, 0)) == -1) {
 398            perror("control_socket");
 399            return FALSE;
 400         }
 401
 402         if (bind (src->control_sock, (struct sockaddr *) &src->myaddr, sizeof (src->myaddr)) == -1) {
 403            perror("control_bind");
 404            return FALSE;
 405         }
 406
 407         if (listen (src->control_sock, 5) == -1) {
 408            perror("listen");
 409            return FALSE;
 410         }
 411
 412         fcntl (src->control_sock, F_SETFL, O_NONBLOCK);
 413
 414         break;
 415     case CONTROL_UDP:
 416         if ((src->control_sock = socket (AF_INET, SOCK_DGRAM, 0)) == -1) {
 417            perror("socket");
 418            return FALSE;
 419         }
 420
 421         if (bind (src->control_sock, (struct sockaddr *) &src->myaddr, sizeof (src->myaddr)) == -1)
 422         {
 423             perror("control_bind");
 424             return FALSE;
 425         }
 426         /* We can only do broadcast in udp */
 427         bc_val = 1;
 428         setsockopt (src->control_sock, SOL_SOCKET, SO_BROADCAST, &bc_val, sizeof (bc_val));
 429         break;
 430     case CONTROL_NONE:
 431         GST_FLAG_SET (src, GST_UDPSRC_OPEN);
 432         return TRUE;
 433         break;
 434     default:
 435         return FALSE;
 436         break;
 437   }
 438
 439   GST_FLAG_SET (src, GST_UDPSRC_OPEN);
 440
 441   return TRUE;
 442 }
 443
 444 static void
 445 gst_udpsrc_close (GstUDPSrc *src)
 446 {
 447   if (src->sock != -1) {
 448     close (src->sock);
 449     src->sock = -1;
 450   }
 451   if (src->control_sock != -1) {
 452     close (src->control_sock);
 453     src->control_sock = -1;
 454   }
 455
 456   GST_FLAG_UNSET (src, GST_UDPSRC_OPEN);
 457 }
 458
 459 static GstElementStateReturn
 460 gst_udpsrc_change_state (GstElement *element)
 461 {
 462   g_return_val_if_fail (GST_IS_UDPSRC (element), GST_STATE_FAILURE);
 463
 464   if (GST_STATE_PENDING (element) == GST_STATE_NULL) {
 465     if (GST_FLAG_IS_SET (element, GST_UDPSRC_OPEN))
 466       gst_udpsrc_close (GST_UDPSRC (element));
 467   } else {
 468     if (!GST_FLAG_IS_SET (element, GST_UDPSRC_OPEN)) {
 469       if (!gst_udpsrc_init_receive (GST_UDPSRC (element)))
 470         return GST_STATE_FAILURE;
 471     }
 472   }
 473
 474   if (GST_ELEMENT_CLASS (parent_class)->change_state)
 475     return GST_ELEMENT_CLASS (parent_class)->change_state (element);
 476
 477   return GST_STATE_SUCCESS;
 478 }
 479