gst/tcp/gstmultifdsink.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  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Library General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Library General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Library General Public
  17  * License along with this library; if not, write to the
  18  * Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
  19  * Boston, MA 02110-1301, USA.
  20  */
  21
  22 /**
  23  * SECTION:element-multifdsink
  24  * @see_also: tcpserversink
  25  *
  26  * This plugin writes incoming data to a set of file descriptors. The
  27  * file descriptors can be added to multifdsink by emitting the #GstMultiFdSink::add signal.
  28  * For each descriptor added, the #GstMultiFdSink::client-added signal will be called.
  29  *
  30  * The multifdsink element needs to be set into READY, PAUSED or PLAYING state
  31  * before operations such as adding clients are possible.
  32  *
  33  * As of version 0.10.8, a client can also be added with the #GstMultiFdSink::add-full signal
  34  * that allows for more control over what and how much data a client
  35  * initially receives.
  36  *
  37  * Clients can be removed from multifdsink by emitting the #GstMultiFdSink::remove signal. For
  38  * each descriptor removed, the #GstMultiFdSink::client-removed signal will be called. The
  39  * #GstMultiFdSink::client-removed signal can also be fired when multifdsink decides that a
  40  * client is not active anymore or, depending on the value of the
  41  * #GstMultiFdSink:recover-policy property, if the client is reading too slowly.
  42  * In all cases, multifdsink will never close a file descriptor itself.
  43  * The user of multifdsink is responsible for closing all file descriptors.
  44  * This can for example be done in response to the #GstMultiFdSink::client-fd-removed signal.
  45  * Note that multifdsink still has a reference to the file descriptor when the
  46  * #GstMultiFdSink::client-removed signal is emitted, so that "get-stats" can be performed on
  47  * the descriptor; it is therefore not safe to close the file descriptor in
  48  * the #GstMultiFdSink::client-removed signal handler, and you should use the
  49  * #GstMultiFdSink::client-fd-removed signal to safely close the fd.
  50  *
  51  * Multifdsink internally keeps a queue of the incoming buffers and uses a
  52  * separate thread to send the buffers to the clients. This ensures that no
  53  * client write can block the pipeline and that clients can read with different
  54  * speeds.
  55  *
  56  * When adding a client to multifdsink, the #GstMultiFdSink:sync-method property will define
  57  * which buffer in the queued buffers will be sent first to the client. Clients
  58  * can be sent the most recent buffer (which might not be decodable by the
  59  * client if it is not a keyframe), the next keyframe received in
  60  * multifdsink (which can take some time depending on the keyframe rate), or the
  61  * last received keyframe (which will cause a simple burst-on-connect).
  62  * Multifdsink will always keep at least one keyframe in its internal buffers
  63  * when the sync-mode is set to latest-keyframe.
  64  *
  65  * As of version 0.10.8, there are additional values for the #GstMultiFdSink:sync-method
  66  * property to allow finer control over burst-on-connect behaviour. By selecting
  67  * the 'burst' method a minimum burst size can be chosen, 'burst-keyframe'
  68  * additionally requires that the burst begin with a keyframe, and
  69  * 'burst-with-keyframe' attempts to burst beginning with a keyframe, but will
  70  * prefer a minimum burst size even if it requires not starting with a keyframe.
  71  *
  72  * Multifdsink can be instructed to keep at least a minimum amount of data
  73  * expressed in time or byte units in its internal queues with the
  74  * #GstMultiFdSink:time-min and #GstMultiFdSink:bytes-min properties respectively.
  75  * These properties are useful if the application adds clients with the
  76  * #GstMultiFdSink::add-full signal to make sure that a burst connect can
  77  * actually be honored.
  78  *
  79  * When streaming data, clients are allowed to read at a different rate than
  80  * the rate at which multifdsink receives data. If the client is reading too
  81  * fast, no data will be send to the client until multifdsink receives more
  82  * data. If the client, however, reads too slowly, data for that client will be
  83  * queued up in multifdsink. Two properties control the amount of data
  84  * (buffers) that is queued in multifdsink: #GstMultiFdSink:buffers-max and
  85  * #GstMultiFdSink:buffers-soft-max. A client that falls behind by
  86  * #GstMultiFdSink:buffers-max is removed from multifdsink forcibly.
  87  *
  88  * A client with a lag of at least #GstMultiFdSink:buffers-soft-max enters the recovery
  89  * procedure which is controlled with the #GstMultiFdSink:recover-policy property.
  90  * A recover policy of NONE will do nothing, RESYNC_LATEST will send the most recently
  91  * received buffer as the next buffer for the client, RESYNC_SOFT_LIMIT
  92  * positions the client to the soft limit in the buffer queue and
  93  * RESYNC_KEYFRAME positions the client at the most recent keyframe in the
  94  * buffer queue.
  95  *
  96  * multifdsink will by default synchronize on the clock before serving the
  97  * buffers to the clients. This behaviour can be disabled by setting the sync
  98  * property to FALSE. Multifdsink will by default not do QoS and will never
  99  * drop late buffers.
 100  *
 101  * Last reviewed on 2006-09-12 (0.10.10)
 102  */
 103
 104 #ifdef HAVE_CONFIG_H
 105 #include "config.h"
 106 #endif
 107
 108 #include <gst/gst-i18n-plugin.h>
 109
 110 #include <sys/ioctl.h>
 111
 112 #ifdef HAVE_UNISTD_H
 113 #include <unistd.h>
 114 #endif
 115
 116 #include <string.h>
 117 #include <fcntl.h>
 118 #include <sys/types.h>
 119 #include <sys/socket.h>
 120 #include <sys/stat.h>
 121 #include <netinet/in.h>
 122
 123 #ifdef HAVE_FIONREAD_IN_SYS_FILIO
 124 #include <sys/filio.h>
 125 #endif
 126
 127 #include "gstmultifdsink.h"
 128 #include "gsttcp-marshal.h"
 129
 130 #define NOT_IMPLEMENTED 0
 131
 132 GST_DEBUG_CATEGORY_STATIC (multifdsink_debug);
 133 #define GST_CAT_DEFAULT (multifdsink_debug)
 134
 135 /* MultiFdSink signals and args */
 136 enum
 137 {
 138   /* methods */
 139   SIGNAL_ADD,
 140   SIGNAL_ADD_BURST,
 141   SIGNAL_REMOVE,
 142   SIGNAL_REMOVE_FLUSH,
 143   SIGNAL_GET_STATS,
 144
 145   /* signals */
 146   SIGNAL_CLIENT_ADDED,
 147   SIGNAL_CLIENT_REMOVED,
 148   SIGNAL_CLIENT_FD_REMOVED,
 149
 150   LAST_SIGNAL
 151 };
 152
 153 /* this is really arbitrarily chosen */
 154 #define DEFAULT_HANDLE_READ             TRUE
 155
 156 enum
 157 {
 158   PROP_0,
 159   PROP_HANDLE_READ,
 160   PROP_LAST
 161 };
 162
 163 static void gst_multi_fd_sink_stop_pre (GstMultiHandleSink * mhsink);
 164 static void gst_multi_fd_sink_stop_post (GstMultiHandleSink * mhsink);
 165 static gboolean gst_multi_fd_sink_start_pre (GstMultiHandleSink * mhsink);
 166 static gpointer gst_multi_fd_sink_thread (GstMultiHandleSink * mhsink);
 167
 168 static void gst_multi_fd_sink_add (GstMultiFdSink * sink, int fd);
 169 static void gst_multi_fd_sink_add_full (GstMultiFdSink * sink, int fd,
 170     GstSyncMethod sync, GstFormat min_format, guint64 min_value,
 171     GstFormat max_format, guint64 max_value);
 172 static void gst_multi_fd_sink_remove (GstMultiFdSink * sink, int fd);
 173 static void gst_multi_fd_sink_remove_flush (GstMultiFdSink * sink, int fd);
 174 static GstStructure *gst_multi_fd_sink_get_stats (GstMultiFdSink * sink,
 175     int fd);
 176
 177 static void gst_multi_fd_sink_emit_client_added (GstMultiHandleSink * mhsink,
 178     GstMultiSinkHandle handle);
 179 static void gst_multi_fd_sink_emit_client_removed (GstMultiHandleSink * mhsink,
 180     GstMultiSinkHandle handle, GstClientStatus status);
 181
 182 static GstMultiHandleClient *gst_multi_fd_sink_new_client (GstMultiHandleSink *
 183     mhsink, GstMultiSinkHandle handle, GstSyncMethod sync_method);
 184 static void gst_multi_fd_sink_client_free (GstMultiHandleSink * m,
 185     GstMultiHandleClient * client);
 186 static int gst_multi_fd_sink_client_get_fd (GstMultiHandleClient * client);
 187 static void gst_multi_fd_sink_handle_debug (GstMultiSinkHandle handle,
 188     gchar debug[30]);
 189 static gpointer gst_multi_fd_sink_handle_hash_key (GstMultiSinkHandle handle);
 190 static void gst_multi_fd_sink_hash_adding (GstMultiHandleSink * mhsink,
 191     GstMultiHandleClient * mhclient);
 192 static void gst_multi_fd_sink_hash_removing (GstMultiHandleSink * mhsink,
 193     GstMultiHandleClient * mhclient);
 194
 195 static void gst_multi_fd_sink_hash_changed (GstMultiHandleSink * mhsink);
 196
 197 static void gst_multi_fd_sink_set_property (GObject * object, guint prop_id,
 198     const GValue * value, GParamSpec * pspec);
 199 static void gst_multi_fd_sink_get_property (GObject * object, guint prop_id,
 200     GValue * value, GParamSpec * pspec);
 201
 202 #define gst_multi_fd_sink_parent_class parent_class
 203 G_DEFINE_TYPE (GstMultiFdSink, gst_multi_fd_sink, GST_TYPE_MULTI_HANDLE_SINK);
 204
 205 static guint gst_multi_fd_sink_signals[LAST_SIGNAL] = { 0 };
 206
 207 static void
 208 gst_multi_fd_sink_class_init (GstMultiFdSinkClass * klass)
 209 {
 210   GObjectClass *gobject_class;
 211   GstElementClass *gstelement_class;
 212   GstMultiHandleSinkClass *gstmultihandlesink_class;
 213
 214   gobject_class = (GObjectClass *) klass;
 215   gstelement_class = (GstElementClass *) klass;
 216   gstmultihandlesink_class = (GstMultiHandleSinkClass *) klass;
 217
 218   gobject_class->set_property = gst_multi_fd_sink_set_property;
 219   gobject_class->get_property = gst_multi_fd_sink_get_property;
 220
 221   /**
 222    * GstMultiFdSink::handle-read
 223    *
 224    * Handle read requests from clients and discard the data.
 225    *
 226    * Since: 0.10.23
 227    */
 228   g_object_class_install_property (gobject_class, PROP_HANDLE_READ,
 229       g_param_spec_boolean ("handle-read", "Handle Read",
 230           "Handle client reads and discard the data",
 231           DEFAULT_HANDLE_READ, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 232
 233   /**
 234    * GstMultiFdSink::add:
 235    * @gstmultifdsink: the multifdsink element to emit this signal on
 236    * @fd:             the file descriptor to add to multifdsink
 237    *
 238    * Hand the given open file descriptor to multifdsink to write to.
 239    */
 240   gst_multi_fd_sink_signals[SIGNAL_ADD] =
 241       g_signal_new ("add", G_TYPE_FROM_CLASS (klass),
 242       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 243       G_STRUCT_OFFSET (GstMultiFdSinkClass, add), NULL, NULL,
 244       g_cclosure_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT);
 245   /**
 246    * GstMultiFdSink::add-full:
 247    * @gstmultifdsink:  the multifdsink element to emit this signal on
 248    * @fd:              the file descriptor to add to multifdsink
 249    * @sync:            the sync method to use
 250    * @format_min:      the format of @value_min
 251    * @value_min:       the minimum amount of data to burst expressed in
 252    *                   @format_min units.
 253    * @format_max:      the format of @value_max
 254    * @value_max:       the maximum amount of data to burst expressed in
 255    *                   @format_max units.
 256    *
 257    * Hand the given open file descriptor to multifdsink to write to and
 258    * specify the burst parameters for the new connection.
 259    */
 260   gst_multi_fd_sink_signals[SIGNAL_ADD_BURST] =
 261       g_signal_new ("add-full", G_TYPE_FROM_CLASS (klass),
 262       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 263       G_STRUCT_OFFSET (GstMultiFdSinkClass, add_full), NULL, NULL,
 264       gst_tcp_marshal_VOID__INT_ENUM_INT_UINT64_INT_UINT64, G_TYPE_NONE, 6,
 265       G_TYPE_INT, GST_TYPE_SYNC_METHOD, GST_TYPE_FORMAT, G_TYPE_UINT64,
 266       GST_TYPE_FORMAT, G_TYPE_UINT64);
 267   /**
 268    * GstMultiFdSink::remove:
 269    * @gstmultifdsink: the multifdsink element to emit this signal on
 270    * @fd:             the file descriptor to remove from multifdsink
 271    *
 272    * Remove the given open file descriptor from multifdsink.
 273    */
 274   gst_multi_fd_sink_signals[SIGNAL_REMOVE] =
 275       g_signal_new ("remove", G_TYPE_FROM_CLASS (klass),
 276       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 277       G_STRUCT_OFFSET (GstMultiFdSinkClass, remove), NULL, NULL,
 278       gst_tcp_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT);
 279   /**
 280    * GstMultiFdSink::remove-flush:
 281    * @gstmultifdsink: the multifdsink element to emit this signal on
 282    * @fd:             the file descriptor to remove from multifdsink
 283    *
 284    * Remove the given open file descriptor from multifdsink after flushing all
 285    * the pending data to the fd.
 286    */
 287   gst_multi_fd_sink_signals[SIGNAL_REMOVE_FLUSH] =
 288       g_signal_new ("remove-flush", G_TYPE_FROM_CLASS (klass),
 289       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 290       G_STRUCT_OFFSET (GstMultiFdSinkClass, remove_flush), NULL, NULL,
 291       gst_tcp_marshal_VOID__INT, G_TYPE_NONE, 1, G_TYPE_INT);
 292
 293   /**
 294    * GstMultiFdSink::get-stats:
 295    * @gstmultifdsink: the multifdsink element to emit this signal on
 296    * @fd:             the file descriptor to get stats of from multifdsink
 297    *
 298    * Get statistics about @fd. This function returns a GValueArray to ease
 299    * automatic wrapping for bindings.
 300    *
 301    * Returns: a GValueArray with the statistics. The array contains guint64
 302    *     values that represent respectively: total number of bytes sent, time
 303    *     when the client was added, time when the client was
 304    *     disconnected/removed, time the client is/was active, last activity
 305    *     time (in epoch seconds), number of buffers dropped.
 306    *     All times are expressed in nanoseconds (GstClockTime).
 307    *     The array can be 0-length if the client was not found.
 308    */
 309   gst_multi_fd_sink_signals[SIGNAL_GET_STATS] =
 310       g_signal_new ("get-stats", G_TYPE_FROM_CLASS (klass),
 311       G_SIGNAL_RUN_LAST | G_SIGNAL_ACTION,
 312       G_STRUCT_OFFSET (GstMultiFdSinkClass, get_stats), NULL, NULL,
 313       gst_tcp_marshal_BOXED__INT, GST_TYPE_STRUCTURE, 1, G_TYPE_INT);
 314
 315   /**
 316    * GstMultiFdSink::client-added:
 317    * @gstmultifdsink: the multifdsink element that emitted this signal
 318    * @fd:             the file descriptor that was added to multifdsink
 319    *
 320    * The given file descriptor was added to multifdsink. This signal will
 321    * be emitted from the streaming thread so application should be prepared
 322    * for that.
 323    */
 324   gst_multi_fd_sink_signals[SIGNAL_CLIENT_ADDED] =
 325       g_signal_new ("client-added", G_TYPE_FROM_CLASS (klass),
 326       G_SIGNAL_RUN_LAST, 0, NULL, NULL, gst_tcp_marshal_VOID__INT, G_TYPE_NONE,
 327       1, G_TYPE_INT);
 328   /**
 329    * GstMultiFdSink::client-removed:
 330    * @gstmultifdsink: the multifdsink element that emitted this signal
 331    * @fd:             the file descriptor that is to be removed from multifdsink
 332    * @status:         the reason why the client was removed
 333    *
 334    * The given file descriptor is about to be removed from multifdsink. This
 335    * signal will be emitted from the streaming thread so applications should
 336    * be prepared for that.
 337    *
 338    * @gstmultifdsink still holds a handle to @fd so it is possible to call
 339    * the get-stats signal from this callback. For the same reason it is
 340    * not safe to close() and reuse @fd in this callback.
 341    */
 342   gst_multi_fd_sink_signals[SIGNAL_CLIENT_REMOVED] =
 343       g_signal_new ("client-removed", G_TYPE_FROM_CLASS (klass),
 344       G_SIGNAL_RUN_LAST, 0, NULL, NULL, gst_tcp_marshal_VOID__INT_ENUM,
 345       G_TYPE_NONE, 2, G_TYPE_INT, GST_TYPE_CLIENT_STATUS);
 346   /**
 347    * GstMultiFdSink::client-fd-removed:
 348    * @gstmultifdsink: the multifdsink element that emitted this signal
 349    * @fd:             the file descriptor that was removed from multifdsink
 350    *
 351    * The given file descriptor was removed from multifdsink. This signal will
 352    * be emitted from the streaming thread so applications should be prepared
 353    * for that.
 354    *
 355    * In this callback, @gstmultifdsink has removed all the information
 356    * associated with @fd and it is therefore not possible to call get-stats
 357    * with @fd. It is however safe to close() and reuse @fd in the callback.
 358    *
 359    * Since: 0.10.7
 360    */
 361   gst_multi_fd_sink_signals[SIGNAL_CLIENT_FD_REMOVED] =
 362       g_signal_new ("client-fd-removed", G_TYPE_FROM_CLASS (klass),
 363       G_SIGNAL_RUN_LAST, 0, NULL, NULL, gst_tcp_marshal_VOID__INT,
 364       G_TYPE_NONE, 1, G_TYPE_INT);
 365
 366   gst_element_class_set_static_metadata (gstelement_class,
 367       "Multi filedescriptor sink", "Sink/Network",
 368       "Send data to multiple filedescriptors",
 369       "Thomas Vander Stichele <thomas at apestaart dot org>, "
 370       "Wim Taymans <wim@fluendo.com>");
 371
 372   klass->add = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_add);
 373   klass->add_full = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_add_full);
 374   klass->remove = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_remove);
 375   klass->remove_flush = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_remove_flush);
 376   klass->get_stats = GST_DEBUG_FUNCPTR (gst_multi_fd_sink_get_stats);
 377
 378   gstmultihandlesink_class->emit_client_added =
 379       gst_multi_fd_sink_emit_client_added;
 380   gstmultihandlesink_class->emit_client_removed =
 381       gst_multi_fd_sink_emit_client_removed;
 382
 383   gstmultihandlesink_class->stop_pre =
 384       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_stop_pre);
 385   gstmultihandlesink_class->stop_post =
 386       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_stop_post);
 387   gstmultihandlesink_class->start_pre =
 388       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_start_pre);
 389   gstmultihandlesink_class->thread =
 390       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_thread);
 391   gstmultihandlesink_class->new_client =
 392       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_new_client);
 393   gstmultihandlesink_class->client_free = gst_multi_fd_sink_client_free;
 394   gstmultihandlesink_class->client_get_fd =
 395       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_client_get_fd);
 396   gstmultihandlesink_class->handle_debug =
 397       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_handle_debug);
 398   gstmultihandlesink_class->handle_hash_key =
 399       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_handle_hash_key);
 400   gstmultihandlesink_class->hash_changed =
 401       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_hash_changed);
 402   gstmultihandlesink_class->hash_adding =
 403       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_hash_adding);
 404   gstmultihandlesink_class->hash_removing =
 405       GST_DEBUG_FUNCPTR (gst_multi_fd_sink_hash_removing);
 406
 407   GST_DEBUG_CATEGORY_INIT (multifdsink_debug, "multifdsink", 0, "FD sink");
 408 }
 409
 410 static void
 411 gst_multi_fd_sink_init (GstMultiFdSink * this)
 412 {
 413   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (this);
 414
 415   mhsink->handle_hash = g_hash_table_new (g_direct_hash, g_direct_equal);
 416
 417   this->handle_read = DEFAULT_HANDLE_READ;
 418 }
 419
 420 /* methods to emit signals */
 421
 422 static void
 423 gst_multi_fd_sink_emit_client_added (GstMultiHandleSink * mhsink,
 424     GstMultiSinkHandle handle)
 425 {
 426   g_signal_emit (mhsink, gst_multi_fd_sink_signals[SIGNAL_CLIENT_ADDED], 0,
 427       handle.fd);
 428 }
 429
 430 static void
 431 gst_multi_fd_sink_emit_client_removed (GstMultiHandleSink * mhsink,
 432     GstMultiSinkHandle handle, GstClientStatus status)
 433 {
 434   g_signal_emit (mhsink, gst_multi_fd_sink_signals[SIGNAL_CLIENT_REMOVED], 0,
 435       handle.fd, status);
 436 }
 437
 438 static void
 439 gst_multi_fd_sink_client_free (GstMultiHandleSink * mhsink,
 440     GstMultiHandleClient * client)
 441 {
 442   g_signal_emit (mhsink, gst_multi_fd_sink_signals[SIGNAL_CLIENT_FD_REMOVED],
 443       0, client->handle.fd);
 444 }
 445
 446 /* action signals */
 447
 448 static void
 449 gst_multi_fd_sink_add (GstMultiFdSink * sink, int fd)
 450 {
 451   GstMultiSinkHandle handle;
 452
 453   handle.fd = fd;
 454   gst_multi_handle_sink_add (GST_MULTI_HANDLE_SINK_CAST (sink), handle);
 455 }
 456
 457 static void
 458 gst_multi_fd_sink_add_full (GstMultiFdSink * sink, int fd,
 459     GstSyncMethod sync, GstFormat min_format, guint64 min_value,
 460     GstFormat max_format, guint64 max_value)
 461 {
 462   GstMultiSinkHandle handle;
 463
 464   handle.fd = fd;
 465   gst_multi_handle_sink_add_full (GST_MULTI_HANDLE_SINK_CAST (sink), handle,
 466       sync, min_format, min_value, max_format, max_value);
 467 }
 468
 469 static void
 470 gst_multi_fd_sink_remove (GstMultiFdSink * sink, int fd)
 471 {
 472   GstMultiSinkHandle handle;
 473
 474   handle.fd = fd;
 475   gst_multi_handle_sink_remove (GST_MULTI_HANDLE_SINK_CAST (sink), handle);
 476 }
 477
 478 static void
 479 gst_multi_fd_sink_remove_flush (GstMultiFdSink * sink, int fd)
 480 {
 481   GstMultiSinkHandle handle;
 482
 483   handle.fd = fd;
 484   gst_multi_handle_sink_remove_flush (GST_MULTI_HANDLE_SINK_CAST (sink),
 485       handle);
 486 }
 487
 488 static GstStructure *
 489 gst_multi_fd_sink_get_stats (GstMultiFdSink * sink, int fd)
 490 {
 491   GstMultiSinkHandle handle;
 492
 493   handle.fd = fd;
 494   return gst_multi_handle_sink_get_stats (GST_MULTI_HANDLE_SINK_CAST (sink),
 495       handle);
 496 }
 497
 498 /* vfuncs */
 499
 500 static GstMultiHandleClient *
 501 gst_multi_fd_sink_new_client (GstMultiHandleSink * mhsink,
 502     GstMultiSinkHandle handle, GstSyncMethod sync_method)
 503 {
 504   struct stat statbuf;
 505   GstTCPClient *client;
 506   GstMultiHandleClient *mhclient;
 507   GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
 508   GstMultiHandleSinkClass *mhsinkclass =
 509       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
 510
 511   /* create client datastructure */
 512   client = g_new0 (GstTCPClient, 1);
 513   mhclient = (GstMultiHandleClient *) client;
 514
 515   mhclient->handle = handle;
 516
 517   gst_poll_fd_init (&client->gfd);
 518   client->gfd.fd = mhclient->handle.fd;
 519
 520   gst_multi_handle_sink_client_init (mhclient, sync_method);
 521   mhsinkclass->handle_debug (handle, mhclient->debug);
 522
 523   /* set the socket to non blocking */
 524   if (fcntl (handle.fd, F_SETFL, O_NONBLOCK) < 0) {
 525     GST_ERROR_OBJECT (mhsink, "failed to make socket %s non-blocking: %s",
 526         mhclient->debug, g_strerror (errno));
 527   }
 528
 529   /* we always read from a client */
 530   gst_poll_add_fd (sink->fdset, &client->gfd);
 531
 532   /* we don't try to read from write only fds */
 533   if (sink->handle_read) {
 534     gint flags;
 535
 536     flags = fcntl (handle.fd, F_GETFL, 0);
 537     if ((flags & O_ACCMODE) != O_WRONLY) {
 538       gst_poll_fd_ctl_read (sink->fdset, &client->gfd, TRUE);
 539     }
 540   }
 541   /* figure out the mode, can't use send() for non sockets */
 542   if (fstat (handle.fd, &statbuf) == 0 && S_ISSOCK (statbuf.st_mode)) {
 543     client->is_socket = TRUE;
 544     gst_multi_handle_sink_setup_dscp_client (mhsink, mhclient);
 545   }
 546
 547   return mhclient;
 548 }
 549
 550 static int
 551 gst_multi_fd_sink_client_get_fd (GstMultiHandleClient * client)
 552 {
 553   GstTCPClient *tclient = (GstTCPClient *) client;
 554
 555   return tclient->gfd.fd;
 556 }
 557
 558 static void
 559 gst_multi_fd_sink_handle_debug (GstMultiSinkHandle handle, gchar debug[30])
 560 {
 561   g_snprintf (debug, 30, "[fd %5d]", handle.fd);
 562 }
 563
 564 static gpointer
 565 gst_multi_fd_sink_handle_hash_key (GstMultiSinkHandle handle)
 566 {
 567   return GINT_TO_POINTER (handle.fd);
 568 }
 569
 570 static void
 571 gst_multi_fd_sink_hash_changed (GstMultiHandleSink * mhsink)
 572 {
 573   GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
 574
 575   gst_poll_restart (sink->fdset);
 576 }
 577
 578 /* handle a read on a client fd,
 579  * which either indicates a close or should be ignored
 580  * returns FALSE if some error occured or the client closed. */
 581 static gboolean
 582 gst_multi_fd_sink_handle_client_read (GstMultiFdSink * sink,
 583     GstTCPClient * client)
 584 {
 585   int avail, fd;
 586   gboolean ret;
 587   GstMultiHandleClient *mhclient = (GstMultiHandleClient *) client;
 588
 589   fd = client->gfd.fd;
 590
 591   if (ioctl (fd, FIONREAD, &avail) < 0)
 592     goto ioctl_failed;
 593
 594   GST_DEBUG_OBJECT (sink, "%s select reports client read of %d bytes",
 595       mhclient->debug, avail);
 596
 597   ret = TRUE;
 598
 599   if (avail == 0) {
 600     /* client sent close, so remove it */
 601     GST_DEBUG_OBJECT (sink, "%s client asked for close, removing",
 602         mhclient->debug);
 603     mhclient->status = GST_CLIENT_STATUS_CLOSED;
 604     ret = FALSE;
 605   } else if (avail < 0) {
 606     GST_WARNING_OBJECT (sink, "%s avail < 0, removing", mhclient->debug);
 607     mhclient->status = GST_CLIENT_STATUS_ERROR;
 608     ret = FALSE;
 609   } else {
 610     guint8 dummy[512];
 611     gint nread;
 612
 613     /* just Read 'n' Drop, could also just drop the client as it's not supposed
 614      * to write to us except for closing the socket, I guess it's because we
 615      * like to listen to our customers. */
 616     do {
 617       /* this is the maximum we can read */
 618       gint to_read = MIN (avail, 512);
 619
 620       GST_DEBUG_OBJECT (sink, "%s client wants us to read %d bytes",
 621           mhclient->debug, to_read);
 622
 623       nread = read (fd, dummy, to_read);
 624       if (nread < -1) {
 625         GST_WARNING_OBJECT (sink, "%s could not read %d bytes: %s (%d)",
 626             mhclient->debug, to_read, g_strerror (errno), errno);
 627         mhclient->status = GST_CLIENT_STATUS_ERROR;
 628         ret = FALSE;
 629         break;
 630       } else if (nread == 0) {
 631         GST_WARNING_OBJECT (sink, "%s 0 bytes in read, removing",
 632             mhclient->debug);
 633         mhclient->status = GST_CLIENT_STATUS_ERROR;
 634         ret = FALSE;
 635         break;
 636       }
 637       avail -= nread;
 638     }
 639     while (avail > 0);
 640   }
 641   return ret;
 642
 643   /* ERRORS */
 644 ioctl_failed:
 645   {
 646     GST_WARNING_OBJECT (sink, "%s ioctl failed: %s (%d)",
 647         mhclient->debug, g_strerror (errno), errno);
 648     mhclient->status = GST_CLIENT_STATUS_ERROR;
 649     return FALSE;
 650   }
 651 }
 652
 653 /* Handle a write on a client,
 654  * which indicates a read request from a client.
 655  *
 656  * For each client we maintain a queue of GstBuffers that contain the raw bytes
 657  * we need to send to the client.
 658  *
 659  * We first check to see if we need to send streamheaders. If so, we queue them.
 660  *
 661  * Then we run into the main loop that tries to send as many buffers as
 662  * possible. It will first exhaust the mhclient->sending queue and if the queue
 663  * is empty, it will pick a buffer from the global queue.
 664  *
 665  * Sending the buffers from the mhclient->sending queue is basically writing
 666  * the bytes to the socket and maintaining a count of the bytes that were
 667  * sent. When the buffer is completely sent, it is removed from the
 668  * mhclient->sending queue and we try to pick a new buffer for sending.
 669  *
 670  * When the sending returns a partial buffer we stop sending more data as
 671  * the next send operation could block.
 672  *
 673  * This functions returns FALSE if some error occured.
 674  */
 675 static gboolean
 676 gst_multi_fd_sink_handle_client_write (GstMultiFdSink * sink,
 677     GstTCPClient * client)
 678 {
 679   gboolean more;
 680   gboolean flushing;
 681   GstClockTime now;
 682   GTimeVal nowtv;
 683   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (sink);
 684   GstMultiHandleSinkClass *mhsinkclass =
 685       GST_MULTI_HANDLE_SINK_GET_CLASS (mhsink);
 686   GstMultiHandleClient *mhclient = (GstMultiHandleClient *) client;
 687   int fd = mhclient->handle.fd;
 688
 689   g_get_current_time (&nowtv);
 690   now = GST_TIMEVAL_TO_TIME (nowtv);
 691
 692   flushing = mhclient->status == GST_CLIENT_STATUS_FLUSHING;
 693
 694   more = TRUE;
 695   do {
 696     gint maxsize;
 697
 698     if (!mhclient->sending) {
 699       /* client is not working on a buffer */
 700       if (mhclient->bufpos == -1) {
 701         /* client is too fast, remove from write queue until new buffer is
 702          * available */
 703         /* FIXME: specific */
 704         gst_poll_fd_ctl_write (sink->fdset, &client->gfd, FALSE);
 705
 706         /* if we flushed out all of the client buffers, we can stop */
 707         if (mhclient->flushcount == 0)
 708           goto flushed;
 709
 710         return TRUE;
 711       } else {
 712         /* client can pick a buffer from the global queue */
 713         GstBuffer *buf;
 714         GstClockTime timestamp;
 715
 716         /* for new connections, we need to find a good spot in the
 717          * bufqueue to start streaming from */
 718         if (mhclient->new_connection && !flushing) {
 719           gint position =
 720               gst_multi_handle_sink_new_client_position (mhsink, mhclient);
 721
 722           if (position >= 0) {
 723             /* we got a valid spot in the queue */
 724             mhclient->new_connection = FALSE;
 725             mhclient->bufpos = position;
 726           } else {
 727             /* cannot send data to this client yet */
 728             /* FIXME: specific */
 729             gst_poll_fd_ctl_write (sink->fdset, &client->gfd, FALSE);
 730             return TRUE;
 731           }
 732         }
 733
 734         /* we flushed all remaining buffers, no need to get a new one */
 735         if (mhclient->flushcount == 0)
 736           goto flushed;
 737
 738         /* grab buffer */
 739         buf = g_array_index (mhsink->bufqueue, GstBuffer *, mhclient->bufpos);
 740         mhclient->bufpos--;
 741
 742         /* update stats */
 743         timestamp = GST_BUFFER_TIMESTAMP (buf);
 744         if (mhclient->first_buffer_ts == GST_CLOCK_TIME_NONE)
 745           mhclient->first_buffer_ts = timestamp;
 746         if (timestamp != -1)
 747           mhclient->last_buffer_ts = timestamp;
 748
 749         /* decrease flushcount */
 750         if (mhclient->flushcount != -1)
 751           mhclient->flushcount--;
 752
 753         GST_LOG_OBJECT (sink, "%s client %p at position %d",
 754             mhclient->debug, client, mhclient->bufpos);
 755
 756         /* queueing a buffer will ref it */
 757         mhsinkclass->client_queue_buffer (mhsink, mhclient, buf);
 758
 759         /* need to start from the first byte for this new buffer */
 760         mhclient->bufoffset = 0;
 761       }
 762     }
 763
 764     /* see if we need to send something */
 765     if (mhclient->sending) {
 766       ssize_t wrote;
 767       GstBuffer *head;
 768       GstMapInfo info;
 769       guint8 *data;
 770
 771       /* pick first buffer from list */
 772       head = GST_BUFFER (mhclient->sending->data);
 773
 774       if (!gst_buffer_map (head, &info, GST_MAP_READ))
 775         g_return_val_if_reached (FALSE);
 776
 777       data = info.data;
 778       maxsize = info.size - mhclient->bufoffset;
 779
 780       /* FIXME: specific */
 781       /* try to write the complete buffer */
 782 #ifdef MSG_NOSIGNAL
 783 #define FLAGS MSG_NOSIGNAL
 784 #else
 785 #define FLAGS 0
 786 #endif
 787       if (client->is_socket) {
 788         wrote = send (fd, data + mhclient->bufoffset, maxsize, FLAGS);
 789       } else {
 790         wrote = write (fd, data + mhclient->bufoffset, maxsize);
 791       }
 792       gst_buffer_unmap (head, &info);
 793
 794       if (wrote < 0) {
 795         /* hmm error.. */
 796         if (errno == EAGAIN) {
 797           /* nothing serious, resource was unavailable, try again later */
 798           more = FALSE;
 799         } else if (errno == ECONNRESET) {
 800           goto connection_reset;
 801         } else {
 802           goto write_error;
 803         }
 804       } else {
 805         if (wrote < maxsize) {
 806           /* partial write means that the client cannot read more and we should
 807            * stop sending more */
 808           GST_LOG_OBJECT (sink,
 809               "partial write on %s of %" G_GSSIZE_FORMAT " bytes",
 810               mhclient->debug, wrote);
 811           mhclient->bufoffset += wrote;
 812           more = FALSE;
 813         } else {
 814           /* complete buffer was written, we can proceed to the next one */
 815           mhclient->sending = g_slist_remove (mhclient->sending, head);
 816           gst_buffer_unref (head);
 817           /* make sure we start from byte 0 for the next buffer */
 818           mhclient->bufoffset = 0;
 819         }
 820         /* update stats */
 821         mhclient->bytes_sent += wrote;
 822         mhclient->last_activity_time = now;
 823         mhsink->bytes_served += wrote;
 824       }
 825     }
 826   } while (more);
 827
 828   return TRUE;
 829
 830   /* ERRORS */
 831 flushed:
 832   {
 833     GST_DEBUG_OBJECT (sink, "%s flushed, removing", mhclient->debug);
 834     mhclient->status = GST_CLIENT_STATUS_REMOVED;
 835     return FALSE;
 836   }
 837 connection_reset:
 838   {
 839     GST_DEBUG_OBJECT (sink, "%s connection reset by peer, removing",
 840         mhclient->debug);
 841     mhclient->status = GST_CLIENT_STATUS_CLOSED;
 842     return FALSE;
 843   }
 844 write_error:
 845   {
 846     GST_WARNING_OBJECT (sink,
 847         "%s could not write, removing client: %s (%d)", mhclient->debug,
 848         g_strerror (errno), errno);
 849     mhclient->status = GST_CLIENT_STATUS_ERROR;
 850     return FALSE;
 851   }
 852 }
 853
 854 static void
 855 gst_multi_fd_sink_hash_adding (GstMultiHandleSink * mhsink,
 856     GstMultiHandleClient * mhclient)
 857 {
 858   GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
 859   GstTCPClient *client = (GstTCPClient *) mhclient;
 860
 861   gst_poll_fd_ctl_write (sink->fdset, &client->gfd, TRUE);
 862 }
 863
 864 static void
 865 gst_multi_fd_sink_hash_removing (GstMultiHandleSink * mhsink,
 866     GstMultiHandleClient * mhclient)
 867 {
 868   GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
 869   GstTCPClient *client = (GstTCPClient *) mhclient;
 870
 871   gst_poll_remove_fd (sink->fdset, &client->gfd);
 872 }
 873
 874
 875 /* Handle the clients. Basically does a blocking select for one
 876  * of the client fds to become read or writable. We also have a
 877  * filedescriptor to receive commands on that we need to check.
 878  *
 879  * After going out of the select call, we read and write to all
 880  * clients that can do so. Badly behaving clients are put on a
 881  * garbage list and removed.
 882  */
 883 static void
 884 gst_multi_fd_sink_handle_clients (GstMultiFdSink * sink)
 885 {
 886   int result;
 887   GList *clients, *next;
 888   gboolean try_again;
 889   GstMultiFdSinkClass *fclass;
 890   guint cookie;
 891   GstMultiHandleSink *mhsink = GST_MULTI_HANDLE_SINK (sink);
 892   int fd;
 893
 894
 895   fclass = GST_MULTI_FD_SINK_GET_CLASS (sink);
 896
 897   do {
 898     try_again = FALSE;
 899
 900     /* check for:
 901      * - server socket input (ie, new client connections)
 902      * - client socket input (ie, clients saying goodbye)
 903      * - client socket output (ie, client reads)          */
 904     GST_LOG_OBJECT (sink, "waiting on action on fdset");
 905
 906     result =
 907         gst_poll_wait (sink->fdset,
 908         mhsink->timeout != 0 ? mhsink->timeout : GST_CLOCK_TIME_NONE);
 909
 910     /* Handle the special case in which the sink is not receiving more buffers
 911      * and will not disconnect inactive client in the streaming thread. */
 912     if (G_UNLIKELY (result == 0)) {
 913       GstClockTime now;
 914       GTimeVal nowtv;
 915
 916       g_get_current_time (&nowtv);
 917       now = GST_TIMEVAL_TO_TIME (nowtv);
 918
 919       CLIENTS_LOCK (mhsink);
 920       for (clients = mhsink->clients; clients; clients = next) {
 921         GstTCPClient *client;
 922         GstMultiHandleClient *mhclient;
 923
 924         client = (GstTCPClient *) clients->data;
 925         mhclient = (GstMultiHandleClient *) client;
 926         next = g_list_next (clients);
 927         if (mhsink->timeout > 0
 928             && now - mhclient->last_activity_time > mhsink->timeout) {
 929           mhclient->status = GST_CLIENT_STATUS_SLOW;
 930           gst_multi_handle_sink_remove_client_link (mhsink, clients);
 931         }
 932       }
 933       CLIENTS_UNLOCK (mhsink);
 934       return;
 935     } else if (result < 0) {
 936       GST_WARNING_OBJECT (sink, "wait failed: %s (%d)", g_strerror (errno),
 937           errno);
 938       if (errno == EBADF) {
 939         /* ok, so one or more of the fds is invalid. We loop over them to find
 940          * the ones that give an error to the F_GETFL fcntl. */
 941         CLIENTS_LOCK (mhsink);
 942       restart:
 943         cookie = mhsink->clients_cookie;
 944         for (clients = mhsink->clients; clients; clients = next) {
 945           GstTCPClient *client;
 946           GstMultiHandleClient *mhclient;
 947           long flags;
 948           int res;
 949
 950           if (cookie != mhsink->clients_cookie) {
 951             GST_DEBUG_OBJECT (sink, "Cookie changed finding bad fd");
 952             goto restart;
 953           }
 954
 955           client = (GstTCPClient *) clients->data;
 956           mhclient = (GstMultiHandleClient *) client;
 957           next = g_list_next (clients);
 958
 959           fd = client->gfd.fd;
 960
 961           res = fcntl (fd, F_GETFL, &flags);
 962           if (res == -1) {
 963             GST_WARNING_OBJECT (sink, "fnctl failed for %d, removing: %s (%d)",
 964                 fd, g_strerror (errno), errno);
 965             if (errno == EBADF) {
 966               mhclient->status = GST_CLIENT_STATUS_ERROR;
 967               /* releases the CLIENTS lock */
 968               gst_multi_handle_sink_remove_client_link (mhsink, clients);
 969             }
 970           }
 971         }
 972         CLIENTS_UNLOCK (mhsink);
 973         /* after this, go back in the select loop as the read/writefds
 974          * are not valid */
 975         try_again = TRUE;
 976       } else if (errno == EINTR) {
 977         /* interrupted system call, just redo the wait */
 978         try_again = TRUE;
 979       } else if (errno == EBUSY) {
 980         /* the call to gst_poll_wait() was flushed */
 981         return;
 982       } else {
 983         /* this is quite bad... */
 984         GST_ELEMENT_ERROR (sink, RESOURCE, READ, (NULL),
 985             ("select failed: %s (%d)", g_strerror (errno), errno));
 986         return;
 987       }
 988     } else {
 989       GST_LOG_OBJECT (sink, "wait done: %d sockets with events", result);
 990     }
 991   } while (try_again);
 992
 993   /* subclasses can check fdset with this virtual function */
 994   if (fclass->wait)
 995     fclass->wait (sink, sink->fdset);
 996
 997   /* Check the clients */
 998   CLIENTS_LOCK (mhsink);
 999
1000 restart2:
1001   cookie = mhsink->clients_cookie;
1002   for (clients = mhsink->clients; clients; clients = next) {
1003     GstTCPClient *client;
1004     GstMultiHandleClient *mhclient;
1005
1006     if (mhsink->clients_cookie != cookie) {
1007       GST_DEBUG_OBJECT (sink, "Restarting loop, cookie out of date");
1008       goto restart2;
1009     }
1010
1011     client = (GstTCPClient *) clients->data;
1012     mhclient = (GstMultiHandleClient *) client;
1013     next = g_list_next (clients);
1014
1015     if (mhclient->status != GST_CLIENT_STATUS_FLUSHING
1016         && mhclient->status != GST_CLIENT_STATUS_OK) {
1017       gst_multi_handle_sink_remove_client_link (mhsink, clients);
1018       continue;
1019     }
1020
1021     if (gst_poll_fd_has_closed (sink->fdset, &client->gfd)) {
1022       mhclient->status = GST_CLIENT_STATUS_CLOSED;
1023       gst_multi_handle_sink_remove_client_link (mhsink, clients);
1024       continue;
1025     }
1026     if (gst_poll_fd_has_error (sink->fdset, &client->gfd)) {
1027       GST_WARNING_OBJECT (sink, "gst_poll_fd_has_error for %d", client->gfd.fd);
1028       mhclient->status = GST_CLIENT_STATUS_ERROR;
1029       gst_multi_handle_sink_remove_client_link (mhsink, clients);
1030       continue;
1031     }
1032     if (gst_poll_fd_can_read (sink->fdset, &client->gfd)) {
1033       /* handle client read */
1034       if (!gst_multi_fd_sink_handle_client_read (sink, client)) {
1035         gst_multi_handle_sink_remove_client_link (mhsink, clients);
1036         continue;
1037       }
1038     }
1039     if (gst_poll_fd_can_write (sink->fdset, &client->gfd)) {
1040       /* handle client write */
1041       if (!gst_multi_fd_sink_handle_client_write (sink, client)) {
1042         gst_multi_handle_sink_remove_client_link (mhsink, clients);
1043         continue;
1044       }
1045     }
1046   }
1047   CLIENTS_UNLOCK (mhsink);
1048 }
1049
1050 /* we handle the client communication in another thread so that we do not block
1051  * the gstreamer thread while we select() on the client fds */
1052 static gpointer
1053 gst_multi_fd_sink_thread (GstMultiHandleSink * mhsink)
1054 {
1055   GstMultiFdSink *sink = GST_MULTI_FD_SINK (mhsink);
1056
1057   while (mhsink->running) {
1058     gst_multi_fd_sink_handle_clients (sink);
1059   }
1060   return NULL;
1061 }
1062
1063 static void
1064 gst_multi_fd_sink_set_property (GObject * object, guint prop_id,
1065     const GValue * value, GParamSpec * pspec)
1066 {
1067   GstMultiFdSink *multifdsink;
1068
1069   multifdsink = GST_MULTI_FD_SINK (object);
1070
1071   switch (prop_id) {
1072     case PROP_HANDLE_READ:
1073       multifdsink->handle_read = g_value_get_boolean (value);
1074       break;
1075
1076     default:
1077       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1078       break;
1079   }
1080 }
1081
1082 static void
1083 gst_multi_fd_sink_get_property (GObject * object, guint prop_id, GValue * value,
1084     GParamSpec * pspec)
1085 {
1086   GstMultiFdSink *multifdsink;
1087
1088   multifdsink = GST_MULTI_FD_SINK (object);
1089
1090   switch (prop_id) {
1091     case PROP_HANDLE_READ:
1092       g_value_set_boolean (value, multifdsink->handle_read);
1093       break;
1094
1095     default:
1096       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
1097       break;
1098   }
1099 }
1100
1101 static gboolean
1102 gst_multi_fd_sink_start_pre (GstMultiHandleSink * mhsink)
1103 {
1104   GstMultiFdSink *mfsink = GST_MULTI_FD_SINK (mhsink);
1105
1106   GST_INFO_OBJECT (mfsink, "starting");
1107   if ((mfsink->fdset = gst_poll_new (TRUE)) == NULL)
1108     goto socket_pair;
1109
1110   return TRUE;
1111
1112   /* ERRORS */
1113 socket_pair:
1114   {
1115     GST_ELEMENT_ERROR (mfsink, RESOURCE, OPEN_READ_WRITE, (NULL),
1116         GST_ERROR_SYSTEM);
1117     return FALSE;
1118   }
1119 }
1120
1121 static gboolean
1122 multifdsink_hash_remove (gpointer key, gpointer value, gpointer data)
1123 {
1124   return TRUE;
1125 }
1126
1127 static void
1128 gst_multi_fd_sink_stop_pre (GstMultiHandleSink * mhsink)
1129 {
1130   GstMultiFdSink *mfsink = GST_MULTI_FD_SINK (mhsink);
1131
1132   gst_poll_set_flushing (mfsink->fdset, TRUE);
1133 }
1134
1135 static void
1136 gst_multi_fd_sink_stop_post (GstMultiHandleSink * mhsink)
1137 {
1138   GstMultiFdSink *mfsink = GST_MULTI_FD_SINK (mhsink);
1139
1140   if (mfsink->fdset) {
1141     gst_poll_free (mfsink->fdset);
1142     mfsink->fdset = NULL;
1143   }
1144   g_hash_table_foreach_remove (mhsink->handle_hash, multifdsink_hash_remove,
1145       mfsink);
1146 }