gst/gstnicesink.c

   1
   2 /*
   3  * This file is part of the Nice GLib ICE library.
   4  *
   5  * (C) 2006, 2007 Collabora Ltd.
   6  *  Contact: Dafydd Harries
   7  * (C) 2006, 2007 Nokia Corporation. All rights reserved.
   8  *  Contact: Kai Vehmanen
   9  *
  10  * The contents of this file are subject to the Mozilla Public License Version
  11  * 1.1 (the "License"); you may not use this file except in compliance with
  12  * the License. You may obtain a copy of the License at
  13  * http://www.mozilla.org/MPL/
  14  *
  15  * Software distributed under the License is distributed on an "AS IS" basis,
  16  * WITHOUT WARRANTY OF ANY KIND, either express or implied. See the License
  17  * for the specific language governing rights and limitations under the
  18  * License.
  19  *
  20  * The Original Code is the Nice GLib ICE library.
  21  *
  22  * The Initial Developers of the Original Code are Collabora Ltd and Nokia
  23  * Corporation. All Rights Reserved.
  24  *
  25  * Contributors:
  26  *   Dafydd Harries, Collabora Ltd.
  27  *
  28  * Alternatively, the contents of this file may be used under the terms of the
  29  * the GNU Lesser General Public License Version 2.1 (the "LGPL"), in which
  30  * case the provisions of LGPL are applicable instead of those above. If you
  31  * wish to allow use of your version of this file only under the terms of the
  32  * LGPL and not to allow others to use your version of this file under the
  33  * MPL, indicate your decision by deleting the provisions above and replace
  34  * them with the notice and other provisions required by the LGPL. If you do
  35  * not delete the provisions above, a recipient may use your version of this
  36  * file under either the MPL or the LGPL.
  37  */
  38 #ifdef HAVE_CONFIG_H
  39 # include "config.h"
  40 #endif
  41
  42 #include "gstnicesink.h"
  43
  44
  45 GST_DEBUG_CATEGORY_STATIC (nicesink_debug);
  46 #define GST_CAT_DEFAULT nicesink_debug
  47
  48 static GstFlowReturn
  49 gst_nice_sink_render (
  50   GstBaseSink *basesink,
  51   GstBuffer *buffer);
  52 #if GST_CHECK_VERSION (1,0,0)
  53 static GstFlowReturn
  54 gst_nice_sink_render_list (
  55   GstBaseSink *basesink,
  56   GstBufferList *buffer_list);
  57 #endif
  58
  59 static gboolean
  60 gst_nice_sink_unlock (GstBaseSink *basesink);
  61
  62 static gboolean
  63 gst_nice_sink_unlock_stop (GstBaseSink *basesink);
  64
  65 static void
  66 _reliable_transport_writable (
  67     NiceAgent *agent,
  68     guint stream_id,
  69     guint component_id,
  70     GstNiceSink *sink);
  71
  72 static void
  73 gst_nice_sink_set_property (
  74   GObject *object,
  75   guint prop_id,
  76   const GValue *value,
  77   GParamSpec *pspec);
  78
  79 static void
  80 gst_nice_sink_get_property (
  81   GObject *object,
  82   guint prop_id,
  83   GValue *value,
  84   GParamSpec *pspec);
  85
  86 static void
  87 gst_nice_sink_dispose (GObject *object);
  88 #if GST_CHECK_VERSION (1,0,0)
  89 static void
  90 gst_nice_sink_finalize (GObject *object);
  91 #endif
  92
  93 static GstStateChangeReturn
  94 gst_nice_sink_change_state (
  95     GstElement * element,
  96     GstStateChange transition);
  97
  98 static GstStaticPadTemplate gst_nice_sink_sink_template =
  99 GST_STATIC_PAD_TEMPLATE (
 100     "sink",
 101     GST_PAD_SINK,
 102     GST_PAD_ALWAYS,
 103     GST_STATIC_CAPS_ANY);
 104
 105 G_DEFINE_TYPE (GstNiceSink, gst_nice_sink, GST_TYPE_BASE_SINK);
 106
 107 enum
 108 {
 109   PROP_AGENT = 1,
 110   PROP_STREAM,
 111   PROP_COMPONENT
 112 };
 113
 114 static void
 115 gst_nice_sink_class_init (GstNiceSinkClass *klass)
 116 {
 117   GstBaseSinkClass *gstbasesink_class;
 118   GstElementClass *gstelement_class;
 119   GObjectClass *gobject_class;
 120
 121   GST_DEBUG_CATEGORY_INIT (nicesink_debug, "nicesink",
 122       0, "libnice sink");
 123
 124   gstbasesink_class = (GstBaseSinkClass *) klass;
 125   gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_nice_sink_render);
 126 #if GST_CHECK_VERSION (1,0,0)
 127   gstbasesink_class->render_list = GST_DEBUG_FUNCPTR (gst_nice_sink_render_list);
 128 #endif
 129   gstbasesink_class->unlock = GST_DEBUG_FUNCPTR (gst_nice_sink_unlock);
 130   gstbasesink_class->unlock_stop = GST_DEBUG_FUNCPTR (gst_nice_sink_unlock_stop);
 131
 132   gobject_class = (GObjectClass *) klass;
 133   gobject_class->set_property = gst_nice_sink_set_property;
 134   gobject_class->get_property = gst_nice_sink_get_property;
 135   gobject_class->dispose = gst_nice_sink_dispose;
 136 #if GST_CHECK_VERSION (1,0,0)
 137   gobject_class->finalize = gst_nice_sink_finalize;
 138 #endif
 139
 140   gstelement_class = (GstElementClass *) klass;
 141   gstelement_class->change_state = gst_nice_sink_change_state;
 142
 143   gst_element_class_add_pad_template (gstelement_class,
 144       gst_static_pad_template_get (&gst_nice_sink_sink_template));
 145 #if GST_CHECK_VERSION (1,0,0)
 146   gst_element_class_set_metadata (gstelement_class,
 147 #else
 148   gst_element_class_set_details_simple (gstelement_class,
 149 #endif
 150     "ICE sink",
 151     "Sink",
 152     "Interactive UDP connectivity establishment",
 153     "Dafydd Harries <dafydd.harries@collabora.co.uk>");
 154
 155
 156   g_object_class_install_property (gobject_class, PROP_AGENT,
 157       g_param_spec_object (
 158          "agent",
 159          "Agent",
 160          "The NiceAgent this source is bound to",
 161          NICE_TYPE_AGENT,
 162          G_PARAM_READWRITE));
 163
 164   g_object_class_install_property (gobject_class, PROP_STREAM,
 165       g_param_spec_uint (
 166          "stream",
 167          "Stream ID",
 168          "The ID of the stream to read from",
 169          0,
 170          G_MAXUINT,
 171          0,
 172          G_PARAM_READWRITE));
 173
 174   g_object_class_install_property (gobject_class, PROP_COMPONENT,
 175       g_param_spec_uint (
 176          "component",
 177          "Component ID",
 178          "The ID of the component to read from",
 179          0,
 180          G_MAXUINT,
 181          0,
 182          G_PARAM_READWRITE));
 183 }
 184
 185 static void
 186 gst_nice_sink_init (GstNiceSink *sink)
 187 {
 188 #if GST_CHECK_VERSION (1,0,0)
 189   guint max_mem;
 190 #endif
 191
 192   g_cond_init (&sink->writable_cond);
 193
 194 #if GST_CHECK_VERSION (1,0,0)
 195   /* pre-allocate OutputVector, MapInfo and OutputMessage arrays
 196    * for use in the render and render_list functions */
 197   max_mem = gst_buffer_get_max_memory ();
 198
 199   sink->n_vecs = max_mem;
 200   sink->vecs = g_new (GOutputVector, sink->n_vecs);
 201
 202   sink->n_maps = max_mem;
 203   sink->maps = g_new (GstMapInfo, sink->n_maps);
 204
 205   sink->n_messages = 1;
 206   sink->messages = g_new (NiceOutputMessage, sink->n_messages);
 207 #endif
 208
 209 #if GST_CHECK_VERSION (1,12,0)
 210   gst_base_sink_set_drop_out_of_segment (GST_BASE_SINK (sink), FALSE);
 211 #endif
 212 }
 213
 214 static void
 215 _reliable_transport_writable (NiceAgent *agent, guint stream_id,
 216     guint component_id, GstNiceSink *sink)
 217 {
 218   GST_OBJECT_LOCK (sink);
 219   if (stream_id == sink->stream_id && component_id == sink->component_id) {
 220     g_cond_broadcast (&sink->writable_cond);
 221   }
 222   GST_OBJECT_UNLOCK (sink);
 223 }
 224
 225 #if GST_CHECK_VERSION (1,0,0)
 226 static gsize
 227 fill_vectors (GOutputVector * vecs, GstMapInfo * maps, guint n, GstBuffer * buf)
 228 {
 229   GstMemory *mem;
 230   gsize size = 0;
 231   guint i;
 232
 233   g_assert (gst_buffer_n_memory (buf) == n);
 234
 235   for (i = 0; i < n; ++i) {
 236     mem = gst_buffer_peek_memory (buf, i);
 237     if (gst_memory_map (mem, &maps[i], GST_MAP_READ)) {
 238       vecs[i].buffer = maps[i].data;
 239       vecs[i].size = maps[i].size;
 240     } else {
 241       GST_WARNING ("Failed to map memory %p for reading", mem);
 242       vecs[i].buffer = "";
 243       vecs[i].size = 0;
 244     }
 245     size += vecs[i].size;
 246   }
 247
 248   return size;
 249 }
 250
 251 /* Buffer list code written by
 252  *   Tim-Philipp Müller <tim@centricular.com>
 253  * taken from
 254  *   gst-plugins-good/gst/udp/gstmultiudpsink.c
 255  */
 256 static GstFlowReturn
 257 gst_nice_sink_render_buffers (GstNiceSink * sink, GstBuffer ** buffers,
 258     guint num_buffers, guint8 * mem_nums, guint total_mem_num)
 259 {
 260   NiceOutputMessage *msgs;
 261   GOutputVector *vecs;
 262   GstMapInfo *map_infos;
 263   guint i, mem;
 264   guint written = 0;
 265   gint ret;
 266   GstFlowReturn flow_ret = GST_FLOW_OK;
 267
 268   GST_LOG_OBJECT (sink, "%u buffers, %u memories -> to be sent",
 269       num_buffers, total_mem_num);
 270
 271   if (sink->n_vecs < total_mem_num) {
 272     sink->n_vecs = GST_ROUND_UP_16 (total_mem_num);
 273     g_free (sink->vecs);
 274     sink->vecs = g_new (GOutputVector, sink->n_vecs);
 275   }
 276   vecs = sink->vecs;
 277
 278   if (sink->n_maps < total_mem_num) {
 279     sink->n_maps = GST_ROUND_UP_16 (total_mem_num);
 280     g_free (sink->maps);
 281     sink->maps = g_new (GstMapInfo, sink->n_maps);
 282   }
 283   map_infos = sink->maps;
 284
 285   if (sink->n_messages < num_buffers) {
 286     sink->n_messages = GST_ROUND_UP_16 (num_buffers);
 287     g_free (sink->messages);
 288     sink->messages = g_new (NiceOutputMessage, sink->n_messages);
 289   }
 290   msgs = sink->messages;
 291
 292   for (i = 0, mem = 0; i < num_buffers; ++i) {
 293     fill_vectors (&vecs[mem], &map_infos[mem], mem_nums[i], buffers[i]);
 294     msgs[i].buffers = &vecs[mem];
 295     msgs[i].n_buffers = mem_nums[i];
 296     mem += mem_nums[i];
 297   }
 298
 299   GST_OBJECT_LOCK (sink);
 300   do {
 301     ret = nice_agent_send_messages_nonblocking(sink->agent, sink->stream_id,
 302         sink->component_id, msgs + written, num_buffers - written, NULL, NULL);
 303
 304     if (ret > 0)
 305       written += ret;
 306
 307     if (sink->reliable && written < num_buffers)
 308       g_cond_wait (&sink->writable_cond, GST_OBJECT_GET_LOCK (sink));
 309
 310     if (sink->flushing) {
 311       flow_ret = GST_FLOW_FLUSHING;
 312       break;
 313     }
 314   } while (sink->reliable && written < num_buffers);
 315   GST_OBJECT_UNLOCK (sink);
 316
 317   for (i = 0; i < mem; ++i)
 318     gst_memory_unmap (map_infos[i].memory, &map_infos[i]);
 319
 320   return flow_ret;
 321 }
 322 #endif
 323
 324 static GstFlowReturn
 325 gst_nice_sink_render (GstBaseSink *basesink, GstBuffer *buffer)
 326 {
 327   GstNiceSink *nicesink = GST_NICE_SINK (basesink);
 328   GstFlowReturn flow_ret = GST_FLOW_OK;
 329 #if GST_CHECK_VERSION (1,0,0)
 330   guint8 n_mem;
 331
 332   n_mem = gst_buffer_n_memory (buffer);
 333
 334   if (n_mem > 0) {
 335     flow_ret = gst_nice_sink_render_buffers (nicesink, &buffer, 1, &n_mem,
 336         n_mem);
 337   }
 338
 339 #else
 340   guint written = 0;
 341   gint ret;
 342   gchar *data = NULL;
 343   guint size = 0;
 344
 345   data = (gchar *) GST_BUFFER_DATA (buffer);
 346   size = GST_BUFFER_SIZE (buffer);
 347
 348   GST_OBJECT_LOCK (nicesink);
 349   do {
 350     ret = nice_agent_send (nicesink->agent, nicesink->stream_id,
 351         nicesink->component_id, size - written, data + written);
 352     if (ret > 0)
 353       written += ret;
 354
 355     if (nicesink->reliable && written < size)
 356       g_cond_wait (&nicesink->writable_cond, GST_OBJECT_GET_LOCK (nicesink));
 357     if (nicesink->flushing) {
 358       flow_ret = GST_FLOW_WRONG_STATE;
 359       break;
 360     }
 361   } while (nicesink->reliable && written < size);
 362   GST_OBJECT_UNLOCK (nicesink);
 363
 364 #endif
 365   return flow_ret;
 366 }
 367
 368 #if GST_CHECK_VERSION (1,0,0)
 369 static GstFlowReturn
 370 gst_nice_sink_render_list (GstBaseSink *basesink, GstBufferList *buffer_list)
 371 {
 372   GstNiceSink *nicesink = GST_NICE_SINK (basesink);
 373   GstBuffer **buffers;
 374   GstFlowReturn flow_ret = GST_FLOW_OK;
 375   guint8 *mem_nums;
 376   guint total_mems;
 377   guint i, num_buffers;
 378
 379   num_buffers = gst_buffer_list_length (buffer_list);
 380   if (num_buffers == 0)
 381     goto no_data;
 382
 383   buffers = g_newa (GstBuffer *, num_buffers);
 384   mem_nums = g_newa (guint8, num_buffers);
 385   for (i = 0, total_mems = 0; i < num_buffers; ++i) {
 386     buffers[i] = gst_buffer_list_get (buffer_list, i);
 387     mem_nums[i] = gst_buffer_n_memory (buffers[i]);
 388     total_mems += mem_nums[i];
 389   }
 390
 391   flow_ret = gst_nice_sink_render_buffers (nicesink, buffers, num_buffers,
 392       mem_nums, total_mems);
 393
 394   return flow_ret;
 395
 396 no_data:
 397   {
 398     GST_LOG_OBJECT (nicesink, "empty buffer");
 399     return GST_FLOW_OK;
 400   }
 401
 402   return flow_ret;
 403 }
 404 #endif
 405
 406 static gboolean gst_nice_sink_unlock (GstBaseSink *basesink)
 407 {
 408   GstNiceSink *nicesink = GST_NICE_SINK (basesink);
 409
 410   GST_OBJECT_LOCK (nicesink);
 411   nicesink->flushing = TRUE;
 412   g_cond_broadcast (&nicesink->writable_cond);
 413   GST_OBJECT_UNLOCK (nicesink);
 414
 415   return TRUE;
 416 }
 417
 418 static gboolean gst_nice_sink_unlock_stop (GstBaseSink *basesink)
 419 {
 420   GstNiceSink *nicesink = GST_NICE_SINK (basesink);
 421
 422   GST_OBJECT_LOCK (nicesink);
 423   nicesink->flushing = FALSE;
 424   GST_OBJECT_UNLOCK (nicesink);
 425
 426   return TRUE;
 427 }
 428
 429 static void
 430 gst_nice_sink_dispose (GObject *object)
 431 {
 432   GstNiceSink *sink = GST_NICE_SINK (object);
 433
 434   if (sink->agent && sink->writable_id)
 435     g_signal_handler_disconnect (sink->agent, sink->writable_id);
 436   sink->writable_id = 0;
 437   g_clear_object (&sink->agent);
 438
 439   g_cond_clear (&sink->writable_cond);
 440
 441   G_OBJECT_CLASS (gst_nice_sink_parent_class)->dispose (object);
 442 }
 443
 444 #if GST_CHECK_VERSION (1,0,0)
 445 static void
 446 gst_nice_sink_finalize (GObject *object)
 447 {
 448   GstNiceSink *sink = GST_NICE_SINK (object);
 449
 450   g_free (sink->vecs);
 451   sink->vecs = NULL;
 452   sink->n_vecs = 0;
 453   g_free (sink->maps);
 454   sink->maps = NULL;
 455   sink->n_maps = 0;
 456   g_free (sink->messages);
 457   sink->messages = NULL;
 458   sink->n_messages = 0;
 459
 460   G_OBJECT_CLASS (gst_nice_sink_parent_class)->finalize (object);
 461 }
 462 #endif
 463
 464 static void
 465 gst_nice_sink_set_property (
 466   GObject *object,
 467   guint prop_id,
 468   const GValue *value,
 469   GParamSpec *pspec)
 470 {
 471   GstNiceSink *sink = GST_NICE_SINK (object);
 472
 473   switch (prop_id)
 474     {
 475     case PROP_AGENT:
 476       if (sink->agent) {
 477         GST_ERROR_OBJECT (object,
 478             "Changing the agent on a nice sink not allowed");
 479       } else {
 480         sink->agent = g_value_dup_object (value);
 481         g_object_get (sink->agent, "reliable", &sink->reliable, NULL);
 482         if (sink->reliable)
 483           sink->writable_id = g_signal_connect (sink->agent,
 484               "reliable-transport-writable",
 485               (GCallback) _reliable_transport_writable, sink);
 486       }
 487       break;
 488
 489     case PROP_STREAM:
 490       GST_OBJECT_LOCK (sink);
 491       sink->stream_id = g_value_get_uint (value);
 492       GST_OBJECT_UNLOCK (sink);
 493       break;
 494
 495     case PROP_COMPONENT:
 496       {
 497         guint new_component_id = g_value_get_uint (value);
 498         GST_OBJECT_LOCK (sink);
 499         if (sink->component_id != new_component_id) {
 500           sink->component_id = new_component_id;
 501           g_cond_broadcast (&sink->writable_cond);
 502         }
 503         GST_OBJECT_UNLOCK (sink);
 504       }
 505       break;
 506
 507     default:
 508       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 509       break;
 510     }
 511 }
 512
 513 static void
 514 gst_nice_sink_get_property (
 515   GObject *object,
 516   guint prop_id,
 517   GValue *value,
 518   GParamSpec *pspec)
 519 {
 520   GstNiceSink *sink = GST_NICE_SINK (object);
 521
 522   switch (prop_id)
 523     {
 524     case PROP_AGENT:
 525       g_value_set_object (value, sink->agent);
 526       break;
 527
 528     case PROP_STREAM:
 529       GST_OBJECT_LOCK (sink);
 530       g_value_set_uint (value, sink->stream_id);
 531       GST_OBJECT_UNLOCK (sink);
 532       break;
 533
 534     case PROP_COMPONENT:
 535       GST_OBJECT_LOCK (sink);
 536       g_value_set_uint (value, sink->component_id);
 537       GST_OBJECT_UNLOCK (sink);
 538       break;
 539
 540     default:
 541       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 542       break;
 543     }
 544 }
 545
 546 static GstStateChangeReturn
 547 gst_nice_sink_change_state (GstElement * element, GstStateChange transition)
 548 {
 549   GstNiceSink *sink;
 550   GstStateChangeReturn ret;
 551
 552   sink = GST_NICE_SINK (element);
 553
 554   switch (transition) {
 555     case GST_STATE_CHANGE_NULL_TO_READY:
 556       if (sink->agent == NULL)
 557         {
 558           GST_ERROR_OBJECT (element,
 559               "Trying to start Nice sink without an agent set");
 560           return GST_STATE_CHANGE_FAILURE;
 561         }
 562       else if (sink->stream_id == 0)
 563           {
 564             GST_ERROR_OBJECT (element,
 565                 "Trying to start Nice sink without a stream set");
 566             return GST_STATE_CHANGE_FAILURE;
 567           }
 568       else if (sink->component_id == 0)
 569           {
 570             GST_ERROR_OBJECT (element,
 571                 "Trying to start Nice sink without a component set");
 572             return GST_STATE_CHANGE_FAILURE;
 573           }
 574       break;
 575     case GST_STATE_CHANGE_READY_TO_PAUSED:
 576     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
 577     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
 578     case GST_STATE_CHANGE_PAUSED_TO_READY:
 579     case GST_STATE_CHANGE_READY_TO_NULL:
 580     default:
 581       break;
 582   }
 583
 584   ret = GST_ELEMENT_CLASS (gst_nice_sink_parent_class)->change_state (element,
 585       transition);
 586
 587   return ret;
 588 }