libs/gst/net/gstnettimeprovider.c

   1 /* GStreamer
   2  * Copyright (C) 2005 Andy Wingo <wingo@pobox.com>
   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  * SECTION:gstnettimeprovider
  21  * @short_description: Special object that exposed the time of a clock
  22  *                     on the network.
  23  * @see_also: #GstClock, #GstNetClientClock, #GstPipeline
  24  *
  25  * This object exposes the time of a #GstClock on the network.
  26  *
  27  * A #GstNetTimeProvider is created with gst_net_time_provider_new() which
  28  * takes a #GstClock, an address and a port number as arguments.
  29  *
  30  * After creating the object, a client clock such as #GstNetClientClock can
  31  * query the exposed clock over the network for its values.
  32  *
  33  * The #GstNetTimeProvider typically wraps the clock used by a #GstPipeline.
  34  *
  35  * Last reviewed on 2005-11-23 (0.9.5)
  36  */
  37
  38 #ifdef HAVE_CONFIG_H
  39 #include "config.h"
  40 #endif
  41
  42 #include "gstnettimeprovider.h"
  43 #include "gstnettimepacket.h"
  44
  45 #include <glib.h>
  46
  47 #ifdef HAVE_UNISTD_H
  48 #include <unistd.h>
  49 #endif
  50
  51 #if defined (_MSC_VER) && _MSC_VER >= 1400
  52 #include <io.h>
  53 #endif
  54
  55 #ifndef G_OS_WIN32
  56 #include <sys/ioctl.h>
  57 #endif
  58
  59 #ifdef HAVE_FIONREAD_IN_SYS_FILIO
  60 #include <sys/filio.h>
  61 #endif
  62
  63 GST_DEBUG_CATEGORY_STATIC (ntp_debug);
  64 #define GST_CAT_DEFAULT (ntp_debug)
  65
  66 #ifdef G_OS_WIN32
  67 #define close(sock) closesocket(sock)
  68 #endif
  69
  70 #define DEFAULT_ADDRESS         "0.0.0.0"
  71 #define DEFAULT_PORT            5637
  72
  73 #define IS_ACTIVE(self) (g_atomic_int_get (&((self)->active)))
  74
  75 #ifdef G_OS_WIN32
  76 #define setsockopt(sock, sol_flags, reuse_flags, ru, sizeofru) setsockopt (sock, sol_flags, reuse_flags, (char *)ru, sizeofru)
  77 #endif
  78 enum
  79 {
  80   PROP_0,
  81   PROP_PORT,
  82   PROP_ADDRESS,
  83   PROP_CLOCK,
  84   PROP_ACTIVE
  85       /* FILL ME */
  86 };
  87
  88 #define GST_NET_TIME_PROVIDER_GET_PRIVATE(obj)  \
  89    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_NET_TIME_PROVIDER, GstNetTimeProviderPrivate))
  90
  91 struct _GstNetTimeProviderPrivate
  92 {
  93   GstPollFD sock;
  94   GstPoll *fdset;
  95 };
  96
  97 static gboolean gst_net_time_provider_start (GstNetTimeProvider * bself);
  98 static void gst_net_time_provider_stop (GstNetTimeProvider * bself);
  99
 100 static gpointer gst_net_time_provider_thread (gpointer data);
 101
 102 static void gst_net_time_provider_finalize (GObject * object);
 103 static void gst_net_time_provider_set_property (GObject * object, guint prop_id,
 104     const GValue * value, GParamSpec * pspec);
 105 static void gst_net_time_provider_get_property (GObject * object, guint prop_id,
 106     GValue * value, GParamSpec * pspec);
 107
 108 #define _do_init \
 109   GST_DEBUG_CATEGORY_INIT (ntp_debug, "nettime", 0, "Network time provider");
 110
 111 #define gst_net_time_provider_parent_class parent_class
 112 G_DEFINE_TYPE_WITH_CODE (GstNetTimeProvider, gst_net_time_provider,
 113     GST_TYPE_OBJECT, _do_init);
 114
 115 #ifdef G_OS_WIN32
 116 static int
 117 inet_aton (const char *c, struct in_addr *paddr)
 118 {
 119   paddr->s_addr = inet_addr (c);
 120   if (paddr->s_addr == INADDR_NONE)
 121     return 0;
 122
 123   return 1;
 124 }
 125 #endif
 126
 127 static void
 128 gst_net_time_provider_class_init (GstNetTimeProviderClass * klass)
 129 {
 130   GObjectClass *gobject_class;
 131
 132   gobject_class = G_OBJECT_CLASS (klass);
 133
 134   g_assert (sizeof (GstClockTime) == 8);
 135
 136   g_type_class_add_private (klass, sizeof (GstNetTimeProviderPrivate));
 137
 138   gobject_class->finalize = gst_net_time_provider_finalize;
 139   gobject_class->set_property = gst_net_time_provider_set_property;
 140   gobject_class->get_property = gst_net_time_provider_get_property;
 141
 142   g_object_class_install_property (gobject_class, PROP_PORT,
 143       g_param_spec_int ("port", "port",
 144           "The port to receive the packets from, 0=allocate", 0, G_MAXUINT16,
 145           DEFAULT_PORT, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 146   g_object_class_install_property (gobject_class, PROP_ADDRESS,
 147       g_param_spec_string ("address", "address",
 148           "The address to bind on, as a dotted quad (x.x.x.x)",
 149           DEFAULT_ADDRESS, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 150   g_object_class_install_property (gobject_class, PROP_CLOCK,
 151       g_param_spec_object ("clock", "Clock",
 152           "The clock to export over the network", GST_TYPE_CLOCK,
 153           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 154   g_object_class_install_property (gobject_class, PROP_ACTIVE,
 155       g_param_spec_boolean ("active", "Active",
 156           "TRUE if the clock will respond to queries over the network", TRUE,
 157           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 158 }
 159
 160 static void
 161 gst_net_time_provider_init (GstNetTimeProvider * self)
 162 {
 163 #ifdef G_OS_WIN32
 164   WSADATA w;
 165   int error = WSAStartup (0x0202, &w);
 166
 167   if (error) {
 168     GST_DEBUG_OBJECT (self, "Error on WSAStartup");
 169   }
 170   if (w.wVersion != 0x0202) {
 171     WSACleanup ();
 172   }
 173 #endif
 174   self->priv = GST_NET_TIME_PROVIDER_GET_PRIVATE (self);
 175
 176   self->port = DEFAULT_PORT;
 177   self->priv->sock.fd = -1;
 178   self->address = g_strdup (DEFAULT_ADDRESS);
 179   self->thread = NULL;
 180   self->active = TRUE;
 181 }
 182
 183 static void
 184 gst_net_time_provider_finalize (GObject * object)
 185 {
 186   GstNetTimeProvider *self = GST_NET_TIME_PROVIDER (object);
 187
 188   if (self->thread) {
 189     gst_net_time_provider_stop (self);
 190     g_assert (self->thread == NULL);
 191   }
 192
 193   if (self->priv->fdset) {
 194     gst_poll_free (self->priv->fdset);
 195     self->priv->fdset = NULL;
 196   }
 197
 198   g_free (self->address);
 199   self->address = NULL;
 200
 201   if (self->clock)
 202     gst_object_unref (self->clock);
 203   self->clock = NULL;
 204
 205 #ifdef G_OS_WIN32
 206   WSACleanup ();
 207 #endif
 208
 209   G_OBJECT_CLASS (parent_class)->finalize (object);
 210 }
 211
 212 static gpointer
 213 gst_net_time_provider_thread (gpointer data)
 214 {
 215   GstNetTimeProvider *self = data;
 216   struct sockaddr_in tmpaddr;
 217   socklen_t len;
 218   GstNetTimePacket *packet;
 219   gint ret;
 220
 221   while (TRUE) {
 222     GST_LOG_OBJECT (self, "doing select");
 223     ret = gst_poll_wait (self->priv->fdset, GST_CLOCK_TIME_NONE);
 224     GST_LOG_OBJECT (self, "select returned %d", ret);
 225
 226     if (ret <= 0) {
 227       if (errno == EBUSY) {
 228         GST_LOG_OBJECT (self, "stop");
 229         goto stopped;
 230       } else if (errno != EAGAIN && errno != EINTR)
 231         goto select_error;
 232       else
 233         continue;
 234     } else {
 235       /* got data in */
 236       len = sizeof (struct sockaddr);
 237
 238       packet = gst_net_time_packet_receive (self->priv->sock.fd,
 239           (struct sockaddr *) &tmpaddr, &len);
 240
 241       if (!packet)
 242         goto receive_error;
 243
 244       if (IS_ACTIVE (self)) {
 245         /* do what we were asked to and send the packet back */
 246         packet->remote_time = gst_clock_get_time (self->clock);
 247
 248         /* ignore errors */
 249         gst_net_time_packet_send (packet, self->priv->sock.fd,
 250             (struct sockaddr *) &tmpaddr, len);
 251       }
 252
 253       g_free (packet);
 254
 255       continue;
 256     }
 257
 258     g_assert_not_reached ();
 259
 260     /* log errors and keep going */
 261   select_error:
 262     {
 263       GST_DEBUG_OBJECT (self, "select error %d: %s (%d)", ret,
 264           g_strerror (errno), errno);
 265       continue;
 266     }
 267   stopped:
 268     {
 269       GST_DEBUG_OBJECT (self, "shutting down");
 270       /* close socket */
 271       return NULL;
 272     }
 273   receive_error:
 274     {
 275       GST_DEBUG_OBJECT (self, "receive error");
 276       continue;
 277     }
 278
 279     g_assert_not_reached ();
 280
 281   }
 282
 283   g_assert_not_reached ();
 284
 285   return NULL;
 286 }
 287
 288 static void
 289 gst_net_time_provider_set_property (GObject * object, guint prop_id,
 290     const GValue * value, GParamSpec * pspec)
 291 {
 292   GstNetTimeProvider *self = GST_NET_TIME_PROVIDER (object);
 293   GstClock **clock_p = &self->clock;
 294
 295   switch (prop_id) {
 296     case PROP_PORT:
 297       self->port = g_value_get_int (value);
 298       break;
 299     case PROP_ADDRESS:
 300       g_free (self->address);
 301       if (g_value_get_string (value) == NULL)
 302         self->address = g_strdup (DEFAULT_ADDRESS);
 303       else
 304         self->address = g_strdup (g_value_get_string (value));
 305       break;
 306     case PROP_CLOCK:
 307       gst_object_replace ((GstObject **) clock_p,
 308           (GstObject *) g_value_get_object (value));
 309       break;
 310     case PROP_ACTIVE:
 311       g_atomic_int_set (&self->active, g_value_get_boolean (value));
 312       break;
 313     default:
 314       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 315       break;
 316   }
 317 }
 318
 319 static void
 320 gst_net_time_provider_get_property (GObject * object, guint prop_id,
 321     GValue * value, GParamSpec * pspec)
 322 {
 323   GstNetTimeProvider *self = GST_NET_TIME_PROVIDER (object);
 324
 325   switch (prop_id) {
 326     case PROP_PORT:
 327       g_value_set_int (value, self->port);
 328       break;
 329     case PROP_ADDRESS:
 330       g_value_set_string (value, self->address);
 331       break;
 332     case PROP_CLOCK:
 333       g_value_set_object (value, self->clock);
 334       break;
 335     case PROP_ACTIVE:
 336       g_value_set_boolean (value, IS_ACTIVE (self));
 337       break;
 338     default:
 339       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 340       break;
 341   }
 342 }
 343
 344 static gboolean
 345 gst_net_time_provider_start (GstNetTimeProvider * self)
 346 {
 347   gint ru;
 348   struct sockaddr_in my_addr;
 349   socklen_t len;
 350   int port;
 351   gint ret;
 352   GError *error;
 353
 354   if ((ret = socket (AF_INET, SOCK_DGRAM, 0)) < 0)
 355     goto no_socket;
 356
 357   self->priv->sock.fd = ret;
 358
 359   ru = 1;
 360   ret =
 361       setsockopt (self->priv->sock.fd, SOL_SOCKET, SO_REUSEADDR, &ru,
 362       sizeof (ru));
 363   if (ret < 0)
 364     goto setsockopt_error;
 365
 366   memset (&my_addr, 0, sizeof (my_addr));
 367   my_addr.sin_family = AF_INET; /* host byte order */
 368   my_addr.sin_port = htons ((gint16) self->port);       /* short, network byte order */
 369   my_addr.sin_addr.s_addr = INADDR_ANY;
 370   if (self->address) {
 371     ret = inet_aton (self->address, &my_addr.sin_addr);
 372     if (ret == 0)
 373       goto invalid_address_error;
 374   }
 375
 376   GST_DEBUG_OBJECT (self, "binding on port %d", self->port);
 377   ret =
 378       bind (self->priv->sock.fd, (struct sockaddr *) &my_addr,
 379       sizeof (my_addr));
 380   if (ret < 0)
 381     goto bind_error;
 382
 383   len = sizeof (my_addr);
 384   ret = getsockname (self->priv->sock.fd, (struct sockaddr *) &my_addr, &len);
 385   if (ret < 0)
 386     goto getsockname_error;
 387
 388   port = ntohs (my_addr.sin_port);
 389   GST_DEBUG_OBJECT (self, "bound, on port %d", port);
 390
 391   if (port != self->port) {
 392     self->port = port;
 393     GST_DEBUG_OBJECT (self, "notifying %d", port);
 394     g_object_notify (G_OBJECT (self), "port");
 395   }
 396
 397   gst_poll_add_fd (self->priv->fdset, &self->priv->sock);
 398   gst_poll_fd_ctl_read (self->priv->fdset, &self->priv->sock, TRUE);
 399
 400   self->thread = g_thread_create (gst_net_time_provider_thread, self, TRUE,
 401       &error);
 402   if (!self->thread)
 403     goto no_thread;
 404
 405   return TRUE;
 406
 407   /* ERRORS */
 408 no_socket:
 409   {
 410     GST_ERROR_OBJECT (self, "socket failed %d: %s (%d)", ret,
 411         g_strerror (errno), errno);
 412     return FALSE;
 413   }
 414 setsockopt_error:
 415   {
 416     close (self->priv->sock.fd);
 417     self->priv->sock.fd = -1;
 418     GST_ERROR_OBJECT (self, "setsockopt failed %d: %s (%d)", ret,
 419         g_strerror (errno), errno);
 420     return FALSE;
 421   }
 422 invalid_address_error:
 423   {
 424     close (self->priv->sock.fd);
 425     self->priv->sock.fd = -1;
 426     GST_ERROR_OBJECT (self, "invalid network address %s: %s (%d)",
 427         self->address, g_strerror (errno), errno);
 428     return FALSE;
 429   }
 430 bind_error:
 431   {
 432     close (self->priv->sock.fd);
 433     self->priv->sock.fd = -1;
 434     GST_ERROR_OBJECT (self, "bind failed %d: %s (%d)", ret,
 435         g_strerror (errno), errno);
 436     return FALSE;
 437   }
 438 getsockname_error:
 439   {
 440     close (self->priv->sock.fd);
 441     self->priv->sock.fd = -1;
 442     GST_ERROR_OBJECT (self, "getsockname failed %d: %s (%d)", ret,
 443         g_strerror (errno), errno);
 444     return FALSE;
 445   }
 446 no_thread:
 447   {
 448     gst_poll_remove_fd (self->priv->fdset, &self->priv->sock);
 449     close (self->priv->sock.fd);
 450     self->priv->sock.fd = -1;
 451     GST_ERROR_OBJECT (self, "could not create thread: %s", error->message);
 452     g_error_free (error);
 453     return FALSE;
 454   }
 455 }
 456
 457 static void
 458 gst_net_time_provider_stop (GstNetTimeProvider * self)
 459 {
 460   gst_poll_set_flushing (self->priv->fdset, TRUE);
 461   g_thread_join (self->thread);
 462   self->thread = NULL;
 463
 464   if (self->priv->sock.fd != -1) {
 465     gst_poll_remove_fd (self->priv->fdset, &self->priv->sock);
 466     close (self->priv->sock.fd);
 467     self->priv->sock.fd = -1;
 468   }
 469 }
 470
 471 /**
 472  * gst_net_time_provider_new:
 473  * @clock: a #GstClock to export over the network
 474  * @address: an address to bind on as a dotted quad (xxx.xxx.xxx.xxx), or NULL
 475  *           to bind to all addresses
 476  * @port: a port to bind on, or 0 to let the kernel choose
 477  *
 478  * Allows network clients to get the current time of @clock.
 479  *
 480  * Returns: the new #GstNetTimeProvider, or NULL on error
 481  */
 482 GstNetTimeProvider *
 483 gst_net_time_provider_new (GstClock * clock, const gchar * address, gint port)
 484 {
 485   GstNetTimeProvider *ret;
 486
 487   g_return_val_if_fail (clock && GST_IS_CLOCK (clock), NULL);
 488   g_return_val_if_fail (port >= 0 && port <= G_MAXUINT16, NULL);
 489
 490   ret = g_object_new (GST_TYPE_NET_TIME_PROVIDER, "clock", clock, "address",
 491       address, "port", port, NULL);
 492
 493   if ((ret->priv->fdset = gst_poll_new (TRUE)) == NULL)
 494     goto no_fdset;
 495
 496   if (!gst_net_time_provider_start (ret))
 497     goto failed_start;
 498
 499   /* all systems go, cap'n */
 500   return ret;
 501
 502 no_fdset:
 503   {
 504     GST_ERROR_OBJECT (ret, "could not create an fdset: %s (%d)",
 505         g_strerror (errno), errno);
 506     gst_object_unref (ret);
 507     return NULL;
 508   }
 509 failed_start:
 510   {
 511     /* already printed a nice error */
 512     gst_object_unref (ret);
 513     return NULL;
 514   }
 515 }