gst/tcp/gstmultihandlesink.c

   1 /* GStreamer
   2  * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>
   3  * Copyright (C) <2004> Thomas Vander Stichele <thomas at apestaart dot org>
   4  * Copyright (C) 2006 Wim Taymans <wim at fluendo dot com>
   5  * Copyright (C) <2011> Collabora Ltd.
   6  *     Author: Sebastian Dröge <sebastian.droege@collabora.co.uk>
   7  *
   8  * This library is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Library General Public
  10  * License as published by the Free Software Foundation; either
  11  * version 2 of the License, or (at your option) any later version.
  12  *
  13  * This library is distributed in the hope that it will be useful,
  14  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  15  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  16  * Library General Public License for more details.
  17  *
  18  * You should have received a copy of the GNU Library General Public
  19  * License along with this library; if not, write to the
  20  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  21  * Boston, MA 02110-1301, USA.
  22  */
  23
  24 /**
  25  * SECTION:element-multihandlesink
  26  * @title: multihandlesink
  27  * @see_also: tcpserversink
  28  *
  29  * This plugin writes incoming data to a set of file descriptors. The
  30  * file descriptors can be added to multihandlesink by emitting the #GstMultiHandleSink::add signal.
  31  * For each descriptor added, the #GstMultiHandleSink::client-added signal will be called.
  32  *
  33  * A client can also be added with the #GstMultiHandleSink::add-full signal
  34  * that allows for more control over what and how much data a client
  35  * initially receives.
  36  *
  37  * Clients can be removed from multihandlesink by emitting the #GstMultiHandleSink::remove signal. For
  38  * each descriptor removed, the #GstMultiHandleSink::client-removed signal will be called. The
  39  * #GstMultiHandleSink::client-removed signal can also be fired when multihandlesink decides that a
  40  * client is not active anymore or, depending on the value of the
  41  * #GstMultiHandleSink:recover-policy property, if the client is reading too slowly.
  42  * In all cases, multihandlesink will never close a file descriptor itself.
  43  * The user of multihandlesink is responsible for closing all file descriptors.
  44  * This can for example be done in response to the #GstMultiHandleSink::client-fd-removed signal.
  45  * Note that multihandlesink still has a reference to the file descriptor when the
  46  * #GstMultiHandleSink::client-removed signal is emitted, so that "get-stats" can be performed on
  47  * the descriptor; it is therefore not safe to close the file descriptor in
  48  * the #GstMultiHandleSink::client-removed signal handler, and you should use the
  49  * #GstMultiHandleSink::client-fd-removed signal to safely close the fd.
  50  *
  51  * Multisocketsink internally keeps a queue of the incoming buffers and uses a
  52  * separate thread to send the buffers to the clients. This ensures that no
  53  * client write can block the pipeline and that clients can read with different
  54  * speeds.
  55  *
  56  * When adding a client to multihandlesink, the #GstMultiHandleSink:sync-method property will define
  57  * which buffer in the queued buffers will be sent first to the client. Clients
  58  * can be sent the most recent buffer (which might not be decodable by the
  59  * client if it is not a keyframe), the next keyframe received in
  60  * multihandlesink (which can take some time depending on the keyframe rate), or the
  61  * last received keyframe (which will cause a simple burst-on-connect).
  62  * Multisocketsink will always keep at least one keyframe in its internal buffers
  63  * when the sync-mode is set to latest-keyframe.
  64  *
  65  * There are additional values for the #GstMultiHandleSink:sync-method
  66  * property to allow finer control over burst-on-connect behaviour. By selecting
  67  * the 'burst' method a minimum burst size can be chosen, 'burst-keyframe'
  68  * additionally requires that the burst begin with a keyframe, and
  69  * 'burst-with-keyframe' attempts to burst beginning with a keyframe, but will
  70  * prefer a minimum burst size even if it requires not starting with a keyframe.
  71  *
  72  * Multisocketsink can be instructed to keep at least a minimum amount of data
  73  * expressed in time or byte units in its internal queues with the
  74  * #GstMultiHandleSink:time-min and #GstMultiHandleSink:bytes-min properties respectively.
  75  * These properties are useful if the application adds clients with the
  76  * #GstMultiHandleSink::add-full signal to make sure that a burst connect can
  77  * actually be honored.
  78  *
  79  * When streaming data, clients are allowed to read at a different rate than
  80  * the rate at which multihandlesink receives data. If the client is reading too
  81  * fast, no data will be send to the client until multihandlesink receives more
  82  * data. If the client, however, reads too slowly, data for that client will be
  83  * queued up in multihandlesink. Two properties control the amount of data
  84  * (buffers) that is queued in multihandlesink: #GstMultiHandleSink:buffers-max and
  85  * #GstMultiHandleSink:buffers-soft-max. A client that falls behind by
  86  * #GstMultiHandleSink:buffers-max is removed from multihandlesink forcibly.
  87  *
  88  * A client with a lag of at least #GstMultiHandleSink:buffers-soft-max enters the recovery
  89  * procedure which is controlled with the #GstMultiHandleSink:recover-policy property.
  90  * A recover policy of NONE will do nothing, RESYNC_LATEST will send the most recently
  91  * received buffer as the next buffer for the client, RESYNC_SOFT_LIMIT
  92  * positions the client to the soft limit in the buffer queue and
  93  * RESYNC_KEYFRAME positions the client at the most recent keyframe in the
  94  * buffer queue.
  95  *
  96  * multihandlesink will by default synchronize on the clock before serving the
  97  * buffers to the clients. This behaviour can be disabled by setting the sync
  98  * property to FALSE. Multisocketsink will by default not do QoS and will never
  99  * drop late buffers.
 100  */
 101
 102 #ifdef HAVE_CONFIG_H
 103 #include "config.h"
 104 #endif
 105
 106 #include <gst/gst-i18n-plugin.h>
 107
 108 #include "gstmultihandlesink.h"
 109
 110 #ifdef HAVE_SYS_SOCKET_H
 111 #include <sys/socket.h>
 112 #endif
 113
 114 #ifndef G_OS_WIN32
 115 #include <netinet/in.h>
 116 #endif
 117
 118 #include <string.h>
 119
 120 #define NOT_IMPLEMENTED 0
 121
 122 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
 123     GST_PAD_SINK,
 124     GST_PAD_ALWAYS,
 125     GST_STATIC_CAPS_ANY);
 126
 127 GST_DEBUG_CATEGORY_STATIC (multihandlesink_debug);
 128 #define GST_CAT_DEFAULT (multihandlesink_debug)
 129
 130 /* MultiHandleSink signals and args */
 131 enum
 132 {
 133   GST_MULTI_SINK_LAST_SIGNAL,
 134
 135   /* methods */
 136   SIGNAL_ADD,
 137   SIGNAL_ADD_BURST,
 138   SIGNAL_CLEAR,
 139
 140   /* signals */
 141   SIGNAL_CLIENT_ADDED,
 142   SIGNAL_CLIENT_REMOVED,
 143   SIGNAL_CLIENT_SOCKET_REMOVED,
 144
 145   LAST_SIGNAL
 146 };
 147
 148
 149 /* this is really arbitrarily chosen */
 150 #define DEFAULT_BUFFERS_MAX             -1
 151 #define DEFAULT_BUFFERS_SOFT_MAX        -1
 152 #define DEFAULT_TIME_MIN                -1
 153 #define DEFAULT_BYTES_MIN               -1
 154 #define DEFAULT_BUFFERS_MIN             -1
 155 #define DEFAULT_UNIT_FORMAT               GST_FORMAT_BUFFERS
 156 #define DEFAULT_UNITS_MAX               -1
 157 #define DEFAULT_UNITS_SOFT_MAX          -1
 158 #define DEFAULT_RECOVER_POLICY          GST_RECOVER_POLICY_NONE
 159 #define DEFAULT_TIMEOUT                 0
 160 #define DEFAULT_SYNC_METHOD             GST_SYNC_METHOD_LATEST
 161
 162 #define DEFAULT_BURST_FORMAT            GST_FORMAT_UNDEFINED
 163 #define DEFAULT_BURST_VALUE             0
 164
 165 #define DEFAULT_QOS_DSCP                -1
 166
 167 #define DEFAULT_RESEND_STREAMHEADER      TRUE
 168
 169 enum
 170 {
 171   PROP_0,
 172   PROP_BUFFERS_QUEUED,
 173   PROP_BYTES_QUEUED,
 174   PROP_TIME_QUEUED,
 175
 176   PROP_UNIT_FORMAT,
 177   PROP_UNITS_MAX,
 178   PROP_UNITS_SOFT_MAX,
 179
 180   PROP_BUFFERS_MAX,
 181   PROP_BUFFERS_SOFT_MAX,
 182
 183   PROP_TIME_MIN,
 184   PROP_BYTES_MIN,
 185   PROP_BUFFERS_MIN,
 186
 187   PROP_RECOVER_POLICY,
 188   PROP_TIMEOUT,
 189   PROP_SYNC_METHOD,
 190   PROP_BYTES_TO_SERVE,
 191   PROP_BYTES_SERVED,
 192
 193   PROP_BURST_FORMAT,
 194   PROP_BURST_VALUE,
 195
 196   PROP_QOS_DSCP,
 197
 198   PROP_RESEND_STREAMHEADER,
 199
 200   PROP_NUM_HANDLES
 201 };
 202
 203 GType
 204 gst_multi_handle_sink_recover_policy_get_type (void)
 205 {
 206   static GType recover_policy_type = 0;
 207   static const GEnumValue recover_policy[] = {
 208     {GST_RECOVER_POLICY_NONE,
 209         "Do not try to recover", "none"},
 210     {GST_RECOVER_POLICY_RESYNC_LATEST,
 211         "Resync client to latest buffer", "latest"},
 212     {GST_RECOVER_POLICY_RESYNC_SOFT_LIMIT,
 213         "Resync client to soft limit", "soft-limit"},
 214     {GST_RECOVER_POLICY_RESYNC_KEYFRAME,
 215         "Resync client to most recent keyframe", "keyframe"},
 216     {0, NULL, NULL},
 217   };
 218
 219   if (!recover_policy_type) {
 220     recover_policy_type =
 221         g_enum_register_static ("GstMultiHandleSinkRecoverPolicy",
 222         recover_policy);
 223   }
 224   return recover_policy_type;
 225 }
 226
 227 GType
 228 gst_multi_handle_sink_sync_method_get_type (void)
 229 {
 230   static GType sync_method_type = 0;
 231   static const GEnumValue sync_method[] = {
 232     {GST_SYNC_METHOD_LATEST,
 233         "Serve starting from the latest buffer", "latest"},
 234     {GST_SYNC_METHOD_NEXT_KEYFRAME,
 235         "Serve starting from the next keyframe", "next-keyframe"},
 236     {GST_SYNC_METHOD_LATEST_KEYFRAME,
 237           "Serve everything since the latest keyframe (burst)",
 238         "latest-keyframe"},
 239     {GST_SYNC_METHOD_BURST, "Serve burst-value data to client", "burst"},
 240     {GST_SYNC_METHOD_BURST_KEYFRAME,
 241           "Serve burst-value data starting on a keyframe",
 242         "burst-keyframe"},
 243     {GST_SYNC_METHOD_BURST_WITH_KEYFRAME,
 244           "Serve burst-value data preferably starting on a keyframe",
 245         "burst-with-keyframe"},
 246     {0, NULL, NULL},
 247   };
 248
 249   if (!sync_method_type) {
 250     sync_method_type =
 251         g_enum_register_static ("GstMultiHandleSinkSyncMethod", sync_method);
 252   }
 253   return sync_method_type;
 254 }
 255
 256 GType
 257 gst_multi_handle_sink_client_status_get_type (void)
 258 {
 259   static GType client_status_type = 0;
 260   static const GEnumValue client_status[] = {
 261     {GST_CLIENT_STATUS_OK, "ok", "ok"},
 262     {GST_CLIENT_STATUS_CLOSED, "Closed", "closed"},
 263     {GST_CLIENT_STATUS_REMOVED, "Removed", "removed"},
 264     {GST_CLIENT_STATUS_SLOW, "Too slow", "slow"},
 265     {GST_CLIENT_STATUS_ERROR, "Error", "error"},
 266     {GST_CLIENT_STATUS_DUPLICATE, "Duplicate", "duplicate"},
 267     {GST_CLIENT_STATUS_FLUSHING, "Flushing", "flushing"},
 268     {0, NULL, NULL},
 269   };
 270
 271   if (!client_status_type) {
 272     client_status_type =
 273         g_enum_register_static ("GstMultiHandleSinkClientStatus",
 274         client_status);
 275   }
 276   return client_status_type;
 277 }
 278
 279 static void gst_multi_handle_sink_finalize (GObject * object);
 280 static void gst_multi_handle_sink_clear (GstMultiHandleSink * mhsink);
 281
 282 static GstFlowReturn gst_multi_handle_sink_render (GstBaseSink * bsink,
 283     GstBuffer * buf);
 284 static void gst_multi_handle_sink_queue_buffer (GstMultiHandleSink * mhsink,
 285     GstBuffer * buffer);
 286 static gboolean gst_multi_handle_sink_client_queue_buffer (GstMultiHandleSink *
 287     mhsink, GstMultiHandleClient * mhclient, GstBuffer * buffer);
 288 static GstStateChangeReturn gst_multi_handle_sink_change_state (GstElement *
 289     element, GstStateChange transition);
 290
 291 static void gst_multi_handle_sink_set_property (GObject * object, guint prop_id,
 292     const GValue * value, GParamSpec * pspec);
 293 static void gst_multi_handle_sink_get_property (GObject * object, guint prop_id,
 294     GValue * value, GParamSpec * pspec);
 295
 296 #define gst_multi_handle_sink_parent_class parent_class
 297 G_DEFINE_TYPE (GstMultiHandleSink, gst_multi_handle_sink, GST_TYPE_BASE_SINK);
 298
 299 static guint gst_multi_handle_sink_signals[LAST_SIGNAL] = { 0 };
 300
 301 static gint
 302 find_syncframe (GstMultiHandleSink * sink, gint idx, gint direction);
 303 #define find_next_syncframe(s,i)        find_syncframe(s,i,1)
 304 #define find_prev_syncframe(s,i)        find_syncframe(s,i,-1)
 305 static gboolean is_sync_frame (GstMultiHandleSink * sink, GstBuffer * buffer);
 306 static gboolean gst_multi_handle_sink_stop (GstBaseSink * bsink);
 307 static gboolean gst_multi_handle_sink_start (GstBaseSink * bsink);
 308 static gint get_buffers_max (GstMultiHandleSink * sink, gint64 max);
 309 static gint
 310 gst_multi_handle_sink_recover_client (GstMultiHandleSink * sink,
 311     GstMultiHandleClient * client);
 312 static void gst_multi_handle_sink_setup_dscp (GstMultiHandleSink * mhsink);
 313 static gboolean
 314 find_limits (GstMultiHandleSink * sink,
 315     gint * min_idx, gint bytes_min, gint buffers_min, gint64 time_min,
 316     gint * max_idx, gint bytes_max, gint buffers_max, gint64 time_max);
 317
 318
 319 static void
 320 gst_multi_handle_sink_class_init (GstMultiHandleSinkClass * klass)
 321 {
 322   GObjectClass *gobject_class;
 323   GstElementClass *gstelement_class;
 324   GstBaseSinkClass *gstbasesink_class;
 325
 326   gobject_class = (GObjectClass *) klass;
 327   gstelement_class = (GstElementClass *) klass;
 328   gstbasesink_class = (GstBaseSinkClass *) klass;
 329
 330   gobject_class->set_property = gst_multi_handle_sink_set_property;
 331   gobject_class->get_property = gst_multi_handle_sink_get_property;
 332   gobject_class->finalize = gst_multi_handle_sink_finalize;
 333
 334   g_object_class_install_property (gobject_class, PROP_BUFFERS_MAX,
 335       g_param_spec_int ("buffers-max", "Buffers max",
 336           "max number of buffers to queue for a client (-1 = no limit)", -1,
 337           G_MAXINT, DEFAULT_BUFFERS_MAX,
 338           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 339   g_object_class_install_property (gobject_class, PROP_BUFFERS_SOFT_MAX,
 340       g_param_spec_int ("buffers-soft-max", "Buffers soft max",
 341           "Recover client when going over this limit (-1 = no limit)", -1,
 342           G_MAXINT, DEFAULT_BUFFERS_SOFT_MAX,
 343           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 344
 345   g_object_class_install_property (gobject_class, PROP_BYTES_MIN,
 346       g_param_spec_int ("bytes-min", "Bytes min",
 347           "min number of bytes to queue (-1 = as little as possible)", -1,
 348           G_MAXINT, DEFAULT_BYTES_MIN,
 349           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 350   g_object_class_install_property (gobject_class, PROP_TIME_MIN,
 351       g_param_spec_int64 ("time-min", "Time min",
 352           "min amount of time to queue (in nanoseconds) "
 353           "(-1 = as little as possible)", -1, G_MAXINT64, DEFAULT_TIME_MIN,
 354           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 355   g_object_class_install_property (gobject_class, PROP_BUFFERS_MIN,
 356       g_param_spec_int ("buffers-min", "Buffers min",
 357           "min number of buffers to queue (-1 = as few as possible)", -1,
 358           G_MAXINT, DEFAULT_BUFFERS_MIN,
 359           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 360
 361   g_object_class_install_property (gobject_class, PROP_UNIT_FORMAT,
 362       g_param_spec_enum ("unit-format", "Units format",
 363           "The unit to measure the max/soft-max/queued properties",
 364           GST_TYPE_FORMAT, DEFAULT_UNIT_FORMAT,
 365           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 366   g_object_class_install_property (gobject_class, PROP_UNITS_MAX,
 367       g_param_spec_int64 ("units-max", "Units max",
 368           "max number of units to queue (-1 = no limit)", -1, G_MAXINT64,
 369           DEFAULT_UNITS_MAX, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 370   g_object_class_install_property (gobject_class, PROP_UNITS_SOFT_MAX,
 371       g_param_spec_int64 ("units-soft-max", "Units soft max",
 372           "Recover client when going over this limit (-1 = no limit)", -1,
 373           G_MAXINT64, DEFAULT_UNITS_SOFT_MAX,
 374           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 375
 376   g_object_class_install_property (gobject_class, PROP_BUFFERS_QUEUED,
 377       g_param_spec_uint ("buffers-queued", "Buffers queued",
 378           "Number of buffers currently queued", 0, G_MAXUINT, 0,
 379           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 380 #if NOT_IMPLEMENTED
 381   g_object_class_install_property (gobject_class, PROP_BYTES_QUEUED,
 382       g_param_spec_uint ("bytes-queued", "Bytes queued",
 383           "Number of bytes currently queued", 0, G_MAXUINT, 0,
 384           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 385   g_object_class_install_property (gobject_class, PROP_TIME_QUEUED,
 386       g_param_spec_uint64 ("time-queued", "Time queued",
 387           "Amount of time currently queued (in nanoseconds)", 0, G_MAXUINT64, 0,
 388           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 389 #endif
 390
 391   g_object_class_install_property (gobject_class, PROP_RECOVER_POLICY,
 392       g_param_spec_enum ("recover-policy", "Recover Policy",
 393           "How to recover when client reaches the soft max",
 394           GST_TYPE_RECOVER_POLICY, DEFAULT_RECOVER_POLICY,
 395           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 396   g_object_class_install_property (gobject_class, PROP_TIMEOUT,
 397       g_param_spec_uint64 ("timeout", "Timeout",
 398           "Maximum inactivity timeout in nanoseconds for a client (0 = no limit)",
 399           0, G_MAXUINT64, DEFAULT_TIMEOUT,
 400           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 401   g_object_class_install_property (gobject_class, PROP_SYNC_METHOD,
 402       g_param_spec_enum ("sync-method", "Sync Method",
 403           "How to sync new clients to the stream", GST_TYPE_SYNC_METHOD,
 404           DEFAULT_SYNC_METHOD, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 405   g_object_class_install_property (gobject_class, PROP_BYTES_TO_SERVE,
 406       g_param_spec_uint64 ("bytes-to-serve", "Bytes to serve",
 407           "Number of bytes received to serve to clients", 0, G_MAXUINT64, 0,
 408           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 409   g_object_class_install_property (gobject_class, PROP_BYTES_SERVED,
 410       g_param_spec_uint64 ("bytes-served", "Bytes served",
 411           "Total number of bytes send to all clients", 0, G_MAXUINT64, 0,
 412           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 413
 414   g_object_class_install_property (gobject_class, PROP_BURST_FORMAT,
 415       g_param_spec_enum ("burst-format", "Burst format",
 416           "The format of the burst units (when sync-method is burst[[-with]-keyframe])",
 417           GST_TYPE_FORMAT, DEFAULT_BURST_FORMAT,
 418           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 419   g_object_class_install_property (gobject_class, PROP_BURST_VALUE,
 420       g_param_spec_uint64 ("burst-value", "Burst value",
 421           "The amount of burst expressed in burst-format", 0, G_MAXUINT64,
 422           DEFAULT_BURST_VALUE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 423
 424   g_object_class_install_property (gobject_class, PROP_QOS_DSCP,
 425       g_param_spec_int ("qos-dscp", "QoS diff srv code point",
 426           "Quality of Service, differentiated services code point (-1 default)",
 427           -1, 63, DEFAULT_QOS_DSCP,
 428           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 429
 430   /**
 431    * GstMultiHandleSink::resend-streamheader
 432    *
 433    * Resend the streamheaders to existing clients when they change.
 434    */
 435   g_object_class_install_property (gobject_class, PROP_RESEND_STREAMHEADER,
 436       g_param_spec_boolean ("resend-streamheader", "Resend streamheader",
 437           "Resend the streamheader if it changes in the caps",
 438           DEFAULT_RESEND_STREAMHEADER,
 439           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 440
 441   g_object_class_install_property (gobject_class, PROP_NUM_HANDLES,
 442       g_param_spec_uint ("num-handles", "Number of handles",
 443           "The current number of client handles",
 444           0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 445
 446   /**
 447    * GstMultiHandleSink::clear:
 448    * @gstmultihandlesink: the multihandlesink element to emit this signal on
 449    *
 450    * Remove all sockets from multihandlesink.  Since multihandlesink did not
 451    * open sockets itself, it does not explicitly close the sockets. The application
 452    * should do so by connecting to the client-socket-removed callback.
 453    */
 454   gst_multi_handle_sink_signals[SIGNAL_CLEAR] =
 455       g_signal_new ("clear", G_TYPE_FROM_CLASS (klass),
 456       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 457       G_STRUCT_OFFSET (GstMultiHandleSinkClass, clear), NULL, NULL,
 458       NULL, G_TYPE_NONE, 0);
 459
 460   gst_element_class_add_static_pad_template (gstelement_class, &sinktemplate);
 461
 462   gst_element_class_set_static_metadata (gstelement_class,
 463       "Multi socket sink", "Sink/Network",
 464       "Send data to multiple sockets",
 465       "Thomas Vander Stichele <thomas at apestaart dot org>, "
 466       "Wim Taymans <wim@fluendo.com>, "
 467       "Sebastian Dröge <sebastian.droege@collabora.co.uk>");
 468
 469   gstelement_class->change_state =
 470       GST_DEBUG_FUNCPTR (gst_multi_handle_sink_change_state);
 471
 472   gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_render);
 473   klass->client_queue_buffer =
 474       GST_DEBUG_FUNCPTR (gst_multi_handle_sink_client_queue_buffer);
 475
 476 #if 0
 477   klass->add = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_add);
 478   klass->add_full = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_add_full);
 479   klass->remove = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_remove);
 480   klass->remove_flush = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_remove_flush);
 481 #endif
 482
 483   klass->clear = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_clear);
 484
 485   GST_DEBUG_CATEGORY_INIT (multihandlesink_debug, "multihandlesink", 0,
 486       "Multi socket sink");
 487
 488   gst_type_mark_as_plugin_api (GST_TYPE_RECOVER_POLICY, 0);
 489   gst_type_mark_as_plugin_api (GST_TYPE_SYNC_METHOD, 0);
 490   gst_type_mark_as_plugin_api (GST_TYPE_CLIENT_STATUS, 0);
 491 }
 492
 493 static void
 494 gst_multi_handle_sink_init (GstMultiHandleSink * this)
 495 {
 496   GST_OBJECT_FLAG_UNSET (this, GST_MULTI_HANDLE_SINK_OPEN);
 497
 498   CLIENTS_LOCK_INIT (this);
 499   this->clients = NULL;
 500
 501   this->bufqueue = g_array_new (FALSE, TRUE, sizeof (GstBuffer *));
 502   this->unit_format = DEFAULT_UNIT_FORMAT;
 503   this->units_max = DEFAULT_UNITS_MAX;
 504   this->units_soft_max = DEFAULT_UNITS_SOFT_MAX;
 505   this->time_min = DEFAULT_TIME_MIN;
 506   this->bytes_min = DEFAULT_BYTES_MIN;
 507   this->buffers_min = DEFAULT_BUFFERS_MIN;
 508   this->recover_policy = DEFAULT_RECOVER_POLICY;
 509
 510   this->timeout = DEFAULT_TIMEOUT;
 511   this->def_sync_method = DEFAULT_SYNC_METHOD;
 512
 513   this->def_burst_format = DEFAULT_BURST_FORMAT;
 514   this->def_burst_value = DEFAULT_BURST_VALUE;
 515
 516   this->qos_dscp = DEFAULT_QOS_DSCP;
 517
 518   this->resend_streamheader = DEFAULT_RESEND_STREAMHEADER;
 519 }
 520
 521 static void
 522 gst_multi_handle_sink_finalize (GObject * object)
 523 {
 524   GstMultiHandleSink *this;
 525
 526   this = GST_MULTI_HANDLE_SINK (object);
 527
 528   CLIENTS_LOCK_CLEAR (this);
 529   g_array_free (this->bufqueue, TRUE);
 530   g_hash_table_destroy (this->handle_hash);
 531
 532   G_OBJECT_CLASS (parent_class)->finalize (object);
 533 }
 534
 535 gint
 536 gst_multi_handle_sink_setup_dscp_client (GstMultiHandleSink * sink,
 537     GstMultiHandleClient * client)
 538 {
 539 #if !defined(IP_TOS) || !defined(HAVE_SYS_SOCKET_H)
 540   return 0;
 541 #else
 542   gint tos;
 543   gint ret;
 544   int fd;
 545   union gst_sockaddr
 546   {
 547     struct sockaddr sa;
 548     struct sockaddr_in6 sa_in6;
 549     struct sockaddr_storage sa_stor;
 550   } sa;
 551   socklen_t slen = sizeof (sa);
 552   gint af;
 553   GstMultiHandleSinkClass *mhsinkclass = GST_MULTI_HANDLE_SINK_GET_CLASS (sink);
 554
 555   /* don't touch */
 556   if (sink->qos_dscp < 0)
 557     return 0;
 558
 559   fd = mhsinkclass->client_get_fd (client);
 560
 561   if ((ret = getsockname (fd, &sa.sa, &slen)) < 0) {
 562     GST_DEBUG_OBJECT (sink, "could not get sockname: %s", g_strerror (errno));
 563     return ret;
 564   }
 565
 566   af = sa.sa.sa_family;
 567
 568   /* if this is an IPv4-mapped address then do IPv4 QoS */
 569   if (af == AF_INET6) {
 570
 571     GST_DEBUG_OBJECT (sink, "check IP6 socket");
 572     if (IN6_IS_ADDR_V4MAPPED (&(sa.sa_in6.sin6_addr))) {
 573       GST_DEBUG_OBJECT (sink, "mapped to IPV4");
 574       af = AF_INET;
 575     }
 576   }
 577
 578   /* extract and shift 6 bits of the DSCP */
 579   tos = (sink->qos_dscp & 0x3f) << 2;
 580
 581   switch (af) {
 582     case AF_INET:
 583       ret = setsockopt (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos));
 584       break;
 585     case AF_INET6:
 586 #ifdef IPV6_TCLASS
 587       ret = setsockopt (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos));
 588       break;
 589 #endif
 590     default:
 591       ret = 0;
 592       GST_ERROR_OBJECT (sink, "unsupported AF");
 593       break;
 594   }
 595   if (ret)
 596     GST_DEBUG_OBJECT (sink, "could not set DSCP: %s", g_strerror (errno));
 597
 598   return ret;
 599 #endif
 600 }
 601
 602 void
 603 gst_multi_handle_sink_client_init (GstMultiHandleClient * client,
 604     GstSyncMethod sync_method)
 605 {
 606   client->status = GST_CLIENT_STATUS_OK;
 607   client->bufpos = -1;
 608   client->flushcount = -1;
 609   client->bufoffset = 0;
 610   client->sending = NULL;
 611   client->bytes_sent = 0;
 612   client->dropped_buffers = 0;
 613   client->avg_queue_size = 0;
 614   client->first_buffer_ts = GST_CLOCK_TIME_NONE;
 615   client->last_buffer_ts = GST_CLOCK_TIME_NONE;
 616   client->new_connection = TRUE;
 617   client->sync_method = sync_method;
 618   client->currently_removing = FALSE;
 619
 620   /* update start time */
 621   client->connect_time = g_get_real_time () * GST_USECOND;
 622   client->disconnect_time = 0;
 623   /* set last activity time to connect time */
 624   client->last_activity_time = client->connect_time;
 625 }
 626
 627 static void
 628 gst_multi_handle_sink_setup_dscp (GstMultiHandleSink * mhsink)
 629 {
 630   GList *clients;
 631
 632   CLIENTS_LOCK (mhsink);
 633   for (clients = mhsink->clients; clients; clients = clients->next) {
 634     GstMultiHandleClient *client;
 635
 636     client = clients->data;
 637
 638     gst_multi_handle_sink_setup_dscp_client (mhsink, client);
 639   }
 640   CLIENTS_UNLOCK (mhsink);
 641 }
 642
 643 void
 644 gst_multi_handle_sink_add_full (GstMultiHandleSink * sink,
 645     GstMultiSinkHandle handle, GstSyncMethod sync_method, GstFormat min_format,
 646     guint64 min_value, GstFormat max_format, guint64 max_value)
 647 {
 648   GstMultiHandleClient *mhclient;
 649   GList *clink;
 650   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (sink);
 651   gchar debug[30];
 652   GstMultiHandleSinkClass *mhsinkclass =
 653       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
 654
 655   if (!sink->running) {
 656     g_warning ("Element %s must be set to READY, PAUSED or PLAYING state "
 657         "before clients can be added", GST_OBJECT_NAME (sink));
 658     return;
 659   }
 660
 661   mhsinkclass->handle_debug (handle, debug);
 662   GST_DEBUG_OBJECT (sink, "%s adding client, sync_method %d, "
 663       "min_format %d, min_value %" G_GUINT64_FORMAT
 664       ", max_format %d, max_value %" G_GUINT64_FORMAT, debug,
 665       sync_method, min_format, min_value, max_format, max_value);
 666
 667   /* do limits check if we can */
 668   if (min_format == max_format) {
 669     if (max_value != -1 && min_value != -1 && max_value < min_value)
 670       goto wrong_limits;
 671   }
 672
 673   CLIENTS_LOCK (sink);
 674
 675   /* check the hash to find a duplicate handle */
 676   clink = g_hash_table_lookup (mhsink->handle_hash,
 677       mhsinkclass->handle_hash_key (handle));
 678   if (clink != NULL)
 679     goto duplicate;
 680
 681   /* We do not take ownership of @handle in this function, but we can't take a
 682    * reference directly as we don't know the concrete type of the handle.
 683    * GstMultiHandleSink relies on the derived class to take a reference for us
 684    * in new_client: */
 685   mhclient = mhsinkclass->new_client (mhsink, handle, sync_method);
 686
 687   /* we can add the handle now */
 688   clink = mhsink->clients = g_list_prepend (mhsink->clients, mhclient);
 689   g_hash_table_insert (mhsink->handle_hash,
 690       mhsinkclass->handle_hash_key (mhclient->handle), clink);
 691   mhsink->clients_cookie++;
 692
 693
 694   mhclient->burst_min_format = min_format;
 695   mhclient->burst_min_value = min_value;
 696   mhclient->burst_max_format = max_format;
 697   mhclient->burst_max_value = max_value;
 698
 699   if (mhsinkclass->hash_changed)
 700     mhsinkclass->hash_changed (mhsink);
 701
 702   CLIENTS_UNLOCK (sink);
 703
 704   mhsinkclass->emit_client_added (mhsink, handle);
 705
 706   return;
 707
 708   /* errors */
 709 wrong_limits:
 710   {
 711     GST_WARNING_OBJECT (sink,
 712         "%s wrong values min =%" G_GUINT64_FORMAT ", max=%"
 713         G_GUINT64_FORMAT ", unit %d specified when adding client",
 714         debug, min_value, max_value, min_format);
 715     return;
 716   }
 717 duplicate:
 718   {
 719     CLIENTS_UNLOCK (sink);
 720     GST_WARNING_OBJECT (sink, "%s duplicate client found, refusing", debug);
 721     mhsinkclass->emit_client_removed (mhsink, handle,
 722         GST_CLIENT_STATUS_DUPLICATE);
 723     return;
 724   }
 725 }
 726
 727 /* "add" signal implementation */
 728 void
 729 gst_multi_handle_sink_add (GstMultiHandleSink * sink, GstMultiSinkHandle handle)
 730 {
 731   gst_multi_handle_sink_add_full (sink, handle, sink->def_sync_method,
 732       sink->def_burst_format, sink->def_burst_value, sink->def_burst_format,
 733       -1);
 734 }
 735
 736 /* "remove" signal implementation */
 737 void
 738 gst_multi_handle_sink_remove (GstMultiHandleSink * sink,
 739     GstMultiSinkHandle handle)
 740 {
 741   GList *clink;
 742   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (sink);
 743   GstMultiHandleSinkClass *mhsinkclass =
 744       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
 745   gchar debug[30];
 746
 747   mhsinkclass->handle_debug (handle, debug);
 748
 749   GST_DEBUG_OBJECT (sink, "%s removing client", debug);
 750
 751   CLIENTS_LOCK (sink);
 752   clink = g_hash_table_lookup (mhsink->handle_hash,
 753       mhsinkclass->handle_hash_key (handle));
 754   if (clink != NULL) {
 755     GstMultiHandleClient *mhclient = (GstMultiHandleClient *) clink->data;
 756
 757     if (mhclient->status != GST_CLIENT_STATUS_OK) {
 758       GST_INFO_OBJECT (sink,
 759           "%s Client already disconnecting with status %d",
 760           debug, mhclient->status);
 761       goto done;
 762     }
 763
 764     mhclient->status = GST_CLIENT_STATUS_REMOVED;
 765     gst_multi_handle_sink_remove_client_link (GST_MULTI_HANDLE_SINK (sink),
 766         clink);
 767     if (mhsinkclass->hash_changed)
 768       mhsinkclass->hash_changed (mhsink);
 769   } else {
 770     GST_WARNING_OBJECT (sink, "%s no client with this handle found!", debug);
 771   }
 772
 773 done:
 774   CLIENTS_UNLOCK (sink);
 775 }
 776
 777 /* "remove-flush" signal implementation */
 778 void
 779 gst_multi_handle_sink_remove_flush (GstMultiHandleSink * sink,
 780     GstMultiSinkHandle handle)
 781 {
 782   GList *clink;
 783   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (sink);
 784   GstMultiHandleSinkClass *mhsinkclass =
 785       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
 786   gchar debug[30];
 787
 788   mhsinkclass->handle_debug (handle, debug);
 789
 790   GST_DEBUG_OBJECT (sink, "%s flushing client", debug);
 791
 792   CLIENTS_LOCK (sink);
 793   clink = g_hash_table_lookup (mhsink->handle_hash,
 794       mhsinkclass->handle_hash_key (handle));
 795   if (clink != NULL) {
 796     GstMultiHandleClient *mhclient = (GstMultiHandleClient *) clink->data;
 797
 798     if (mhclient->status != GST_CLIENT_STATUS_OK) {
 799       GST_INFO_OBJECT (sink,
 800           "%s Client already disconnecting with status %d",
 801           mhclient->debug, mhclient->status);
 802       goto done;
 803     }
 804
 805     /* take the position of the client as the number of buffers left to flush.
 806      * If the client was at position -1, we flush 0 buffers, 0 == flush 1
 807      * buffer, etc... */
 808     mhclient->flushcount = mhclient->bufpos + 1;
 809     /* mark client as flushing. We can not remove the client right away because
 810      * it might have some buffers to flush in the ->sending queue. */
 811     mhclient->status = GST_CLIENT_STATUS_FLUSHING;
 812   } else {
 813     GST_WARNING_OBJECT (sink, "%s no client with this handle found!", debug);
 814   }
 815 done:
 816   CLIENTS_UNLOCK (sink);
 817 }
 818
 819 /* can be called both through the signal (i.e. from any thread) or when
 820  * stopping, after the writing thread has shut down */
 821 static void
 822 gst_multi_handle_sink_clear (GstMultiHandleSink * mhsink)
 823 {
 824   GList *clients, *next;
 825   guint32 cookie;
 826   GstMultiHandleSinkClass *mhsinkclass =
 827       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
 828
 829   GST_DEBUG_OBJECT (mhsink, "clearing all clients");
 830
 831   CLIENTS_LOCK (mhsink);
 832 restart:
 833   cookie = mhsink->clients_cookie;
 834   for (clients = mhsink->clients; clients; clients = next) {
 835     GstMultiHandleClient *mhclient;
 836
 837     if (cookie != mhsink->clients_cookie) {
 838       GST_DEBUG_OBJECT (mhsink, "cookie changed while removing all clients");
 839       goto restart;
 840     }
 841
 842     mhclient = (GstMultiHandleClient *) clients->data;
 843     next = g_list_next (clients);
 844
 845     mhclient->status = GST_CLIENT_STATUS_REMOVED;
 846     /* the next call changes the list, which is why we iterate
 847      * with a temporary next pointer */
 848     gst_multi_handle_sink_remove_client_link (mhsink, clients);
 849   }
 850   if (mhsinkclass->hash_changed)
 851     mhsinkclass->hash_changed (mhsink);
 852
 853   CLIENTS_UNLOCK (mhsink);
 854 }
 855
 856
 857 /* "get-stats" signal implementation
 858  */
 859 GstStructure *
 860 gst_multi_handle_sink_get_stats (GstMultiHandleSink * sink,
 861     GstMultiSinkHandle handle)
 862 {
 863   GstMultiHandleClient *client;
 864   GstStructure *result = NULL;
 865   GList *clink;
 866   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (sink);
 867   GstMultiHandleSinkClass *mhsinkclass =
 868       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
 869   gchar debug[30];
 870
 871   mhsinkclass->handle_debug (handle, debug);
 872
 873   CLIENTS_LOCK (sink);
 874   clink = g_hash_table_lookup (mhsink->handle_hash,
 875       mhsinkclass->handle_hash_key (handle));
 876   if (clink == NULL)
 877     goto noclient;
 878
 879   client = clink->data;
 880   if (client != NULL) {
 881     GstMultiHandleClient *mhclient = (GstMultiHandleClient *) client;
 882     guint64 interval;
 883
 884     result = gst_structure_new_empty ("multihandlesink-stats");
 885
 886     if (mhclient->disconnect_time == 0) {
 887       interval = (g_get_real_time () * GST_USECOND) - mhclient->connect_time;
 888     } else {
 889       interval = mhclient->disconnect_time - mhclient->connect_time;
 890     }
 891
 892     gst_structure_set (result,
 893         "bytes-sent", G_TYPE_UINT64, mhclient->bytes_sent,
 894         "connect-time", G_TYPE_UINT64, mhclient->connect_time,
 895         "disconnect-time", G_TYPE_UINT64, mhclient->disconnect_time,
 896         "connect-duration", G_TYPE_UINT64, interval,
 897         "last-activity-time", G_TYPE_UINT64, mhclient->last_activity_time,
 898         "buffers-dropped", G_TYPE_UINT64, mhclient->dropped_buffers,
 899         "first-buffer-ts", G_TYPE_UINT64, mhclient->first_buffer_ts,
 900         "last-buffer-ts", G_TYPE_UINT64, mhclient->last_buffer_ts, NULL);
 901   }
 902
 903 noclient:
 904   CLIENTS_UNLOCK (sink);
 905
 906   /* python doesn't like a NULL pointer yet */
 907   if (result == NULL) {
 908     GST_WARNING_OBJECT (sink, "%s no client with this found!", debug);
 909     result = gst_structure_new_empty ("multihandlesink-stats");
 910   }
 911
 912   return result;
 913 }
 914
 915 /* should be called with the clientslock held.
 916  * Note that we don't close the fd as we didn't open it in the first
 917  * place. An application should connect to the client-fd-removed signal and
 918  * close the fd itself.
 919  */
 920 void
 921 gst_multi_handle_sink_remove_client_link (GstMultiHandleSink * sink,
 922     GList * link)
 923 {
 924   GstMultiHandleClient *mhclient = (GstMultiHandleClient *) link->data;
 925   GstMultiHandleSinkClass *mhsinkclass = GST_MULTI_HANDLE_SINK_GET_CLASS (sink);
 926
 927   if (mhclient->currently_removing) {
 928     GST_WARNING_OBJECT (sink, "%s client is already being removed",
 929         mhclient->debug);
 930     return;
 931   } else {
 932     mhclient->currently_removing = TRUE;
 933   }
 934
 935   /* FIXME: if we keep track of ip we can log it here and signal */
 936   switch (mhclient->status) {
 937     case GST_CLIENT_STATUS_OK:
 938       GST_WARNING_OBJECT (sink, "%s removing client %p for no reason",
 939           mhclient->debug, mhclient);
 940       break;
 941     case GST_CLIENT_STATUS_CLOSED:
 942       GST_DEBUG_OBJECT (sink, "%s removing client %p because of close",
 943           mhclient->debug, mhclient);
 944       break;
 945     case GST_CLIENT_STATUS_REMOVED:
 946       GST_DEBUG_OBJECT (sink,
 947           "%s removing client %p because the app removed it", mhclient->debug,
 948           mhclient);
 949       break;
 950     case GST_CLIENT_STATUS_SLOW:
 951       GST_INFO_OBJECT (sink,
 952           "%s removing client %p because it was too slow", mhclient->debug,
 953           mhclient);
 954       break;
 955     case GST_CLIENT_STATUS_ERROR:
 956       GST_WARNING_OBJECT (sink,
 957           "%s removing client %p because of error", mhclient->debug, mhclient);
 958       break;
 959     case GST_CLIENT_STATUS_FLUSHING:
 960     default:
 961       GST_WARNING_OBJECT (sink,
 962           "%s removing client %p with invalid reason %d", mhclient->debug,
 963           mhclient, mhclient->status);
 964       break;
 965   }
 966
 967   mhsinkclass->hash_removing (sink, mhclient);
 968
 969   mhclient->disconnect_time = g_get_real_time () * GST_USECOND;
 970
 971   /* free client buffers */
 972   g_slist_foreach (mhclient->sending, (GFunc) gst_mini_object_unref, NULL);
 973   g_slist_free (mhclient->sending);
 974   mhclient->sending = NULL;
 975
 976   if (mhclient->caps)
 977     gst_caps_unref (mhclient->caps);
 978   mhclient->caps = NULL;
 979
 980   /* unlock the mutex before signaling because the signal handler
 981    * might query some properties */
 982   CLIENTS_UNLOCK (sink);
 983
 984   mhsinkclass->emit_client_removed (sink, mhclient->handle, mhclient->status);
 985
 986   /* lock again before we remove the client completely */
 987   CLIENTS_LOCK (sink);
 988
 989   /* handle cannot be reused in the above signal callback so we can safely
 990    * remove it from the hashtable here */
 991   if (!g_hash_table_remove (sink->handle_hash,
 992           mhsinkclass->handle_hash_key (mhclient->handle))) {
 993     GST_WARNING_OBJECT (sink,
 994         "%s error removing client %p from hash", mhclient->debug, mhclient);
 995   }
 996   /* after releasing the lock above, the link could be invalid, more
 997    * precisely, the next and prev pointers could point to invalid list
 998    * links. One optimisation could be to add a cookie to the linked list
 999    * and take a shortcut when it did not change between unlocking and locking
1000    * our mutex. For now we just walk the list again. */
1001   sink->clients = g_list_remove (sink->clients, mhclient);
1002   sink->clients_cookie++;
1003
1004   if (mhsinkclass->removed)
1005     mhsinkclass->removed (sink, mhclient->handle);
1006
1007   CLIENTS_UNLOCK (sink);
1008
1009   /* sub-class must implement this to emit the client-$handle-removed signal */
1010   g_assert (mhsinkclass->client_free != NULL);
1011
1012   /* and the handle is really gone now */
1013   mhsinkclass->client_free (sink, mhclient);
1014
1015   g_free (mhclient);
1016
1017   CLIENTS_LOCK (sink);
1018 }
1019
1020 static gboolean
1021 gst_multi_handle_sink_client_queue_buffer (GstMultiHandleSink * mhsink,
1022     GstMultiHandleClient * mhclient, GstBuffer * buffer)
1023 {
1024   GstMultiHandleSink *sink = GST_MULTI_HANDLE_SINK (mhsink);
1025   GstCaps *caps;
1026
1027   /* TRUE: send them if the new caps have them */
1028   gboolean send_streamheader = FALSE;
1029   GstStructure *s;
1030
1031   /* before we queue the buffer, we check if we need to queue streamheader
1032    * buffers (because it's a new client, or because they changed) */
1033   caps = gst_pad_get_current_caps (GST_BASE_SINK_PAD (sink));
1034
1035   if (!mhclient->caps) {
1036     if (caps) {
1037       GST_DEBUG_OBJECT (sink,
1038           "%s no previous caps for this client, send streamheader",
1039           mhclient->debug);
1040       send_streamheader = TRUE;
1041       mhclient->caps = gst_caps_ref (caps);
1042     }
1043   } else {
1044     /* there were previous caps recorded, so compare */
1045     if (!gst_caps_is_equal (caps, mhclient->caps)) {
1046       const GValue *sh1, *sh2;
1047
1048       /* caps are not equal, but could still have the same streamheader */
1049       s = gst_caps_get_structure (caps, 0);
1050       if (!gst_structure_has_field (s, "streamheader")) {
1051         /* no new streamheader, so nothing new to send */
1052         GST_DEBUG_OBJECT (sink,
1053             "%s new caps do not have streamheader, not sending",
1054             mhclient->debug);
1055       } else {
1056         /* there is a new streamheader */
1057         s = gst_caps_get_structure (mhclient->caps, 0);
1058         if (!gst_structure_has_field (s, "streamheader")) {
1059           /* no previous streamheader, so send the new one */
1060           GST_DEBUG_OBJECT (sink,
1061               "%s previous caps did not have streamheader, sending",
1062               mhclient->debug);
1063           send_streamheader = TRUE;
1064         } else {
1065           /* both old and new caps have streamheader set */
1066           if (!mhsink->resend_streamheader) {
1067             GST_DEBUG_OBJECT (sink,
1068                 "%s asked to not resend the streamheader, not sending",
1069                 mhclient->debug);
1070             send_streamheader = FALSE;
1071           } else {
1072             sh1 = gst_structure_get_value (s, "streamheader");
1073             s = gst_caps_get_structure (caps, 0);
1074             sh2 = gst_structure_get_value (s, "streamheader");
1075             if (gst_value_compare (sh1, sh2) != GST_VALUE_EQUAL) {
1076               GST_DEBUG_OBJECT (sink,
1077                   "%s new streamheader different from old, sending",
1078                   mhclient->debug);
1079               send_streamheader = TRUE;
1080             }
1081           }
1082         }
1083       }
1084     }
1085     /* Replace the old caps */
1086     gst_caps_replace (&mhclient->caps, caps);
1087   }
1088
1089   if (G_UNLIKELY (send_streamheader)) {
1090     const GValue *sh;
1091     GArray *buffers;
1092     int i;
1093
1094     GST_LOG_OBJECT (sink,
1095         "%s sending streamheader from caps %" GST_PTR_FORMAT,
1096         mhclient->debug, caps);
1097     s = gst_caps_get_structure (caps, 0);
1098     if (!gst_structure_has_field (s, "streamheader")) {
1099       GST_DEBUG_OBJECT (sink,
1100           "%s no new streamheader, so nothing to send", mhclient->debug);
1101     } else {
1102       GST_LOG_OBJECT (sink,
1103           "%s sending streamheader from caps %" GST_PTR_FORMAT,
1104           mhclient->debug, caps);
1105       sh = gst_structure_get_value (s, "streamheader");
1106       g_assert (G_VALUE_TYPE (sh) == GST_TYPE_ARRAY);
1107       buffers = g_value_peek_pointer (sh);
1108       GST_DEBUG_OBJECT (sink, "%d streamheader buffers", buffers->len);
1109       for (i = 0; i < buffers->len; ++i) {
1110         GValue *bufval;
1111         GstBuffer *buffer;
1112
1113         bufval = &g_array_index (buffers, GValue, i);
1114         g_assert (G_VALUE_TYPE (bufval) == GST_TYPE_BUFFER);
1115         buffer = g_value_peek_pointer (bufval);
1116         GST_DEBUG_OBJECT (sink,
1117             "%s queueing streamheader buffer of length %" G_GSIZE_FORMAT,
1118             mhclient->debug, gst_buffer_get_size (buffer));
1119         gst_buffer_ref (buffer);
1120
1121         mhclient->sending = g_slist_append (mhclient->sending, buffer);
1122       }
1123     }
1124   }
1125
1126   if (caps)
1127     gst_caps_unref (caps);
1128   caps = NULL;
1129
1130   GST_LOG_OBJECT (sink, "%s queueing buffer of length %" G_GSIZE_FORMAT,
1131       mhclient->debug, gst_buffer_get_size (buffer));
1132
1133   gst_buffer_ref (buffer);
1134   mhclient->sending = g_slist_append (mhclient->sending, buffer);
1135
1136   return TRUE;
1137 }
1138
1139 static gboolean
1140 is_sync_frame (GstMultiHandleSink * sink, GstBuffer * buffer)
1141 {
1142   if (GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT))
1143     return FALSE;
1144   return TRUE;
1145 }
1146
1147 /* find the keyframe in the list of buffers starting the
1148  * search from @idx. @direction as -1 will search backwards,
1149  * 1 will search forwards.
1150  * Returns: the index or -1 if there is no keyframe after idx.
1151  */
1152 gint
1153 find_syncframe (GstMultiHandleSink * sink, gint idx, gint direction)
1154 {
1155   gint i, len, result;
1156
1157   /* take length of queued buffers */
1158   len = sink->bufqueue->len;
1159
1160   /* assume we don't find a keyframe */
1161   result = -1;
1162
1163   /* then loop over all buffers to find the first keyframe */
1164   for (i = idx; i >= 0 && i < len; i += direction) {
1165     GstBuffer *buf;
1166
1167     buf = g_array_index (sink->bufqueue, GstBuffer *, i);
1168     if (is_sync_frame (sink, buf)) {
1169       GST_LOG_OBJECT (sink, "found keyframe at %d from %d, direction %d",
1170           i, idx, direction);
1171       result = i;
1172       break;
1173     }
1174   }
1175   return result;
1176 }
1177
1178 /* Get the number of buffers from the buffer queue needed to satisfy
1179  * the maximum max in the configured units.
1180  * If units are not BUFFERS, and there are insufficient buffers in the
1181  * queue to satisfy the limit, return len(queue) + 1 */
1182 gint
1183 get_buffers_max (GstMultiHandleSink * sink, gint64 max)
1184 {
1185   switch (sink->unit_format) {
1186     case GST_FORMAT_BUFFERS:
1187       return max;
1188     case GST_FORMAT_TIME:
1189     {
1190       GstBuffer *buf;
1191       int i;
1192       int len;
1193       gint64 diff;
1194       GstClockTime first = GST_CLOCK_TIME_NONE;
1195
1196       len = sink->bufqueue->len;
1197
1198       for (i = 0; i < len; i++) {
1199         buf = g_array_index (sink->bufqueue, GstBuffer *, i);
1200         if (GST_BUFFER_TIMESTAMP_IS_VALID (buf)) {
1201           if (first == -1)
1202             first = GST_BUFFER_TIMESTAMP (buf);
1203
1204           diff = first - GST_BUFFER_TIMESTAMP (buf);
1205
1206           if (diff > max)
1207             return i + 1;
1208         }
1209       }
1210       return len + 1;
1211     }
1212     case GST_FORMAT_BYTES:
1213     {
1214       GstBuffer *buf;
1215       int i;
1216       int len;
1217       gint acc = 0;
1218
1219       len = sink->bufqueue->len;
1220
1221       for (i = 0; i < len; i++) {
1222         buf = g_array_index (sink->bufqueue, GstBuffer *, i);
1223         acc += gst_buffer_get_size (buf);
1224
1225         if (acc > max)
1226           return i + 1;
1227       }
1228       return len + 1;
1229     }
1230     default:
1231       return max;
1232   }
1233 }
1234
1235 /* find the positions in the buffer queue where *_min and *_max
1236  * is satisfied
1237  */
1238 /* count the amount of data in the buffers and return the index
1239  * that satisfies the given limits.
1240  *
1241  * Returns: index @idx in the buffer queue so that the given limits are
1242  * satisfied. TRUE if all the limits could be satisfied, FALSE if not
1243  * enough data was in the queue.
1244  *
1245  * FIXME, this code might now work if any of the units is in buffers...
1246  */
1247 gboolean
1248 find_limits (GstMultiHandleSink * sink,
1249     gint * min_idx, gint bytes_min, gint buffers_min, gint64 time_min,
1250     gint * max_idx, gint bytes_max, gint buffers_max, gint64 time_max)
1251 {
1252   GstClockTime first, time;
1253   gint i, len, bytes;
1254   gboolean result, max_hit;
1255
1256   /* take length of queue */
1257   len = sink->bufqueue->len;
1258
1259   /* this must hold */
1260   g_assert (len > 0);
1261
1262   GST_LOG_OBJECT (sink,
1263       "bytes_min %d, buffers_min %d, time_min %" GST_TIME_FORMAT
1264       ", bytes_max %d, buffers_max %d, time_max %" GST_TIME_FORMAT, bytes_min,
1265       buffers_min, GST_TIME_ARGS (time_min), bytes_max, buffers_max,
1266       GST_TIME_ARGS (time_max));
1267
1268   /* do the trivial buffer limit test */
1269   if (buffers_min != -1 && len < buffers_min) {
1270     *min_idx = len - 1;
1271     *max_idx = len - 1;
1272     return FALSE;
1273   }
1274
1275   result = FALSE;
1276   /* else count bytes and time */
1277   first = -1;
1278   bytes = 0;
1279   /* unset limits */
1280   *min_idx = -1;
1281   *max_idx = -1;
1282   max_hit = FALSE;
1283
1284   i = 0;
1285   /* loop through the buffers, when a limit is ok, mark it
1286    * as -1, we have at least one buffer in the queue. */
1287   do {
1288     GstBuffer *buf;
1289
1290     /* if we checked all min limits, update result */
1291     if (bytes_min == -1 && time_min == -1 && *min_idx == -1) {
1292       /* don't go below 0 */
1293       *min_idx = MAX (i - 1, 0);
1294     }
1295     /* if we reached one max limit break out */
1296     if (max_hit) {
1297       /* i > 0 when we get here, we subtract one to get the position
1298        * of the previous buffer. */
1299       *max_idx = i - 1;
1300       /* we have valid complete result if we found a min_idx too */
1301       result = *min_idx != -1;
1302       break;
1303     }
1304     buf = g_array_index (sink->bufqueue, GstBuffer *, i);
1305
1306     bytes += gst_buffer_get_size (buf);
1307
1308     /* take timestamp and save for the base first timestamp */
1309     if ((time = GST_BUFFER_TIMESTAMP (buf)) != -1) {
1310       GST_LOG_OBJECT (sink, "Ts %" GST_TIME_FORMAT " on buffer",
1311           GST_TIME_ARGS (time));
1312       if (first == -1)
1313         first = time;
1314
1315       /* increase max usage if we did not fill enough. Note that
1316        * buffers are sorted from new to old, so the first timestamp is
1317        * bigger than the next one. */
1318       if (time_min != -1 && first - time >= time_min)
1319         time_min = -1;
1320       if (time_max != -1 && first - time >= time_max)
1321         max_hit = TRUE;
1322     } else {
1323       GST_LOG_OBJECT (sink, "No timestamp on buffer");
1324     }
1325     /* time is OK or unknown, check and increase if not enough bytes */
1326     if (bytes_min != -1) {
1327       if (bytes >= bytes_min)
1328         bytes_min = -1;
1329     }
1330     if (bytes_max != -1) {
1331       if (bytes >= bytes_max) {
1332         max_hit = TRUE;
1333       }
1334     }
1335     i++;
1336   }
1337   while (i < len);
1338
1339   /* if we did not hit the max or min limit, set to buffer size */
1340   if (*max_idx == -1)
1341     *max_idx = len - 1;
1342   /* make sure min does not exceed max */
1343   if (*min_idx == -1)
1344     *min_idx = *max_idx;
1345
1346   return result;
1347 }
1348
1349 /* parse the unit/value pair and assign it to the result value of the
1350  * right type, leave the other values untouched
1351  *
1352  * Returns: FALSE if the unit is unknown or undefined. TRUE otherwise.
1353  */
1354 static gboolean
1355 assign_value (GstFormat format, guint64 value, gint * bytes, gint * buffers,
1356     GstClockTime * time)
1357 {
1358   gboolean res = TRUE;
1359
1360   /* set only the limit of the given format to the given value */
1361   switch (format) {
1362     case GST_FORMAT_BUFFERS:
1363       *buffers = (gint) value;
1364       break;
1365     case GST_FORMAT_TIME:
1366       *time = value;
1367       break;
1368     case GST_FORMAT_BYTES:
1369       *bytes = (gint) value;
1370       break;
1371     case GST_FORMAT_UNDEFINED:
1372     default:
1373       res = FALSE;
1374       break;
1375   }
1376   return res;
1377 }
1378
1379 /* count the index in the buffer queue to satisfy the given unit
1380  * and value pair starting from buffer at index 0.
1381  *
1382  * Returns: TRUE if there was enough data in the queue to satisfy the
1383  * burst values. @idx contains the index in the buffer that contains enough
1384  * data to satisfy the limits or the last buffer in the queue when the
1385  * function returns FALSE.
1386  */
1387 static gboolean
1388 count_burst_unit (GstMultiHandleSink * sink, gint * min_idx,
1389     GstFormat min_format, guint64 min_value, gint * max_idx,
1390     GstFormat max_format, guint64 max_value)
1391 {
1392   gint bytes_min = -1, buffers_min = -1;
1393   gint bytes_max = -1, buffers_max = -1;
1394   GstClockTime time_min = GST_CLOCK_TIME_NONE, time_max = GST_CLOCK_TIME_NONE;
1395
1396   assign_value (min_format, min_value, &bytes_min, &buffers_min, &time_min);
1397   assign_value (max_format, max_value, &bytes_max, &buffers_max, &time_max);
1398
1399   return find_limits (sink, min_idx, bytes_min, buffers_min, time_min,
1400       max_idx, bytes_max, buffers_max, time_max);
1401 }
1402
1403 /* decide where in the current buffer queue this new client should start
1404  * receiving buffers from.
1405  * This function is called whenever a client is connected and has not yet
1406  * received a buffer.
1407  * If this returns -1, it means that we haven't found a good point to
1408  * start streaming from yet, and this function should be called again later
1409  * when more buffers have arrived.
1410  */
1411 gint
1412 gst_multi_handle_sink_new_client_position (GstMultiHandleSink * sink,
1413     GstMultiHandleClient * client)
1414 {
1415   gint result;
1416
1417   GST_DEBUG_OBJECT (sink,
1418       "%s new client, deciding where to start in queue", client->debug);
1419   GST_DEBUG_OBJECT (sink, "queue is currently %d buffers long",
1420       sink->bufqueue->len);
1421   switch (client->sync_method) {
1422     case GST_SYNC_METHOD_LATEST:
1423       /* no syncing, we are happy with whatever the client is going to get */
1424       result = client->bufpos;
1425       GST_DEBUG_OBJECT (sink,
1426           "%s SYNC_METHOD_LATEST, position %d", client->debug, result);
1427       break;
1428     case GST_SYNC_METHOD_NEXT_KEYFRAME:
1429     {
1430       /* if one of the new buffers (between client->bufpos and 0) in the queue
1431        * is a sync point, we can proceed, otherwise we need to keep waiting */
1432       GST_LOG_OBJECT (sink,
1433           "%s new client, bufpos %d, waiting for keyframe",
1434           client->debug, client->bufpos);
1435
1436       result = find_prev_syncframe (sink, client->bufpos);
1437       if (result != -1) {
1438         GST_DEBUG_OBJECT (sink,
1439             "%s SYNC_METHOD_NEXT_KEYFRAME: result %d", client->debug, result);
1440         break;
1441       }
1442
1443       /* client is not on a syncbuffer, need to skip these buffers and
1444        * wait some more */
1445       GST_LOG_OBJECT (sink,
1446           "%s new client, skipping buffer(s), no syncpoint found",
1447           client->debug);
1448       client->bufpos = -1;
1449       break;
1450     }
1451     case GST_SYNC_METHOD_LATEST_KEYFRAME:
1452     {
1453       GST_DEBUG_OBJECT (sink, "%s SYNC_METHOD_LATEST_KEYFRAME", client->debug);
1454
1455       /* for new clients we initially scan the complete buffer queue for
1456        * a sync point when a buffer is added. If we don't find a keyframe,
1457        * we need to wait for the next keyframe and so we change the client's
1458        * sync method to GST_SYNC_METHOD_NEXT_KEYFRAME.
1459        */
1460       result = find_next_syncframe (sink, 0);
1461       if (result != -1) {
1462         GST_DEBUG_OBJECT (sink,
1463             "%s SYNC_METHOD_LATEST_KEYFRAME: result %d", client->debug, result);
1464         break;
1465       }
1466
1467       GST_DEBUG_OBJECT (sink,
1468           "%s SYNC_METHOD_LATEST_KEYFRAME: no keyframe found, "
1469           "switching to SYNC_METHOD_NEXT_KEYFRAME", client->debug);
1470       /* throw client to the waiting state */
1471       client->bufpos = -1;
1472       /* and make client sync to next keyframe */
1473       client->sync_method = GST_SYNC_METHOD_NEXT_KEYFRAME;
1474       break;
1475     }
1476     case GST_SYNC_METHOD_BURST:
1477     {
1478       gboolean ok;
1479       gint max;
1480
1481       /* move to the position where we satisfy the client's burst
1482        * parameters. If we could not satisfy the parameters because there
1483        * is not enough data, we just send what we have (which is in result).
1484        * We use the max value to limit the search
1485        */
1486       ok = count_burst_unit (sink, &result, client->burst_min_format,
1487           client->burst_min_value, &max, client->burst_max_format,
1488           client->burst_max_value);
1489       GST_DEBUG_OBJECT (sink,
1490           "%s SYNC_METHOD_BURST: burst_unit returned %d, result %d",
1491           client->debug, ok, result);
1492
1493       GST_LOG_OBJECT (sink, "min %d, max %d", result, max);
1494
1495       /* we hit the max and it is below the min, use that then */
1496       if (max != -1 && max <= result) {
1497         result = MAX (max - 1, 0);
1498         GST_DEBUG_OBJECT (sink,
1499             "%s SYNC_METHOD_BURST: result above max, taken down to %d",
1500             client->debug, result);
1501       }
1502       break;
1503     }
1504     case GST_SYNC_METHOD_BURST_KEYFRAME:
1505     {
1506       gint min_idx, max_idx;
1507       gint next_syncframe, prev_syncframe;
1508
1509       /* BURST_KEYFRAME:
1510        *
1511        * _always_ start sending a keyframe to the client. We first search
1512        * a keyframe between min/max limits. If there is none, we send it the
1513        * last keyframe before min. If there is none, the behaviour is like
1514        * NEXT_KEYFRAME.
1515        */
1516       /* gather burst limits */
1517       count_burst_unit (sink, &min_idx, client->burst_min_format,
1518           client->burst_min_value, &max_idx, client->burst_max_format,
1519           client->burst_max_value);
1520
1521       GST_LOG_OBJECT (sink, "min %d, max %d", min_idx, max_idx);
1522
1523       /* first find a keyframe after min_idx */
1524       next_syncframe = find_next_syncframe (sink, min_idx);
1525       if (next_syncframe != -1 && next_syncframe < max_idx) {
1526         /* we have a valid keyframe and it's below the max */
1527         GST_LOG_OBJECT (sink, "found keyframe in min/max limits");
1528         result = next_syncframe;
1529         break;
1530       }
1531
1532       /* no valid keyframe, try to find one below min */
1533       prev_syncframe = find_prev_syncframe (sink, min_idx);
1534       if (prev_syncframe != -1) {
1535         GST_WARNING_OBJECT (sink,
1536             "using keyframe below min in BURST_KEYFRAME sync mode");
1537         result = prev_syncframe;
1538         break;
1539       }
1540
1541       /* no prev keyframe or not enough data  */
1542       GST_WARNING_OBJECT (sink,
1543           "no prev keyframe found in BURST_KEYFRAME sync mode, waiting for next");
1544
1545       /* throw client to the waiting state */
1546       client->bufpos = -1;
1547       /* and make client sync to next keyframe */
1548       client->sync_method = GST_SYNC_METHOD_NEXT_KEYFRAME;
1549       result = -1;
1550       break;
1551     }
1552     case GST_SYNC_METHOD_BURST_WITH_KEYFRAME:
1553     {
1554       gint min_idx, max_idx;
1555       gint next_syncframe;
1556
1557       /* BURST_WITH_KEYFRAME:
1558        *
1559        * try to start sending a keyframe to the client. We first search
1560        * a keyframe between min/max limits. If there is none, we send it the
1561        * amount of data up 'till min.
1562        */
1563       /* gather enough data to burst */
1564       count_burst_unit (sink, &min_idx, client->burst_min_format,
1565           client->burst_min_value, &max_idx, client->burst_max_format,
1566           client->burst_max_value);
1567
1568       GST_LOG_OBJECT (sink, "min %d, max %d", min_idx, max_idx);
1569
1570       /* first find a keyframe after min_idx */
1571       next_syncframe = find_next_syncframe (sink, min_idx);
1572       if (next_syncframe != -1 && next_syncframe < max_idx) {
1573         /* we have a valid keyframe and it's below the max */
1574         GST_LOG_OBJECT (sink, "found keyframe in min/max limits");
1575         result = next_syncframe;
1576         break;
1577       }
1578
1579       /* no keyframe, send data from min_idx */
1580       GST_WARNING_OBJECT (sink, "using min in BURST_WITH_KEYFRAME sync mode");
1581
1582       /* make sure we don't go over the max limit */
1583       if (max_idx != -1 && max_idx <= min_idx) {
1584         result = MAX (max_idx - 1, 0);
1585       } else {
1586         result = min_idx;
1587       }
1588
1589       break;
1590     }
1591     default:
1592       g_warning ("unknown sync method %d", client->sync_method);
1593       result = client->bufpos;
1594       break;
1595   }
1596   return result;
1597 }
1598
1599 /* calculate the new position for a client after recovery. This function
1600  * does not update the client position but merely returns the required
1601  * position.
1602  */
1603 gint
1604 gst_multi_handle_sink_recover_client (GstMultiHandleSink * sink,
1605     GstMultiHandleClient * client)
1606 {
1607   gint newbufpos;
1608
1609   GST_WARNING_OBJECT (sink,
1610       "%s client %p is lagging at %d, recover using policy %d",
1611       client->debug, client, client->bufpos, sink->recover_policy);
1612
1613   switch (sink->recover_policy) {
1614     case GST_RECOVER_POLICY_NONE:
1615       /* do nothing, client will catch up or get kicked out when it reaches
1616        * the hard max */
1617       newbufpos = client->bufpos;
1618       break;
1619     case GST_RECOVER_POLICY_RESYNC_LATEST:
1620       /* move to beginning of queue */
1621       newbufpos = -1;
1622       break;
1623     case GST_RECOVER_POLICY_RESYNC_SOFT_LIMIT:
1624       /* move to beginning of soft max */
1625       newbufpos = get_buffers_max (sink, sink->units_soft_max);
1626       break;
1627     case GST_RECOVER_POLICY_RESYNC_KEYFRAME:
1628       /* find keyframe in buffers, we search backwards to find the
1629        * closest keyframe relative to what this client already received. */
1630       newbufpos = MIN (sink->bufqueue->len - 1,
1631           get_buffers_max (sink, sink->units_soft_max) - 1);
1632
1633       while (newbufpos >= 0) {
1634         GstBuffer *buf;
1635
1636         buf = g_array_index (sink->bufqueue, GstBuffer *, newbufpos);
1637         if (is_sync_frame (sink, buf)) {
1638           /* found a buffer that is not a delta unit */
1639           break;
1640         }
1641         newbufpos--;
1642       }
1643       break;
1644     default:
1645       /* unknown recovery procedure */
1646       newbufpos = get_buffers_max (sink, sink->units_soft_max);
1647       break;
1648   }
1649   return newbufpos;
1650 }
1651
1652 /* Queue a buffer on the global queue.
1653  *
1654  * This function adds the buffer to the front of a GArray. It removes the
1655  * tail buffer if the max queue size is exceeded, unreffing the queued buffer.
1656  * Note that unreffing the buffer is not a problem as clients who
1657  * started writing out this buffer will still have a reference to it in the
1658  * mhclient->sending queue.
1659  *
1660  * After adding the buffer, we update all client positions in the queue. If
1661  * a client moves over the soft max, we start the recovery procedure for this
1662  * slow client. If it goes over the hard max, it is put into the slow list
1663  * and removed.
1664  *
1665  * Special care is taken of clients that were waiting for a new buffer (they
1666  * had a position of -1) because they can proceed after adding this new buffer.
1667  * This is done by adding the client back into the write fd_set and signaling
1668  * the select thread that the fd_set changed.
1669  */
1670 static void
1671 gst_multi_handle_sink_queue_buffer (GstMultiHandleSink * mhsink,
1672     GstBuffer * buffer)
1673 {
1674   GList *clients, *next;
1675   gint queuelen;
1676   gboolean hash_changed = FALSE;
1677   gint max_buffer_usage;
1678   gint i;
1679   GstClockTime now;
1680   gint max_buffers, soft_max_buffers;
1681   guint cookie;
1682   GstMultiHandleSink *sink = GST_MULTI_HANDLE_SINK (mhsink);
1683   GstMultiHandleSinkClass *mhsinkclass =
1684       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
1685
1686   CLIENTS_LOCK (mhsink);
1687   /* add buffer to queue */
1688   g_array_prepend_val (mhsink->bufqueue, buffer);
1689   queuelen = mhsink->bufqueue->len;
1690
1691   if (mhsink->units_max > 0)
1692     max_buffers = get_buffers_max (mhsink, mhsink->units_max);
1693   else
1694     max_buffers = -1;
1695
1696   if (mhsink->units_soft_max > 0)
1697     soft_max_buffers = get_buffers_max (mhsink, mhsink->units_soft_max);
1698   else
1699     soft_max_buffers = -1;
1700   GST_LOG_OBJECT (sink, "Using max %d, softmax %d", max_buffers,
1701       soft_max_buffers);
1702
1703   /* then loop over the clients and update the positions */
1704   cookie = mhsink->clients_cookie;
1705   for (clients = mhsink->clients; clients; clients = clients->next) {
1706     GstMultiHandleClient *mhclient = clients->data;
1707
1708     mhclient->bufpos++;
1709     GST_LOG_OBJECT (sink, "%s client %p at position %d",
1710         mhclient->debug, mhclient, mhclient->bufpos);
1711
1712     /* check soft max if needed, recover client */
1713     if (soft_max_buffers > 0 && mhclient->bufpos >= soft_max_buffers) {
1714       gint newpos;
1715
1716       newpos = gst_multi_handle_sink_recover_client (mhsink, mhclient);
1717       if (newpos != mhclient->bufpos) {
1718         mhclient->dropped_buffers += mhclient->bufpos - newpos;
1719         mhclient->bufpos = newpos;
1720         mhclient->discont = TRUE;
1721         GST_INFO_OBJECT (sink, "%s client %p position reset to %d",
1722             mhclient->debug, mhclient, mhclient->bufpos);
1723       } else {
1724         GST_INFO_OBJECT (sink,
1725             "%s client %p not recovering position", mhclient->debug, mhclient);
1726       }
1727     }
1728   }
1729
1730   max_buffer_usage = 0;
1731   now = g_get_real_time () * GST_USECOND;
1732
1733   /* now check for new or slow clients */
1734 restart:
1735   cookie = mhsink->clients_cookie;
1736   for (clients = mhsink->clients; clients; clients = next) {
1737     GstMultiHandleClient *mhclient = clients->data;
1738
1739     if (cookie != mhsink->clients_cookie) {
1740       GST_DEBUG_OBJECT (sink, "Clients cookie outdated, restarting");
1741       goto restart;
1742     }
1743
1744     next = g_list_next (clients);
1745
1746     /* check hard max and timeout, remove client */
1747     if ((max_buffers > 0 && mhclient->bufpos >= max_buffers) ||
1748         (mhsink->timeout > 0
1749             && now - mhclient->last_activity_time > mhsink->timeout)) {
1750       /* remove client */
1751       GST_WARNING_OBJECT (sink, "%s client %p is too slow, removing",
1752           mhclient->debug, mhclient);
1753       /* remove the client, the handle set will be cleared and the select thread
1754        * will be signaled */
1755       mhclient->status = GST_CLIENT_STATUS_SLOW;
1756       /* set client to invalid position while being removed */
1757       mhclient->bufpos = -1;
1758       gst_multi_handle_sink_remove_client_link (mhsink, clients);
1759       hash_changed = TRUE;
1760       continue;
1761     } else if (mhclient->bufpos == 0 || mhclient->new_connection) {
1762       /* can send data to this client now. need to signal the select thread that
1763        * the handle_set changed */
1764       mhsinkclass->hash_adding (mhsink, mhclient);
1765       hash_changed = TRUE;
1766     }
1767
1768     /* keep track of maximum buffer usage */
1769     if (mhclient->bufpos > max_buffer_usage) {
1770       max_buffer_usage = mhclient->bufpos;
1771     }
1772   }
1773
1774   /* make sure we respect bytes-min, buffers-min and time-min when they are set */
1775   {
1776     gint usage, max;
1777
1778     GST_LOG_OBJECT (sink,
1779         "extending queue %d to respect time_min %" GST_TIME_FORMAT
1780         ", bytes_min %d, buffers_min %d", max_buffer_usage,
1781         GST_TIME_ARGS (mhsink->time_min), mhsink->bytes_min,
1782         mhsink->buffers_min);
1783
1784     /* get index where the limits are ok, we don't really care if all limits
1785      * are ok, we just queue as much as we need. We also don't compare against
1786      * the max limits. */
1787     find_limits (mhsink, &usage, mhsink->bytes_min, mhsink->buffers_min,
1788         mhsink->time_min, &max, -1, -1, -1);
1789
1790     max_buffer_usage = MAX (max_buffer_usage, usage);
1791     GST_LOG_OBJECT (sink, "extended queue to %d", max_buffer_usage);
1792   }
1793
1794   /* now look for sync points and make sure there is at least one
1795    * sync point in the queue. We only do this if the LATEST_KEYFRAME or
1796    * BURST_KEYFRAME mode is selected */
1797   if (mhsink->def_sync_method == GST_SYNC_METHOD_LATEST_KEYFRAME ||
1798       mhsink->def_sync_method == GST_SYNC_METHOD_BURST_KEYFRAME) {
1799     /* no point in searching beyond the queue length */
1800     gint limit = queuelen;
1801     GstBuffer *buf;
1802
1803     /* no point in searching beyond the soft-max if any. */
1804     if (soft_max_buffers > 0) {
1805       limit = MIN (limit, soft_max_buffers);
1806     }
1807     GST_LOG_OBJECT (sink,
1808         "extending queue to include sync point, now at %d, limit is %d",
1809         max_buffer_usage, limit);
1810     for (i = 0; i < limit; i++) {
1811       buf = g_array_index (mhsink->bufqueue, GstBuffer *, i);
1812       if (is_sync_frame (mhsink, buf)) {
1813         /* found a sync frame, now extend the buffer usage to
1814          * include at least this frame. */
1815         max_buffer_usage = MAX (max_buffer_usage, i);
1816         break;
1817       }
1818     }
1819     GST_LOG_OBJECT (sink, "max buffer usage is now %d", max_buffer_usage);
1820   }
1821
1822   GST_LOG_OBJECT (sink, "len %d, usage %d", queuelen, max_buffer_usage);
1823
1824   /* nobody is referencing units after max_buffer_usage so we can
1825    * remove them from the queue. We remove them in reverse order as
1826    * this is the most optimal for GArray. */
1827   for (i = queuelen - 1; i > max_buffer_usage; i--) {
1828     GstBuffer *old;
1829
1830     /* queue exceeded max size */
1831     queuelen--;
1832     old = g_array_index (mhsink->bufqueue, GstBuffer *, i);
1833     mhsink->bufqueue = g_array_remove_index (mhsink->bufqueue, i);
1834
1835     /* unref tail buffer */
1836     gst_buffer_unref (old);
1837   }
1838   /* save for stats */
1839   mhsink->buffers_queued = max_buffer_usage + 1;
1840   CLIENTS_UNLOCK (sink);
1841
1842   /* and send a signal to thread if handle_set changed */
1843   if (hash_changed && mhsinkclass->hash_changed) {
1844     mhsinkclass->hash_changed (mhsink);
1845   }
1846 }
1847
1848 static gboolean
1849 buffer_is_in_caps (GstMultiHandleSink * sink, GstBuffer * buf)
1850 {
1851   GstCaps *caps;
1852   GstStructure *s;
1853   const GValue *v;
1854
1855   caps = gst_pad_get_current_caps (GST_BASE_SINK_PAD (sink));
1856   if (!caps)
1857     return FALSE;
1858   s = gst_caps_get_structure (caps, 0);
1859   if (!gst_structure_has_field (s, "streamheader")) {
1860     gst_caps_unref (caps);
1861     return FALSE;
1862   }
1863
1864   v = gst_structure_get_value (s, "streamheader");
1865   if (GST_VALUE_HOLDS_ARRAY (v)) {
1866     guint n = gst_value_array_get_size (v);
1867     guint i;
1868     GstMapInfo map;
1869
1870     gst_buffer_map (buf, &map, GST_MAP_READ);
1871
1872     for (i = 0; i < n; i++) {
1873       const GValue *v2 = gst_value_array_get_value (v, i);
1874       GstBuffer *buf2;
1875       GstMapInfo map2;
1876
1877       if (!GST_VALUE_HOLDS_BUFFER (v2))
1878         continue;
1879
1880       buf2 = gst_value_get_buffer (v2);
1881       if (buf == buf2) {
1882         gst_caps_unref (caps);
1883         return TRUE;
1884       }
1885       gst_buffer_map (buf2, &map2, GST_MAP_READ);
1886       if (map.size == map2.size && memcmp (map.data, map2.data, map.size) == 0) {
1887         gst_buffer_unmap (buf2, &map2);
1888         gst_buffer_unmap (buf, &map);
1889         gst_caps_unref (caps);
1890         return TRUE;
1891       }
1892       gst_buffer_unmap (buf2, &map2);
1893     }
1894     gst_buffer_unmap (buf, &map);
1895   }
1896
1897   gst_caps_unref (caps);
1898
1899   return FALSE;
1900 }
1901
1902 static GstFlowReturn
1903 gst_multi_handle_sink_render (GstBaseSink * bsink, GstBuffer * buf)
1904 {
1905   gboolean is_header, in_caps;
1906 #if 0
1907   GstCaps *bufcaps, *padcaps;
1908 #endif
1909
1910   GstMultiHandleSink *sink = GST_MULTI_HANDLE_SINK (bsink);
1911
1912   g_return_val_if_fail (GST_OBJECT_FLAG_IS_SET (sink,
1913           GST_MULTI_HANDLE_SINK_OPEN), GST_FLOW_FLUSHING);
1914
1915 #if 0
1916   /* since we check every buffer for streamheader caps, we need to make
1917    * sure every buffer has caps set */
1918   bufcaps = gst_buffer_get_caps (buf);
1919   padcaps = GST_PAD_CAPS (GST_BASE_SINK_PAD (bsink));
1920
1921   /* make sure we have caps on the pad */
1922   if (!padcaps && !bufcaps)
1923     goto no_caps;
1924 #endif
1925
1926   /* get HEADER first, code below might mess with the flags */
1927   is_header = GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_HEADER);
1928   in_caps = is_header && buffer_is_in_caps (sink, buf);
1929
1930 #if 0
1931   /* stamp the buffer with previous caps if no caps set */
1932   if (!bufcaps) {
1933     if (!gst_buffer_is_writable (buf)) {
1934       /* metadata is not writable, copy will be made and original buffer
1935        * will be unreffed so we need to ref so that we don't lose the
1936        * buffer in the render method. */
1937       gst_buffer_ref (buf);
1938       /* the new buffer is ours only, we keep it out of the scope of this
1939        * function */
1940       buf = gst_buffer_make_writable (buf);
1941     } else {
1942       /* else the metadata is writable, we ref because we keep the buffer
1943        * out of the scope of this method */
1944       gst_buffer_ref (buf);
1945     }
1946     /* buffer metadata is writable now, set the caps */
1947     gst_buffer_set_caps (buf, padcaps);
1948   } else {
1949     gst_caps_unref (bufcaps);
1950
1951     /* since we keep this buffer out of the scope of this method */
1952     gst_buffer_ref (buf);
1953   }
1954 #endif
1955   gst_buffer_ref (buf);
1956
1957   GST_LOG_OBJECT (sink, "received buffer %p, in_caps: %s, offset %"
1958       G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT
1959       ", timestamp %" GST_TIME_FORMAT ", duration %" GST_TIME_FORMAT,
1960       buf, in_caps ? "yes" : "no", GST_BUFFER_OFFSET (buf),
1961       GST_BUFFER_OFFSET_END (buf),
1962       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)),
1963       GST_TIME_ARGS (GST_BUFFER_DURATION (buf)));
1964
1965   /* if the incoming buffer is a streamheader from the caps, then we assume for now
1966    * it's a streamheader that needs to be sent to each new client.
1967    *
1968    * We don't send the buffer to the client, since streamheaders are sent
1969    * separately when necessary. */
1970   if (in_caps) {
1971     GST_DEBUG_OBJECT (sink, "ignoring HEADER buffer with length %"
1972         G_GSIZE_FORMAT, gst_buffer_get_size (buf));
1973     gst_buffer_unref (buf);
1974   } else {
1975     /* queue the buffer, this is a regular data buffer. */
1976     gst_multi_handle_sink_queue_buffer (sink, buf);
1977
1978     sink->bytes_to_serve += gst_buffer_get_size (buf);
1979   }
1980   return GST_FLOW_OK;
1981
1982   /* ERRORS */
1983 #if 0
1984 no_caps:
1985   {
1986     GST_ELEMENT_ERROR (sink, CORE, NEGOTIATION, (NULL),
1987         ("Received first buffer without caps set"));
1988     return GST_FLOW_NOT_NEGOTIATED;
1989   }
1990 #endif
1991 }
1992
1993 static void
1994 gst_multi_handle_sink_set_property (GObject * object, guint prop_id,
1995     const GValue * value, GParamSpec * pspec)
1996 {
1997   GstMultiHandleSink *multihandlesink;
1998
1999   multihandlesink = GST_MULTI_HANDLE_SINK (object);
2000
2001   switch (prop_id) {
2002     case PROP_BUFFERS_MAX:
2003       multihandlesink->units_max = g_value_get_int (value);
2004       break;
2005     case PROP_BUFFERS_SOFT_MAX:
2006       multihandlesink->units_soft_max = g_value_get_int (value);
2007       break;
2008     case PROP_TIME_MIN:
2009       multihandlesink->time_min = g_value_get_int64 (value);
2010       break;
2011     case PROP_BYTES_MIN:
2012       multihandlesink->bytes_min = g_value_get_int (value);
2013       break;
2014     case PROP_BUFFERS_MIN:
2015       multihandlesink->buffers_min = g_value_get_int (value);
2016       break;
2017     case PROP_UNIT_FORMAT:
2018       multihandlesink->unit_format = g_value_get_enum (value);
2019       break;
2020     case PROP_UNITS_MAX:
2021       multihandlesink->units_max = g_value_get_int64 (value);
2022       break;
2023     case PROP_UNITS_SOFT_MAX:
2024       multihandlesink->units_soft_max = g_value_get_int64 (value);
2025       break;
2026     case PROP_RECOVER_POLICY:
2027       multihandlesink->recover_policy = g_value_get_enum (value);
2028       break;
2029     case PROP_TIMEOUT:
2030       multihandlesink->timeout = g_value_get_uint64 (value);
2031       break;
2032     case PROP_SYNC_METHOD:
2033       multihandlesink->def_sync_method = g_value_get_enum (value);
2034       break;
2035     case PROP_BURST_FORMAT:
2036       multihandlesink->def_burst_format = g_value_get_enum (value);
2037       break;
2038     case PROP_BURST_VALUE:
2039       multihandlesink->def_burst_value = g_value_get_uint64 (value);
2040       break;
2041     case PROP_QOS_DSCP:
2042       multihandlesink->qos_dscp = g_value_get_int (value);
2043       gst_multi_handle_sink_setup_dscp (multihandlesink);
2044       break;
2045
2046     case PROP_RESEND_STREAMHEADER:
2047       multihandlesink->resend_streamheader = g_value_get_boolean (value);
2048       break;
2049
2050     default:
2051       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2052       break;
2053   }
2054 }
2055
2056 static void
2057 gst_multi_handle_sink_get_property (GObject * object, guint prop_id,
2058     GValue * value, GParamSpec * pspec)
2059 {
2060   GstMultiHandleSink *multihandlesink;
2061
2062   multihandlesink = GST_MULTI_HANDLE_SINK (object);
2063
2064   switch (prop_id) {
2065     case PROP_BUFFERS_MAX:
2066       g_value_set_int (value, multihandlesink->units_max);
2067       break;
2068     case PROP_BUFFERS_SOFT_MAX:
2069       g_value_set_int (value, multihandlesink->units_soft_max);
2070       break;
2071     case PROP_TIME_MIN:
2072       g_value_set_int64 (value, multihandlesink->time_min);
2073       break;
2074     case PROP_BYTES_MIN:
2075       g_value_set_int (value, multihandlesink->bytes_min);
2076       break;
2077     case PROP_BUFFERS_MIN:
2078       g_value_set_int (value, multihandlesink->buffers_min);
2079       break;
2080     case PROP_BUFFERS_QUEUED:
2081       g_value_set_uint (value, multihandlesink->buffers_queued);
2082       break;
2083     case PROP_BYTES_QUEUED:
2084       g_value_set_uint (value, multihandlesink->bytes_queued);
2085       break;
2086     case PROP_TIME_QUEUED:
2087       g_value_set_uint64 (value, multihandlesink->time_queued);
2088       break;
2089     case PROP_UNIT_FORMAT:
2090       g_value_set_enum (value, multihandlesink->unit_format);
2091       break;
2092     case PROP_UNITS_MAX:
2093       g_value_set_int64 (value, multihandlesink->units_max);
2094       break;
2095     case PROP_UNITS_SOFT_MAX:
2096       g_value_set_int64 (value, multihandlesink->units_soft_max);
2097       break;
2098     case PROP_RECOVER_POLICY:
2099       g_value_set_enum (value, multihandlesink->recover_policy);
2100       break;
2101     case PROP_TIMEOUT:
2102       g_value_set_uint64 (value, multihandlesink->timeout);
2103       break;
2104     case PROP_SYNC_METHOD:
2105       g_value_set_enum (value, multihandlesink->def_sync_method);
2106       break;
2107     case PROP_BYTES_TO_SERVE:
2108       g_value_set_uint64 (value, multihandlesink->bytes_to_serve);
2109       break;
2110     case PROP_BYTES_SERVED:
2111       g_value_set_uint64 (value, multihandlesink->bytes_served);
2112       break;
2113     case PROP_BURST_FORMAT:
2114       g_value_set_enum (value, multihandlesink->def_burst_format);
2115       break;
2116     case PROP_BURST_VALUE:
2117       g_value_set_uint64 (value, multihandlesink->def_burst_value);
2118       break;
2119     case PROP_QOS_DSCP:
2120       g_value_set_int (value, multihandlesink->qos_dscp);
2121       break;
2122     case PROP_RESEND_STREAMHEADER:
2123       g_value_set_boolean (value, multihandlesink->resend_streamheader);
2124       break;
2125     case PROP_NUM_HANDLES:
2126       g_value_set_uint (value,
2127           g_hash_table_size (multihandlesink->handle_hash));
2128       break;
2129     default:
2130       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2131       break;
2132   }
2133 }
2134
2135 /* create a socket for sending to remote machine */
2136 static gboolean
2137 gst_multi_handle_sink_start (GstBaseSink * bsink)
2138 {
2139   GstMultiHandleSinkClass *mhsclass;
2140   GstMultiHandleSink *mhsink;
2141
2142   if (GST_OBJECT_FLAG_IS_SET (bsink, GST_MULTI_HANDLE_SINK_OPEN))
2143     return TRUE;
2144
2145   mhsink = GST_MULTI_HANDLE_SINK (bsink);
2146   mhsclass = GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
2147
2148   if (!mhsclass->start_pre (mhsink))
2149     return FALSE;
2150
2151   mhsink->bytes_to_serve = 0;
2152   mhsink->bytes_served = 0;
2153
2154   if (mhsclass->init) {
2155     mhsclass->init (mhsink);
2156   }
2157
2158   mhsink->running = TRUE;
2159
2160   mhsink->thread = g_thread_new ("multihandlesink",
2161       (GThreadFunc) mhsclass->thread, mhsink);
2162
2163   GST_OBJECT_FLAG_SET (bsink, GST_MULTI_HANDLE_SINK_OPEN);
2164
2165   return TRUE;
2166 }
2167
2168 static gboolean
2169 gst_multi_handle_sink_stop (GstBaseSink * bsink)
2170 {
2171   GstMultiHandleSinkClass *mhclass;
2172   GstBuffer *buf;
2173   gint i;
2174   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (bsink);
2175
2176   mhclass = GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
2177
2178   if (!GST_OBJECT_FLAG_IS_SET (bsink, GST_MULTI_HANDLE_SINK_OPEN))
2179     return TRUE;
2180
2181   mhsink->running = FALSE;
2182
2183   mhclass->stop_pre (mhsink);
2184
2185   if (mhsink->thread) {
2186     GST_DEBUG_OBJECT (mhsink, "joining thread");
2187     g_thread_join (mhsink->thread);
2188     GST_DEBUG_OBJECT (mhsink, "joined thread");
2189     mhsink->thread = NULL;
2190   }
2191
2192   /* free the clients */
2193   mhclass->clear (GST_MULTI_HANDLE_SINK (mhsink));
2194
2195   if (mhclass->close)
2196     mhclass->close (mhsink);
2197
2198   mhclass->stop_post (mhsink);
2199
2200   /* remove all queued buffers */
2201   if (mhsink->bufqueue) {
2202     GST_DEBUG_OBJECT (mhsink, "Emptying bufqueue with %d buffers",
2203         mhsink->bufqueue->len);
2204     for (i = mhsink->bufqueue->len - 1; i >= 0; --i) {
2205       buf = g_array_index (mhsink->bufqueue, GstBuffer *, i);
2206       GST_LOG_OBJECT (mhsink, "Removing buffer %p (%d) with refcount %d", buf,
2207           i, GST_MINI_OBJECT_REFCOUNT (buf));
2208       gst_buffer_unref (buf);
2209       mhsink->bufqueue = g_array_remove_index (mhsink->bufqueue, i);
2210     }
2211     /* freeing the array is done in _finalize */
2212   }
2213   GST_OBJECT_FLAG_UNSET (mhsink, GST_MULTI_HANDLE_SINK_OPEN);
2214
2215   return TRUE;
2216 }
2217
2218 static GstStateChangeReturn
2219 gst_multi_handle_sink_change_state (GstElement * element,
2220     GstStateChange transition)
2221 {
2222   GstMultiHandleSink *sink;
2223   GstStateChangeReturn ret;
2224
2225   sink = GST_MULTI_HANDLE_SINK (element);
2226
2227   /* we disallow changing the state from the streaming thread */
2228   if (g_thread_self () == sink->thread) {
2229     g_warning
2230         ("\nTrying to change %s's state from its streaming thread would deadlock.\n"
2231         "You cannot change the state of an element from its streaming\n"
2232         "thread. Use g_idle_add() or post a GstMessage on the bus to\n"
2233         "schedule the state change from the main thread.\n",
2234         GST_ELEMENT_NAME (sink));
2235
2236     return GST_STATE_CHANGE_FAILURE;
2237   }
2238
2239   switch (transition) {
2240     case GST_STATE_CHANGE_NULL_TO_READY:
2241       if (!gst_multi_handle_sink_start (GST_BASE_SINK (sink)))
2242         goto start_failed;
2243       break;
2244     case GST_STATE_CHANGE_READY_TO_PAUSED:
2245       break;
2246     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
2247       break;
2248     default:
2249       break;
2250   }
2251
2252   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
2253
2254   switch (transition) {
2255     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
2256       break;
2257     case GST_STATE_CHANGE_PAUSED_TO_READY:
2258       break;
2259     case GST_STATE_CHANGE_READY_TO_NULL:
2260       gst_multi_handle_sink_stop (GST_BASE_SINK (sink));
2261       break;
2262     default:
2263       break;
2264   }
2265   return ret;
2266
2267   /* ERRORS */
2268 start_failed:
2269   {
2270     /* error message was posted */
2271     return GST_STATE_CHANGE_FAILURE;
2272   }
2273 }