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., 51 Franklin St, Fifth Floor,
  17  * Boston, MA 02110-1301, USA.
  18  */
  19 /**
  20  * SECTION:gstnettimeprovider
  21  * @title: GstNetTimeProvider
  22  * @short_description: Special object that exposed the time of a clock
  23  *                     on the network.
  24  * @see_also: #GstClock, #GstNetClientClock, #GstPipeline
  25  *
  26  * This object exposes the time of a #GstClock on the network.
  27  *
  28  * A #GstNetTimeProvider is created with gst_net_time_provider_new() which
  29  * takes a #GstClock, an address and a port number as arguments.
  30  *
  31  * After creating the object, a client clock such as #GstNetClientClock can
  32  * query the exposed clock over the network for its values.
  33  *
  34  * The #GstNetTimeProvider typically wraps the clock used by a #GstPipeline.
  35  */
  36
  37 #ifdef HAVE_CONFIG_H
  38 #include "config.h"
  39 #endif
  40
  41 #include "gstnettimeprovider.h"
  42 #include "gstnettimepacket.h"
  43
  44 GST_DEBUG_CATEGORY_STATIC (ntp_debug);
  45 #define GST_CAT_DEFAULT (ntp_debug)
  46
  47 #define DEFAULT_ADDRESS         "0.0.0.0"
  48 #define DEFAULT_PORT            5637
  49
  50 #define IS_ACTIVE(self) (g_atomic_int_get (&((self)->priv->active)))
  51
  52 enum
  53 {
  54   PROP_0,
  55   PROP_PORT,
  56   PROP_ADDRESS,
  57   PROP_CLOCK,
  58   PROP_ACTIVE
  59 };
  60
  61 #define GST_NET_TIME_PROVIDER_GET_PRIVATE(obj)  \
  62    (G_TYPE_INSTANCE_GET_PRIVATE ((obj), GST_TYPE_NET_TIME_PROVIDER, GstNetTimeProviderPrivate))
  63
  64 struct _GstNetTimeProviderPrivate
  65 {
  66   gchar *address;
  67   int port;
  68
  69   GThread *thread;
  70
  71   GstClock *clock;
  72
  73   gboolean active;              /* ATOMIC */
  74
  75   GSocket *socket;
  76   GCancellable *cancel;
  77   gboolean made_cancel_fd;
  78 };
  79
  80 static void gst_net_time_provider_initable_iface_init (gpointer g_iface);
  81
  82 static gboolean gst_net_time_provider_start (GstNetTimeProvider * bself,
  83     GError ** error);
  84 static void gst_net_time_provider_stop (GstNetTimeProvider * bself);
  85
  86 static gpointer gst_net_time_provider_thread (gpointer data);
  87
  88 static void gst_net_time_provider_finalize (GObject * object);
  89 static void gst_net_time_provider_set_property (GObject * object, guint prop_id,
  90     const GValue * value, GParamSpec * pspec);
  91 static void gst_net_time_provider_get_property (GObject * object, guint prop_id,
  92     GValue * value, GParamSpec * pspec);
  93
  94 #define _do_init \
  95   GST_DEBUG_CATEGORY_INIT (ntp_debug, "nettime", 0, "Network time provider"); \
  96   G_IMPLEMENT_INTERFACE (G_TYPE_INITABLE, gst_net_time_provider_initable_iface_init)
  97
  98 #define gst_net_time_provider_parent_class parent_class
  99 G_DEFINE_TYPE_WITH_CODE (GstNetTimeProvider, gst_net_time_provider,
 100     GST_TYPE_OBJECT, _do_init);
 101
 102 static void
 103 gst_net_time_provider_class_init (GstNetTimeProviderClass * klass)
 104 {
 105   GObjectClass *gobject_class;
 106
 107   gobject_class = G_OBJECT_CLASS (klass);
 108
 109   g_assert (sizeof (GstClockTime) == 8);
 110
 111   g_type_class_add_private (klass, sizeof (GstNetTimeProviderPrivate));
 112
 113   gobject_class->finalize = gst_net_time_provider_finalize;
 114   gobject_class->set_property = gst_net_time_provider_set_property;
 115   gobject_class->get_property = gst_net_time_provider_get_property;
 116
 117   g_object_class_install_property (gobject_class, PROP_PORT,
 118       g_param_spec_int ("port", "port",
 119           "The port to receive the packets from, 0=allocate", 0, G_MAXUINT16,
 120           DEFAULT_PORT,
 121           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 122   g_object_class_install_property (gobject_class, PROP_ADDRESS,
 123       g_param_spec_string ("address", "address",
 124           "The address to bind on, as a dotted quad (x.x.x.x)", DEFAULT_ADDRESS,
 125           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 126   g_object_class_install_property (gobject_class, PROP_CLOCK,
 127       g_param_spec_object ("clock", "Clock",
 128           "The clock to export over the network", GST_TYPE_CLOCK,
 129           G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 130   g_object_class_install_property (gobject_class, PROP_ACTIVE,
 131       g_param_spec_boolean ("active", "Active",
 132           "TRUE if the clock will respond to queries over the network", TRUE,
 133           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 134 }
 135
 136 static void
 137 gst_net_time_provider_init (GstNetTimeProvider * self)
 138 {
 139   self->priv = GST_NET_TIME_PROVIDER_GET_PRIVATE (self);
 140
 141   self->priv->port = DEFAULT_PORT;
 142   self->priv->address = g_strdup (DEFAULT_ADDRESS);
 143   self->priv->thread = NULL;
 144   self->priv->active = TRUE;
 145 }
 146
 147 static void
 148 gst_net_time_provider_finalize (GObject * object)
 149 {
 150   GstNetTimeProvider *self = GST_NET_TIME_PROVIDER (object);
 151
 152   if (self->priv->thread) {
 153     gst_net_time_provider_stop (self);
 154     g_assert (self->priv->thread == NULL);
 155   }
 156
 157   g_free (self->priv->address);
 158   self->priv->address = NULL;
 159
 160   if (self->priv->clock)
 161     gst_object_unref (self->priv->clock);
 162   self->priv->clock = NULL;
 163
 164   G_OBJECT_CLASS (parent_class)->finalize (object);
 165 }
 166
 167 static gpointer
 168 gst_net_time_provider_thread (gpointer data)
 169 {
 170   GstNetTimeProvider *self = data;
 171   GCancellable *cancel = self->priv->cancel;
 172   GSocket *socket = self->priv->socket;
 173   GstNetTimePacket *packet;
 174   GError *err = NULL;
 175
 176   GST_INFO_OBJECT (self, "time provider thread is running");
 177
 178   while (TRUE) {
 179     GSocketAddress *sender_addr = NULL;
 180
 181     GST_LOG_OBJECT (self, "waiting on socket");
 182     if (!g_socket_condition_wait (socket, G_IO_IN, cancel, &err)) {
 183       GST_INFO_OBJECT (self, "socket error: %s", err->message);
 184
 185       if (err->code == G_IO_ERROR_CANCELLED)
 186         break;
 187
 188       /* try again */
 189       g_usleep (G_USEC_PER_SEC / 10);
 190       g_error_free (err);
 191       err = NULL;
 192       continue;
 193     }
 194
 195     /* got data in */
 196     packet = gst_net_time_packet_receive (socket, &sender_addr, &err);
 197
 198     if (err != NULL) {
 199       GST_DEBUG_OBJECT (self, "receive error: %s", err->message);
 200       g_usleep (G_USEC_PER_SEC / 10);
 201       g_error_free (err);
 202       err = NULL;
 203       continue;
 204     }
 205
 206     if (IS_ACTIVE (self)) {
 207       /* do what we were asked to and send the packet back */
 208       packet->remote_time = gst_clock_get_time (self->priv->clock);
 209
 210       /* ignore errors */
 211       gst_net_time_packet_send (packet, socket, sender_addr, NULL);
 212       g_object_unref (sender_addr);
 213       g_free (packet);
 214     }
 215   }
 216
 217   g_error_free (err);
 218
 219   GST_INFO_OBJECT (self, "time provider thread is stopping");
 220   return NULL;
 221 }
 222
 223 static void
 224 gst_net_time_provider_set_property (GObject * object, guint prop_id,
 225     const GValue * value, GParamSpec * pspec)
 226 {
 227   GstNetTimeProvider *self = GST_NET_TIME_PROVIDER (object);
 228   GstClock **clock_p = &self->priv->clock;
 229
 230   switch (prop_id) {
 231     case PROP_PORT:
 232       self->priv->port = g_value_get_int (value);
 233       break;
 234     case PROP_ADDRESS:
 235       g_free (self->priv->address);
 236       if (g_value_get_string (value) == NULL)
 237         self->priv->address = g_strdup (DEFAULT_ADDRESS);
 238       else
 239         self->priv->address = g_strdup (g_value_get_string (value));
 240       break;
 241     case PROP_CLOCK:
 242       gst_object_replace ((GstObject **) clock_p,
 243           (GstObject *) g_value_get_object (value));
 244       break;
 245     case PROP_ACTIVE:
 246       g_atomic_int_set (&self->priv->active, g_value_get_boolean (value));
 247       break;
 248     default:
 249       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 250       break;
 251   }
 252 }
 253
 254 static void
 255 gst_net_time_provider_get_property (GObject * object, guint prop_id,
 256     GValue * value, GParamSpec * pspec)
 257 {
 258   GstNetTimeProvider *self = GST_NET_TIME_PROVIDER (object);
 259
 260   switch (prop_id) {
 261     case PROP_PORT:
 262       g_value_set_int (value, self->priv->port);
 263       break;
 264     case PROP_ADDRESS:
 265       g_value_set_string (value, self->priv->address);
 266       break;
 267     case PROP_CLOCK:
 268       g_value_set_object (value, self->priv->clock);
 269       break;
 270     case PROP_ACTIVE:
 271       g_value_set_boolean (value, IS_ACTIVE (self));
 272       break;
 273     default:
 274       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 275       break;
 276   }
 277 }
 278
 279 static gboolean
 280 gst_net_time_provider_start (GstNetTimeProvider * self, GError ** error)
 281 {
 282   GSocketAddress *socket_addr, *bound_addr;
 283   GInetAddress *inet_addr;
 284   GPollFD dummy_pollfd;
 285   GSocket *socket;
 286   int port;
 287   gchar *address;
 288   GError *err = NULL;
 289
 290   if (self->priv->address) {
 291     inet_addr = g_inet_address_new_from_string (self->priv->address);
 292     if (inet_addr == NULL) {
 293       err =
 294           g_error_new (G_IO_ERROR, G_IO_ERROR_FAILED,
 295           "Failed to parse address '%s'", self->priv->address);
 296       goto invalid_address;
 297     }
 298   } else {
 299     inet_addr = g_inet_address_new_any (G_SOCKET_FAMILY_IPV4);
 300   }
 301
 302   GST_LOG_OBJECT (self, "creating socket");
 303   socket = g_socket_new (g_inet_address_get_family (inet_addr),
 304       G_SOCKET_TYPE_DATAGRAM, G_SOCKET_PROTOCOL_UDP, &err);
 305
 306   if (!socket)
 307     goto no_socket;
 308
 309   GST_DEBUG_OBJECT (self, "binding on port %d", self->priv->port);
 310   socket_addr = g_inet_socket_address_new (inet_addr, self->priv->port);
 311   if (!g_socket_bind (socket, socket_addr, TRUE, &err)) {
 312     g_object_unref (socket_addr);
 313     g_object_unref (inet_addr);
 314     goto bind_error;
 315   }
 316   g_object_unref (socket_addr);
 317   g_object_unref (inet_addr);
 318
 319   bound_addr = g_socket_get_local_address (socket, NULL);
 320   port = g_inet_socket_address_get_port (G_INET_SOCKET_ADDRESS (bound_addr));
 321   inet_addr =
 322       g_inet_socket_address_get_address (G_INET_SOCKET_ADDRESS (bound_addr));
 323   address = g_inet_address_to_string (inet_addr);
 324
 325   if (g_strcmp0 (address, self->priv->address)) {
 326     g_free (self->priv->address);
 327     self->priv->address = address;
 328     GST_DEBUG_OBJECT (self, "notifying address %s", address);
 329     g_object_notify (G_OBJECT (self), "address");
 330   } else {
 331     g_free (address);
 332   }
 333   if (port != self->priv->port) {
 334     self->priv->port = port;
 335     GST_DEBUG_OBJECT (self, "notifying port %d", port);
 336     g_object_notify (G_OBJECT (self), "port");
 337   }
 338   GST_DEBUG_OBJECT (self, "bound on UDP address %s, port %d",
 339       self->priv->address, port);
 340   g_object_unref (bound_addr);
 341
 342   self->priv->socket = socket;
 343   self->priv->cancel = g_cancellable_new ();
 344   self->priv->made_cancel_fd =
 345       g_cancellable_make_pollfd (self->priv->cancel, &dummy_pollfd);
 346
 347   self->priv->thread = g_thread_try_new ("GstNetTimeProvider",
 348       gst_net_time_provider_thread, self, &err);
 349
 350   if (!self->priv->thread)
 351     goto no_thread;
 352
 353   return TRUE;
 354
 355   /* ERRORS */
 356 invalid_address:
 357   {
 358     GST_ERROR_OBJECT (self, "invalid address: %s", self->priv->address);
 359     g_propagate_error (error, err);
 360     return FALSE;
 361   }
 362 no_socket:
 363   {
 364     GST_ERROR_OBJECT (self, "could not create socket: %s", err->message);
 365     g_propagate_error (error, err);
 366     g_object_unref (inet_addr);
 367     return FALSE;
 368   }
 369 bind_error:
 370   {
 371     GST_ERROR_OBJECT (self, "bind failed: %s", err->message);
 372     g_propagate_error (error, err);
 373     g_object_unref (socket);
 374     return FALSE;
 375   }
 376 no_thread:
 377   {
 378     GST_ERROR_OBJECT (self, "could not create thread: %s", err->message);
 379     g_propagate_error (error, err);
 380     g_object_unref (self->priv->socket);
 381     self->priv->socket = NULL;
 382     g_object_unref (self->priv->cancel);
 383     self->priv->cancel = NULL;
 384     return FALSE;
 385   }
 386 }
 387
 388 static void
 389 gst_net_time_provider_stop (GstNetTimeProvider * self)
 390 {
 391   g_return_if_fail (self->priv->thread != NULL);
 392
 393   GST_INFO_OBJECT (self, "stopping..");
 394   g_cancellable_cancel (self->priv->cancel);
 395
 396   g_thread_join (self->priv->thread);
 397   self->priv->thread = NULL;
 398
 399   if (self->priv->made_cancel_fd)
 400     g_cancellable_release_fd (self->priv->cancel);
 401
 402   g_object_unref (self->priv->cancel);
 403   self->priv->cancel = NULL;
 404
 405   g_object_unref (self->priv->socket);
 406   self->priv->socket = NULL;
 407
 408   GST_INFO_OBJECT (self, "stopped");
 409 }
 410
 411 static gboolean
 412 gst_net_time_provider_initable_init (GInitable * initable,
 413     GCancellable * cancellable, GError ** error)
 414 {
 415   GstNetTimeProvider *self = GST_NET_TIME_PROVIDER (initable);
 416
 417   return gst_net_time_provider_start (self, error);
 418 }
 419
 420 static void
 421 gst_net_time_provider_initable_iface_init (gpointer g_iface)
 422 {
 423   GInitableIface *iface = g_iface;
 424
 425   iface->init = gst_net_time_provider_initable_init;
 426 }
 427
 428 /**
 429  * gst_net_time_provider_new:
 430  * @clock: a #GstClock to export over the network
 431  * @address: (allow-none): an address to bind on as a dotted quad
 432  *           (xxx.xxx.xxx.xxx), IPv6 address, or NULL to bind to all addresses
 433  * @port: a port to bind on, or 0 to let the kernel choose
 434  *
 435  * Allows network clients to get the current time of @clock.
 436  *
 437  * Returns: (transfer full): the new #GstNetTimeProvider, or NULL on error
 438  */
 439 GstNetTimeProvider *
 440 gst_net_time_provider_new (GstClock * clock, const gchar * address, gint port)
 441 {
 442   GstNetTimeProvider *ret;
 443
 444   g_return_val_if_fail (clock && GST_IS_CLOCK (clock), NULL);
 445   g_return_val_if_fail (port >= 0 && port <= G_MAXUINT16, NULL);
 446
 447   ret =
 448       g_initable_new (GST_TYPE_NET_TIME_PROVIDER, NULL, NULL, "clock", clock,
 449       "address", address, "port", port, NULL);
 450
 451   /* Clear floating flag */
 452   g_object_ref_sink (ret);
 453
 454   return ret;
 455 }