subprojects/gst-plugins-bad/ext/srt/gstsrtsrc.c

   1 /* GStreamer
   2  * Copyright (C) 2018, Collabora Ltd.
   3  * Copyright (C) 2018, SK Telecom, Co., Ltd.
   4  *   Author: Jeongseok Kim <jeongseok.kim@sk.com>
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  19  * Boston, MA 02110-1301, USA.
  20  */
  21
  22 /**
  23  * SECTION:element-srtsrc
  24  * @title: srtsrc
  25  *
  26  * srtsrc is a network source that reads [SRT](http://www.srtalliance.org/)
  27  * packets from the network.
  28  *
  29  * ## Examples
  30  * |[
  31  * gst-launch-1.0 -v srtsrc uri="srt://127.0.0.1:7001" ! fakesink
  32  * ]| This pipeline shows how to connect SRT server by setting #GstSRTSrc:uri property.
  33  *
  34  * |[
  35  * gst-launch-1.0 -v srtsrc uri="srt://:7001?mode=listener" ! fakesink
  36  * ]| This pipeline shows how to wait SRT connection by setting #GstSRTSrc:uri property.
  37  *
  38  * |[
  39  * gst-launch-1.0 -v srtclientsrc uri="srt://192.168.1.10:7001?mode=rendez-vous" ! fakesink
  40  * ]| This pipeline shows how to connect SRT server by setting #GstSRTSrc:uri property and using the rendez-vous mode.
  41  *
  42  */
  43
  44 #ifdef HAVE_CONFIG_H
  45 #include <config.h>
  46 #endif
  47
  48 #include "gstsrtelements.h"
  49 #include "gstsrtsrc.h"
  50
  51 static GstStaticPadTemplate src_template = GST_STATIC_PAD_TEMPLATE ("src",
  52     GST_PAD_SRC,
  53     GST_PAD_ALWAYS,
  54     GST_STATIC_CAPS_ANY);
  55
  56 #define GST_CAT_DEFAULT gst_debug_srt_src
  57 GST_DEBUG_CATEGORY (GST_CAT_DEFAULT);
  58
  59 enum
  60 {
  61   SIG_CALLER_ADDED,
  62   SIG_CALLER_REMOVED,
  63   SIG_CALLER_REJECTED,
  64   SIG_CALLER_CONNECTING,
  65   LAST_SIGNAL
  66 };
  67
  68 static guint signals[LAST_SIGNAL] = { 0 };
  69
  70 static void gst_srt_src_uri_handler_init (gpointer g_iface,
  71     gpointer iface_data);
  72 static gchar *gst_srt_src_uri_get_uri (GstURIHandler * handler);
  73 static gboolean gst_srt_src_uri_set_uri (GstURIHandler * handler,
  74     const gchar * uri, GError ** error);
  75 static gboolean src_default_caller_connecting (GstSRTSrc * self,
  76     GSocketAddress * addr, const gchar * username, gpointer data);
  77 static gboolean src_authentication_accumulator (GSignalInvocationHint * ihint,
  78     GValue * return_accu, const GValue * handler_return, gpointer data);
  79
  80 #define gst_srt_src_parent_class parent_class
  81 G_DEFINE_TYPE_WITH_CODE (GstSRTSrc, gst_srt_src,
  82     GST_TYPE_PUSH_SRC,
  83     G_IMPLEMENT_INTERFACE (GST_TYPE_URI_HANDLER, gst_srt_src_uri_handler_init)
  84     GST_DEBUG_CATEGORY_INIT (GST_CAT_DEFAULT, "srtsrc", 0, "SRT Source"));
  85 GST_ELEMENT_REGISTER_DEFINE_WITH_CODE (srtsrc, "srtsrc", GST_RANK_PRIMARY,
  86     GST_TYPE_SRT_SRC, srt_element_init (plugin));
  87
  88 static gboolean
  89 src_default_caller_connecting (GstSRTSrc * self,
  90     GSocketAddress * addr, const gchar * stream_id, gpointer data)
  91 {
  92   /* Accept all connections. */
  93   return TRUE;
  94 }
  95
  96 static gboolean
  97 src_authentication_accumulator (GSignalInvocationHint * ihint,
  98     GValue * return_accu, const GValue * handler_return, gpointer data)
  99 {
 100   gboolean ret = g_value_get_boolean (handler_return);
 101   /* Handlers return TRUE on authentication success and we want to stop on
 102    * the first failure. */
 103   g_value_set_boolean (return_accu, ret);
 104   return ret;
 105 }
 106
 107 static gboolean
 108 gst_srt_src_start (GstBaseSrc * bsrc)
 109 {
 110   GstSRTSrc *self = GST_SRT_SRC (bsrc);
 111   GError *error = NULL;
 112   gboolean ret = FALSE;
 113
 114   ret = gst_srt_object_open (self->srtobject, self->cancellable, &error);
 115
 116   if (!ret) {
 117     /* ensure error is posted since state change will fail */
 118     GST_ELEMENT_ERROR (self, RESOURCE, OPEN_READ, (NULL),
 119         ("Failed to open SRT: %s", error->message));
 120     g_clear_error (&error);
 121   }
 122
 123   /* Reset expected pktseq */
 124   self->next_pktseq = 0;
 125
 126   return ret;
 127 }
 128
 129 static gboolean
 130 gst_srt_src_stop (GstBaseSrc * bsrc)
 131 {
 132   GstSRTSrc *self = GST_SRT_SRC (bsrc);
 133
 134   gst_srt_object_close (self->srtobject);
 135
 136   return TRUE;
 137 }
 138
 139 static GstFlowReturn
 140 gst_srt_src_fill (GstPushSrc * src, GstBuffer * outbuf)
 141 {
 142   GstSRTSrc *self = GST_SRT_SRC (src);
 143   GstFlowReturn ret = GST_FLOW_OK;
 144   GstMapInfo info;
 145   GError *err = NULL;
 146   gssize recv_len;
 147   GstClock *clock;
 148   GstClockTime base_time;
 149   GstClockTime capture_time;
 150   GstClockTimeDiff delay;
 151   int64_t srt_time;
 152   SRT_MSGCTRL mctrl;
 153
 154   if (g_cancellable_is_cancelled (self->cancellable)) {
 155     ret = GST_FLOW_FLUSHING;
 156   }
 157
 158   if (!gst_buffer_map (outbuf, &info, GST_MAP_WRITE)) {
 159     GST_ELEMENT_ERROR (src, RESOURCE, READ,
 160         ("Could not map the buffer for writing "), (NULL));
 161     ret = GST_FLOW_ERROR;
 162     goto out;
 163   }
 164
 165   /* Get clock and values */
 166   clock = gst_element_get_clock (GST_ELEMENT (src));
 167   if (!clock) {
 168     GST_DEBUG_OBJECT (src, "Clock missing, flushing");
 169     return GST_FLOW_FLUSHING;
 170   }
 171
 172   base_time = gst_element_get_base_time (GST_ELEMENT (src));
 173
 174   recv_len = gst_srt_object_read (self->srtobject, info.data,
 175       gst_buffer_get_size (outbuf), self->cancellable, &err, &mctrl);
 176
 177   /* Capture clock values ASAP */
 178   capture_time = gst_clock_get_time (clock);
 179 #if SRT_VERSION_VALUE >= 0x10402
 180   /* Use SRT clock value if available (SRT > 1.4.2) */
 181   srt_time = srt_time_now ();
 182 #else
 183   /* Else use the unix epoch monotonic clock */
 184   srt_time = g_get_real_time ();
 185 #endif
 186   gst_object_unref (clock);
 187
 188   gst_buffer_unmap (outbuf, &info);
 189
 190   GST_LOG_OBJECT (src,
 191       "recv_len:%" G_GSIZE_FORMAT " pktseq:%d msgno:%d srctime:%"
 192       G_GINT64_FORMAT, recv_len, mctrl.pktseq, mctrl.msgno, mctrl.srctime);
 193
 194   if (g_cancellable_is_cancelled (self->cancellable)) {
 195     ret = GST_FLOW_FLUSHING;
 196     goto out;
 197   }
 198
 199   if (recv_len < 0) {
 200     GST_ELEMENT_ERROR (src, RESOURCE, READ, (NULL), ("%s", err->message));
 201     ret = GST_FLOW_ERROR;
 202     g_clear_error (&err);
 203     goto out;
 204   } else if (recv_len == 0) {
 205     ret = GST_FLOW_EOS;
 206     goto out;
 207   }
 208
 209   /* Detect discontinuities */
 210   if (mctrl.pktseq != self->next_pktseq) {
 211     GST_WARNING_OBJECT (src, "discont detected %d (expected: %d)",
 212         mctrl.pktseq, self->next_pktseq);
 213     GST_BUFFER_FLAG_SET (outbuf, GST_BUFFER_FLAG_DISCONT);
 214   }
 215   /* pktseq is a 31bit field */
 216   self->next_pktseq = (mctrl.pktseq + 1) % G_MAXINT32;
 217
 218   /* 0 means we do not have a srctime */
 219   if (mctrl.srctime != 0)
 220     delay = (srt_time - mctrl.srctime) * GST_USECOND;
 221   else
 222     delay = 0;
 223
 224   GST_LOG_OBJECT (src, "delay: %" GST_STIME_FORMAT, GST_STIME_ARGS (delay));
 225
 226   if (delay < 0) {
 227     GST_WARNING_OBJECT (src,
 228         "Calculated SRT delay %" GST_STIME_FORMAT " is negative, clamping to 0",
 229         GST_STIME_ARGS (delay));
 230     delay = 0;
 231   }
 232
 233   /* Subtract the base_time (since the pipeline started) ... */
 234   if (capture_time > base_time)
 235     capture_time -= base_time;
 236   else
 237     capture_time = 0;
 238   /* And adjust by the delay */
 239   if (capture_time > delay)
 240     capture_time -= delay;
 241   else
 242     capture_time = 0;
 243   GST_BUFFER_TIMESTAMP (outbuf) = capture_time;
 244
 245   gst_buffer_resize (outbuf, 0, recv_len);
 246
 247   GST_LOG_OBJECT (src,
 248       "filled buffer from _get of size %" G_GSIZE_FORMAT ", ts %"
 249       GST_TIME_FORMAT ", dur %" GST_TIME_FORMAT
 250       ", offset %" G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT,
 251       gst_buffer_get_size (outbuf),
 252       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (outbuf)),
 253       GST_TIME_ARGS (GST_BUFFER_DURATION (outbuf)),
 254       GST_BUFFER_OFFSET (outbuf), GST_BUFFER_OFFSET_END (outbuf));
 255
 256 out:
 257   return ret;
 258 }
 259
 260 static void
 261 gst_srt_src_init (GstSRTSrc * self)
 262 {
 263   self->srtobject = gst_srt_object_new (GST_ELEMENT (self));
 264   self->cancellable = g_cancellable_new ();
 265
 266   gst_base_src_set_format (GST_BASE_SRC (self), GST_FORMAT_TIME);
 267   gst_base_src_set_live (GST_BASE_SRC (self), TRUE);
 268   /* We do the timing ourselves */
 269   gst_base_src_set_do_timestamp (GST_BASE_SRC (self), FALSE);
 270
 271   gst_srt_object_set_uri (self->srtobject, GST_SRT_DEFAULT_URI, NULL);
 272
 273 }
 274
 275 static void
 276 gst_srt_src_finalize (GObject * object)
 277 {
 278   GstSRTSrc *self = GST_SRT_SRC (object);
 279
 280   g_clear_object (&self->cancellable);
 281   gst_srt_object_destroy (self->srtobject);
 282
 283   G_OBJECT_CLASS (parent_class)->finalize (object);
 284 }
 285
 286 static gboolean
 287 gst_srt_src_unlock (GstBaseSrc * bsrc)
 288 {
 289   GstSRTSrc *self = GST_SRT_SRC (bsrc);
 290
 291   gst_srt_object_wakeup (self->srtobject, self->cancellable);
 292
 293   return TRUE;
 294 }
 295
 296 static gboolean
 297 gst_srt_src_unlock_stop (GstBaseSrc * bsrc)
 298 {
 299   GstSRTSrc *self = GST_SRT_SRC (bsrc);
 300
 301   g_cancellable_reset (self->cancellable);
 302
 303   return TRUE;
 304 }
 305
 306 static void
 307 gst_srt_src_set_property (GObject * object,
 308     guint prop_id, const GValue * value, GParamSpec * pspec)
 309 {
 310   GstSRTSrc *self = GST_SRT_SRC (object);
 311
 312   if (!gst_srt_object_set_property_helper (self->srtobject, prop_id, value,
 313           pspec)) {
 314     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 315   }
 316 }
 317
 318 static void
 319 gst_srt_src_get_property (GObject * object,
 320     guint prop_id, GValue * value, GParamSpec * pspec)
 321 {
 322   GstSRTSrc *self = GST_SRT_SRC (object);
 323
 324   if (!gst_srt_object_get_property_helper (self->srtobject, prop_id, value,
 325           pspec)) {
 326     G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 327   }
 328 }
 329
 330 static gboolean
 331 gst_srt_src_query (GstBaseSrc * basesrc, GstQuery * query)
 332 {
 333   GstSRTSrc *self = GST_SRT_SRC (basesrc);
 334
 335   if (GST_QUERY_TYPE (query) == GST_QUERY_LATENCY) {
 336     gint latency;
 337     if (!gst_structure_get_int (self->srtobject->parameters, "latency",
 338             &latency))
 339       latency = GST_SRT_DEFAULT_LATENCY;
 340     gst_query_set_latency (query, TRUE, latency * GST_MSECOND,
 341         latency * GST_MSECOND);
 342     return TRUE;
 343   } else {
 344     return GST_BASE_SRC_CLASS (parent_class)->query (basesrc, query);
 345   }
 346 }
 347
 348 static void
 349 gst_srt_src_class_init (GstSRTSrcClass * klass)
 350 {
 351   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 352   GstElementClass *gstelement_class = GST_ELEMENT_CLASS (klass);
 353   GstBaseSrcClass *gstbasesrc_class = GST_BASE_SRC_CLASS (klass);
 354   GstPushSrcClass *gstpushsrc_class = GST_PUSH_SRC_CLASS (klass);
 355
 356   gobject_class->set_property = gst_srt_src_set_property;
 357   gobject_class->get_property = gst_srt_src_get_property;
 358   gobject_class->finalize = gst_srt_src_finalize;
 359   klass->caller_connecting = src_default_caller_connecting;
 360
 361   /**
 362    * GstSRTSrc::caller-added:
 363    * @gstsrtsrc: the srtsrc element that emitted this signal
 364    * @unused: always zero (for ABI compatibility with previous versions)
 365    * @addr: the #GSocketAddress of the new caller
 366    *
 367    * A new caller has connected to srtsrc.
 368    */
 369   signals[SIG_CALLER_ADDED] =
 370       g_signal_new ("caller-added", G_TYPE_FROM_CLASS (klass),
 371       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTSrcClass, caller_added),
 372       NULL, NULL, NULL, G_TYPE_NONE, 2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
 373
 374   /**
 375    * GstSRTSrc::caller-removed:
 376    * @gstsrtsrc: the srtsrc element that emitted this signal
 377    * @unused: always zero (for ABI compatibility with previous versions)
 378    * @addr: the #GSocketAddress of the caller
 379    *
 380    * The given caller has disconnected.
 381    */
 382   signals[SIG_CALLER_REMOVED] =
 383       g_signal_new ("caller-removed", G_TYPE_FROM_CLASS (klass),
 384       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTSrcClass,
 385           caller_added), NULL, NULL, NULL, G_TYPE_NONE,
 386       2, G_TYPE_INT, G_TYPE_SOCKET_ADDRESS);
 387
 388   /**
 389    * GstSRTSrc::caller-rejected:
 390    * @gstsrtsrc: the srtsrc element that emitted this signal
 391    * @addr: the #GSocketAddress that describes the client socket
 392    * @stream_id: the stream Id to which the caller wants to connect
 393    *
 394    * A caller's connection to srtsrc in listener mode has been rejected.
 395    *
 396    * Since: 1.20
 397    *
 398    */
 399   signals[SIG_CALLER_REJECTED] =
 400       g_signal_new ("caller-rejected", G_TYPE_FROM_CLASS (klass),
 401       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTSrcClass, caller_rejected),
 402       NULL, NULL, NULL, G_TYPE_NONE, 2, G_TYPE_SOCKET_ADDRESS, G_TYPE_STRING);
 403
 404   /**
 405    * GstSRTSrc::caller-connecting:
 406    * @gstsrtsrc: the srtsrc element that emitted this signal
 407    * @addr: the #GSocketAddress that describes the client socket
 408    * @stream_id: the stream Id to which the caller wants to connect
 409    *
 410    * Whether to accept or reject a caller's connection to srtsrc in listener mode.
 411    * The Caller's connection is rejected if the callback returns FALSE, else
 412    * the connection is accepeted.
 413    *
 414    * Since: 1.20
 415    *
 416    */
 417   signals[SIG_CALLER_CONNECTING] =
 418       g_signal_new ("caller-connecting", G_TYPE_FROM_CLASS (klass),
 419       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstSRTSrcClass, caller_connecting),
 420       src_authentication_accumulator, NULL, NULL, G_TYPE_BOOLEAN,
 421       2, G_TYPE_SOCKET_ADDRESS, G_TYPE_STRING);
 422
 423   gst_srt_object_install_properties_helper (gobject_class);
 424
 425   gst_element_class_add_static_pad_template (gstelement_class, &src_template);
 426   gst_element_class_set_metadata (gstelement_class,
 427       "SRT source", "Source/Network",
 428       "Receive data over the network via SRT",
 429       "Justin Kim <justin.joy.9to5@gmail.com>");
 430
 431   gstbasesrc_class->start = GST_DEBUG_FUNCPTR (gst_srt_src_start);
 432   gstbasesrc_class->stop = GST_DEBUG_FUNCPTR (gst_srt_src_stop);
 433   gstbasesrc_class->unlock = GST_DEBUG_FUNCPTR (gst_srt_src_unlock);
 434   gstbasesrc_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_srt_src_unlock_stop);
 435   gstbasesrc_class->query = GST_DEBUG_FUNCPTR (gst_srt_src_query);
 436
 437   gstpushsrc_class->fill = GST_DEBUG_FUNCPTR (gst_srt_src_fill);
 438 }
 439
 440 static GstURIType
 441 gst_srt_src_uri_get_type (GType type)
 442 {
 443   return GST_URI_SRC;
 444 }
 445
 446 static const gchar *const *
 447 gst_srt_src_uri_get_protocols (GType type)
 448 {
 449   static const gchar *protocols[] = { GST_SRT_DEFAULT_URI_SCHEME, NULL };
 450
 451   return protocols;
 452 }
 453
 454 static gchar *
 455 gst_srt_src_uri_get_uri (GstURIHandler * handler)
 456 {
 457   gchar *uri_str;
 458   GstSRTSrc *self = GST_SRT_SRC (handler);
 459
 460   GST_OBJECT_LOCK (self);
 461   uri_str = gst_uri_to_string (self->srtobject->uri);
 462   GST_OBJECT_UNLOCK (self);
 463
 464   return uri_str;
 465 }
 466
 467 static gboolean
 468 gst_srt_src_uri_set_uri (GstURIHandler * handler,
 469     const gchar * uri, GError ** error)
 470 {
 471   GstSRTSrc *self = GST_SRT_SRC (handler);
 472   gboolean ret;
 473
 474   GST_OBJECT_LOCK (self);
 475   ret = gst_srt_object_set_uri (self->srtobject, uri, error);
 476   GST_OBJECT_UNLOCK (self);
 477
 478   return ret;
 479 }
 480
 481 static void
 482 gst_srt_src_uri_handler_init (gpointer g_iface, gpointer iface_data)
 483 {
 484   GstURIHandlerInterface *iface = (GstURIHandlerInterface *) g_iface;
 485
 486   iface->get_type = gst_srt_src_uri_get_type;
 487   iface->get_protocols = gst_srt_src_uri_get_protocols;
 488   iface->get_uri = gst_srt_src_uri_get_uri;
 489   iface->set_uri = gst_srt_src_uri_set_uri;
 490 }