/* GStreamer

 * Copyright (C) <1999> Erik Walthinsen <omega@cse.ogi.edu>

 * Copyright (C) <2004> Thomas Vander Stichele <thomas at apestaart dot org>

 * Copyright (C) 2006 Wim Taymans <wim at fluendo dot com>

 *

 * This library is free software; you can redistribute it and/or

 * modify it under the terms of the GNU Library General Public

 * License as published by the Free Software Foundation; either

 * version 2 of the License, or (at your option) any later version.

 *

 * This library is distributed in the hope that it will be useful,

 * but WITHOUT ANY WARRANTY; without even the implied warranty of

 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

 * Library General Public License for more details.

 *

 * You should have received a copy of the GNU Library General Public

 * License along with this library; if not, write to the

 * Free Software Foundation, Inc., 59 Temple Place - Suite 330,

 * Boston, MA 02111-1307, USA.

 */

/**

 * SECTION:element-multifdsink

 * @see_also: tcpserversink

 *

 * This plugin writes incoming data to a set of file descriptors. The

 * that allows for more control over what and how much data a client 

 * initially receives.

 * client is not active anymore or, depending on the value of the

 * In all cases, multifdsink will never close a file descriptor itself.

 * The user of multifdsink is responsible for closing all file descriptors.

 * Note that multifdsink still has a reference to the file descriptor when the

 * the descriptor; it is therefore not safe to close the file descriptor in

 * Multifdsink internally keeps a queue of the incoming buffers and uses a

 * separate thread to send the buffers to the clients. This ensures that no

 * client write can block the pipeline and that clients can read with different

 * speeds.

 * which buffer in the queued buffers will be sent first to the client. Clients 

 * can be sent the most recent buffer (which might not be decodable by the 

 * client if it is not a keyframe), the next keyframe received in 

 * multifdsink (which can take some time depending on the keyframe rate), or the

 * last received keyframe (which will cause a simple burst-on-connect). 

 * Multifdsink will always keep at least one keyframe in its internal buffers

 * when the sync-mode is set to latest-keyframe.

 * property to allow finer control over burst-on-connect behaviour. By selecting

 * the 'burst' method a minimum burst size can be chosen, 'burst-keyframe'

 * additionally requires that the burst begin with a keyframe, and 

 * 'burst-with-keyframe' attempts to burst beginning with a keyframe, but will

 * prefer a minimum burst size even if it requires not starting with a keyframe.

 * Multifdsink can be instructed to keep at least a minimum amount of data

 * expressed in time or byte units in its internal queues with the the 

 * When streaming data, clients are allowed to read at a different rate than

 * the rate at which multifdsink receives data. If the client is reading too

 * fast, no data will be send to the client until multifdsink receives more

 * data. If the client, however, reads too slowly, data for that client will be 

 * queued up in multifdsink. Two properties control the amount of data 

 * received buffer as the next buffer for the client, RESYNC_SOFT_LIMIT

 * positions the client to the soft limit in the buffer queue and

 * RESYNC_KEYFRAME positions the client at the most recent keyframe in the

 * buffer queue.

 * multifdsink will by default synchronize on the clock before serving the 

 * buffers to the clients. This behaviour can be disabled by setting the sync 

 * property to FALSE. Multifdsink will by default not do QoS and will never

 * drop late buffers.

 *

 * Last reviewed on 2006-09-12 (0.10.10)

 */

#ifdef HAVE_CONFIG_H

#include "config.h"

#endif

#include <gst/gst-i18n-plugin.h>

#include <sys/ioctl.h>

#include <unistd.h>

#include <fcntl.h>

#include <sys/types.h>

#include <sys/socket.h>

#include <sys/stat.h>

#ifdef HAVE_FIONREAD_IN_SYS_FILIO

#include <sys/filio.h>

#endif

#include "gstmultifdsink.h"

#include "gsttcp-marshal.h"

#define NOT_IMPLEMENTED 0

/* elementfactory information */

static const GstElementDetails gst_multi_fd_sink_details =

GST_ELEMENT_DETAILS ("Multi filedescriptor sink",

    "Sink/Network",

    "Send data to multiple filedescriptors",

    "Thomas Vander Stichele <thomas at apestaart dot org>, "

    "Wim Taymans <wim@fluendo.com>");

static GstStaticPadTemplate sinktemplate = GST_STATIC_PAD_TEMPLATE ("sink",

    GST_PAD_SINK,

    GST_PAD_ALWAYS,

    GST_STATIC_CAPS_ANY);

GST_DEBUG_CATEGORY_STATIC (multifdsink_debug);

#define GST_CAT_DEFAULT (multifdsink_debug)

/* MultiFdSink signals and args */

enum

{

  /* methods */

  SIGNAL_ADD,

  SIGNAL_ADD_BURST,

  SIGNAL_REMOVE,

  SIGNAL_REMOVE_FLUSH,

  SIGNAL_CLEAR,

  SIGNAL_GET_STATS,

  /* signals */

  SIGNAL_CLIENT_ADDED,

  SIGNAL_CLIENT_REMOVED,

  SIGNAL_CLIENT_FD_REMOVED,

  LAST_SIGNAL

};

/* this is really arbitrarily chosen */

#define DEFAULT_PROTOCOL                GST_TCP_PROTOCOL_NONE

#define DEFAULT_MODE                    1

#define DEFAULT_BUFFERS_MAX             -1

#define DEFAULT_BUFFERS_SOFT_MAX        -1

#define DEFAULT_TIME_MIN                -1

#define DEFAULT_BYTES_MIN               -1

#define DEFAULT_BUFFERS_MIN             -1

#define DEFAULT_UNITS_MAX               -1

#define DEFAULT_UNITS_SOFT_MAX          -1

#define DEFAULT_RECOVER_POLICY          GST_RECOVER_POLICY_NONE

#define DEFAULT_TIMEOUT                 0

#define DEFAULT_SYNC_METHOD             GST_SYNC_METHOD_LATEST

#define DEFAULT_BURST_VALUE             0

enum

{

  PROP_0,

  PROP_PROTOCOL,

  PROP_MODE,

  PROP_BUFFERS_QUEUED,

  PROP_BYTES_QUEUED,

  PROP_TIME_QUEUED,

  PROP_UNIT_TYPE,

  PROP_UNITS_MAX,

  PROP_UNITS_SOFT_MAX,

  PROP_BUFFERS_MAX,

  PROP_BUFFERS_SOFT_MAX,

  PROP_TIME_MIN,

  PROP_BYTES_MIN,

  PROP_BUFFERS_MIN,

  PROP_RECOVER_POLICY,

  PROP_TIMEOUT,

  PROP_SYNC_METHOD,

  PROP_BYTES_TO_SERVE,

  PROP_BYTES_SERVED,

  PROP_BURST_UNIT,

  PROP_BURST_VALUE,

};

/* For backward compat, we can't really select the poll mode anymore with

 * GstPoll. */

#define GST_TYPE_FDSET_MODE (gst_fdset_mode_get_type())

static GType

gst_fdset_mode_get_type (void)

{

  static GType fdset_mode_type = 0;

  static const GEnumValue fdset_mode[] = {

    {0, "Select", "select"},

    {1, "Poll", "poll"},

    {2, "EPoll", "epoll"},

    {0, NULL, NULL},

  };

  if (!fdset_mode_type) {

    fdset_mode_type = g_enum_register_static ("GstFDSetMode", fdset_mode);

  }

  return fdset_mode_type;

}

#define GST_TYPE_RECOVER_POLICY (gst_recover_policy_get_type())

static GType

gst_recover_policy_get_type (void)

{

  static GType recover_policy_type = 0;

  static const GEnumValue recover_policy[] = {

    {GST_RECOVER_POLICY_NONE,

        "Do not try to recover", "none"},

    {GST_RECOVER_POLICY_RESYNC_LATEST,

        "Resync client to latest buffer", "latest"},

    {GST_RECOVER_POLICY_RESYNC_SOFT_LIMIT,

        "Resync client to soft limit", "soft-limit"},

    {GST_RECOVER_POLICY_RESYNC_KEYFRAME,

        "Resync client to most recent keyframe", "keyframe"},

    {0, NULL, NULL},

  };

  if (!recover_policy_type) {

    recover_policy_type =

        g_enum_register_static ("GstRecoverPolicy", recover_policy);

  }

  return recover_policy_type;

}

#define GST_TYPE_SYNC_METHOD (gst_sync_method_get_type())

static GType

gst_sync_method_get_type (void)

{

  static GType sync_method_type = 0;

  static const GEnumValue sync_method[] = {

    {GST_SYNC_METHOD_LATEST,

        "Serve starting from the latest buffer", "latest"},

    {GST_SYNC_METHOD_NEXT_KEYFRAME,

        "Serve starting from the next keyframe", "next-keyframe"},

    {GST_SYNC_METHOD_LATEST_KEYFRAME,

          "Serve everything since the latest keyframe (burst)",

        "latest-keyframe"},

    {GST_SYNC_METHOD_BURST, "Serve burst-value data to client", "burst"},

    {GST_SYNC_METHOD_BURST_KEYFRAME,

          "Serve burst-value data starting on a keyframe",

        "burst-keyframe"},

    {GST_SYNC_METHOD_BURST_WITH_KEYFRAME,

          "Serve burst-value data preferably starting on a keyframe",

        "burst-with-keyframe"},

    {0, NULL, NULL},

  };

  if (!sync_method_type) {

    sync_method_type = g_enum_register_static ("GstSyncMethod", sync_method);

  }

  return sync_method_type;

}

#define GST_TYPE_UNIT_TYPE (gst_unit_type_get_type())

static GType

gst_unit_type_get_type (void)

{

  static GType unit_type_type = 0;

  static const GEnumValue unit_type[] = {

    {0, NULL, NULL},

  };

  if (!unit_type_type) {

    unit_type_type = g_enum_register_static ("GstTCPUnitType", unit_type);

  }

  return unit_type_type;

}

#define GST_TYPE_CLIENT_STATUS (gst_client_status_get_type())

static GType

gst_client_status_get_type (void)

{

  static GType client_status_type = 0;

  static const GEnumValue client_status[] = {

    {GST_CLIENT_STATUS_OK, "ok", "ok"},

    {GST_CLIENT_STATUS_CLOSED, "Closed", "closed"},

    {GST_CLIENT_STATUS_REMOVED, "Removed", "removed"},

    {GST_CLIENT_STATUS_SLOW, "Too slow", "slow"},

    {GST_CLIENT_STATUS_ERROR, "Error", "error"},

    {GST_CLIENT_STATUS_DUPLICATE, "Duplicate", "duplicate"},

    {GST_CLIENT_STATUS_FLUSHING, "Flushing", "flushing"},

    {0, NULL, NULL},

  };

  if (!client_status_type) {

    client_status_type =

        g_enum_register_static ("GstClientStatus", client_status);

  }

  return client_status_type;

}

static void gst_multi_fd_sink_finalize (GObject * object);

static void gst_multi_fd_sink_remove_client_link (GstMultiFdSink * sink,

    GList * link);

static GstFlowReturn gst_multi_fd_sink_render (GstBaseSink * bsink,

    GstBuffer * buf);

static GstStateChangeReturn gst_multi_fd_sink_change_state (GstElement *

    element, GstStateChange transition);

static void gst_multi_fd_sink_set_property (GObject * object, guint prop_id,

    const GValue * value, GParamSpec * pspec);

static void gst_multi_fd_sink_get_property (GObject * object, guint prop_id,

    GValue * value, GParamSpec * pspec);

GST_BOILERPLATE (GstMultiFdSink, gst_multi_fd_sink, GstBaseSink,

    GST_TYPE_BASE_SINK);

static guint gst_multi_fd_sink_signals[LAST_SIGNAL] = { 0 };

static void

gst_multi_fd_sink_base_init (gpointer g_class)

{

  GstElementClass *element_class = GST_ELEMENT_CLASS (g_class);

  gst_element_class_add_pad_template (element_class,

      gst_static_pad_template_get (&sinktemplate));

  gst_element_class_set_details (element_class, &gst_multi_fd_sink_details);

}

static void

gst_multi_fd_sink_class_init (GstMultiFdSinkClass * klass)

{

  GObjectClass *gobject_class;

  GstElementClass *gstelement_class;

  GstBaseSinkClass *gstbasesink_class;

  gobject_class = (GObjectClass *) klass;

  gstelement_class = (GstElementClass *) klass;

  gstbasesink_class = (GstBaseSinkClass *) klass;

  gobject_class->set_property = gst_multi_fd_sink_set_property;

  gobject_class->get_property = gst_multi_fd_sink_get_property;

  gobject_class->finalize = gst_multi_fd_sink_finalize;

  g_object_class_install_property (gobject_class, PROP_PROTOCOL,

      g_param_spec_enum ("protocol", "Protocol", "The protocol to wrap data in",

  /**

   * GstMultiFdSink::mode

   *

   * The mode for selecting activity on the fds. 

   *

   * This property is deprecated since 0.10.18, if will now automatically

   * select and use the most optimal method.

   */

  g_object_class_install_property (gobject_class, PROP_MODE,

      g_param_spec_enum ("mode", "Mode",

          "The mode for selecting activity on the fds (deprecated)",

  g_object_class_install_property (gobject_class, PROP_BUFFERS_MAX,

      g_param_spec_int ("buffers-max", "Buffers max",

          "max number of buffers to queue for a client (-1 = no limit)", -1,

          "Recover client when going over this limit (-1 = no limit)", -1,

  g_object_class_install_property (gobject_class, PROP_BYTES_MIN,

      g_param_spec_int ("bytes-min", "Bytes min",

          "min number of bytes to queue (-1 = as little as possible)", -1,

  g_object_class_install_property (gobject_class, PROP_TIME_MIN,

      g_param_spec_int64 ("time-min", "Time min",

          "min number of time to queue (-1 = as little as possible)", -1,

  g_object_class_install_property (gobject_class, PROP_BUFFERS_MIN,

      g_param_spec_int ("buffers-min", "Buffers min",

          "min number of buffers to queue (-1 = as few as possible)", -1,

  g_object_class_install_property (gobject_class, PROP_UNIT_TYPE,

      g_param_spec_enum ("unit-type", "Units type",

          "The unit to measure the max/soft-max/queued properties",

  g_object_class_install_property (gobject_class, PROP_UNITS_MAX,

      g_param_spec_int64 ("units-max", "Units max",

          "max number of units to queue (-1 = no limit)", -1, G_MAXINT64,

  g_object_class_install_property (gobject_class, PROP_UNITS_SOFT_MAX,

      g_param_spec_int64 ("units-soft-max", "Units soft max",

          "Recover client when going over this limit (-1 = no limit)", -1,

  g_object_class_install_property (gobject_class, PROP_BUFFERS_QUEUED,

      g_param_spec_uint ("buffers-queued", "Buffers queued",

          "Number of buffers currently queued", 0, G_MAXUINT, 0,

#if NOT_IMPLEMENTED

  g_object_class_install_property (gobject_class, PROP_BYTES_QUEUED,

      g_param_spec_uint ("bytes-queued", "Bytes queued",

          "Number of bytes currently queued", 0, G_MAXUINT, 0,

  g_object_class_install_property (gobject_class, PROP_TIME_QUEUED,

      g_param_spec_uint64 ("time-queued", "Time queued",

          "Number of time currently queued", 0, G_MAXUINT64, 0,

#endif

  g_object_class_install_property (gobject_class, PROP_RECOVER_POLICY,

      g_param_spec_enum ("recover-policy", "Recover Policy",

          "How to recover when client reaches the soft max",

  g_object_class_install_property (gobject_class, PROP_TIMEOUT,

      g_param_spec_uint64 ("timeout", "Timeout",

          "Maximum inactivity timeout in nanoseconds for a client (0 = no limit)",

  g_object_class_install_property (gobject_class, PROP_SYNC_METHOD,

      g_param_spec_enum ("sync-method", "Sync Method",

  g_object_class_install_property (gobject_class, PROP_BYTES_TO_SERVE,

      g_param_spec_uint64 ("bytes-to-serve", "Bytes to serve",

          "Number of bytes received to serve to clients", 0, G_MAXUINT64, 0,

  g_object_class_install_property (gobject_class, PROP_BYTES_SERVED,

      g_param_spec_uint64 ("bytes-served", "Bytes served",

          "Total number of bytes send to all clients", 0, G_MAXUINT64, 0,

  g_object_class_install_property (gobject_class, PROP_BURST_UNIT,

      g_param_spec_enum ("burst-unit", "Burst unit",

          "The format of the burst units (when sync-method is burst[[-with]-keyframe])",

  g_object_class_install_property (gobject_class, PROP_BURST_VALUE,

      g_param_spec_uint64 ("burst-value", "Burst value",