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., 59 Temple Place - Suite 330,
  21  * Boston, MA 02111-1307, USA.
  22  */
  23
  24 /**
  25  * SECTION:element-multihandlesink
  26  * @see_also: tcpserversink
  27  *
  28  * This plugin writes incoming data to a set of file descriptors. The
  29  * file descriptors can be added to multihandlesink by emitting the #GstMultiHandleSink::add signal.
  30  * For each descriptor added, the #GstMultiHandleSink::client-added signal will be called.
  31  *
  32  * As of version 0.10.8, a client can also be added with the #GstMultiHandleSink::add-full signal
  33  * that allows for more control over what and how much data a client
  34  * initially receives.
  35  *
  36  * Clients can be removed from multihandlesink by emitting the #GstMultiHandleSink::remove signal. For
  37  * each descriptor removed, the #GstMultiHandleSink::client-removed signal will be called. The
  38  * #GstMultiHandleSink::client-removed signal can also be fired when multihandlesink decides that a
  39  * client is not active anymore or, depending on the value of the
  40  * #GstMultiHandleSink:recover-policy property, if the client is reading too slowly.
  41  * In all cases, multihandlesink will never close a file descriptor itself.
  42  * The user of multihandlesink is responsible for closing all file descriptors.
  43  * This can for example be done in response to the #GstMultiHandleSink::client-fd-removed signal.
  44  * Note that multihandlesink still has a reference to the file descriptor when the
  45  * #GstMultiHandleSink::client-removed signal is emitted, so that "get-stats" can be performed on
  46  * the descriptor; it is therefore not safe to close the file descriptor in
  47  * the #GstMultiHandleSink::client-removed signal handler, and you should use the
  48  * #GstMultiHandleSink::client-fd-removed signal to safely close the fd.
  49  *
  50  * Multisocketsink internally keeps a queue of the incoming buffers and uses a
  51  * separate thread to send the buffers to the clients. This ensures that no
  52  * client write can block the pipeline and that clients can read with different
  53  * speeds.
  54  *
  55  * When adding a client to multihandlesink, the #GstMultiHandleSink:sync-method property will define
  56  * which buffer in the queued buffers will be sent first to the client. Clients
  57  * can be sent the most recent buffer (which might not be decodable by the
  58  * client if it is not a keyframe), the next keyframe received in
  59  * multihandlesink (which can take some time depending on the keyframe rate), or the
  60  * last received keyframe (which will cause a simple burst-on-connect).
  61  * Multisocketsink will always keep at least one keyframe in its internal buffers
  62  * when the sync-mode is set to latest-keyframe.
  63  *
  64  * As of version 0.10.8, there are additional values for the #GstMultiHandleSink:sync-method
  65  * property to allow finer control over burst-on-connect behaviour. By selecting
  66  * the 'burst' method a minimum burst size can be chosen, 'burst-keyframe'
  67  * additionally requires that the burst begin with a keyframe, and
  68  * 'burst-with-keyframe' attempts to burst beginning with a keyframe, but will
  69  * prefer a minimum burst size even if it requires not starting with a keyframe.
  70  *
  71  * Multisocketsink can be instructed to keep at least a minimum amount of data
  72  * expressed in time or byte units in its internal queues with the
  73  * #GstMultiHandleSink:time-min and #GstMultiHandleSink:bytes-min properties respectively.
  74  * These properties are useful if the application adds clients with the
  75  * #GstMultiHandleSink::add-full signal to make sure that a burst connect can
  76  * actually be honored.
  77  *
  78  * When streaming data, clients are allowed to read at a different rate than
  79  * the rate at which multihandlesink receives data. If the client is reading too
  80  * fast, no data will be send to the client until multihandlesink receives more
  81  * data. If the client, however, reads too slowly, data for that client will be
  82  * queued up in multihandlesink. Two properties control the amount of data
  83  * (buffers) that is queued in multihandlesink: #GstMultiHandleSink:buffers-max and
  84  * #GstMultiHandleSink:buffers-soft-max. A client that falls behind by
  85  * #GstMultiHandleSink:buffers-max is removed from multihandlesink forcibly.
  86  *
  87  * A client with a lag of at least #GstMultiHandleSink:buffers-soft-max enters the recovery
  88  * procedure which is controlled with the #GstMultiHandleSink:recover-policy property.
  89  * A recover policy of NONE will do nothing, RESYNC_LATEST will send the most recently
  90  * received buffer as the next buffer for the client, RESYNC_SOFT_LIMIT
  91  * positions the client to the soft limit in the buffer queue and
  92  * RESYNC_KEYFRAME positions the client at the most recent keyframe in the
  93  * buffer queue.
  94  *
  95  * multihandlesink will by default synchronize on the clock before serving the
  96  * buffers to the clients. This behaviour can be disabled by setting the sync
  97  * property to FALSE. Multisocketsink will by default not do QoS and will never
  98  * drop late buffers.
  99  *
 100  * Last reviewed on 2006-09-12 (0.10.10)
 101  */
 102
 103 #ifdef HAVE_CONFIG_H
 104 #include "config.h"
 105 #endif
 106
 107 #include <gst/gst-i18n-plugin.h>
 108
 109 #include "gstmultihandlesink.h"
 110 #include "gsttcp-marshal.h"
 111
 112 #ifndef G_OS_WIN32
 113 #include <netinet/in.h>
 114 #endif
 115
 116 #define NOT_IMPLEMENTED 0
 117
 118 static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",
 119     GST_PAD_SINK,
 120     GST_PAD_ALWAYS,
 121     GST_STATIC_CAPS_ANY);
 122
 123 GST_DEBUG_CATEGORY_STATIC (multihandlesink_debug);
 124 #define GST_CAT_DEFAULT (multihandlesink_debug)
 125
 126 /* MultiHandleSink signals and args */
 127 enum
 128 {
 129   /* methods */
 130   SIGNAL_ADD,
 131   SIGNAL_ADD_BURST,
 132   SIGNAL_REMOVE,
 133   SIGNAL_REMOVE_FLUSH,
 134   SIGNAL_CLEAR,
 135   SIGNAL_GET_STATS,
 136
 137   /* signals */
 138   SIGNAL_CLIENT_ADDED,
 139   SIGNAL_CLIENT_REMOVED,
 140   SIGNAL_CLIENT_SOCKET_REMOVED,
 141
 142   LAST_SIGNAL
 143 };
 144
 145
 146 /* this is really arbitrarily chosen */
 147 #define DEFAULT_BUFFERS_MAX             -1
 148 #define DEFAULT_BUFFERS_SOFT_MAX        -1
 149 #define DEFAULT_TIME_MIN                -1
 150 #define DEFAULT_BYTES_MIN               -1
 151 #define DEFAULT_BUFFERS_MIN             -1
 152 #define DEFAULT_UNIT_FORMAT               GST_FORMAT_BUFFERS
 153 #define DEFAULT_UNITS_MAX               -1
 154 #define DEFAULT_UNITS_SOFT_MAX          -1
 155 #define DEFAULT_RECOVER_POLICY          GST_RECOVER_POLICY_NONE
 156 #define DEFAULT_TIMEOUT                 0
 157 #define DEFAULT_SYNC_METHOD             GST_SYNC_METHOD_LATEST
 158
 159 #define DEFAULT_BURST_FORMAT            GST_FORMAT_UNDEFINED
 160 #define DEFAULT_BURST_VALUE             0
 161
 162 #define DEFAULT_QOS_DSCP                -1
 163
 164 #define DEFAULT_RESEND_STREAMHEADER      TRUE
 165
 166 enum
 167 {
 168   PROP_0,
 169   PROP_BUFFERS_QUEUED,
 170   PROP_BYTES_QUEUED,
 171   PROP_TIME_QUEUED,
 172
 173   PROP_UNIT_FORMAT,
 174   PROP_UNITS_MAX,
 175   PROP_UNITS_SOFT_MAX,
 176
 177   PROP_BUFFERS_MAX,
 178   PROP_BUFFERS_SOFT_MAX,
 179
 180   PROP_TIME_MIN,
 181   PROP_BYTES_MIN,
 182   PROP_BUFFERS_MIN,
 183
 184   PROP_RECOVER_POLICY,
 185   PROP_TIMEOUT,
 186   PROP_SYNC_METHOD,
 187   PROP_BYTES_TO_SERVE,
 188   PROP_BYTES_SERVED,
 189
 190   PROP_BURST_FORMAT,
 191   PROP_BURST_VALUE,
 192
 193   PROP_QOS_DSCP,
 194
 195   PROP_RESEND_STREAMHEADER,
 196
 197 #if 0
 198   PROP_NUM_SOCKETS,
 199 #endif
 200
 201   PROP_LAST
 202 };
 203
 204 // FIXME: make static again when refactored
 205 //#define GST_TYPE_RECOVER_POLICY (gst_multi_handle_sink_recover_policy_get_type())
 206 //static GType
 207 GType
 208 gst_multi_handle_sink_recover_policy_get_type (void)
 209 {
 210   static GType recover_policy_type = 0;
 211   static const GEnumValue recover_policy[] = {
 212     {GST_RECOVER_POLICY_NONE,
 213         "Do not try to recover", "none"},
 214     {GST_RECOVER_POLICY_RESYNC_LATEST,
 215         "Resync client to latest buffer", "latest"},
 216     {GST_RECOVER_POLICY_RESYNC_SOFT_LIMIT,
 217         "Resync client to soft limit", "soft-limit"},
 218     {GST_RECOVER_POLICY_RESYNC_KEYFRAME,
 219         "Resync client to most recent keyframe", "keyframe"},
 220     {0, NULL, NULL},
 221   };
 222
 223   if (!recover_policy_type) {
 224     recover_policy_type =
 225         g_enum_register_static ("GstMultiHandleSinkRecoverPolicy",
 226         recover_policy);
 227   }
 228   return recover_policy_type;
 229 }
 230
 231 // FIXME: make static again after refactoring
 232 //#define GST_TYPE_SYNC_METHOD (gst_multi_handle_sink_sync_method_get_type())
 233 //static GType
 234 GType
 235 gst_multi_handle_sink_sync_method_get_type (void)
 236 {
 237   static GType sync_method_type = 0;
 238   static const GEnumValue sync_method[] = {
 239     {GST_SYNC_METHOD_LATEST,
 240         "Serve starting from the latest buffer", "latest"},
 241     {GST_SYNC_METHOD_NEXT_KEYFRAME,
 242         "Serve starting from the next keyframe", "next-keyframe"},
 243     {GST_SYNC_METHOD_LATEST_KEYFRAME,
 244           "Serve everything since the latest keyframe (burst)",
 245         "latest-keyframe"},
 246     {GST_SYNC_METHOD_BURST, "Serve burst-value data to client", "burst"},
 247     {GST_SYNC_METHOD_BURST_KEYFRAME,
 248           "Serve burst-value data starting on a keyframe",
 249         "burst-keyframe"},
 250     {GST_SYNC_METHOD_BURST_WITH_KEYFRAME,
 251           "Serve burst-value data preferably starting on a keyframe",
 252         "burst-with-keyframe"},
 253     {0, NULL, NULL},
 254   };
 255
 256   if (!sync_method_type) {
 257     sync_method_type =
 258         g_enum_register_static ("GstMultiHandleSinkSyncMethod", sync_method);
 259   }
 260   return sync_method_type;
 261 }
 262
 263 // FIXME: make static again after refactoring
 264 //#define GST_TYPE_CLIENT_STATUS (gst_multi_handle_sink_client_status_get_type())
 265 //static GType
 266 GType
 267 gst_multi_handle_sink_client_status_get_type (void)
 268 {
 269   static GType client_status_type = 0;
 270   static const GEnumValue client_status[] = {
 271     {GST_CLIENT_STATUS_OK, "ok", "ok"},
 272     {GST_CLIENT_STATUS_CLOSED, "Closed", "closed"},
 273     {GST_CLIENT_STATUS_REMOVED, "Removed", "removed"},
 274     {GST_CLIENT_STATUS_SLOW, "Too slow", "slow"},
 275     {GST_CLIENT_STATUS_ERROR, "Error", "error"},
 276     {GST_CLIENT_STATUS_DUPLICATE, "Duplicate", "duplicate"},
 277     {GST_CLIENT_STATUS_FLUSHING, "Flushing", "flushing"},
 278     {0, NULL, NULL},
 279   };
 280
 281   if (!client_status_type) {
 282     client_status_type =
 283         g_enum_register_static ("GstMultiHandleSinkClientStatus",
 284         client_status);
 285   }
 286   return client_status_type;
 287 }
 288
 289 static void gst_multi_handle_sink_finalize (GObject * object);
 290 static void gst_multi_handle_sink_clear (GstMultiHandleSink * mhsink);
 291
 292 #if 0
 293 static void gst_multi_handle_sink_remove_client_link (GstMultiHandleSink * sink,
 294     GList * link);
 295 static gboolean gst_multi_handle_sink_socket_condition (GSocket * socket,
 296     GIOCondition condition, GstMultiHandleSink * sink);
 297 #endif
 298
 299 static GstFlowReturn gst_multi_handle_sink_render (GstBaseSink * bsink,
 300     GstBuffer * buf);
 301 #if 0
 302 static gboolean gst_multi_handle_sink_unlock (GstBaseSink * bsink);
 303 static gboolean gst_multi_handle_sink_unlock_stop (GstBaseSink * bsink);
 304 #endif
 305 static GstStateChangeReturn gst_multi_handle_sink_change_state (GstElement *
 306     element, GstStateChange transition);
 307
 308 static void gst_multi_handle_sink_set_property (GObject * object, guint prop_id,
 309     const GValue * value, GParamSpec * pspec);
 310 static void gst_multi_handle_sink_get_property (GObject * object, guint prop_id,
 311     GValue * value, GParamSpec * pspec);
 312
 313 #define gst_multi_handle_sink_parent_class parent_class
 314 G_DEFINE_TYPE (GstMultiHandleSink, gst_multi_handle_sink, GST_TYPE_BASE_SINK);
 315
 316 static guint gst_multi_handle_sink_signals[LAST_SIGNAL] = { 0 };
 317
 318 static void
 319 gst_multi_handle_sink_class_init (GstMultiHandleSinkClass * klass)
 320 {
 321   GObjectClass *gobject_class;
 322   GstElementClass *gstelement_class;
 323   GstBaseSinkClass *gstbasesink_class;
 324
 325   gobject_class = (GObjectClass *) klass;
 326   gstelement_class = (GstElementClass *) klass;
 327   gstbasesink_class = (GstBaseSinkClass *) klass;
 328
 329   gobject_class->set_property = gst_multi_handle_sink_set_property;
 330   gobject_class->get_property = gst_multi_handle_sink_get_property;
 331   gobject_class->finalize = gst_multi_handle_sink_finalize;
 332
 333 #if 0
 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 #endif
 340 #if 0
 341   g_object_class_install_property (gobject_class, PROP_BUFFERS_SOFT_MAX,
 342       g_param_spec_int ("buffers-soft-max", "Buffers soft max",
 343           "Recover client when going over this limit (-1 = no limit)", -1,
 344           G_MAXINT, DEFAULT_BUFFERS_SOFT_MAX,
 345           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 346 #endif
 347
 348   g_object_class_install_property (gobject_class, PROP_BYTES_MIN,
 349       g_param_spec_int ("bytes-min", "Bytes min",
 350           "min number of bytes to queue (-1 = as little as possible)", -1,
 351           G_MAXINT, DEFAULT_BYTES_MIN,
 352           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 353   g_object_class_install_property (gobject_class, PROP_TIME_MIN,
 354       g_param_spec_int64 ("time-min", "Time min",
 355           "min number of time to queue (-1 = as little as possible)", -1,
 356           G_MAXINT64, DEFAULT_TIME_MIN,
 357           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 358   g_object_class_install_property (gobject_class, PROP_BUFFERS_MIN,
 359       g_param_spec_int ("buffers-min", "Buffers min",
 360           "min number of buffers to queue (-1 = as few as possible)", -1,
 361           G_MAXINT, DEFAULT_BUFFERS_MIN,
 362           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 363
 364 #if 0
 365   g_object_class_install_property (gobject_class, PROP_UNIT_FORMAT,
 366       g_param_spec_enum ("unit-format", "Units format",
 367           "The unit to measure the max/soft-max/queued properties",
 368           GST_TYPE_FORMAT, DEFAULT_UNIT_FORMAT,
 369           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 370   g_object_class_install_property (gobject_class, PROP_UNITS_MAX,
 371       g_param_spec_int64 ("units-max", "Units max",
 372           "max number of units to queue (-1 = no limit)", -1, G_MAXINT64,
 373           DEFAULT_UNITS_MAX, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 374   g_object_class_install_property (gobject_class, PROP_UNITS_SOFT_MAX,
 375       g_param_spec_int64 ("units-soft-max", "Units soft max",
 376           "Recover client when going over this limit (-1 = no limit)", -1,
 377           G_MAXINT64, DEFAULT_UNITS_SOFT_MAX,
 378           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 379 #endif
 380
 381   g_object_class_install_property (gobject_class, PROP_BUFFERS_QUEUED,
 382       g_param_spec_uint ("buffers-queued", "Buffers queued",
 383           "Number of buffers currently queued", 0, G_MAXUINT, 0,
 384           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 385 #if NOT_IMPLEMENTED
 386   g_object_class_install_property (gobject_class, PROP_BYTES_QUEUED,
 387       g_param_spec_uint ("bytes-queued", "Bytes queued",
 388           "Number of bytes currently queued", 0, G_MAXUINT, 0,
 389           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 390   g_object_class_install_property (gobject_class, PROP_TIME_QUEUED,
 391       g_param_spec_uint64 ("time-queued", "Time queued",
 392           "Number of time currently queued", 0, G_MAXUINT64, 0,
 393           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 394 #endif
 395
 396   g_object_class_install_property (gobject_class, PROP_RECOVER_POLICY,
 397       g_param_spec_enum ("recover-policy", "Recover Policy",
 398           "How to recover when client reaches the soft max",
 399           GST_TYPE_RECOVER_POLICY, DEFAULT_RECOVER_POLICY,
 400           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 401   g_object_class_install_property (gobject_class, PROP_TIMEOUT,
 402       g_param_spec_uint64 ("timeout", "Timeout",
 403           "Maximum inactivity timeout in nanoseconds for a client (0 = no limit)",
 404           0, G_MAXUINT64, DEFAULT_TIMEOUT,
 405           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 406   g_object_class_install_property (gobject_class, PROP_SYNC_METHOD,
 407       g_param_spec_enum ("sync-method", "Sync Method",
 408           "How to sync new clients to the stream", GST_TYPE_SYNC_METHOD,
 409           DEFAULT_SYNC_METHOD, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 410   g_object_class_install_property (gobject_class, PROP_BYTES_TO_SERVE,
 411       g_param_spec_uint64 ("bytes-to-serve", "Bytes to serve",
 412           "Number of bytes received to serve to clients", 0, G_MAXUINT64, 0,
 413           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 414   g_object_class_install_property (gobject_class, PROP_BYTES_SERVED,
 415       g_param_spec_uint64 ("bytes-served", "Bytes served",
 416           "Total number of bytes send to all clients", 0, G_MAXUINT64, 0,
 417           G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 418
 419   g_object_class_install_property (gobject_class, PROP_BURST_FORMAT,
 420       g_param_spec_enum ("burst-format", "Burst format",
 421           "The format of the burst units (when sync-method is burst[[-with]-keyframe])",
 422           GST_TYPE_FORMAT, DEFAULT_BURST_FORMAT,
 423           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 424   g_object_class_install_property (gobject_class, PROP_BURST_VALUE,
 425       g_param_spec_uint64 ("burst-value", "Burst value",
 426           "The amount of burst expressed in burst-unit", 0, G_MAXUINT64,
 427           DEFAULT_BURST_VALUE, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 428
 429   g_object_class_install_property (gobject_class, PROP_QOS_DSCP,
 430       g_param_spec_int ("qos-dscp", "QoS diff srv code point",
 431           "Quality of Service, differentiated services code point (-1 default)",
 432           -1, 63, DEFAULT_QOS_DSCP,
 433           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 434
 435 #if 0
 436   /**
 437    * GstMultiHandleSink::handle-read
 438    *
 439    * Handle read requests from clients and discard the data.
 440    *
 441    * Since: 0.10.23
 442    */
 443   g_object_class_install_property (gobject_class, PROP_HANDLE_READ,
 444       g_param_spec_boolean ("handle-read", "Handle Read",
 445           "Handle client reads and discard the data",
 446           DEFAULT_HANDLE_READ, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 447 #endif
 448   /**
 449    * GstMultiHandleSink::resend-streamheader
 450    *
 451    * Resend the streamheaders to existing clients when they change.
 452    *
 453    * Since: 0.10.23
 454    */
 455   g_object_class_install_property (gobject_class, PROP_RESEND_STREAMHEADER,
 456       g_param_spec_boolean ("resend-streamheader", "Resend streamheader",
 457           "Resend the streamheader if it changes in the caps",
 458           DEFAULT_RESEND_STREAMHEADER,
 459           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 460
 461 #if 0
 462   g_object_class_install_property (gobject_class, PROP_NUM_SOCKETS,
 463       g_param_spec_uint ("num-sockets", "Number of sockets",
 464           "The current number of client sockets",
 465           0, G_MAXUINT, 0, G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 466
 467   /**
 468    * GstMultiHandleSink::add:
 469    * @gstmultihandlesink: the multihandlesink element to emit this signal on
 470    * @socket:             the socket to add to multihandlesink
 471    *
 472    * Hand the given open socket to multihandlesink to write to.
 473    */
 474   gst_multi_handle_sink_signals[SIGNAL_ADD] =
 475       g_signal_new ("add", G_TYPE_FROM_CLASS (klass),
 476       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 477       G_STRUCT_OFFSET (GstMultiHandleSinkClass, add), NULL, NULL,
 478       g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1, G_TYPE_SOCKET);
 479   /**
 480    * GstMultiHandleSink::add-full:
 481    * @gstmultihandlesink: the multihandlesink element to emit this signal on
 482    * @socket:         the socket to add to multihandlesink
 483    * @sync:           the sync method to use
 484    * @format_min:     the format of @value_min
 485    * @value_min:      the minimum amount of data to burst expressed in
 486    *                  @format_min units.
 487    * @format_max:     the format of @value_max
 488    * @value_max:      the maximum amount of data to burst expressed in
 489    *                  @format_max units.
 490    *
 491    * Hand the given open socket to multihandlesink to write to and
 492    * specify the burst parameters for the new connection.
 493    */
 494   gst_multi_handle_sink_signals[SIGNAL_ADD_BURST] =
 495       g_signal_new ("add-full", G_TYPE_FROM_CLASS (klass),
 496       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 497       G_STRUCT_OFFSET (GstMultiHandleSinkClass, add_full), NULL, NULL,
 498       gst_tcp_marshal_VOID__OBJECT_ENUM_ENUM_UINT64_ENUM_UINT64, G_TYPE_NONE, 6,
 499       G_TYPE_SOCKET, GST_TYPE_SYNC_METHOD, GST_TYPE_FORMAT, G_TYPE_UINT64,
 500       GST_TYPE_FORMAT, G_TYPE_UINT64);
 501   /**
 502    * GstMultiHandleSink::remove:
 503    * @gstmultihandlesink: the multihandlesink element to emit this signal on
 504    * @socket:             the socket to remove from multihandlesink
 505    *
 506    * Remove the given open socket from multihandlesink.
 507    */
 508   gst_multi_handle_sink_signals[SIGNAL_REMOVE] =
 509       g_signal_new ("remove", G_TYPE_FROM_CLASS (klass),
 510       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 511       G_STRUCT_OFFSET (GstMultiHandleSinkClass, remove), NULL, NULL,
 512       g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1, G_TYPE_SOCKET);
 513   /**
 514    * GstMultiHandleSink::remove-flush:
 515    * @gstmultihandlesink: the multihandlesink element to emit this signal on
 516    * @socket:             the socket to remove from multihandlesink
 517    *
 518    * Remove the given open socket from multihandlesink after flushing all
 519    * the pending data to the socket.
 520    */
 521   gst_multi_handle_sink_signals[SIGNAL_REMOVE_FLUSH] =
 522       g_signal_new ("remove-flush", G_TYPE_FROM_CLASS (klass),
 523       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 524       G_STRUCT_OFFSET (GstMultiHandleSinkClass, remove_flush), NULL, NULL,
 525       g_cclosure_marshal_VOID__OBJECT, G_TYPE_NONE, 1, G_TYPE_SOCKET);
 526 #endif
 527   /**
 528    * GstMultiHandleSink::clear:
 529    * @gstmultihandlesink: the multihandlesink element to emit this signal on
 530    *
 531    * Remove all sockets from multihandlesink.  Since multihandlesink did not
 532    * open sockets itself, it does not explicitly close the sockets. The application
 533    * should do so by connecting to the client-socket-removed callback.
 534    */
 535   gst_multi_handle_sink_signals[SIGNAL_CLEAR] =
 536       g_signal_new ("clear", G_TYPE_FROM_CLASS (klass),
 537       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 538       G_STRUCT_OFFSET (GstMultiHandleSinkClass, clear), NULL, NULL,
 539       g_cclosure_marshal_VOID__VOID, G_TYPE_NONE, 0);
 540
 541 #if 0
 542   /**
 543    * GstMultiHandleSink::get-stats:
 544    * @gstmultihandlesink: the multihandlesink element to emit this signal on
 545    * @socket:             the socket to get stats of from multihandlesink
 546    *
 547    * Get statistics about @socket. This function returns a GstStructure.
 548    *
 549    * Returns: a GstStructure with the statistics. The structure contains
 550    *     values that represent: total number of bytes sent, time
 551    *     when the client was added, time when the client was
 552    *     disconnected/removed, time the client is/was active, last activity
 553    *     time (in epoch seconds), number of buffers dropped.
 554    *     All times are expressed in nanoseconds (GstClockTime).
 555    */
 556   gst_multi_handle_sink_signals[SIGNAL_GET_STATS] =
 557       g_signal_new ("get-stats", G_TYPE_FROM_CLASS (klass),
 558       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 559       G_STRUCT_OFFSET (GstMultiHandleSinkClass, get_stats), NULL, NULL,
 560       gst_tcp_marshal_BOXED__OBJECT, GST_TYPE_STRUCTURE, 1, G_TYPE_SOCKET);
 561
 562   /**
 563    * GstMultiHandleSink::client-added:
 564    * @gstmultihandlesink: the multihandlesink element that emitted this signal
 565    * @socket:             the socket that was added to multihandlesink
 566    *
 567    * The given socket was added to multihandlesink. This signal will
 568    * be emitted from the streaming thread so application should be prepared
 569    * for that.
 570    */
 571   gst_multi_handle_sink_signals[SIGNAL_CLIENT_ADDED] =
 572       g_signal_new ("client-added", G_TYPE_FROM_CLASS (klass),
 573       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiHandleSinkClass,
 574           client_added), NULL, NULL, g_cclosure_marshal_VOID__OBJECT,
 575       G_TYPE_NONE, 1, G_TYPE_OBJECT);
 576   /**
 577    * GstMultiHandleSink::client-removed:
 578    * @gstmultihandlesink: the multihandlesink element that emitted this signal
 579    * @socket:             the socket that is to be removed from multihandlesink
 580    * @status:             the reason why the client was removed
 581    *
 582    * The given socket is about to be removed from multihandlesink. This
 583    * signal will be emitted from the streaming thread so applications should
 584    * be prepared for that.
 585    *
 586    * @gstmultihandlesink still holds a handle to @socket so it is possible to call
 587    * the get-stats signal from this callback. For the same reason it is
 588    * not safe to close() and reuse @socket in this callback.
 589    */
 590   gst_multi_handle_sink_signals[SIGNAL_CLIENT_REMOVED] =
 591       g_signal_new ("client-removed", G_TYPE_FROM_CLASS (klass),
 592       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiHandleSinkClass,
 593           client_removed), NULL, NULL, gst_tcp_marshal_VOID__OBJECT_ENUM,
 594       G_TYPE_NONE, 2, G_TYPE_INT, GST_TYPE_CLIENT_STATUS);
 595   /**
 596    * GstMultiHandleSink::client-socket-removed:
 597    * @gstmultihandlesink: the multihandlesink element that emitted this signal
 598    * @socket:             the socket that was removed from multihandlesink
 599    *
 600    * The given socket was removed from multihandlesink. This signal will
 601    * be emitted from the streaming thread so applications should be prepared
 602    * for that.
 603    *
 604    * In this callback, @gstmultihandlesink has removed all the information
 605    * associated with @socket and it is therefore not possible to call get-stats
 606    * with @socket. It is however safe to close() and reuse @fd in the callback.
 607    *
 608    * Since: 0.10.7
 609    */
 610   gst_multi_handle_sink_signals[SIGNAL_CLIENT_SOCKET_REMOVED] =
 611       g_signal_new ("client-socket-removed", G_TYPE_FROM_CLASS (klass),
 612       G_SIGNAL_RUN_LAST, G_STRUCT_OFFSET (GstMultiHandleSinkClass,
 613           client_socket_removed), NULL, NULL, g_cclosure_marshal_VOID__OBJECT,
 614       G_TYPE_NONE, 1, G_TYPE_SOCKET);
 615 #endif
 616
 617   gst_element_class_add_pad_template (gstelement_class,
 618       gst_static_pad_template_get (&sinktemplate));
 619
 620   gst_element_class_set_details_simple (gstelement_class,
 621       "Multi socket sink", "Sink/Network",
 622       "Send data to multiple sockets",
 623       "Thomas Vander Stichele <thomas at apestaart dot org>, "
 624       "Wim Taymans <wim@fluendo.com>, "
 625       "Sebastian Dröge <sebastian.droege@collabora.co.uk>");
 626
 627   gstelement_class->change_state =
 628       GST_DEBUG_FUNCPTR (gst_multi_handle_sink_change_state);
 629
 630   gstbasesink_class->render = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_render);
 631 #if 0
 632   gstbasesink_class->unlock = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_unlock);
 633   gstbasesink_class->unlock_stop =
 634       GST_DEBUG_FUNCPTR (gst_multi_handle_sink_unlock_stop);
 635 #endif
 636
 637 #if 0
 638   klass->add = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_add);
 639   klass->add_full = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_add_full);
 640   klass->remove = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_remove);
 641   klass->remove_flush = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_remove_flush);
 642 #endif
 643   klass->clear = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_clear);
 644 #if 0
 645   klass->get_stats = GST_DEBUG_FUNCPTR (gst_multi_handle_sink_get_stats);
 646 #endif
 647
 648   GST_DEBUG_CATEGORY_INIT (multihandlesink_debug, "multihandlesink", 0,
 649       "Multi socket sink");
 650 }
 651
 652 static void
 653 gst_multi_handle_sink_init (GstMultiHandleSink * this)
 654 {
 655   GST_OBJECT_FLAG_UNSET (this, GST_MULTI_HANDLE_SINK_OPEN);
 656
 657   CLIENTS_LOCK_INIT (this);
 658   this->clients = NULL;
 659 #if 0
 660   this->socket_hash = g_hash_table_new (g_direct_hash, g_int_equal);
 661 #endif
 662
 663   this->bufqueue = g_array_new (FALSE, TRUE, sizeof (GstBuffer *));
 664   this->unit_format = DEFAULT_UNIT_FORMAT;
 665   this->units_max = DEFAULT_UNITS_MAX;
 666   this->units_soft_max = DEFAULT_UNITS_SOFT_MAX;
 667   this->time_min = DEFAULT_TIME_MIN;
 668   this->bytes_min = DEFAULT_BYTES_MIN;
 669   this->buffers_min = DEFAULT_BUFFERS_MIN;
 670   this->recover_policy = DEFAULT_RECOVER_POLICY;
 671
 672   this->timeout = DEFAULT_TIMEOUT;
 673   this->def_sync_method = DEFAULT_SYNC_METHOD;
 674
 675   this->def_burst_format = DEFAULT_BURST_FORMAT;
 676   this->def_burst_value = DEFAULT_BURST_VALUE;
 677
 678   this->qos_dscp = DEFAULT_QOS_DSCP;
 679
 680   this->resend_streamheader = DEFAULT_RESEND_STREAMHEADER;
 681 }
 682
 683 static void
 684 gst_multi_handle_sink_finalize (GObject * object)
 685 {
 686   GstMultiHandleSink *this;
 687
 688   this = GST_MULTI_HANDLE_SINK (object);
 689
 690   CLIENTS_LOCK_CLEAR (this);
 691   g_array_free (this->bufqueue, TRUE);
 692
 693   G_OBJECT_CLASS (parent_class)->finalize (object);
 694 }
 695
 696 gint
 697 gst_multi_handle_sink_setup_dscp_client (GstMultiHandleSink * sink,
 698     GstMultiHandleClient * client)
 699 {
 700 #ifndef IP_TOS
 701   return 0;
 702 #else
 703   gint tos;
 704   gint ret;
 705   int fd;
 706   union gst_sockaddr
 707   {
 708     struct sockaddr sa;
 709     struct sockaddr_in6 sa_in6;
 710     struct sockaddr_storage sa_stor;
 711   } sa;
 712   socklen_t slen = sizeof (sa);
 713   gint af;
 714   GstMultiHandleSinkClass *mhsinkclass = GST_MULTI_HANDLE_SINK_GET_CLASS (sink);
 715
 716   /* don't touch */
 717   if (sink->qos_dscp < 0)
 718     return 0;
 719
 720   fd = mhsinkclass->client_get_fd (client);
 721
 722   if ((ret = getsockname (fd, &sa.sa, &slen)) < 0) {
 723     GST_DEBUG_OBJECT (sink, "could not get sockname: %s", g_strerror (errno));
 724     return ret;
 725   }
 726
 727   af = sa.sa.sa_family;
 728
 729   /* if this is an IPv4-mapped address then do IPv4 QoS */
 730   if (af == AF_INET6) {
 731
 732     GST_DEBUG_OBJECT (sink, "check IP6 socket");
 733     if (IN6_IS_ADDR_V4MAPPED (&(sa.sa_in6.sin6_addr))) {
 734       GST_DEBUG_OBJECT (sink, "mapped to IPV4");
 735       af = AF_INET;
 736     }
 737   }
 738
 739   /* extract and shift 6 bits of the DSCP */
 740   tos = (sink->qos_dscp & 0x3f) << 2;
 741
 742   switch (af) {
 743     case AF_INET:
 744       ret = setsockopt (fd, IPPROTO_IP, IP_TOS, &tos, sizeof (tos));
 745       break;
 746     case AF_INET6:
 747 #ifdef IPV6_TCLASS
 748       ret = setsockopt (fd, IPPROTO_IPV6, IPV6_TCLASS, &tos, sizeof (tos));
 749       break;
 750 #endif
 751     default:
 752       ret = 0;
 753       GST_ERROR_OBJECT (sink, "unsupported AF");
 754       break;
 755   }
 756   if (ret)
 757     GST_DEBUG_OBJECT (sink, "could not set DSCP: %s", g_strerror (errno));
 758
 759   return ret;
 760 #endif
 761 }
 762
 763 void
 764 gst_multi_handle_sink_client_init (GstMultiHandleClient * client,
 765     GstSyncMethod sync_method)
 766 {
 767   GTimeVal now;
 768
 769   client->status = GST_CLIENT_STATUS_OK;
 770   client->bufpos = -1;
 771   client->flushcount = -1;
 772   client->bufoffset = 0;
 773   client->sending = NULL;
 774   client->bytes_sent = 0;
 775   client->dropped_buffers = 0;
 776   client->avg_queue_size = 0;
 777   client->first_buffer_ts = GST_CLOCK_TIME_NONE;
 778   client->last_buffer_ts = GST_CLOCK_TIME_NONE;
 779   client->new_connection = TRUE;
 780   client->sync_method = sync_method;
 781   client->currently_removing = FALSE;
 782
 783   /* update start time */
 784   g_get_current_time (&now);
 785   client->connect_time = GST_TIMEVAL_TO_TIME (now);
 786   client->disconnect_time = 0;
 787   /* set last activity time to connect time */
 788   client->last_activity_time = client->connect_time;
 789 }
 790
 791 void
 792 gst_multi_handle_sink_setup_dscp (GstMultiHandleSink * mhsink)
 793 {
 794   GList *clients;
 795
 796   CLIENTS_LOCK (mhsink);
 797   for (clients = mhsink->clients; clients; clients = clients->next) {
 798     GstMultiHandleClient *client;
 799
 800     client = clients->data;
 801
 802     gst_multi_handle_sink_setup_dscp_client (mhsink, client);
 803   }
 804   CLIENTS_UNLOCK (mhsink);
 805 }
 806
 807
 808 #if 0
 809 /* "add-full" signal implementation */
 810 void
 811 gst_multi_handle_sink_add_full (GstMultiHandleSink * sink, GSocket * socket,
 812     GstSyncMethod sync_method, GstFormat min_format, guint64 min_value,
 813     GstFormat max_format, guint64 max_value)
 814 {
 815   GstSocketClient *client;
 816   GList *clink;
 817   GTimeVal now;
 818
 819   GST_DEBUG_OBJECT (sink, "[socket %p] adding client, sync_method %d, "
 820       "min_format %d, min_value %" G_GUINT64_FORMAT
 821       ", max_format %d, max_value %" G_GUINT64_FORMAT, socket,
 822       sync_method, min_format, min_value, max_format, max_value);
 823
 824   /* do limits check if we can */
 825   if (min_format == max_format) {
 826     if (max_value != -1 && min_value != -1 && max_value < min_value)
 827       goto wrong_limits;
 828   }
 829
 830   /* create client datastructure */
 831   client = g_new0 (GstSocketClient, 1);
 832   client->socket = G_SOCKET (g_object_ref (socket));
 833   client->burst_min_format = min_format;
 834   client->burst_min_value = min_value;
 835   client->burst_max_format = max_format;
 836   client->burst_max_value = max_value;
 837
 838   CLIENTS_LOCK (sink);
 839
 840   /* check the hash to find a duplicate fd */
 841   clink = g_hash_table_lookup (sink->socket_hash, socket);
 842   if (clink != NULL)
 843     goto duplicate;
 844
 845   /* we can add the fd now */
 846   clink = sink->clients = g_list_prepend (sink->clients, client);
 847   g_hash_table_insert (sink->socket_hash, socket, clink);
 848   sink->clients_cookie++;
 849
 850   /* set the socket to non blocking */
 851   g_socket_set_blocking (socket, FALSE);
 852
 853   /* we always read from a client */
 854   if (sink->main_context) {
 855     client->source =
 856         g_socket_create_source (client->socket,
 857         G_IO_IN | G_IO_OUT | G_IO_PRI | G_IO_ERR | G_IO_HUP, sink->cancellable);
 858     g_source_set_callback (client->source,
 859         (GSourceFunc) gst_multi_handle_sink_socket_condition,
 860         gst_object_ref (sink), (GDestroyNotify) gst_object_unref);
 861     g_source_attach (client->source, sink->main_context);
 862   }
 863
 864   setup_dscp_client (sink, client);
 865
 866   CLIENTS_UNLOCK (sink);
 867
 868   g_signal_emit (G_OBJECT (sink),
 869       gst_multi_handle_sink_signals[SIGNAL_CLIENT_ADDED], 0, socket);
 870
 871   return;
 872
 873   /* errors */
 874 wrong_limits:
 875   {
 876     GST_WARNING_OBJECT (sink,
 877         "[socket %p] wrong values min =%" G_GUINT64_FORMAT ", max=%"
 878         G_GUINT64_FORMAT ", format %d specified when adding client", socket,
 879         min_value, max_value, min_format);
 880     return;
 881   }
 882 duplicate:
 883   {
 884     client->status = GST_CLIENT_STATUS_DUPLICATE;
 885     CLIENTS_UNLOCK (sink);
 886     GST_WARNING_OBJECT (sink, "[socket %p] duplicate client found, refusing",
 887         socket);
 888     g_signal_emit (G_OBJECT (sink),
 889         gst_multi_handle_sink_signals[SIGNAL_CLIENT_REMOVED], 0, socket,
 890         client->status);
 891     g_free (client);
 892     return;
 893   }
 894 }
 895
 896 /* "add" signal implementation */
 897 void
 898 gst_multi_handle_sink_add (GstMultiHandleSink * sink, GSocket * socket)
 899 {
 900   gst_multi_handle_sink_add_full (sink, socket, sink->def_sync_method,
 901       sink->def_burst_format, sink->def_burst_value, sink->def_burst_format,
 902       -1);
 903 }
 904
 905 /* "remove" signal implementation */
 906 void
 907 gst_multi_handle_sink_remove (GstMultiHandleSink * sink, GSocket * socket)
 908 {
 909   GList *clink;
 910
 911   GST_DEBUG_OBJECT (sink, "[socket %p] removing client", socket);
 912
 913   CLIENTS_LOCK (sink);
 914   clink = g_hash_table_lookup (sink->socket_hash, socket);
 915   if (clink != NULL) {
 916     GstSocketClient *client = clink->data;
 917
 918     if (client->status != GST_CLIENT_STATUS_OK) {
 919       GST_INFO_OBJECT (sink,
 920           "[socket %p] Client already disconnecting with status %d",
 921           socket, client->status);
 922       goto done;
 923     }
 924
 925     client->status = GST_CLIENT_STATUS_REMOVED;
 926     gst_multi_handle_sink_remove_client_link (sink, clink);
 927   } else {
 928     GST_WARNING_OBJECT (sink, "[socket %p] no client with this socket found!",
 929         socket);
 930   }
 931
 932 done:
 933   CLIENTS_UNLOCK (sink);
 934 }
 935
 936 /* "remove-flush" signal implementation */
 937 void
 938 gst_multi_handle_sink_remove_flush (GstMultiHandleSink * sink, GSocket * socket)
 939 {
 940   GList *clink;
 941
 942   GST_DEBUG_OBJECT (sink, "[socket %p] flushing client", socket);
 943
 944   CLIENTS_LOCK (sink);
 945   clink = g_hash_table_lookup (sink->socket_hash, socket);
 946   if (clink != NULL) {
 947     GstSocketClient *client = clink->data;
 948
 949     if (client->status != GST_CLIENT_STATUS_OK) {
 950       GST_INFO_OBJECT (sink,
 951           "[socket %p] Client already disconnecting with status %d",
 952           socket, client->status);
 953       goto done;
 954     }
 955
 956     /* take the position of the client as the number of buffers left to flush.
 957      * If the client was at position -1, we flush 0 buffers, 0 == flush 1
 958      * buffer, etc... */
 959     client->flushcount = client->bufpos + 1;
 960     /* mark client as flushing. We can not remove the client right away because
 961      * it might have some buffers to flush in the ->sending queue. */
 962     client->status = GST_CLIENT_STATUS_FLUSHING;
 963   } else {
 964     GST_WARNING_OBJECT (sink, "[socket %p] no client with this fd found!",
 965         socket);
 966   }
 967 done:
 968   CLIENTS_UNLOCK (sink);
 969 }
 970 #endif
 971
 972 /* can be called both through the signal (i.e. from any thread) or when
 973  * stopping, after the writing thread has shut down */
 974 static void
 975 gst_multi_handle_sink_clear (GstMultiHandleSink * mhsink)
 976 {
 977   GList *clients, *next;
 978   guint32 cookie;
 979   GstMultiHandleSinkClass *mhsinkclass =
 980       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
 981
 982   GST_DEBUG_OBJECT (mhsink, "clearing all clients");
 983
 984   CLIENTS_LOCK (mhsink);
 985 restart:
 986   cookie = mhsink->clients_cookie;
 987   for (clients = mhsink->clients; clients; clients = next) {
 988     GstMultiHandleClient *mhclient;
 989
 990     if (cookie != mhsink->clients_cookie) {
 991       GST_DEBUG_OBJECT (mhsink, "cookie changed while removing all clients");
 992       goto restart;
 993     }
 994
 995     mhclient = (GstMultiHandleClient *) clients->data;
 996     next = g_list_next (clients);
 997
 998     mhclient->status = GST_CLIENT_STATUS_REMOVED;
 999     /* the next call changes the list, which is why we iterate
1000      * with a temporary next pointer */
1001     mhsinkclass->remove_client_link (mhsink, clients);
1002   }
1003   if (mhsinkclass->clear_post)
1004     mhsinkclass->clear_post (mhsink);
1005
1006   CLIENTS_UNLOCK (mhsink);
1007 }
1008
1009 #if 0
1010 /* "get-stats" signal implementation
1011  */
1012 GstStructure *
1013 gst_multi_handle_sink_get_stats (GstMultiHandleSink * sink, GSocket * socket)
1014 {
1015   GstSocketClient *client;
1016   GstStructure *result = NULL;
1017   GList *clink;
1018
1019   CLIENTS_LOCK (sink);
1020   clink = g_hash_table_lookup (sink->socket_hash, socket);
1021   if (clink == NULL)
1022     goto noclient;
1023
1024   client = clink->data;
1025   if (client != NULL) {
1026     guint64 interval;
1027
1028     result = gst_structure_new_empty ("multihandlesink-stats");
1029
1030     if (client->disconnect_time == 0) {
1031       GTimeVal nowtv;
1032
1033       g_get_current_time (&nowtv);
1034
1035       interval = GST_TIMEVAL_TO_TIME (nowtv) - client->connect_time;
1036     } else {
1037       interval = client->disconnect_time - client->connect_time;
1038     }
1039
1040     gst_structure_set (result,
1041         "bytes-sent", G_TYPE_UINT64, client->bytes_sent,
1042         "connect-time", G_TYPE_UINT64, client->connect_time,
1043         "disconnect-time", G_TYPE_UINT64, client->disconnect_time,
1044         "connected-duration", G_TYPE_UINT64, interval,
1045         "last-activatity-time", G_TYPE_UINT64, client->last_activity_time,
1046         "dropped-buffers", G_TYPE_UINT64, client->dropped_buffers,
1047         "first-buffer-ts", G_TYPE_UINT64, client->first_buffer_ts,
1048         "last-buffer-ts", G_TYPE_UINT64, client->last_buffer_ts, NULL);
1049   }
1050
1051 noclient:
1052   CLIENTS_UNLOCK (sink);
1053
1054   /* python doesn't like a NULL pointer yet */
1055   if (result == NULL) {
1056     GST_WARNING_OBJECT (sink, "[socket %p] no client with this found!", socket);
1057     result = gst_structure_new_empty ("multihandlesink-stats");
1058   }
1059
1060   return result;
1061 }
1062 #endif
1063
1064 #if 0
1065 /* should be called with the clientslock held.
1066  * Note that we don't close the fd as we didn't open it in the first
1067  * place. An application should connect to the client-fd-removed signal and
1068  * close the fd itself.
1069  */
1070 static void
1071 gst_multi_handle_sink_remove_client_link (GstMultiHandleSink * sink,
1072     GList * link)
1073 {
1074   GSocket *socket;
1075   GTimeVal now;
1076   GstSocketClient *client = link->data;
1077   GstMultiHandleSinkClass *fclass;
1078
1079   fclass = GST_MULTI_HANDLE_SINK_GET_CLASS (sink);
1080
1081   socket = client->socket;
1082
1083   if (client->currently_removing) {
1084     GST_WARNING_OBJECT (sink, "[socket %p] client is already being removed",
1085         socket);
1086     return;
1087   } else {
1088     client->currently_removing = TRUE;
1089   }
1090
1091   /* FIXME: if we keep track of ip we can log it here and signal */
1092   switch (client->status) {
1093     case GST_CLIENT_STATUS_OK:
1094       GST_WARNING_OBJECT (sink, "[socket %p] removing client %p for no reason",
1095           socket, client);
1096       break;
1097     case GST_CLIENT_STATUS_CLOSED:
1098       GST_DEBUG_OBJECT (sink, "[socket %p] removing client %p because of close",
1099           socket, client);
1100       break;
1101     case GST_CLIENT_STATUS_REMOVED:
1102       GST_DEBUG_OBJECT (sink,
1103           "[socket %p] removing client %p because the app removed it", socket,
1104           client);
1105       break;
1106     case GST_CLIENT_STATUS_SLOW:
1107       GST_INFO_OBJECT (sink,
1108           "[socket %p] removing client %p because it was too slow", socket,
1109           client);
1110       break;
1111     case GST_CLIENT_STATUS_ERROR:
1112       GST_WARNING_OBJECT (sink,
1113           "[socket %p] removing client %p because of error", socket, client);
1114       break;
1115     case GST_CLIENT_STATUS_FLUSHING:
1116     default:
1117       GST_WARNING_OBJECT (sink,
1118           "[socket %p] removing client %p with invalid reason %d", socket,
1119           client, client->status);
1120       break;
1121   }
1122
1123   if (client->source) {
1124     g_source_destroy (client->source);
1125     g_source_unref (client->source);
1126     client->source = NULL;
1127   }
1128
1129   g_get_current_time (&now);
1130   client->disconnect_time = GST_TIMEVAL_TO_TIME (now);
1131
1132   /* free client buffers */
1133   g_slist_foreach (client->sending, (GFunc) gst_mini_object_unref, NULL);
1134   g_slist_free (client->sending);
1135   client->sending = NULL;
1136
1137   if (client->caps)
1138     gst_caps_unref (client->caps);
1139   client->caps = NULL;
1140
1141   /* unlock the mutex before signaling because the signal handler
1142    * might query some properties */
1143   CLIENTS_UNLOCK (sink);
1144
1145   g_signal_emit (G_OBJECT (sink),
1146       gst_multi_handle_sink_signals[SIGNAL_CLIENT_REMOVED], 0, socket,
1147       client->status);
1148
1149   /* lock again before we remove the client completely */
1150   CLIENTS_LOCK (sink);
1151
1152   /* fd cannot be reused in the above signal callback so we can safely
1153    * remove it from the hashtable here */
1154   if (!g_hash_table_remove (sink->socket_hash, socket)) {
1155     GST_WARNING_OBJECT (sink,
1156         "[socket %p] error removing client %p from hash", socket, client);
1157   }
1158   /* after releasing the lock above, the link could be invalid, more
1159    * precisely, the next and prev pointers could point to invalid list
1160    * links. One optimisation could be to add a cookie to the linked list
1161    * and take a shortcut when it did not change between unlocking and locking
1162    * our mutex. For now we just walk the list again. */
1163   sink->clients = g_list_remove (sink->clients, client);
1164   sink->clients_cookie++;
1165
1166   if (fclass->removed)
1167     fclass->removed (sink, socket);
1168
1169   g_free (client);
1170   CLIENTS_UNLOCK (sink);
1171
1172   /* and the fd is really gone now */
1173   g_signal_emit (G_OBJECT (sink),
1174       gst_multi_handle_sink_signals[SIGNAL_CLIENT_SOCKET_REMOVED], 0, socket);
1175   g_object_unref (socket);
1176
1177   CLIENTS_LOCK (sink);
1178 }
1179 #endif
1180
1181 #if 0
1182 /* handle a read on a client socket,
1183  * which either indicates a close or should be ignored
1184  * returns FALSE if some error occured or the client closed. */
1185 static gboolean
1186 gst_multi_handle_sink_handle_client_read (GstMultiHandleSink * sink,
1187     GstSocketClient * client)
1188 {
1189   gboolean ret;
1190   gchar dummy[256];
1191   gssize nread;
1192   GError *err = NULL;
1193   gboolean first = TRUE;
1194
1195   GST_DEBUG_OBJECT (sink, "[socket %p] select reports client read",
1196       client->socket);
1197
1198   ret = TRUE;
1199
1200   /* just Read 'n' Drop, could also just drop the client as it's not supposed
1201    * to write to us except for closing the socket, I guess it's because we
1202    * like to listen to our customers. */
1203   do {
1204     gssize navail;
1205
1206     GST_DEBUG_OBJECT (sink, "[socket %p] client wants us to read",
1207         client->socket);
1208
1209     navail = g_socket_get_available_bytes (client->socket);
1210     if (navail < 0)
1211       break;
1212
1213     nread =
1214         g_socket_receive (client->socket, dummy, MIN (navail, sizeof (dummy)),
1215         sink->cancellable, &err);
1216     if (first && nread == 0) {
1217       /* client sent close, so remove it */
1218       GST_DEBUG_OBJECT (sink, "[socket %p] client asked for close, removing",
1219           client->socket);
1220       client->status = GST_CLIENT_STATUS_CLOSED;
1221       ret = FALSE;
1222     } else if (nread < 0) {
1223       GST_WARNING_OBJECT (sink, "[socket %p] could not read: %s",
1224           client->socket, err->message);
1225       client->status = GST_CLIENT_STATUS_ERROR;
1226       ret = FALSE;
1227       break;
1228     }
1229     first = FALSE;
1230   } while (nread > 0);
1231   g_clear_error (&err);
1232
1233   return ret;
1234 }
1235 #endif
1236
1237 #if 0
1238 /* queue the given buffer for the given client */
1239 static gboolean
1240 gst_multi_handle_sink_client_queue_buffer (GstMultiHandleSink * sink,
1241     GstSocketClient * client, GstBuffer * buffer)
1242 {
1243   GstCaps *caps;
1244
1245   /* TRUE: send them if the new caps have them */
1246   gboolean send_streamheader = FALSE;
1247   GstStructure *s;
1248
1249   /* before we queue the buffer, we check if we need to queue streamheader
1250    * buffers (because it's a new client, or because they changed) */
1251   caps = gst_pad_get_current_caps (GST_BASE_SINK_PAD (sink));
1252
1253   if (!client->caps) {
1254     GST_DEBUG_OBJECT (sink,
1255         "[socket %p] no previous caps for this client, send streamheader",
1256         client->socket);
1257     send_streamheader = TRUE;
1258     client->caps = gst_caps_ref (caps);
1259   } else {
1260     /* there were previous caps recorded, so compare */
1261     if (!gst_caps_is_equal (caps, client->caps)) {
1262       const GValue *sh1, *sh2;
1263
1264       /* caps are not equal, but could still have the same streamheader */
1265       s = gst_caps_get_structure (caps, 0);
1266       if (!gst_structure_has_field (s, "streamheader")) {
1267         /* no new streamheader, so nothing new to send */
1268         GST_DEBUG_OBJECT (sink,
1269             "[socket %p] new caps do not have streamheader, not sending",
1270             client->socket);
1271       } else {
1272         /* there is a new streamheader */
1273         s = gst_caps_get_structure (client->caps, 0);
1274         if (!gst_structure_has_field (s, "streamheader")) {
1275           /* no previous streamheader, so send the new one */
1276           GST_DEBUG_OBJECT (sink,
1277               "[socket %p] previous caps did not have streamheader, sending",
1278               client->socket);
1279           send_streamheader = TRUE;
1280         } else {
1281           /* both old and new caps have streamheader set */
1282           if (!sink->resend_streamheader) {
1283             GST_DEBUG_OBJECT (sink,
1284                 "[socket %p] asked to not resend the streamheader, not sending",
1285                 client->socket);
1286             send_streamheader = FALSE;
1287           } else {
1288             sh1 = gst_structure_get_value (s, "streamheader");
1289             s = gst_caps_get_structure (caps, 0);
1290             sh2 = gst_structure_get_value (s, "streamheader");
1291             if (gst_value_compare (sh1, sh2) != GST_VALUE_EQUAL) {
1292               GST_DEBUG_OBJECT (sink,
1293                   "[socket %p] new streamheader different from old, sending",
1294                   client->socket);
1295               send_streamheader = TRUE;
1296             }
1297           }
1298         }
1299       }
1300     }
1301     /* Replace the old caps */
1302     gst_caps_unref (client->caps);
1303     client->caps = gst_caps_ref (caps);
1304   }
1305
1306   if (G_UNLIKELY (send_streamheader)) {
1307     const GValue *sh;
1308     GArray *buffers;
1309     int i;
1310
1311     GST_LOG_OBJECT (sink,
1312         "[socket %p] sending streamheader from caps %" GST_PTR_FORMAT,
1313         client->socket, caps);
1314     s = gst_caps_get_structure (caps, 0);
1315     if (!gst_structure_has_field (s, "streamheader")) {
1316       GST_DEBUG_OBJECT (sink,
1317           "[socket %p] no new streamheader, so nothing to send",
1318           client->socket);
1319     } else {
1320       GST_LOG_OBJECT (sink,
1321           "[socket %p] sending streamheader from caps %" GST_PTR_FORMAT,
1322           client->socket, caps);
1323       sh = gst_structure_get_value (s, "streamheader");
1324       g_assert (G_VALUE_TYPE (sh) == GST_TYPE_ARRAY);
1325       buffers = g_value_peek_pointer (sh);
1326       GST_DEBUG_OBJECT (sink, "%d streamheader buffers", buffers->len);
1327       for (i = 0; i < buffers->len; ++i) {
1328         GValue *bufval;
1329         GstBuffer *buffer;
1330
1331         bufval = &g_array_index (buffers, GValue, i);
1332         g_assert (G_VALUE_TYPE (bufval) == GST_TYPE_BUFFER);
1333         buffer = g_value_peek_pointer (bufval);
1334         GST_DEBUG_OBJECT (sink,
1335             "[socket %p] queueing streamheader buffer of length %"
1336             G_GSIZE_FORMAT, client->socket, gst_buffer_get_size (buffer));
1337         gst_buffer_ref (buffer);
1338
1339         client->sending = g_slist_append (client->sending, buffer);
1340       }
1341     }
1342   }
1343
1344   gst_caps_unref (caps);
1345   caps = NULL;
1346
1347   GST_LOG_OBJECT (sink,
1348       "[socket %p] queueing buffer of length %" G_GSIZE_FORMAT, client->socket,
1349       gst_buffer_get_size (buffer));
1350
1351   gst_buffer_ref (buffer);
1352   client->sending = g_slist_append (client->sending, buffer);
1353
1354   return TRUE;
1355 }
1356 #endif
1357
1358 // FIXME: privatize again ?
1359 gboolean
1360 is_sync_frame (GstMultiHandleSink * sink, GstBuffer * buffer)
1361 {
1362   if (GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_DELTA_UNIT)) {
1363     return FALSE;
1364   } else if (!GST_BUFFER_FLAG_IS_SET (buffer, GST_BUFFER_FLAG_HEADER)) {
1365     return TRUE;
1366   }
1367
1368   return FALSE;
1369 }
1370
1371 /* find the keyframe in the list of buffers starting the
1372  * search from @idx. @direction as -1 will search backwards,
1373  * 1 will search forwards.
1374  * Returns: the index or -1 if there is no keyframe after idx.
1375  */
1376 gint
1377 find_syncframe (GstMultiHandleSink * sink, gint idx, gint direction)
1378 {
1379   gint i, len, result;
1380
1381   /* take length of queued buffers */
1382   len = sink->bufqueue->len;
1383
1384   /* assume we don't find a keyframe */
1385   result = -1;
1386
1387   /* then loop over all buffers to find the first keyframe */
1388   for (i = idx; i >= 0 && i < len; i += direction) {
1389     GstBuffer *buf;
1390
1391     buf = g_array_index (sink->bufqueue, GstBuffer *, i);
1392     if (is_sync_frame (sink, buf)) {
1393       GST_LOG_OBJECT (sink, "found keyframe at %d from %d, direction %d",
1394           i, idx, direction);
1395       result = i;
1396       break;
1397     }
1398   }
1399   return result;
1400 }
1401
1402 /* Get the number of buffers from the buffer queue needed to satisfy
1403  * the maximum max in the configured units.
1404  * If units are not BUFFERS, and there are insufficient buffers in the
1405  * queue to satify the limit, return len(queue) + 1 */
1406 gint
1407 get_buffers_max (GstMultiHandleSink * sink, gint64 max)
1408 {
1409   switch (sink->unit_format) {
1410     case GST_FORMAT_BUFFERS:
1411       return max;
1412     case GST_FORMAT_TIME:
1413     {
1414       GstBuffer *buf;
1415       int i;
1416       int len;
1417       gint64 diff;
1418       GstClockTime first = GST_CLOCK_TIME_NONE;
1419
1420       len = sink->bufqueue->len;
1421
1422       for (i = 0; i < len; i++) {
1423         buf = g_array_index (sink->bufqueue, GstBuffer *, i);
1424         if (GST_BUFFER_TIMESTAMP_IS_VALID (buf)) {
1425           if (first == -1)
1426             first = GST_BUFFER_TIMESTAMP (buf);
1427
1428           diff = first - GST_BUFFER_TIMESTAMP (buf);
1429
1430           if (diff > max)
1431             return i + 1;
1432         }
1433       }
1434       return len + 1;
1435     }
1436     case GST_FORMAT_BYTES:
1437     {
1438       GstBuffer *buf;
1439       int i;
1440       int len;
1441       gint acc = 0;
1442
1443       len = sink->bufqueue->len;
1444
1445       for (i = 0; i < len; i++) {
1446         buf = g_array_index (sink->bufqueue, GstBuffer *, i);
1447         acc += gst_buffer_get_size (buf);
1448
1449         if (acc > max)
1450           return i + 1;
1451       }
1452       return len + 1;
1453     }
1454     default:
1455       return max;
1456   }
1457 }
1458
1459 /* find the positions in the buffer queue where *_min and *_max
1460  * is satisfied
1461  */
1462 /* count the amount of data in the buffers and return the index
1463  * that satifies the given limits.
1464  *
1465  * Returns: index @idx in the buffer queue so that the given limits are
1466  * satisfied. TRUE if all the limits could be satisfied, FALSE if not
1467  * enough data was in the queue.
1468  *
1469  * FIXME, this code might now work if any of the units is in buffers...
1470  */
1471 gboolean
1472 find_limits (GstMultiHandleSink * sink,
1473     gint * min_idx, gint bytes_min, gint buffers_min, gint64 time_min,
1474     gint * max_idx, gint bytes_max, gint buffers_max, gint64 time_max)
1475 {
1476   GstClockTime first, time;
1477   gint i, len, bytes;
1478   gboolean result, max_hit;
1479
1480   /* take length of queue */
1481   len = sink->bufqueue->len;
1482
1483   /* this must hold */
1484   g_assert (len > 0);
1485
1486   GST_LOG_OBJECT (sink,
1487       "bytes_min %d, buffers_min %d, time_min %" GST_TIME_FORMAT
1488       ", bytes_max %d, buffers_max %d, time_max %" GST_TIME_FORMAT, bytes_min,
1489       buffers_min, GST_TIME_ARGS (time_min), bytes_max, buffers_max,
1490       GST_TIME_ARGS (time_max));
1491
1492   /* do the trivial buffer limit test */
1493   if (buffers_min != -1 && len < buffers_min) {
1494     *min_idx = len - 1;
1495     *max_idx = len - 1;
1496     return FALSE;
1497   }
1498
1499   result = FALSE;
1500   /* else count bytes and time */
1501   first = -1;
1502   bytes = 0;
1503   /* unset limits */
1504   *min_idx = -1;
1505   *max_idx = -1;
1506   max_hit = FALSE;
1507
1508   i = 0;
1509   /* loop through the buffers, when a limit is ok, mark it
1510    * as -1, we have at least one buffer in the queue. */
1511   do {
1512     GstBuffer *buf;
1513
1514     /* if we checked all min limits, update result */
1515     if (bytes_min == -1 && time_min == -1 && *min_idx == -1) {
1516       /* don't go below 0 */
1517       *min_idx = MAX (i - 1, 0);
1518     }
1519     /* if we reached one max limit break out */
1520     if (max_hit) {
1521       /* i > 0 when we get here, we subtract one to get the position
1522        * of the previous buffer. */
1523       *max_idx = i - 1;
1524       /* we have valid complete result if we found a min_idx too */
1525       result = *min_idx != -1;
1526       break;
1527     }
1528     buf = g_array_index (sink->bufqueue, GstBuffer *, i);
1529
1530     bytes += gst_buffer_get_size (buf);
1531
1532     /* take timestamp and save for the base first timestamp */
1533     if ((time = GST_BUFFER_TIMESTAMP (buf)) != -1) {
1534       GST_LOG_OBJECT (sink, "Ts %" GST_TIME_FORMAT " on buffer",
1535           GST_TIME_ARGS (time));
1536       if (first == -1)
1537         first = time;
1538
1539       /* increase max usage if we did not fill enough. Note that
1540        * buffers are sorted from new to old, so the first timestamp is
1541        * bigger than the next one. */
1542       if (time_min != -1 && first - time >= time_min)
1543         time_min = -1;
1544       if (time_max != -1 && first - time >= time_max)
1545         max_hit = TRUE;
1546     } else {
1547       GST_LOG_OBJECT (sink, "No timestamp on buffer");
1548     }
1549     /* time is OK or unknown, check and increase if not enough bytes */
1550     if (bytes_min != -1) {
1551       if (bytes >= bytes_min)
1552         bytes_min = -1;
1553     }
1554     if (bytes_max != -1) {
1555       if (bytes >= bytes_max) {
1556         max_hit = TRUE;
1557       }
1558     }
1559     i++;
1560   }
1561   while (i < len);
1562
1563   /* if we did not hit the max or min limit, set to buffer size */
1564   if (*max_idx == -1)
1565     *max_idx = len - 1;
1566   /* make sure min does not exceed max */
1567   if (*min_idx == -1)
1568     *min_idx = *max_idx;
1569
1570   return result;
1571 }
1572
1573 /* parse the unit/value pair and assign it to the result value of the
1574  * right type, leave the other values untouched
1575  *
1576  * Returns: FALSE if the unit is unknown or undefined. TRUE otherwise.
1577  */
1578 static gboolean
1579 assign_value (GstFormat format, guint64 value, gint * bytes, gint * buffers,
1580     GstClockTime * time)
1581 {
1582   gboolean res = TRUE;
1583
1584   /* set only the limit of the given format to the given value */
1585   switch (format) {
1586     case GST_FORMAT_BUFFERS:
1587       *buffers = (gint) value;
1588       break;
1589     case GST_FORMAT_TIME:
1590       *time = value;
1591       break;
1592     case GST_FORMAT_BYTES:
1593       *bytes = (gint) value;
1594       break;
1595     case GST_FORMAT_UNDEFINED:
1596     default:
1597       res = FALSE;
1598       break;
1599   }
1600   return res;
1601 }
1602
1603 /* count the index in the buffer queue to satisfy the given unit
1604  * and value pair starting from buffer at index 0.
1605  *
1606  * Returns: TRUE if there was enough data in the queue to satisfy the
1607  * burst values. @idx contains the index in the buffer that contains enough
1608  * data to satisfy the limits or the last buffer in the queue when the
1609  * function returns FALSE.
1610  */
1611 static gboolean
1612 count_burst_unit (GstMultiHandleSink * sink, gint * min_idx,
1613     GstFormat min_format, guint64 min_value, gint * max_idx,
1614     GstFormat max_format, guint64 max_value)
1615 {
1616   gint bytes_min = -1, buffers_min = -1;
1617   gint bytes_max = -1, buffers_max = -1;
1618   GstClockTime time_min = GST_CLOCK_TIME_NONE, time_max = GST_CLOCK_TIME_NONE;
1619
1620   assign_value (min_format, min_value, &bytes_min, &buffers_min, &time_min);
1621   assign_value (max_format, max_value, &bytes_max, &buffers_max, &time_max);
1622
1623   return find_limits (sink, min_idx, bytes_min, buffers_min, time_min,
1624       max_idx, bytes_max, buffers_max, time_max);
1625 }
1626
1627 /* decide where in the current buffer queue this new client should start
1628  * receiving buffers from.
1629  * This function is called whenever a client is connected and has not yet
1630  * received a buffer.
1631  * If this returns -1, it means that we haven't found a good point to
1632  * start streaming from yet, and this function should be called again later
1633  * when more buffers have arrived.
1634  */
1635 gint
1636 gst_multi_handle_sink_new_client (GstMultiHandleSink * sink,
1637     GstMultiHandleClient * client)
1638 {
1639   gint result;
1640
1641   GST_DEBUG_OBJECT (sink,
1642       "%s new client, deciding where to start in queue", client->debug);
1643   GST_DEBUG_OBJECT (sink, "queue is currently %d buffers long",
1644       sink->bufqueue->len);
1645   switch (client->sync_method) {
1646     case GST_SYNC_METHOD_LATEST:
1647       /* no syncing, we are happy with whatever the client is going to get */
1648       result = client->bufpos;
1649       GST_DEBUG_OBJECT (sink,
1650           "%s SYNC_METHOD_LATEST, position %d", client->debug, result);
1651       break;
1652     case GST_SYNC_METHOD_NEXT_KEYFRAME:
1653     {
1654       /* if one of the new buffers (between client->bufpos and 0) in the queue
1655        * is a sync point, we can proceed, otherwise we need to keep waiting */
1656       GST_LOG_OBJECT (sink,
1657           "%s new client, bufpos %d, waiting for keyframe",
1658           client->debug, client->bufpos);
1659
1660       result = find_prev_syncframe (sink, client->bufpos);
1661       if (result != -1) {
1662         GST_DEBUG_OBJECT (sink,
1663             "%s SYNC_METHOD_NEXT_KEYFRAME: result %d", client->debug, result);
1664         break;
1665       }
1666
1667       /* client is not on a syncbuffer, need to skip these buffers and
1668        * wait some more */
1669       GST_LOG_OBJECT (sink,
1670           "%s new client, skipping buffer(s), no syncpoint found",
1671           client->debug);
1672       client->bufpos = -1;
1673       break;
1674     }
1675     case GST_SYNC_METHOD_LATEST_KEYFRAME:
1676     {
1677       GST_DEBUG_OBJECT (sink, "%s SYNC_METHOD_LATEST_KEYFRAME", client->debug);
1678
1679       /* for new clients we initially scan the complete buffer queue for
1680        * a sync point when a buffer is added. If we don't find a keyframe,
1681        * we need to wait for the next keyframe and so we change the client's
1682        * sync method to GST_SYNC_METHOD_NEXT_KEYFRAME.
1683        */
1684       result = find_next_syncframe (sink, 0);
1685       if (result != -1) {
1686         GST_DEBUG_OBJECT (sink,
1687             "%s SYNC_METHOD_LATEST_KEYFRAME: result %d", client->debug, result);
1688         break;
1689       }
1690
1691       GST_DEBUG_OBJECT (sink,
1692           "%s SYNC_METHOD_LATEST_KEYFRAME: no keyframe found, "
1693           "switching to SYNC_METHOD_NEXT_KEYFRAME", client->debug);
1694       /* throw client to the waiting state */
1695       client->bufpos = -1;
1696       /* and make client sync to next keyframe */
1697       client->sync_method = GST_SYNC_METHOD_NEXT_KEYFRAME;
1698       break;
1699     }
1700     case GST_SYNC_METHOD_BURST:
1701     {
1702       gboolean ok;
1703       gint max;
1704
1705       /* move to the position where we satisfy the client's burst
1706        * parameters. If we could not satisfy the parameters because there
1707        * is not enough data, we just send what we have (which is in result).
1708        * We use the max value to limit the search
1709        */
1710       ok = count_burst_unit (sink, &result, client->burst_min_format,
1711           client->burst_min_value, &max, client->burst_max_format,
1712           client->burst_max_value);
1713       GST_DEBUG_OBJECT (sink,
1714           "%s SYNC_METHOD_BURST: burst_unit returned %d, result %d",
1715           client->debug, ok, result);
1716
1717       GST_LOG_OBJECT (sink, "min %d, max %d", result, max);
1718
1719       /* we hit the max and it is below the min, use that then */
1720       if (max != -1 && max <= result) {
1721         result = MAX (max - 1, 0);
1722         GST_DEBUG_OBJECT (sink,
1723             "%s SYNC_METHOD_BURST: result above max, taken down to %d",
1724             client->debug, result);
1725       }
1726       break;
1727     }
1728     case GST_SYNC_METHOD_BURST_KEYFRAME:
1729     {
1730       gint min_idx, max_idx;
1731       gint next_syncframe, prev_syncframe;
1732
1733       /* BURST_KEYFRAME:
1734        *
1735        * _always_ start sending a keyframe to the client. We first search
1736        * a keyframe between min/max limits. If there is none, we send it the
1737        * last keyframe before min. If there is none, the behaviour is like
1738        * NEXT_KEYFRAME.
1739        */
1740       /* gather burst limits */
1741       count_burst_unit (sink, &min_idx, client->burst_min_format,
1742           client->burst_min_value, &max_idx, client->burst_max_format,
1743           client->burst_max_value);
1744
1745       GST_LOG_OBJECT (sink, "min %d, max %d", min_idx, max_idx);
1746
1747       /* first find a keyframe after min_idx */
1748       next_syncframe = find_next_syncframe (sink, min_idx);
1749       if (next_syncframe != -1 && next_syncframe < max_idx) {
1750         /* we have a valid keyframe and it's below the max */
1751         GST_LOG_OBJECT (sink, "found keyframe in min/max limits");
1752         result = next_syncframe;
1753         break;
1754       }
1755
1756       /* no valid keyframe, try to find one below min */
1757       prev_syncframe = find_prev_syncframe (sink, min_idx);
1758       if (prev_syncframe != -1) {
1759         GST_WARNING_OBJECT (sink,
1760             "using keyframe below min in BURST_KEYFRAME sync mode");
1761         result = prev_syncframe;
1762         break;
1763       }
1764
1765       /* no prev keyframe or not enough data  */
1766       GST_WARNING_OBJECT (sink,
1767           "no prev keyframe found in BURST_KEYFRAME sync mode, waiting for next");
1768
1769       /* throw client to the waiting state */
1770       client->bufpos = -1;
1771       /* and make client sync to next keyframe */
1772       client->sync_method = GST_SYNC_METHOD_NEXT_KEYFRAME;
1773       result = -1;
1774       break;
1775     }
1776     case GST_SYNC_METHOD_BURST_WITH_KEYFRAME:
1777     {
1778       gint min_idx, max_idx;
1779       gint next_syncframe;
1780
1781       /* BURST_WITH_KEYFRAME:
1782        *
1783        * try to start sending a keyframe to the client. We first search
1784        * a keyframe between min/max limits. If there is none, we send it the
1785        * amount of data up 'till min.
1786        */
1787       /* gather enough data to burst */
1788       count_burst_unit (sink, &min_idx, client->burst_min_format,
1789           client->burst_min_value, &max_idx, client->burst_max_format,
1790           client->burst_max_value);
1791
1792       GST_LOG_OBJECT (sink, "min %d, max %d", min_idx, max_idx);
1793
1794       /* first find a keyframe after min_idx */
1795       next_syncframe = find_next_syncframe (sink, min_idx);
1796       if (next_syncframe != -1 && next_syncframe < max_idx) {
1797         /* we have a valid keyframe and it's below the max */
1798         GST_LOG_OBJECT (sink, "found keyframe in min/max limits");
1799         result = next_syncframe;
1800         break;
1801       }
1802
1803       /* no keyframe, send data from min_idx */
1804       GST_WARNING_OBJECT (sink, "using min in BURST_WITH_KEYFRAME sync mode");
1805
1806       /* make sure we don't go over the max limit */
1807       if (max_idx != -1 && max_idx <= min_idx) {
1808         result = MAX (max_idx - 1, 0);
1809       } else {
1810         result = min_idx;
1811       }
1812
1813       break;
1814     }
1815     default:
1816       g_warning ("unknown sync method %d", client->sync_method);
1817       result = client->bufpos;
1818       break;
1819   }
1820   return result;
1821 }
1822
1823 #if 0
1824 /* Handle a write on a client,
1825  * which indicates a read request from a client.
1826  *
1827  * For each client we maintain a queue of GstBuffers that contain the raw bytes
1828  * we need to send to the client.
1829  *
1830  * We first check to see if we need to send streamheaders. If so, we queue them.
1831  *
1832  * Then we run into the main loop that tries to send as many buffers as
1833  * possible. It will first exhaust the client->sending queue and if the queue
1834  * is empty, it will pick a buffer from the global queue.
1835  *
1836  * Sending the buffers from the client->sending queue is basically writing
1837  * the bytes to the socket and maintaining a count of the bytes that were
1838  * sent. When the buffer is completely sent, it is removed from the
1839  * client->sending queue and we try to pick a new buffer for sending.
1840  *
1841  * When the sending returns a partial buffer we stop sending more data as
1842  * the next send operation could block.
1843  *
1844  * This functions returns FALSE if some error occured.
1845  */
1846 static gboolean
1847 gst_multi_handle_sink_handle_client_write (GstMultiHandleSink * sink,
1848     GstSocketClient * client)
1849 {
1850   GSocket *socket = client->socket;
1851   gboolean more;
1852   gboolean flushing;
1853   GstClockTime now;
1854   GTimeVal nowtv;
1855   GError *err = NULL;
1856
1857   g_get_current_time (&nowtv);
1858   now = GST_TIMEVAL_TO_TIME (nowtv);
1859
1860   flushing = client->status == GST_CLIENT_STATUS_FLUSHING;
1861
1862   more = TRUE;
1863   do {
1864     gint maxsize;
1865
1866     if (!client->sending) {
1867       /* client is not working on a buffer */
1868       if (client->bufpos == -1) {
1869         /* client is too fast, remove from write queue until new buffer is
1870          * available */
1871         if (client->source) {
1872           g_source_destroy (client->source);
1873           g_source_unref (client->source);
1874           client->source = NULL;
1875         }
1876         /* if we flushed out all of the client buffers, we can stop */
1877         if (client->flushcount == 0)
1878           goto flushed;
1879
1880         return TRUE;
1881       } else {
1882         /* client can pick a buffer from the global queue */
1883         GstBuffer *buf;
1884         GstClockTime timestamp;
1885
1886         /* for new connections, we need to find a good spot in the
1887          * bufqueue to start streaming from */
1888         if (client->new_connection && !flushing) {
1889           gint position = gst_multi_handle_sink_new_client (sink, client);
1890
1891           if (position >= 0) {
1892             /* we got a valid spot in the queue */
1893             client->new_connection = FALSE;
1894             client->bufpos = position;
1895           } else {
1896             /* cannot send data to this client yet */
1897             if (client->source) {
1898               g_source_destroy (client->source);
1899               g_source_unref (client->source);
1900               client->source = NULL;
1901             }
1902             return TRUE;
1903           }
1904         }
1905
1906         /* we flushed all remaining buffers, no need to get a new one */
1907         if (client->flushcount == 0)
1908           goto flushed;
1909
1910         /* grab buffer */
1911         buf = g_array_index (sink->bufqueue, GstBuffer *, client->bufpos);
1912         client->bufpos--;
1913
1914         /* update stats */
1915         timestamp = GST_BUFFER_TIMESTAMP (buf);
1916         if (client->first_buffer_ts == GST_CLOCK_TIME_NONE)
1917           client->first_buffer_ts = timestamp;
1918         if (timestamp != -1)
1919           client->last_buffer_ts = timestamp;
1920
1921         /* decrease flushcount */
1922         if (client->flushcount != -1)
1923           client->flushcount--;
1924
1925         GST_LOG_OBJECT (sink, "[socket %p] client %p at position %d",
1926             socket, client, client->bufpos);
1927
1928         /* queueing a buffer will ref it */
1929         gst_multi_handle_sink_client_queue_buffer (sink, client, buf);
1930
1931         /* need to start from the first byte for this new buffer */
1932         client->bufoffset = 0;
1933       }
1934     }
1935
1936     /* see if we need to send something */
1937     if (client->sending) {
1938       gssize wrote;
1939       GstBuffer *head;
1940       GstMapInfo map;
1941
1942       /* pick first buffer from list */
1943       head = GST_BUFFER (client->sending->data);
1944
1945       gst_buffer_map (head, &map, GST_MAP_READ);
1946       maxsize = map.size - client->bufoffset;
1947
1948       /* try to write the complete buffer */
1949
1950       wrote =
1951           g_socket_send (socket, (gchar *) map.data + client->bufoffset,
1952           maxsize, sink->cancellable, &err);
1953       gst_buffer_unmap (head, &map);
1954
1955       if (wrote < 0) {
1956         /* hmm error.. */
1957         if (g_error_matches (err, G_IO_ERROR, G_IO_ERROR_CLOSED)) {
1958           goto connection_reset;
1959         } else {
1960           goto write_error;
1961         }
1962       } else {
1963         if (wrote < maxsize) {
1964           /* partial write means that the client cannot read more and we should
1965            * stop sending more */
1966           GST_LOG_OBJECT (sink,
1967               "partial write on %p of %" G_GSSIZE_FORMAT " bytes", socket,
1968               wrote);
1969           client->bufoffset += wrote;
1970           more = FALSE;
1971         } else {
1972           /* complete buffer was written, we can proceed to the next one */
1973           client->sending = g_slist_remove (client->sending, head);
1974           gst_buffer_unref (head);
1975           /* make sure we start from byte 0 for the next buffer */
1976           client->bufoffset = 0;
1977         }
1978         /* update stats */
1979         client->bytes_sent += wrote;
1980         client->last_activity_time = now;
1981         sink->bytes_served += wrote;
1982       }
1983     }
1984   } while (more);
1985
1986   return TRUE;
1987
1988   /* ERRORS */
1989 flushed:
1990   {
1991     GST_DEBUG_OBJECT (sink, "[socket %p] flushed, removing", socket);
1992     client->status = GST_CLIENT_STATUS_REMOVED;
1993     return FALSE;
1994   }
1995 connection_reset:
1996   {
1997     GST_DEBUG_OBJECT (sink, "[socket %p] connection reset by peer, removing",
1998         socket);
1999     client->status = GST_CLIENT_STATUS_CLOSED;
2000     g_clear_error (&err);
2001     return FALSE;
2002   }
2003 write_error:
2004   {
2005     GST_WARNING_OBJECT (sink,
2006         "[socket %p] could not write, removing client: %s", socket,
2007         err->message);
2008     g_clear_error (&err);
2009     client->status = GST_CLIENT_STATUS_ERROR;
2010     return FALSE;
2011   }
2012 }
2013 #endif
2014
2015 /* calculate the new position for a client after recovery. This function
2016  * does not update the client position but merely returns the required
2017  * position.
2018  */
2019 gint
2020 gst_multi_handle_sink_recover_client (GstMultiHandleSink * sink,
2021     GstMultiHandleClient * client)
2022 {
2023   gint newbufpos;
2024
2025   GST_WARNING_OBJECT (sink,
2026       "%s client %p is lagging at %d, recover using policy %d",
2027       client->debug, client, client->bufpos, sink->recover_policy);
2028
2029   switch (sink->recover_policy) {
2030     case GST_RECOVER_POLICY_NONE:
2031       /* do nothing, client will catch up or get kicked out when it reaches
2032        * the hard max */
2033       newbufpos = client->bufpos;
2034       break;
2035     case GST_RECOVER_POLICY_RESYNC_LATEST:
2036       /* move to beginning of queue */
2037       newbufpos = -1;
2038       break;
2039     case GST_RECOVER_POLICY_RESYNC_SOFT_LIMIT:
2040       /* move to beginning of soft max */
2041       newbufpos = get_buffers_max (sink, sink->units_soft_max);
2042       break;
2043     case GST_RECOVER_POLICY_RESYNC_KEYFRAME:
2044       /* find keyframe in buffers, we search backwards to find the
2045        * closest keyframe relative to what this client already received. */
2046       newbufpos = MIN (sink->bufqueue->len - 1,
2047           get_buffers_max (sink, sink->units_soft_max) - 1);
2048
2049       while (newbufpos >= 0) {
2050         GstBuffer *buf;
2051
2052         buf = g_array_index (sink->bufqueue, GstBuffer *, newbufpos);
2053         if (is_sync_frame (sink, buf)) {
2054           /* found a buffer that is not a delta unit */
2055           break;
2056         }
2057         newbufpos--;
2058       }
2059       break;
2060     default:
2061       /* unknown recovery procedure */
2062       newbufpos = get_buffers_max (sink, sink->units_soft_max);
2063       break;
2064   }
2065   return newbufpos;
2066 }
2067
2068 #if 0
2069 /* Queue a buffer on the global queue.
2070  *
2071  * This function adds the buffer to the front of a GArray. It removes the
2072  * tail buffer if the max queue size is exceeded, unreffing the queued buffer.
2073  * Note that unreffing the buffer is not a problem as clients who
2074  * started writing out this buffer will still have a reference to it in the
2075  * client->sending queue.
2076  *
2077  * After adding the buffer, we update all client positions in the queue. If
2078  * a client moves over the soft max, we start the recovery procedure for this
2079  * slow client. If it goes over the hard max, it is put into the slow list
2080  * and removed.
2081  *
2082  * Special care is taken of clients that were waiting for a new buffer (they
2083  * had a position of -1) because they can proceed after adding this new buffer.
2084  * This is done by adding the client back into the write fd_set and signaling
2085  * the select thread that the fd_set changed.
2086  */
2087 static void
2088 gst_multi_handle_sink_queue_buffer (GstMultiHandleSink * sink, GstBuffer * buf)
2089 {
2090   GList *clients, *next;
2091   gint queuelen;
2092   gint max_buffer_usage;
2093   gint i;
2094   GTimeVal nowtv;
2095   GstClockTime now;
2096   gint max_buffers, soft_max_buffers;
2097   guint cookie;
2098
2099   g_get_current_time (&nowtv);
2100   now = GST_TIMEVAL_TO_TIME (nowtv);
2101
2102   CLIENTS_LOCK (sink);
2103   /* add buffer to queue */
2104   gst_buffer_ref (buf);
2105   g_array_prepend_val (sink->bufqueue, buf);
2106   queuelen = sink->bufqueue->len;
2107
2108   if (sink->units_max > 0)
2109     max_buffers = get_buffers_max (sink, sink->units_max);
2110   else
2111     max_buffers = -1;
2112
2113   if (sink->units_soft_max > 0)
2114     soft_max_buffers = get_buffers_max (sink, sink->units_soft_max);
2115   else
2116     soft_max_buffers = -1;
2117   GST_LOG_OBJECT (sink, "Using max %d, softmax %d", max_buffers,
2118       soft_max_buffers);
2119
2120   /* then loop over the clients and update the positions */
2121   max_buffer_usage = 0;
2122
2123 restart:
2124   cookie = sink->clients_cookie;
2125   for (clients = sink->clients; clients; clients = next) {
2126     GstSocketClient *client;
2127
2128     if (cookie != sink->clients_cookie) {
2129       GST_DEBUG_OBJECT (sink, "Clients cookie outdated, restarting");
2130       goto restart;
2131     }
2132
2133     client = clients->data;
2134     next = g_list_next (clients);
2135
2136     client->bufpos++;
2137     GST_LOG_OBJECT (sink, "[socket %p] client %p at position %d",
2138         client->socket, client, client->bufpos);
2139     /* check soft max if needed, recover client */
2140     if (soft_max_buffers > 0 && client->bufpos >= soft_max_buffers) {
2141       gint newpos;
2142
2143       newpos = gst_multi_handle_sink_recover_client (sink, client);
2144       if (newpos != client->bufpos) {
2145         client->dropped_buffers += client->bufpos - newpos;
2146         client->bufpos = newpos;
2147         client->discont = TRUE;
2148         GST_INFO_OBJECT (sink, "[socket %p] client %p position reset to %d",
2149             client->socket, client, client->bufpos);
2150       } else {
2151         GST_INFO_OBJECT (sink,
2152             "[socket %p] client %p not recovering position",
2153             client->socket, client);
2154       }
2155     }
2156     /* check hard max and timeout, remove client */
2157     if ((max_buffers > 0 && client->bufpos >= max_buffers) ||
2158         (sink->timeout > 0
2159             && now - client->last_activity_time > sink->timeout)) {
2160       /* remove client */
2161       GST_WARNING_OBJECT (sink, "[socket %p] client %p is too slow, removing",
2162           client->socket, client);
2163       /* remove the client, the fd set will be cleared and the select thread
2164        * will be signaled */
2165       client->status = GST_CLIENT_STATUS_SLOW;
2166       /* set client to invalid position while being removed */
2167       client->bufpos = -1;
2168       gst_multi_handle_sink_remove_client_link (sink, clients);
2169       continue;
2170     } else if (client->bufpos == 0 || client->new_connection) {
2171       /* can send data to this client now. need to signal the select thread that
2172        * the fd_set changed */
2173       if (!client->source) {
2174         client->source =
2175             g_socket_create_source (client->socket,
2176             G_IO_IN | G_IO_OUT | G_IO_PRI | G_IO_ERR | G_IO_HUP,
2177             sink->cancellable);
2178         g_source_set_callback (client->source,
2179             (GSourceFunc) gst_multi_handle_sink_socket_condition,
2180             gst_object_ref (sink), (GDestroyNotify) gst_object_unref);
2181         g_source_attach (client->source, sink->main_context);
2182       }
2183     }
2184     /* keep track of maximum buffer usage */
2185     if (client->bufpos > max_buffer_usage) {
2186       max_buffer_usage = client->bufpos;
2187     }
2188   }
2189
2190   /* make sure we respect bytes-min, buffers-min and time-min when they are set */
2191   {
2192     gint usage, max;
2193
2194     GST_LOG_OBJECT (sink,
2195         "extending queue %d to respect time_min %" GST_TIME_FORMAT
2196         ", bytes_min %d, buffers_min %d", max_buffer_usage,
2197         GST_TIME_ARGS (sink->time_min), sink->bytes_min, sink->buffers_min);
2198
2199     /* get index where the limits are ok, we don't really care if all limits
2200      * are ok, we just queue as much as we need. We also don't compare against
2201      * the max limits. */
2202     find_limits (sink, &usage, sink->bytes_min, sink->buffers_min,
2203         sink->time_min, &max, -1, -1, -1);
2204
2205     max_buffer_usage = MAX (max_buffer_usage, usage + 1);
2206     GST_LOG_OBJECT (sink, "extended queue to %d", max_buffer_usage);
2207   }
2208
2209   /* now look for sync points and make sure there is at least one
2210    * sync point in the queue. We only do this if the LATEST_KEYFRAME or
2211    * BURST_KEYFRAME mode is selected */
2212   if (sink->def_sync_method == GST_SYNC_METHOD_LATEST_KEYFRAME ||
2213       sink->def_sync_method == GST_SYNC_METHOD_BURST_KEYFRAME) {
2214     /* no point in searching beyond the queue length */
2215     gint limit = queuelen;
2216     GstBuffer *buf;
2217
2218     /* no point in searching beyond the soft-max if any. */
2219     if (soft_max_buffers > 0) {
2220       limit = MIN (limit, soft_max_buffers);
2221     }
2222     GST_LOG_OBJECT (sink,
2223         "extending queue to include sync point, now at %d, limit is %d",
2224         max_buffer_usage, limit);
2225     for (i = 0; i < limit; i++) {
2226       buf = g_array_index (sink->bufqueue, GstBuffer *, i);
2227       if (is_sync_frame (sink, buf)) {
2228         /* found a sync frame, now extend the buffer usage to
2229          * include at least this frame. */
2230         max_buffer_usage = MAX (max_buffer_usage, i);
2231         break;
2232       }
2233     }
2234     GST_LOG_OBJECT (sink, "max buffer usage is now %d", max_buffer_usage);
2235   }
2236
2237   GST_LOG_OBJECT (sink, "len %d, usage %d", queuelen, max_buffer_usage);
2238
2239   /* nobody is referencing units after max_buffer_usage so we can
2240    * remove them from the queue. We remove them in reverse order as
2241    * this is the most optimal for GArray. */
2242   for (i = queuelen - 1; i > max_buffer_usage; i--) {
2243     GstBuffer *old;
2244
2245     /* queue exceeded max size */
2246     queuelen--;
2247     old = g_array_index (sink->bufqueue, GstBuffer *, i);
2248     sink->bufqueue = g_array_remove_index (sink->bufqueue, i);
2249
2250     /* unref tail buffer */
2251     gst_buffer_unref (old);
2252   }
2253   /* save for stats */
2254   sink->buffers_queued = max_buffer_usage;
2255   CLIENTS_UNLOCK (sink);
2256 }
2257 #endif
2258
2259 #if 0
2260 /* Handle the clients. This is called when a socket becomes ready
2261  * to read or writable. Badly behaving clients are put on a
2262  * garbage list and removed.
2263  */
2264 static gboolean
2265 gst_multi_handle_sink_socket_condition (GSocket * socket,
2266     GIOCondition condition, GstMultiHandleSink * sink)
2267 {
2268   GList *clink;
2269   GstSocketClient *client;
2270   gboolean ret = TRUE;
2271
2272   CLIENTS_LOCK (sink);
2273   clink = g_hash_table_lookup (sink->socket_hash, socket);
2274   if (clink == NULL) {
2275     ret = FALSE;
2276     goto done;
2277   }
2278
2279   client = clink->data;
2280
2281   if (client->status != GST_CLIENT_STATUS_FLUSHING
2282       && client->status != GST_CLIENT_STATUS_OK) {
2283     gst_multi_handle_sink_remove_client_link (sink, clink);
2284     ret = FALSE;
2285     goto done;
2286   }
2287
2288   if ((condition & G_IO_ERR)) {
2289     GST_WARNING_OBJECT (sink, "Socket %p has error", client->socket);
2290     client->status = GST_CLIENT_STATUS_ERROR;
2291     gst_multi_handle_sink_remove_client_link (sink, clink);
2292     ret = FALSE;
2293     goto done;
2294   } else if ((condition & G_IO_HUP)) {
2295     client->status = GST_CLIENT_STATUS_CLOSED;
2296     gst_multi_handle_sink_remove_client_link (sink, clink);
2297     ret = FALSE;
2298     goto done;
2299   } else if ((condition & G_IO_IN) || (condition & G_IO_PRI)) {
2300     /* handle client read */
2301     if (!gst_multi_handle_sink_handle_client_read (sink, client)) {
2302       gst_multi_handle_sink_remove_client_link (sink, clink);
2303       ret = FALSE;
2304       goto done;
2305     }
2306   } else if ((condition & G_IO_OUT)) {
2307     /* handle client write */
2308     if (!gst_multi_handle_sink_handle_client_write (sink, client)) {
2309       gst_multi_handle_sink_remove_client_link (sink, clink);
2310       ret = FALSE;
2311       goto done;
2312     }
2313   }
2314
2315 done:
2316   CLIENTS_UNLOCK (sink);
2317
2318   return ret;
2319 }
2320 #endif
2321
2322 #if 0
2323 static gboolean
2324 gst_multi_handle_sink_timeout (GstMultiHandleSink * sink)
2325 {
2326   GstClockTime now;
2327   GTimeVal nowtv;
2328   GList *clients;
2329
2330   g_get_current_time (&nowtv);
2331   now = GST_TIMEVAL_TO_TIME (nowtv);
2332
2333   CLIENTS_LOCK (sink);
2334   for (clients = sink->clients; clients; clients = clients->next) {
2335     GstSocketClient *client;
2336
2337     client = clients->data;
2338     if (sink->timeout > 0 && now - client->last_activity_time > sink->timeout) {
2339       client->status = GST_CLIENT_STATUS_SLOW;
2340       gst_multi_handle_sink_remove_client_link (sink, clients);
2341     }
2342   }
2343   CLIENTS_UNLOCK (sink);
2344
2345   return FALSE;
2346 }
2347 #endif
2348
2349 #if 0
2350 /* we handle the client communication in another thread so that we do not block
2351  * the gstreamer thread while we select() on the client fds */
2352 static gpointer
2353 gst_multi_handle_sink_thread (GstMultiHandleSink * sink)
2354 {
2355   GSource *timeout = NULL;
2356
2357   while (sink->running) {
2358     if (sink->timeout > 0) {
2359       timeout = g_timeout_source_new (sink->timeout / GST_MSECOND);
2360
2361       g_source_set_callback (timeout,
2362           (GSourceFunc) gst_multi_handle_sink_timeout, gst_object_ref (sink),
2363           (GDestroyNotify) gst_object_unref);
2364       g_source_attach (timeout, sink->main_context);
2365     }
2366
2367     /* Returns after handling all pending events or when
2368      * _wakeup() was called. In any case we have to add
2369      * a new timeout because something happened.
2370      */
2371     g_main_context_iteration (sink->main_context, TRUE);
2372
2373     if (timeout) {
2374       g_source_destroy (timeout);
2375       g_source_unref (timeout);
2376     }
2377   }
2378
2379   return NULL;
2380 }
2381 #endif
2382
2383 static GstFlowReturn
2384 gst_multi_handle_sink_render (GstBaseSink * bsink, GstBuffer * buf)
2385 {
2386   gboolean in_caps;
2387 #if 0
2388   GstCaps *bufcaps, *padcaps;
2389 #endif
2390
2391   GstMultiHandleSink *sink = GST_MULTI_HANDLE_SINK (bsink);
2392   GstMultiHandleSinkClass *mhsinkclass = GST_MULTI_HANDLE_SINK_GET_CLASS (sink);
2393
2394   g_return_val_if_fail (GST_OBJECT_FLAG_IS_SET (sink,
2395           GST_MULTI_HANDLE_SINK_OPEN), GST_FLOW_FLUSHING);
2396
2397 #if 0
2398   /* since we check every buffer for streamheader caps, we need to make
2399    * sure every buffer has caps set */
2400   bufcaps = gst_buffer_get_caps (buf);
2401   padcaps = GST_PAD_CAPS (GST_BASE_SINK_PAD (bsink));
2402
2403   /* make sure we have caps on the pad */
2404   if (!padcaps && !bufcaps)
2405     goto no_caps;
2406 #endif
2407
2408   /* get IN_CAPS first, code below might mess with the flags */
2409   in_caps = GST_BUFFER_FLAG_IS_SET (buf, GST_BUFFER_FLAG_HEADER);
2410
2411 #if 0
2412   /* stamp the buffer with previous caps if no caps set */
2413   if (!bufcaps) {
2414     if (!gst_buffer_is_writable (buf)) {
2415       /* metadata is not writable, copy will be made and original buffer
2416        * will be unreffed so we need to ref so that we don't lose the
2417        * buffer in the render method. */
2418       gst_buffer_ref (buf);
2419       /* the new buffer is ours only, we keep it out of the scope of this
2420        * function */
2421       buf = gst_buffer_make_writable (buf);
2422     } else {
2423       /* else the metadata is writable, we ref because we keep the buffer
2424        * out of the scope of this method */
2425       gst_buffer_ref (buf);
2426     }
2427     /* buffer metadata is writable now, set the caps */
2428     gst_buffer_set_caps (buf, padcaps);
2429   } else {
2430     gst_caps_unref (bufcaps);
2431
2432     /* since we keep this buffer out of the scope of this method */
2433     gst_buffer_ref (buf);
2434   }
2435 #endif
2436   gst_buffer_ref (buf);
2437
2438   GST_LOG_OBJECT (sink, "received buffer %p, in_caps: %s, offset %"
2439       G_GINT64_FORMAT ", offset_end %" G_GINT64_FORMAT
2440       ", timestamp %" GST_TIME_FORMAT ", duration %" GST_TIME_FORMAT,
2441       buf, in_caps ? "yes" : "no", GST_BUFFER_OFFSET (buf),
2442       GST_BUFFER_OFFSET_END (buf),
2443       GST_TIME_ARGS (GST_BUFFER_TIMESTAMP (buf)),
2444       GST_TIME_ARGS (GST_BUFFER_DURATION (buf)));
2445
2446   /* if we get IN_CAPS buffers, but the previous buffer was not IN_CAPS,
2447    * it means we're getting new streamheader buffers, and we should clear
2448    * the old ones */
2449   if (in_caps && sink->previous_buffer_in_caps == FALSE) {
2450     GST_DEBUG_OBJECT (sink,
2451         "receiving new HEADER buffers, clearing old streamheader");
2452     g_slist_foreach (sink->streamheader, (GFunc) gst_mini_object_unref, NULL);
2453     g_slist_free (sink->streamheader);
2454     sink->streamheader = NULL;
2455   }
2456
2457   /* save the current in_caps */
2458   sink->previous_buffer_in_caps = in_caps;
2459
2460   /* if the incoming buffer is marked as IN CAPS, then we assume for now
2461    * it's a streamheader that needs to be sent to each new client, so we
2462    * put it on our internal list of streamheader buffers.
2463    * FIXME: we could check if the buffer's contents are in fact part of the
2464    * current streamheader.
2465    *
2466    * We don't send the buffer to the client, since streamheaders are sent
2467    * separately when necessary. */
2468   if (in_caps) {
2469     GST_DEBUG_OBJECT (sink, "appending HEADER buffer with length %"
2470         G_GSIZE_FORMAT " to streamheader", gst_buffer_get_size (buf));
2471     sink->streamheader = g_slist_append (sink->streamheader, buf);
2472   } else {
2473     /* queue the buffer, this is a regular data buffer. */
2474     mhsinkclass->queue_buffer (sink, buf);
2475
2476     sink->bytes_to_serve += gst_buffer_get_size (buf);
2477   }
2478   return GST_FLOW_OK;
2479
2480   /* ERRORS */
2481 #if 0
2482 no_caps:
2483   {
2484     GST_ELEMENT_ERROR (sink, CORE, NEGOTIATION, (NULL),
2485         ("Received first buffer without caps set"));
2486     return GST_FLOW_NOT_NEGOTIATED;
2487   }
2488 #endif
2489 }
2490
2491 static void
2492 gst_multi_handle_sink_set_property (GObject * object, guint prop_id,
2493     const GValue * value, GParamSpec * pspec)
2494 {
2495   GstMultiHandleSink *multihandlesink;
2496
2497   multihandlesink = GST_MULTI_HANDLE_SINK (object);
2498
2499   switch (prop_id) {
2500     case PROP_BUFFERS_MAX:
2501       multihandlesink->units_max = g_value_get_int (value);
2502       break;
2503     case PROP_BUFFERS_SOFT_MAX:
2504       multihandlesink->units_soft_max = g_value_get_int (value);
2505       break;
2506     case PROP_TIME_MIN:
2507       multihandlesink->time_min = g_value_get_int64 (value);
2508       break;
2509     case PROP_BYTES_MIN:
2510       multihandlesink->bytes_min = g_value_get_int (value);
2511       break;
2512     case PROP_BUFFERS_MIN:
2513       multihandlesink->buffers_min = g_value_get_int (value);
2514       break;
2515     case PROP_UNIT_FORMAT:
2516       multihandlesink->unit_format = g_value_get_enum (value);
2517       break;
2518     case PROP_UNITS_MAX:
2519       multihandlesink->units_max = g_value_get_int64 (value);
2520       break;
2521     case PROP_UNITS_SOFT_MAX:
2522       multihandlesink->units_soft_max = g_value_get_int64 (value);
2523       break;
2524     case PROP_RECOVER_POLICY:
2525       multihandlesink->recover_policy = g_value_get_enum (value);
2526       break;
2527     case PROP_TIMEOUT:
2528       multihandlesink->timeout = g_value_get_uint64 (value);
2529       break;
2530     case PROP_SYNC_METHOD:
2531       multihandlesink->def_sync_method = g_value_get_enum (value);
2532       break;
2533     case PROP_BURST_FORMAT:
2534       multihandlesink->def_burst_format = g_value_get_enum (value);
2535       break;
2536     case PROP_BURST_VALUE:
2537       multihandlesink->def_burst_value = g_value_get_uint64 (value);
2538       break;
2539     case PROP_QOS_DSCP:
2540       multihandlesink->qos_dscp = g_value_get_int (value);
2541       gst_multi_handle_sink_setup_dscp (multihandlesink);
2542       break;
2543
2544     case PROP_RESEND_STREAMHEADER:
2545       multihandlesink->resend_streamheader = g_value_get_boolean (value);
2546       break;
2547
2548     default:
2549       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2550       break;
2551   }
2552 }
2553
2554 static void
2555 gst_multi_handle_sink_get_property (GObject * object, guint prop_id,
2556     GValue * value, GParamSpec * pspec)
2557 {
2558   GstMultiHandleSink *multihandlesink;
2559
2560   multihandlesink = GST_MULTI_HANDLE_SINK (object);
2561
2562   switch (prop_id) {
2563     case PROP_BUFFERS_MAX:
2564       g_value_set_int (value, multihandlesink->units_max);
2565       break;
2566     case PROP_BUFFERS_SOFT_MAX:
2567       g_value_set_int (value, multihandlesink->units_soft_max);
2568       break;
2569     case PROP_TIME_MIN:
2570       g_value_set_int64 (value, multihandlesink->time_min);
2571       break;
2572     case PROP_BYTES_MIN:
2573       g_value_set_int (value, multihandlesink->bytes_min);
2574       break;
2575     case PROP_BUFFERS_MIN:
2576       g_value_set_int (value, multihandlesink->buffers_min);
2577       break;
2578     case PROP_BUFFERS_QUEUED:
2579       g_value_set_uint (value, multihandlesink->buffers_queued);
2580       break;
2581     case PROP_BYTES_QUEUED:
2582       g_value_set_uint (value, multihandlesink->bytes_queued);
2583       break;
2584     case PROP_TIME_QUEUED:
2585       g_value_set_uint64 (value, multihandlesink->time_queued);
2586       break;
2587     case PROP_UNIT_FORMAT:
2588       g_value_set_enum (value, multihandlesink->unit_format);
2589       break;
2590     case PROP_UNITS_MAX:
2591       g_value_set_int64 (value, multihandlesink->units_max);
2592       break;
2593     case PROP_UNITS_SOFT_MAX:
2594       g_value_set_int64 (value, multihandlesink->units_soft_max);
2595       break;
2596     case PROP_RECOVER_POLICY:
2597       g_value_set_enum (value, multihandlesink->recover_policy);
2598       break;
2599     case PROP_TIMEOUT:
2600       g_value_set_uint64 (value, multihandlesink->timeout);
2601       break;
2602     case PROP_SYNC_METHOD:
2603       g_value_set_enum (value, multihandlesink->def_sync_method);
2604       break;
2605     case PROP_BYTES_TO_SERVE:
2606       g_value_set_uint64 (value, multihandlesink->bytes_to_serve);
2607       break;
2608     case PROP_BYTES_SERVED:
2609       g_value_set_uint64 (value, multihandlesink->bytes_served);
2610       break;
2611     case PROP_BURST_FORMAT:
2612       g_value_set_enum (value, multihandlesink->def_burst_format);
2613       break;
2614     case PROP_BURST_VALUE:
2615       g_value_set_uint64 (value, multihandlesink->def_burst_value);
2616       break;
2617     case PROP_QOS_DSCP:
2618       g_value_set_int (value, multihandlesink->qos_dscp);
2619       break;
2620     case PROP_RESEND_STREAMHEADER:
2621       g_value_set_boolean (value, multihandlesink->resend_streamheader);
2622       break;
2623 #if 0
2624     case PROP_NUM_SOCKETS:
2625       g_value_set_uint (value,
2626           g_hash_table_size (multihandlesink->socket_hash));
2627       break;
2628 #endif
2629     default:
2630       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
2631       break;
2632   }
2633 }
2634
2635 /* create a socket for sending to remote machine */
2636 gboolean
2637 gst_multi_handle_sink_start (GstBaseSink * bsink)
2638 {
2639   GstMultiHandleSinkClass *mhsclass;
2640   GstMultiHandleSink *mhsink;
2641
2642   if (GST_OBJECT_FLAG_IS_SET (bsink, GST_MULTI_HANDLE_SINK_OPEN))
2643     return TRUE;
2644
2645   mhsink = GST_MULTI_HANDLE_SINK (bsink);
2646   mhsclass = GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
2647
2648   if (!mhsclass->start_pre (mhsink))
2649     return FALSE;
2650
2651   mhsink->streamheader = NULL;
2652   mhsink->bytes_to_serve = 0;
2653   mhsink->bytes_served = 0;
2654
2655   if (mhsclass->init) {
2656     mhsclass->init (mhsink);
2657   }
2658
2659   mhsink->running = TRUE;
2660
2661   mhsink->thread = g_thread_new ("multihandlesink",
2662       (GThreadFunc) mhsclass->thread, mhsink);
2663
2664   GST_OBJECT_FLAG_SET (bsink, GST_MULTI_HANDLE_SINK_OPEN);
2665
2666   return TRUE;
2667 }
2668
2669 #if 0
2670 /* create a socket for sending to remote machine */
2671 static gboolean
2672 gst_multi_handle_sink_start (GstBaseSink * bsink)
2673 {
2674   GstMultiHandleSinkClass *fclass;
2675   GstMultiHandleSink *this;
2676   GList *clients;
2677
2678   if (GST_OBJECT_FLAG_IS_SET (bsink, GST_MULTI_HANDLE_SINK_OPEN))
2679     return TRUE;
2680
2681   this = GST_MULTI_HANDLE_SINK (bsink);
2682   fclass = GST_MULTI_HANDLE_SINK_GET_CLASS (this);
2683
2684   GST_INFO_OBJECT (this, "starting");
2685
2686   this->main_context = g_main_context_new ();
2687
2688   CLIENTS_LOCK (this);
2689   for (clients = this->clients; clients; clients = clients->next) {
2690     GstSocketClient *client;
2691
2692     client = clients->data;
2693     if (client->source)
2694       continue;
2695     client->source =
2696         g_socket_create_source (client->socket,
2697         G_IO_IN | G_IO_OUT | G_IO_PRI | G_IO_ERR | G_IO_HUP, this->cancellable);
2698     g_source_set_callback (client->source,
2699         (GSourceFunc) gst_multi_handle_sink_socket_condition,
2700         gst_object_ref (this), (GDestroyNotify) gst_object_unref);
2701     g_source_attach (client->source, this->main_context);
2702   }
2703   CLIENTS_UNLOCK (this);
2704
2705   this->streamheader = NULL;
2706   this->bytes_to_serve = 0;
2707   this->bytes_served = 0;
2708
2709   if (fclass->init) {
2710     fclass->init (this);
2711   }
2712
2713   this->running = TRUE;
2714
2715   this->thread = g_thread_new ("multisocketsink",
2716       (GThreadFunc) gst_multi_handle_sink_thread, this);
2717
2718   GST_OBJECT_FLAG_SET (this, GST_MULTI_HANDLE_SINK_OPEN);
2719
2720   return TRUE;
2721 }
2722
2723 static gboolean
2724 multisocketsink_hash_remove (gpointer key, gpointer value, gpointer data)
2725 {
2726   return TRUE;
2727 }
2728 #endif
2729
2730 // FIXME: privatize again
2731 gboolean
2732 gst_multi_handle_sink_stop (GstBaseSink * bsink)
2733 {
2734   GstMultiHandleSinkClass *mhclass;
2735   GstBuffer *buf;
2736   gint i;
2737   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (bsink);
2738
2739   mhclass = GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
2740
2741   if (!GST_OBJECT_FLAG_IS_SET (bsink, GST_MULTI_HANDLE_SINK_OPEN))
2742     return TRUE;
2743
2744   mhsink->running = FALSE;
2745
2746   mhclass->stop_pre (mhsink);
2747
2748   if (mhsink->thread) {
2749     GST_DEBUG_OBJECT (mhsink, "joining thread");
2750     g_thread_join (mhsink->thread);
2751     GST_DEBUG_OBJECT (mhsink, "joined thread");
2752     mhsink->thread = NULL;
2753   }
2754
2755   /* free the clients */
2756   mhclass->clear (GST_MULTI_HANDLE_SINK (mhsink));
2757
2758   if (mhsink->streamheader) {
2759     g_slist_foreach (mhsink->streamheader, (GFunc) gst_mini_object_unref, NULL);
2760     g_slist_free (mhsink->streamheader);
2761     mhsink->streamheader = NULL;
2762   }
2763
2764   if (mhclass->close)
2765     mhclass->close (mhsink);
2766
2767   mhclass->stop_post (mhsink);
2768
2769   /* remove all queued buffers */
2770   if (mhsink->bufqueue) {
2771     GST_DEBUG_OBJECT (mhsink, "Emptying bufqueue with %d buffers",
2772         mhsink->bufqueue->len);
2773     for (i = mhsink->bufqueue->len - 1; i >= 0; --i) {
2774       buf = g_array_index (mhsink->bufqueue, GstBuffer *, i);
2775       GST_LOG_OBJECT (mhsink, "Removing buffer %p (%d) with refcount %d", buf,
2776           i, GST_MINI_OBJECT_REFCOUNT (buf));
2777       gst_buffer_unref (buf);
2778       mhsink->bufqueue = g_array_remove_index (mhsink->bufqueue, i);
2779     }
2780     /* freeing the array is done in _finalize */
2781   }
2782   GST_OBJECT_FLAG_UNSET (mhsink, GST_MULTI_HANDLE_SINK_OPEN);
2783
2784   return TRUE;
2785 }
2786
2787
2788 #if 0
2789 static gboolean
2790 gst_multi_handle_sink_stop (GstBaseSink * bsink)
2791 {
2792   GstMultiHandleSinkClass *fclass;
2793   GstMultiHandleSink *this;
2794   GstBuffer *buf;
2795   gint i;
2796
2797   this = GST_MULTI_HANDLE_SINK (bsink);
2798   fclass = GST_MULTI_HANDLE_SINK_GET_CLASS (this);
2799
2800   if (!GST_OBJECT_FLAG_IS_SET (bsink, GST_MULTI_HANDLE_SINK_OPEN))
2801     return TRUE;
2802
2803   this->running = FALSE;
2804
2805   if (this->main_context)
2806     g_main_context_wakeup (this->main_context);
2807
2808   if (this->thread) {
2809     GST_DEBUG_OBJECT (this, "joining thread");
2810     g_thread_join (this->thread);
2811     GST_DEBUG_OBJECT (this, "joined thread");
2812     this->thread = NULL;
2813   }
2814
2815   /* free the clients */
2816   fclass->clear (this);
2817
2818   if (this->streamheader) {
2819     g_slist_foreach (this->streamheader, (GFunc) gst_mini_object_unref, NULL);
2820     g_slist_free (this->streamheader);
2821     this->streamheader = NULL;
2822   }
2823
2824   if (fclass->close)
2825     fclass->close (this);
2826
2827   if (this->main_context) {
2828     g_main_context_unref (this->main_context);
2829     this->main_context = NULL;
2830   }
2831
2832   g_hash_table_foreach_remove (this->socket_hash, multisocketsink_hash_remove,
2833       this);
2834
2835   /* remove all queued buffers */
2836   if (this->bufqueue) {
2837     GST_DEBUG_OBJECT (this, "Emptying bufqueue with %d buffers",
2838         this->bufqueue->len);
2839     for (i = this->bufqueue->len - 1; i >= 0; --i) {
2840       buf = g_array_index (this->bufqueue, GstBuffer *, i);
2841       GST_LOG_OBJECT (this, "Removing buffer %p (%d) with refcount %d", buf, i,
2842           GST_MINI_OBJECT_REFCOUNT (buf));
2843       gst_buffer_unref (buf);
2844       this->bufqueue = g_array_remove_index (this->bufqueue, i);
2845     }
2846     /* freeing the array is done in _finalize */
2847   }
2848   GST_OBJECT_FLAG_UNSET (this, GST_MULTI_HANDLE_SINK_OPEN);
2849
2850   return TRUE;
2851 }
2852 #endif
2853
2854 static GstStateChangeReturn
2855 gst_multi_handle_sink_change_state (GstElement * element,
2856     GstStateChange transition)
2857 {
2858   GstMultiHandleSink *sink;
2859   GstStateChangeReturn ret;
2860
2861   sink = GST_MULTI_HANDLE_SINK (element);
2862
2863   /* we disallow changing the state from the streaming thread */
2864   if (g_thread_self () == sink->thread)
2865     return GST_STATE_CHANGE_FAILURE;
2866
2867
2868   switch (transition) {
2869     case GST_STATE_CHANGE_NULL_TO_READY:
2870       if (!gst_multi_handle_sink_start (GST_BASE_SINK (sink)))
2871         goto start_failed;
2872       break;
2873     case GST_STATE_CHANGE_READY_TO_PAUSED:
2874       break;
2875     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
2876       break;
2877     default:
2878       break;
2879   }
2880
2881   ret = GST_ELEMENT_CLASS (parent_class)->change_state (element, transition);
2882
2883   switch (transition) {
2884     case GST_STATE_CHANGE_PLAYING_TO_PAUSED:
2885       break;
2886     case GST_STATE_CHANGE_PAUSED_TO_READY:
2887       break;
2888     case GST_STATE_CHANGE_READY_TO_NULL:
2889       gst_multi_handle_sink_stop (GST_BASE_SINK (sink));
2890       break;
2891     default:
2892       break;
2893   }
2894   return ret;
2895
2896   /* ERRORS */
2897 start_failed:
2898   {
2899     /* error message was posted */
2900     return GST_STATE_CHANGE_FAILURE;
2901   }
2902 }
2903
2904 #if 0
2905 static gboolean
2906 gst_multi_handle_sink_unlock (GstBaseSink * bsink)
2907 {
2908   GstMultiHandleSink *sink;
2909
2910   sink = GST_MULTI_HANDLE_SINK (bsink);
2911
2912   GST_DEBUG_OBJECT (sink, "set to flushing");
2913   g_cancellable_cancel (sink->cancellable);
2914   if (sink->main_context)
2915     g_main_context_wakeup (sink->main_context);
2916
2917   return TRUE;
2918 }
2919
2920 /* will be called only between calls to start() and stop() */
2921 static gboolean
2922 gst_multi_handle_sink_unlock_stop (GstBaseSink * bsink)
2923 {
2924   GstMultiHandleSink *sink;
2925
2926   sink = GST_MULTI_HANDLE_SINK (bsink);
2927
2928   GST_DEBUG_OBJECT (sink, "unset flushing");
2929   g_cancellable_reset (sink->cancellable);
2930
2931   return TRUE;
2932 }
2933 #endif