<commit_before>/*
 * stream.c - Source for TfStream
 * Copyright (C) 2006-2008 Collabora Ltd.
 * Copyright (C) 2006-2008 Nokia Corporation
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <string.h>
#include <stdlib.h>

#include <telepathy-glib/dbus.h>
#include <telepathy-glib/errors.h>
#include <telepathy-glib/gtypes.h>
#include <telepathy-glib/interfaces.h>

#include <gst/interfaces/xoverlay.h>

#include <gst/farsight/fs-conference-iface.h>

#include "stream.h"
#include "stream-priv.h"
#include "channel.h"
#include "tf-signals-marshal.h"

G_DEFINE_TYPE (TfStream, tf_stream, G_TYPE_OBJECT);

#define DEBUG(stream, format, ...) \
  g_debug ("stream %d (%s) %s: " format, \
    stream->stream_id, \
    (stream->priv->media_type == FS_MEDIA_TYPE_AUDIO) ? "audio" \
                                                      : "video", \
    G_STRFUNC, \
    ##__VA_ARGS__)

#define WARNING(stream, format, ...) \
  g_warning ("stream %d (%s) %s: " format, \
    stream->stream_id, \
    (stream->priv->media_type == FS_MEDIA_TYPE_AUDIO) ? "audio" \
                                                      : "video", \
    G_STRFUNC, \
    ##__VA_ARGS__)

#define STREAM_PRIVATE(o) ((o)->priv)

struct _TfStreamPrivate
{
  TfChannel *channel;
  FsConference *fs_conference;
  FsParticipant *fs_participant;
  FsSession *fs_session;
  FsStream *fs_stream;
  TpMediaStreamType media_type;
  TpMediaStreamDirection direction;
  const TfNatProperties *nat_props;
  GList *local_preferences;

  GError *construction_error;

  TpMediaStreamHandler *stream_handler_proxy;

  FsStreamDirection desired_direction;
  gboolean held;
  TpMediaStreamDirection has_resource;

  GList *local_candidates;

  gboolean send_local_codecs;
  gboolean send_supported_codecs;

  gboolean gathering;
};

enum
{
  CLOSED,
  ERROR_SIGNAL,
  REQUEST_RESOURCE,
  FREE_RESOURCE,
  SRC_PAD_ADDED,
  SIGNAL_COUNT
};

static guint signals[SIGNAL_COUNT] = {0};

/* properties */
enum
{
  PROP_CHANNEL = 1,
  PROP_FARSIGHT_CONFERENCE,
  PROP_FARSIGHT_PARTICIPANT,
  PROP_PROXY,
  PROP_STREAM_ID,
  PROP_MEDIA_TYPE,
  PROP_DIRECTION,
  PROP_NAT_PROPERTIES,
  PROP_SINK_PAD,
  PROP_LOCAL_PREFERENCES
};


static gboolean tf_stream_request_resource (
    TfStream *self,
    TpMediaStreamDirection dir);
static void tf_stream_free_resource (TfStream *self,
    TpMediaStreamDirection dir);

static void add_remote_candidate (TpMediaStreamHandler *proxy,
    const gchar *candidate, const GPtrArray *transports,
    gpointer user_data, GObject *object);

static void remove_remote_candidate (TpMediaStreamHandler *proxy,
    const gchar *candidate,
    gpointer user_data, GObject *object);

static void set_active_candidate_pair (TpMediaStreamHandler *proxy,
    const gchar *native_candidate, const gchar *remote_candidate,
    gpointer user_data, GObject *object);

static void set_remote_candidate_list (TpMediaStreamHandler *proxy,
    const GPtrArray *candidates, gpointer user_data, GObject *object);

static void set_remote_codecs (TpMediaStreamHandler *proxy,
    const GPtrArray *codecs, gpointer user_data, GObject *object);

static void set_stream_playing (TpMediaStreamHandler *proxy, gboolean play,
    gpointer user_data, GObject *object);
static void set_stream_held (TpMediaStreamHandler *proxy, gboolean held,
    gpointer user_data, GObject *object);

static void set_stream_sending (TpMediaStreamHandler *proxy, gboolean play,
    gpointer user_data, GObject *object);

static void start_telephony_event (TpMediaStreamHandler *proxy, guchar event,
    gpointer user_data, GObject *object);

static void stop_telephony_event (TpMediaStreamHandler *proxy,
    gpointer user_data, GObject *object);

static void close (TpMediaStreamHandler *proxy,
    gpointer user_data, GObject *object);

static void invalidated_cb (TpMediaStreamHandler *proxy,
    guint domain, gint code, gchar *message, gpointer user_data);

static FsMediaType tp_media_type_to_fs (TpMediaStreamType type);

static GPtrArray *fs_codecs_to_tp (TfStream *stream,
    const GList *codecs);
static void async_method_callback (TpMediaStreamHandler *proxy G_GNUC_UNUSED,
    const GError *error,
    gpointer user_data,
    GObject *weak_object);

static void cb_fs_stream_src_pad_added (FsStream *fsstream G_GNUC_UNUSED,
    GstPad *pad,
    FsCodec *codec,
    gpointer user_data);
static void
tf_stream_try_sending_codecs (TfStream *stream);


static void
tf_stream_init (TfStream *self)
{
  TfStreamPrivate *priv = G_TYPE_INSTANCE_GET_PRIVATE (self,
      TF_TYPE_STREAM, TfStreamPrivate);

  self->priv = priv;
  priv->has_resource = TP_MEDIA_STREAM_DIRECTION_NONE;
}

static void
tf_stream_get_property (GObject    *object,
    guint       property_id,
    GValue     *value,
    GParamSpec *pspec)
{
  TfStream *self = TF_STREAM (object);

  switch (property_id)
    {
    case PROP_CHANNEL:
      g_value_set_object (value, self->priv->channel);
      break;
    case PROP_FARSIGHT_CONFERENCE:
      g_value_set_object (value, self->priv->fs_conference);
      break;
    case PROP_FARSIGHT_PARTICIPANT:
      g_value_set_object (value, self->priv->fs_participant);
      break;
    case PROP_PROXY:
      g_value_set_object (value, self->priv->stream_handler_proxy);
      break;
    case PROP_STREAM_ID:
      g_value_set_uint (value, self->stream_id);
      break;
    case PROP_MEDIA_TYPE:
      g_value_set_uint (value, self->priv->media_type);
      break;
    case PROP_DIRECTION:
      g_value_set_uint (value, self->priv->direction);
      break;
    case PROP_NAT_PROPERTIES:
      g_value_set_pointer (value,
          (TfNatProperties *) self->priv->nat_props);
      break;
    case PROP_SINK_PAD:
      g_object_get_property (G_OBJECT (self->priv->fs_session),
          "sink-pad", value);
      break;
    case PROP_LOCAL_PREFERENCES:
      g_value_set_boxed (value, self->priv->local_preferences);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
      break;
    }
}

static void
tf_stream_set_property (GObject      *object,
    guint         property_id,
    const GValue *value,
    GParamSpec   *pspec)
{
  TfStream *self = TF_STREAM (object);

  switch (property_id)
    {
    case PROP_CHANNEL:
      self->priv->channel =
          TF_CHANNEL (g_value_get_object (value));
      break;
    case PROP_FARSIGHT_CONFERENCE:
      self->priv->fs_conference =
          FS_CONFERENCE (g_value_dup_object (value));
      break;
    case PROP_FARSIGHT_PARTICIPANT:
      self->priv->fs_participant =
          FS_PARTICIPANT (g_value_dup_object (value));
      break;
    case PROP_PROXY:
      self->priv->stream_handler_proxy =
          TP_MEDIA_STREAM_HANDLER (g_value_dup_object (value));
      break;
    case PROP_STREAM_ID:
      self->stream_id = g_value_get_uint (value);
      break;
    case PROP_MEDIA_TYPE:
      self->priv->media_type = g_value_get_uint (value);
      break;
    case PROP_DIRECTION:
      self->priv->direction = g_value_get_uint (value);
      break;
    case PROP_NAT_PROPERTIES:
      self->priv->nat_props = g_value_get_pointer (value);
      break;
    case PROP_LOCAL_PREFERENCES:
      self->priv->local_preferences = g_value_dup_boxed (value);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
      break;
    }
}

#define MAX_STREAM_TRANS_PARAMS 6

static GObject *
tf_stream_constructor (GType type,
    guint n_props,
    GObjectConstructParam *props)
{
  GObject *obj;
  TfStream *stream;
  TfStreamPrivate *priv;
  TfStreamClass *klass = NULL;
  gchar *transmitter;
  guint n_args = 0;
  GList *preferred_local_candidates = NULL;
  GParameter params[MAX_STREAM_TRANS_PARAMS];

  obj = G_OBJECT_CLASS (tf_stream_parent_class)->
            constructor (type, n_props, props);
  stream = (TfStream *) obj;
  priv = stream->priv;
  klass = TF_STREAM_GET_CLASS(obj);

  g_signal_connect (priv->stream_handler_proxy, "invalidated",
      G_CALLBACK (invalidated_cb), obj);

  tp_cli_media_stream_handler_connect_to_add_remote_candidate
      (priv->stream_handler_proxy, add_remote_candidate, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_remove_remote_candidate
      (priv->stream_handler_proxy, remove_remote_candidate, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_set_active_candidate_pair
      (priv->stream_handler_proxy, set_active_candidate_pair, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_set_remote_candidate_list
      (priv->stream_handler_proxy, set_remote_candidate_list, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_set_remote_codecs
      (priv->stream_handler_proxy, set_remote_codecs, NULL, NULL, obj, NULL);
  tp_cli_media_stream_handler_connect_to_set_stream_playing
      (priv->stream_handler_proxy, set_stream_playing, NULL, NULL, obj, NULL);
  tp_cli_media_stream_handler_connect_to_set_stream_sending
      (priv->stream_handler_proxy, set_stream_sending, NULL, NULL, obj, NULL);
  tp_cli_media_stream_handler_connect_to_set_stream_held
      (priv->stream_handler_proxy, set_stream_held, NULL, NULL, obj, NULL);
  tp_cli_media_stream_handler_connect_to_start_telephony_event
      (priv->stream_handler_proxy, start_telephony_event, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_stop_telephony_event
      (priv->stream_handler_proxy, stop_telephony_event, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_close
      (priv->stream_handler_proxy, close, NULL, NULL, obj, NULL);

  memset (params, 0, sizeof(GParameter) * MAX_STREAM_TRANS_PARAMS);

  if (stream->priv->nat_props == NULL ||
      stream->priv->nat_props->nat_traversal == NULL ||
      !strcmp (stream->priv->nat_props->nat_traversal, "gtalk-p2p"))
    {
      transmitter = "nice";

      params[n_args].name = "compatibility-mode";
      g_value_init (&params[n_args].value, G_TYPE_UINT);
      g_value_set_uint (&params[n_args].value, 1);
      n_args++;
    }
  else
    {
      transmitter = "rawudp";

      if (stream->priv->media_type == TP_MEDIA_STREAM_TYPE_AUDIO)
        preferred_local_candidates = g_list_prepend (NULL,
            fs_candidate_new (NULL, FS_COMPONENT_RTP, FS_CANDIDATE_TYPE_HOST,
                FS_NETWORK_PROTOCOL_UDP, NULL, 7078));
      else if (stream->priv->media_type == TP_MEDIA_STREAM_TYPE_VIDEO)
        preferred_local_candidates = g_list_prepend (NULL,
            fs_candidate_new (NULL, FS_COMPONENT_RTP, FS_CANDIDATE_TYPE_HOST,
                FS_NETWORK_PROTOCOL_UDP, NULL, 9078));
    }

  if (stream->priv->nat_props &&
      stream->priv->nat_props->stun_server &&
      stream->priv->nat_props->stun_port)
    {
      gchar *conn_timeout_str = NULL;

      params[n_args].name = "stun-ip";
      g_value_init (&params[n_args].value, G_TYPE_STRING);
      g_value_set_string (&params[n_args].value,
          stream->priv->nat_props->stun_server);
      n_args++;

      params[n_args].name = "stun-port";
      g_value_init (&params[n_args].value, G_TYPE_UINT);
      g_value_set_uint (&params[n_args].value,
          stream->priv->nat_props->stun_port);
      n_args++;

      conn_timeout_str = getenv ("FS_CONN_TIMEOUT");

      if (conn_timeout_str)
        {
          gint conn_timeout = strtol (conn_timeout_str, NULL, 10);

          params[n_args].name = "stun-timeout";
          g_value_init (&params[n_args].value, G_TYPE_UINT);
          g_value_set_uint (&params[n_args].value, conn_timeout);
          n_args++;
        }
    }

  if (preferred_local_candidates)
    {
      params[n_args].name = "preferred-local-candidates";
      g_value_init (&params[n_args].value, FS_TYPE_CANDIDATE_LIST);
      g_value_take_boxed (&params[n_args].value,
          preferred_local_candidates);
      n_args++;
    }

  stream->priv->fs_session = fs_conference_new_session (
      stream->priv->fs_conference,
      tp_media_type_to_fs (stream->priv->media_type),
      &stream->priv->construction_error);

  if (!stream->priv->fs_session)
    return obj;

  stream->priv->fs_stream = fs_session_new_stream (stream->priv->fs_session,
      stream->priv->fs_participant,
      FS_DIRECTION_NONE,
      transmitter,
      n_args,
      params,
      &stream->priv->construction_error);

  if (!stream->priv->fs_stream)
    return obj;

  if (stream->priv->local_preferences)
    if (!fs_session_set_codec_preferences (stream->priv->fs_session,
            stream->priv->local_preferences,
            &stream->priv->construction_error))
      return obj;


  if (g_object_class_find_property (
          G_OBJECT_GET_CLASS (stream->priv->fs_session),
          "no-rtcp-timeout"))
    g_object_set (stream->priv->fs_session, "no-rtcp-timeout", 0, NULL);

  g_signal_connect (stream->priv->fs_stream, "src-pad-added",
      G_CALLBACK (cb_fs_stream_src_pad_added), stream);

  stream->priv->send_local_codecs = TRUE;
  tf_stream_try_sending_codecs (stream);

  return obj;
}

static void
tf_stream_dispose (GObject *object)
{
  TfStream *stream = TF_STREAM (object);
  TfStreamPrivate *priv = stream->priv;

  if (priv->stream_handler_proxy)
    {
      TpMediaStreamHandler *tmp = priv->stream_handler_proxy;

      g_signal_handlers_disconnect_by_func (
          priv->stream_handler_proxy, invalidated_cb, stream);

      priv->stream_handler_proxy = NULL;
      g_object_unref (tmp);
    }

  if (priv->fs_stream)
    {
      tf_stream_free_resource (stream,
          TP_MEDIA_STREAM_DIRECTION_SEND);

      g_object_unref (priv->fs_stream);

      tf_stream_free_resource (stream,
          TP_MEDIA_STREAM_DIRECTION_RECEIVE);

      priv->fs_stream = NULL;
    }

  if (priv->fs_session)
    {
      g_object_unref (priv->fs_session);
      priv->fs_session = NULL;
    }

  if (priv->local_preferences)
    {
      fs_codec_list_destroy (priv->local_preferences);
      priv->local_preferences = NULL;
    }

  fs_candidate_list_destroy (priv->local_candidates);
  priv->local_candidates = NULL;

  if (G_OBJECT_CLASS (tf_stream_parent_class)->dispose)
    G_OBJECT_CLASS (tf_stream_parent_class)->dispose (object);
}

static void
tf_stream_class_init (TfStreamClass *klass)
{
  GObjectClass *object_class = G_OBJECT_CLASS (klass);

  g_type_class_add_private (klass, sizeof (TfStreamPrivate));

  object_class->set_property = tf_stream_set_property;
  object_class->get_property = tf_stream_get_property;
  object_class->constructor = tf_stream_constructor;
  object_class->dispose = tf_stream_dispose;

  g_object_class_install_property (object_class, PROP_CHANNEL,
      g_param_spec_object ("channel",
          "Telepathy channel",
          "The TfChannel this stream is in",
          TF_TYPE_CHANNEL,
          G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_FARSIGHT_CONFERENCE,
      g_param_spec_object ("farsight-conference",
          "Farsight conference",
          "The Farsight conference this stream will "
          "create streams within.",
          FS_TYPE_CONFERENCE,
          G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_FARSIGHT_PARTICIPANT,
      g_param_spec_object ("farsight-participant",
          "Farsight participant",
          "The Farsight participant this stream will "
          "create streams for.",
          FS_TYPE_PARTICIPANT,
          G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_PROXY,
      g_param_spec_object ("proxy",
          "TpMediaStreamHandler proxy",
          "The stream handler proxy which this stream interacts with.",
          TP_TYPE_MEDIA_STREAM_HANDLER,
          G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE |G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_STREAM_ID,
      g_param_spec_uint ("stream-id",
          "stream ID",
          "A number identifying this stream within "
          "its channel.",
          0, G_MAXUINT, 0,
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_READWRITE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_MEDIA_TYPE,
      g_param_spec_uint ("media-type",
          "stream media type",
          "The Telepathy stream media type"
          " (as a TpStreamMediaType)",
          TP_MEDIA_STREAM_TYPE_AUDIO,
          TP_MEDIA_STREAM_TYPE_VIDEO,
          TP_MEDIA_STREAM_TYPE_AUDIO,
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_READWRITE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_DIRECTION,
      g_param_spec_uint ("direction",
          "stream direction",
          "The Telepathy stream direction"
          " (a TpMediaStreamDirection)",
          TP_MEDIA_STREAM_DIRECTION_NONE,
          TP_MEDIA_STREAM_DIRECTION_BIDIRECTIONAL,
          TP_MEDIA_STREAM_DIRECTION_BIDIRECTIONAL,
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_READWRITE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_NAT_PROPERTIES,
      g_param_spec_pointer ("nat-properties",
          "NAT properties",
          "A pointer to a "
          "TfNatProperties structure "
          "detailing which NAT traversal method "
          "and parameters to use for this stream",
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_WRITABLE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_SINK_PAD,
      g_param_spec_object ("sink-pad",
          "Sink pad for this stream",
          "This sink pad that data has to be sent",
          GST_TYPE_PAD,
          G_PARAM_READABLE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_LOCAL_PREFERENCES,
      g_param_spec_boxed ("codec-preferences",
          "Local codec preferences",
          "A GList of FsCodec representing preferences"
          " to be passed to the"
          " fs_session_set_local_preferences()"
          " function",
          FS_TYPE_CODEC_LIST,
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_READWRITE |
          G_PARAM_STATIC_STRINGS));

  /**
   * TfStream::closed:
   * @stream: the stream that has been closed
   *
   * This signal is emitted when the Close() signal is received from the
   * connection manager.
   */

  signals[CLOSED] =
      g_signal_new ("closed",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          NULL, NULL,
          g_cclosure_marshal_VOID__VOID,
          G_TYPE_NONE, 0);

  /**
   * TfStream::error:
   * @stream: the stream that has been errored
   *
   * This signal is emitted when there is an error on this stream
   */

  signals[ERROR_SIGNAL] =
      g_signal_new ("error",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          NULL, NULL,
          g_cclosure_marshal_VOID__VOID,
          G_TYPE_NONE, 0);

  /**
   * TfStream::request-resource:
   * @stream: the stream requesting the resources
   * @direction: The direction for which this resource is requested
   *  (as a #TpMediaDirection
   *
   * This signal is emitted when the connection manager ask to send or receive
   * media. For example, this can be used allocated an X window or open a
   * camera. The resouces can later be freed on #TfStream::free-resource
   *
   * Returns: %TRUE if the resources requested could be allocated or %FALSE
   * otherwise
   */

  signals[REQUEST_RESOURCE] =
      g_signal_new ("request-resource",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          g_signal_accumulator_true_handled, NULL,
          _tf_marshal_BOOLEAN__UINT,
          G_TYPE_BOOLEAN, 1, G_TYPE_UINT);

  /**
   * TfStream::free-resource:
   * @stream: the stream for which resources can be freed
   * @direction: The direction for which this resource is freed
   *  (as a #TpMediaDirection
   *
   * Emitted when the stream no longer needs a resource allocated
   * from #TfStream::request-resource and it can be freed.
   */

  signals[FREE_RESOURCE] =
      g_signal_new ("free-resource",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          NULL, NULL,
          g_cclosure_marshal_VOID__UINT,
          G_TYPE_NONE, 1, G_TYPE_UINT);

  /**
   * TfStream::src-pad-added:
   * @stream: the stream which has a new pad
   * @pad: The new src pad
   * @codec: the codec for which data is coming out
   *
   * This is emitted when a new src pad comes out. The user must connect
   * this pad to his pipeline.
   */

  signals[SRC_PAD_ADDED] =
      g_signal_new ("src-pad-added",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          NULL, NULL,
          _tf_marshal_VOID__OBJECT_BOXED,
          G_TYPE_NONE, 2, GST_TYPE_PAD, FS_TYPE_CODEC);
}

/* dummy callback handler for async calling calls with no return values */
static void
async_method_callback (TpMediaStreamHandler *proxy G_GNUC_UNUSED,
    const GError *error,
    gpointer user_data,
    GObject *weak_object)
{
  TfStream *self = TF_STREAM (weak_object);

  if (error != NULL)
    {
      g_warning ("Error calling %s: %s", (gchar *) user_data, error->message);
      g_signal_emit (self, signals[ERROR_SIGNAL], 0);
    }
}


static void
cb_fs_new_local_candidate (TfStream *self,
    FsCandidate *candidate)
{
  DEBUG (self, "called");

  self->priv->local_candidates = g_list_append (self->priv->local_candidates,
      fs_candidate_copy (candidate));
}

static void
cb_fs_local_candidates_prepared (TfStream *self)
{
  DEBUG (self, "called");

  while (self->priv->local_candidates)
    {
      GPtrArray *transports = g_ptr_array_new ();
      FsCandidate *candidate =
          g_list_first (self->priv->local_candidates)->data;
      gchar *foundation = g_strdup (candidate->foundation);

      while (candidate)
        {
          GValue transport = { 0, };
          TpMediaStreamBaseProto proto;
          TpMediaStreamTransportType type;
          GList *item = NULL;

          g_value_init (&transport,
              TP_STRUCT_TYPE_MEDIA_STREAM_HANDLER_TRANSPORT);
          g_value_set_static_boxed (&transport,
              dbus_g_type_specialized_construct (
                  TP_STRUCT_TYPE_MEDIA_STREAM_HANDLER_TRANSPORT));

          switch (candidate->proto) {
          case FS_NETWORK_PROTOCOL_UDP:
            proto = TP_MEDIA_STREAM_BASE_PROTO_UDP;
            break;
          case FS_NETWORK_PROTOCOL_TCP:
            proto = TP_MEDIA_STREAM_BASE_PROTO_TCP;
            break;
          default:
            g_critical ("%s: FarsightTransportInfo.proto has an invalid value",
                G_STRFUNC);
            return;
          }

          switch (candidate->type) {
          case FS_CANDIDATE_TYPE_HOST:
            type = TP_MEDIA_STREAM_TRANSPORT_TYPE_LOCAL;
            break;
          case FS_CANDIDATE_TYPE_SRFLX:
          case FS_CANDIDATE_TYPE_PRFLX:
            type = TP_MEDIA_STREAM_TRANSPORT_TYPE_DERIVED;
            break;
          case FS_CANDIDATE_TYPE_RELAY:
            type = TP_MEDIA_STREAM_TRANSPORT_TYPE_RELAY;
            break;
          default:
            g_critical ("%s: FarsightTransportInfo.proto has an invalid value",
                G_STRFUNC);
            return;
          }

          DEBUG (self, "candidate->ip = '%s'", candidate->ip);

          dbus_g_type_struct_set (&transport,
              0, candidate->component_id,
              1, candidate->ip,
              2, candidate->port,
              3, proto,
              4, "RTP",
              5, "AVP",
              6, (double) candidate->priority / 65536.0,
              7, type,
              8, candidate->username,
              9, candidate->password,
              G_MAXUINT);

          g_ptr_array_add (transports, g_value_dup_boxed (&transport));

          self->priv->local_candidates = g_list_remove (
              self->priv->local_candidates, candidate);

          fs_candidate_destroy (candidate);

          for (item = self->priv->local_candidates;
               item;
               item = g_list_next (item))
            {
              FsCandidate *tmpcand = item->data;
              if (!strcmp (tmpcand->foundation, foundation))
                break;
            }
          if (item)
            candidate = item->data;
          else
            candidate = NULL;
        }

      tp_cli_media_stream_handler_call_new_native_candidate (
          self->priv->stream_handler_proxy, -1, foundation, transports,
          async_method_callback,
          "Media.StreamHandler::NativeCandidatesPrepared",
          NULL, (GObject *) self);

      g_free (foundation);
    }

  tp_cli_media_stream_handler_call_native_candidates_prepared (
    self->priv->stream_handler_proxy, -1, async_method_callback,
    "Media.StreamHandler::NativeCandidatesPrepared",
    NULL, (GObject *) self);
}



/*
 * small helper function to help converting a
 * telepathy dbus candidate to a list of FarsightTransportInfos
 * nothing is copied, so always keep the usage of this within a function
 * if you need to do multiple candidates, call this repeatedly and
 * g_list_join them together.
 * Free the list using free_fs_transports
 */
static GList *
tp_transports_to_fs (const gchar* foundation, const GPtrArray *transports)
{
  GList *fs_trans_list = NULL;
  GValueArray *transport;
  FsCandidate *fs_candidate;
  guint i;

  for (i=0; i< transports->len; i++)
    {
      transport = g_ptr_array_index (transports, i);
      FsNetworkProtocol proto;<commit_msg>Update comment to match code<commit_after>/*
 * stream.c - Source for TfStream
 * Copyright (C) 2006-2008 Collabora Ltd.
 * Copyright (C) 2006-2008 Nokia Corporation
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 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
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
 */

#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#include <string.h>
#include <stdlib.h>

#include <telepathy-glib/dbus.h>
#include <telepathy-glib/errors.h>
#include <telepathy-glib/gtypes.h>
#include <telepathy-glib/interfaces.h>

#include <gst/interfaces/xoverlay.h>

#include <gst/farsight/fs-conference-iface.h>

#include "stream.h"
#include "stream-priv.h"
#include "channel.h"
#include "tf-signals-marshal.h"

G_DEFINE_TYPE (TfStream, tf_stream, G_TYPE_OBJECT);

#define DEBUG(stream, format, ...) \
  g_debug ("stream %d (%s) %s: " format, \
    stream->stream_id, \
    (stream->priv->media_type == FS_MEDIA_TYPE_AUDIO) ? "audio" \
                                                      : "video", \
    G_STRFUNC, \
    ##__VA_ARGS__)

#define WARNING(stream, format, ...) \
  g_warning ("stream %d (%s) %s: " format, \
    stream->stream_id, \
    (stream->priv->media_type == FS_MEDIA_TYPE_AUDIO) ? "audio" \
                                                      : "video", \
    G_STRFUNC, \
    ##__VA_ARGS__)

#define STREAM_PRIVATE(o) ((o)->priv)

struct _TfStreamPrivate
{
  TfChannel *channel;
  FsConference *fs_conference;
  FsParticipant *fs_participant;
  FsSession *fs_session;
  FsStream *fs_stream;
  TpMediaStreamType media_type;
  TpMediaStreamDirection direction;
  const TfNatProperties *nat_props;
  GList *local_preferences;

  GError *construction_error;

  TpMediaStreamHandler *stream_handler_proxy;

  FsStreamDirection desired_direction;
  gboolean held;
  TpMediaStreamDirection has_resource;

  GList *local_candidates;

  gboolean send_local_codecs;
  gboolean send_supported_codecs;

  gboolean gathering;
};

enum
{
  CLOSED,
  ERROR_SIGNAL,
  REQUEST_RESOURCE,
  FREE_RESOURCE,
  SRC_PAD_ADDED,
  SIGNAL_COUNT
};

static guint signals[SIGNAL_COUNT] = {0};

/* properties */
enum
{
  PROP_CHANNEL = 1,
  PROP_FARSIGHT_CONFERENCE,
  PROP_FARSIGHT_PARTICIPANT,
  PROP_PROXY,
  PROP_STREAM_ID,
  PROP_MEDIA_TYPE,
  PROP_DIRECTION,
  PROP_NAT_PROPERTIES,
  PROP_SINK_PAD,
  PROP_LOCAL_PREFERENCES
};


static gboolean tf_stream_request_resource (
    TfStream *self,
    TpMediaStreamDirection dir);
static void tf_stream_free_resource (TfStream *self,
    TpMediaStreamDirection dir);

static void add_remote_candidate (TpMediaStreamHandler *proxy,
    const gchar *candidate, const GPtrArray *transports,
    gpointer user_data, GObject *object);

static void remove_remote_candidate (TpMediaStreamHandler *proxy,
    const gchar *candidate,
    gpointer user_data, GObject *object);

static void set_active_candidate_pair (TpMediaStreamHandler *proxy,
    const gchar *native_candidate, const gchar *remote_candidate,
    gpointer user_data, GObject *object);

static void set_remote_candidate_list (TpMediaStreamHandler *proxy,
    const GPtrArray *candidates, gpointer user_data, GObject *object);

static void set_remote_codecs (TpMediaStreamHandler *proxy,
    const GPtrArray *codecs, gpointer user_data, GObject *object);

static void set_stream_playing (TpMediaStreamHandler *proxy, gboolean play,
    gpointer user_data, GObject *object);
static void set_stream_held (TpMediaStreamHandler *proxy, gboolean held,
    gpointer user_data, GObject *object);

static void set_stream_sending (TpMediaStreamHandler *proxy, gboolean play,
    gpointer user_data, GObject *object);

static void start_telephony_event (TpMediaStreamHandler *proxy, guchar event,
    gpointer user_data, GObject *object);

static void stop_telephony_event (TpMediaStreamHandler *proxy,
    gpointer user_data, GObject *object);

static void close (TpMediaStreamHandler *proxy,
    gpointer user_data, GObject *object);

static void invalidated_cb (TpMediaStreamHandler *proxy,
    guint domain, gint code, gchar *message, gpointer user_data);

static FsMediaType tp_media_type_to_fs (TpMediaStreamType type);

static GPtrArray *fs_codecs_to_tp (TfStream *stream,
    const GList *codecs);
static void async_method_callback (TpMediaStreamHandler *proxy G_GNUC_UNUSED,
    const GError *error,
    gpointer user_data,
    GObject *weak_object);

static void cb_fs_stream_src_pad_added (FsStream *fsstream G_GNUC_UNUSED,
    GstPad *pad,
    FsCodec *codec,
    gpointer user_data);
static void
tf_stream_try_sending_codecs (TfStream *stream);


static void
tf_stream_init (TfStream *self)
{
  TfStreamPrivate *priv = G_TYPE_INSTANCE_GET_PRIVATE (self,
      TF_TYPE_STREAM, TfStreamPrivate);

  self->priv = priv;
  priv->has_resource = TP_MEDIA_STREAM_DIRECTION_NONE;
}

static void
tf_stream_get_property (GObject    *object,
    guint       property_id,
    GValue     *value,
    GParamSpec *pspec)
{
  TfStream *self = TF_STREAM (object);

  switch (property_id)
    {
    case PROP_CHANNEL:
      g_value_set_object (value, self->priv->channel);
      break;
    case PROP_FARSIGHT_CONFERENCE:
      g_value_set_object (value, self->priv->fs_conference);
      break;
    case PROP_FARSIGHT_PARTICIPANT:
      g_value_set_object (value, self->priv->fs_participant);
      break;
    case PROP_PROXY:
      g_value_set_object (value, self->priv->stream_handler_proxy);
      break;
    case PROP_STREAM_ID:
      g_value_set_uint (value, self->stream_id);
      break;
    case PROP_MEDIA_TYPE:
      g_value_set_uint (value, self->priv->media_type);
      break;
    case PROP_DIRECTION:
      g_value_set_uint (value, self->priv->direction);
      break;
    case PROP_NAT_PROPERTIES:
      g_value_set_pointer (value,
          (TfNatProperties *) self->priv->nat_props);
      break;
    case PROP_SINK_PAD:
      g_object_get_property (G_OBJECT (self->priv->fs_session),
          "sink-pad", value);
      break;
    case PROP_LOCAL_PREFERENCES:
      g_value_set_boxed (value, self->priv->local_preferences);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
      break;
    }
}

static void
tf_stream_set_property (GObject      *object,
    guint         property_id,
    const GValue *value,
    GParamSpec   *pspec)
{
  TfStream *self = TF_STREAM (object);

  switch (property_id)
    {
    case PROP_CHANNEL:
      self->priv->channel =
          TF_CHANNEL (g_value_get_object (value));
      break;
    case PROP_FARSIGHT_CONFERENCE:
      self->priv->fs_conference =
          FS_CONFERENCE (g_value_dup_object (value));
      break;
    case PROP_FARSIGHT_PARTICIPANT:
      self->priv->fs_participant =
          FS_PARTICIPANT (g_value_dup_object (value));
      break;
    case PROP_PROXY:
      self->priv->stream_handler_proxy =
          TP_MEDIA_STREAM_HANDLER (g_value_dup_object (value));
      break;
    case PROP_STREAM_ID:
      self->stream_id = g_value_get_uint (value);
      break;
    case PROP_MEDIA_TYPE:
      self->priv->media_type = g_value_get_uint (value);
      break;
    case PROP_DIRECTION:
      self->priv->direction = g_value_get_uint (value);
      break;
    case PROP_NAT_PROPERTIES:
      self->priv->nat_props = g_value_get_pointer (value);
      break;
    case PROP_LOCAL_PREFERENCES:
      self->priv->local_preferences = g_value_dup_boxed (value);
      break;
    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, property_id, pspec);
      break;
    }
}

#define MAX_STREAM_TRANS_PARAMS 6

static GObject *
tf_stream_constructor (GType type,
    guint n_props,
    GObjectConstructParam *props)
{
  GObject *obj;
  TfStream *stream;
  TfStreamPrivate *priv;
  TfStreamClass *klass = NULL;
  gchar *transmitter;
  guint n_args = 0;
  GList *preferred_local_candidates = NULL;
  GParameter params[MAX_STREAM_TRANS_PARAMS];

  obj = G_OBJECT_CLASS (tf_stream_parent_class)->
            constructor (type, n_props, props);
  stream = (TfStream *) obj;
  priv = stream->priv;
  klass = TF_STREAM_GET_CLASS(obj);

  g_signal_connect (priv->stream_handler_proxy, "invalidated",
      G_CALLBACK (invalidated_cb), obj);

  tp_cli_media_stream_handler_connect_to_add_remote_candidate
      (priv->stream_handler_proxy, add_remote_candidate, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_remove_remote_candidate
      (priv->stream_handler_proxy, remove_remote_candidate, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_set_active_candidate_pair
      (priv->stream_handler_proxy, set_active_candidate_pair, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_set_remote_candidate_list
      (priv->stream_handler_proxy, set_remote_candidate_list, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_set_remote_codecs
      (priv->stream_handler_proxy, set_remote_codecs, NULL, NULL, obj, NULL);
  tp_cli_media_stream_handler_connect_to_set_stream_playing
      (priv->stream_handler_proxy, set_stream_playing, NULL, NULL, obj, NULL);
  tp_cli_media_stream_handler_connect_to_set_stream_sending
      (priv->stream_handler_proxy, set_stream_sending, NULL, NULL, obj, NULL);
  tp_cli_media_stream_handler_connect_to_set_stream_held
      (priv->stream_handler_proxy, set_stream_held, NULL, NULL, obj, NULL);
  tp_cli_media_stream_handler_connect_to_start_telephony_event
      (priv->stream_handler_proxy, start_telephony_event, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_stop_telephony_event
      (priv->stream_handler_proxy, stop_telephony_event, NULL, NULL, obj,
       NULL);
  tp_cli_media_stream_handler_connect_to_close
      (priv->stream_handler_proxy, close, NULL, NULL, obj, NULL);

  memset (params, 0, sizeof(GParameter) * MAX_STREAM_TRANS_PARAMS);

  if (stream->priv->nat_props == NULL ||
      stream->priv->nat_props->nat_traversal == NULL ||
      !strcmp (stream->priv->nat_props->nat_traversal, "gtalk-p2p"))
    {
      transmitter = "nice";

      params[n_args].name = "compatibility-mode";
      g_value_init (&params[n_args].value, G_TYPE_UINT);
      g_value_set_uint (&params[n_args].value, 1);
      n_args++;
    }
  else
    {
      transmitter = "rawudp";

      if (stream->priv->media_type == TP_MEDIA_STREAM_TYPE_AUDIO)
        preferred_local_candidates = g_list_prepend (NULL,
            fs_candidate_new (NULL, FS_COMPONENT_RTP, FS_CANDIDATE_TYPE_HOST,
                FS_NETWORK_PROTOCOL_UDP, NULL, 7078));
      else if (stream->priv->media_type == TP_MEDIA_STREAM_TYPE_VIDEO)
        preferred_local_candidates = g_list_prepend (NULL,
            fs_candidate_new (NULL, FS_COMPONENT_RTP, FS_CANDIDATE_TYPE_HOST,
                FS_NETWORK_PROTOCOL_UDP, NULL, 9078));
    }

  if (stream->priv->nat_props &&
      stream->priv->nat_props->stun_server &&
      stream->priv->nat_props->stun_port)
    {
      gchar *conn_timeout_str = NULL;

      params[n_args].name = "stun-ip";
      g_value_init (&params[n_args].value, G_TYPE_STRING);
      g_value_set_string (&params[n_args].value,
          stream->priv->nat_props->stun_server);
      n_args++;

      params[n_args].name = "stun-port";
      g_value_init (&params[n_args].value, G_TYPE_UINT);
      g_value_set_uint (&params[n_args].value,
          stream->priv->nat_props->stun_port);
      n_args++;

      conn_timeout_str = getenv ("FS_CONN_TIMEOUT");

      if (conn_timeout_str)
        {
          gint conn_timeout = strtol (conn_timeout_str, NULL, 10);

          params[n_args].name = "stun-timeout";
          g_value_init (&params[n_args].value, G_TYPE_UINT);
          g_value_set_uint (&params[n_args].value, conn_timeout);
          n_args++;
        }
    }

  if (preferred_local_candidates)
    {
      params[n_args].name = "preferred-local-candidates";
      g_value_init (&params[n_args].value, FS_TYPE_CANDIDATE_LIST);
      g_value_take_boxed (&params[n_args].value,
          preferred_local_candidates);
      n_args++;
    }

  stream->priv->fs_session = fs_conference_new_session (
      stream->priv->fs_conference,
      tp_media_type_to_fs (stream->priv->media_type),
      &stream->priv->construction_error);

  if (!stream->priv->fs_session)
    return obj;

  stream->priv->fs_stream = fs_session_new_stream (stream->priv->fs_session,
      stream->priv->fs_participant,
      FS_DIRECTION_NONE,
      transmitter,
      n_args,
      params,
      &stream->priv->construction_error);

  if (!stream->priv->fs_stream)
    return obj;

  if (stream->priv->local_preferences)
    if (!fs_session_set_codec_preferences (stream->priv->fs_session,
            stream->priv->local_preferences,
            &stream->priv->construction_error))
      return obj;


  if (g_object_class_find_property (
          G_OBJECT_GET_CLASS (stream->priv->fs_session),
          "no-rtcp-timeout"))
    g_object_set (stream->priv->fs_session, "no-rtcp-timeout", 0, NULL);

  g_signal_connect (stream->priv->fs_stream, "src-pad-added",
      G_CALLBACK (cb_fs_stream_src_pad_added), stream);

  stream->priv->send_local_codecs = TRUE;
  tf_stream_try_sending_codecs (stream);

  return obj;
}

static void
tf_stream_dispose (GObject *object)
{
  TfStream *stream = TF_STREAM (object);
  TfStreamPrivate *priv = stream->priv;

  if (priv->stream_handler_proxy)
    {
      TpMediaStreamHandler *tmp = priv->stream_handler_proxy;

      g_signal_handlers_disconnect_by_func (
          priv->stream_handler_proxy, invalidated_cb, stream);

      priv->stream_handler_proxy = NULL;
      g_object_unref (tmp);
    }

  if (priv->fs_stream)
    {
      tf_stream_free_resource (stream,
          TP_MEDIA_STREAM_DIRECTION_SEND);

      g_object_unref (priv->fs_stream);

      tf_stream_free_resource (stream,
          TP_MEDIA_STREAM_DIRECTION_RECEIVE);

      priv->fs_stream = NULL;
    }

  if (priv->fs_session)
    {
      g_object_unref (priv->fs_session);
      priv->fs_session = NULL;
    }

  if (priv->local_preferences)
    {
      fs_codec_list_destroy (priv->local_preferences);
      priv->local_preferences = NULL;
    }

  fs_candidate_list_destroy (priv->local_candidates);
  priv->local_candidates = NULL;

  if (G_OBJECT_CLASS (tf_stream_parent_class)->dispose)
    G_OBJECT_CLASS (tf_stream_parent_class)->dispose (object);
}

static void
tf_stream_class_init (TfStreamClass *klass)
{
  GObjectClass *object_class = G_OBJECT_CLASS (klass);

  g_type_class_add_private (klass, sizeof (TfStreamPrivate));

  object_class->set_property = tf_stream_set_property;
  object_class->get_property = tf_stream_get_property;
  object_class->constructor = tf_stream_constructor;
  object_class->dispose = tf_stream_dispose;

  g_object_class_install_property (object_class, PROP_CHANNEL,
      g_param_spec_object ("channel",
          "Telepathy channel",
          "The TfChannel this stream is in",
          TF_TYPE_CHANNEL,
          G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_FARSIGHT_CONFERENCE,
      g_param_spec_object ("farsight-conference",
          "Farsight conference",
          "The Farsight conference this stream will "
          "create streams within.",
          FS_TYPE_CONFERENCE,
          G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_FARSIGHT_PARTICIPANT,
      g_param_spec_object ("farsight-participant",
          "Farsight participant",
          "The Farsight participant this stream will "
          "create streams for.",
          FS_TYPE_PARTICIPANT,
          G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_PROXY,
      g_param_spec_object ("proxy",
          "TpMediaStreamHandler proxy",
          "The stream handler proxy which this stream interacts with.",
          TP_TYPE_MEDIA_STREAM_HANDLER,
          G_PARAM_CONSTRUCT_ONLY | G_PARAM_READWRITE |G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_STREAM_ID,
      g_param_spec_uint ("stream-id",
          "stream ID",
          "A number identifying this stream within "
          "its channel.",
          0, G_MAXUINT, 0,
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_READWRITE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_MEDIA_TYPE,
      g_param_spec_uint ("media-type",
          "stream media type",
          "The Telepathy stream media type"
          " (as a TpStreamMediaType)",
          TP_MEDIA_STREAM_TYPE_AUDIO,
          TP_MEDIA_STREAM_TYPE_VIDEO,
          TP_MEDIA_STREAM_TYPE_AUDIO,
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_READWRITE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_DIRECTION,
      g_param_spec_uint ("direction",
          "stream direction",
          "The Telepathy stream direction"
          " (a TpMediaStreamDirection)",
          TP_MEDIA_STREAM_DIRECTION_NONE,
          TP_MEDIA_STREAM_DIRECTION_BIDIRECTIONAL,
          TP_MEDIA_STREAM_DIRECTION_BIDIRECTIONAL,
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_READWRITE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_NAT_PROPERTIES,
      g_param_spec_pointer ("nat-properties",
          "NAT properties",
          "A pointer to a "
          "TfNatProperties structure "
          "detailing which NAT traversal method "
          "and parameters to use for this stream",
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_WRITABLE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_SINK_PAD,
      g_param_spec_object ("sink-pad",
          "Sink pad for this stream",
          "This sink pad that data has to be sent",
          GST_TYPE_PAD,
          G_PARAM_READABLE |
          G_PARAM_STATIC_STRINGS));

  g_object_class_install_property (object_class, PROP_LOCAL_PREFERENCES,
      g_param_spec_boxed ("codec-preferences",
          "Local codec preferences",
          "A GList of FsCodec representing preferences"
          " to be passed to the"
          " fs_session_set_local_preferences()"
          " function",
          FS_TYPE_CODEC_LIST,
          G_PARAM_CONSTRUCT_ONLY |
          G_PARAM_READWRITE |
          G_PARAM_STATIC_STRINGS));

  /**
   * TfStream::closed:
   * @stream: the stream that has been closed
   *
   * This signal is emitted when the Close() signal is received from the
   * connection manager.
   */

  signals[CLOSED] =
      g_signal_new ("closed",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          NULL, NULL,
          g_cclosure_marshal_VOID__VOID,
          G_TYPE_NONE, 0);

  /**
   * TfStream::error:
   * @stream: the stream that has been errored
   *
   * This signal is emitted when there is an error on this stream
   */

  signals[ERROR_SIGNAL] =
      g_signal_new ("error",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          NULL, NULL,
          g_cclosure_marshal_VOID__VOID,
          G_TYPE_NONE, 0);

  /**
   * TfStream::request-resource:
   * @stream: the stream requesting the resources
   * @direction: The direction for which this resource is requested
   *  (as a #TpMediaDirection
   *
   * This signal is emitted when the connection manager ask to send or receive
   * media. For example, this can be used allocated an X window or open a
   * camera. The resouces can later be freed on #TfStream::free-resource
   *
   * Returns: %TRUE if the resources requested could be allocated or %FALSE
   * otherwise
   */

  signals[REQUEST_RESOURCE] =
      g_signal_new ("request-resource",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          g_signal_accumulator_true_handled, NULL,
          _tf_marshal_BOOLEAN__UINT,
          G_TYPE_BOOLEAN, 1, G_TYPE_UINT);

  /**
   * TfStream::free-resource:
   * @stream: the stream for which resources can be freed
   * @direction: The direction for which this resource is freed
   *  (as a #TpMediaDirection
   *
   * Emitted when the stream no longer needs a resource allocated
   * from #TfStream::request-resource and it can be freed.
   */

  signals[FREE_RESOURCE] =
      g_signal_new ("free-resource",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          NULL, NULL,
          g_cclosure_marshal_VOID__UINT,
          G_TYPE_NONE, 1, G_TYPE_UINT);

  /**
   * TfStream::src-pad-added:
   * @stream: the stream which has a new pad
   * @pad: The new src pad
   * @codec: the codec for which data is coming out
   *
   * This is emitted when a new src pad comes out. The user must connect
   * this pad to his pipeline.
   */

  signals[SRC_PAD_ADDED] =
      g_signal_new ("src-pad-added",
          G_OBJECT_CLASS_TYPE (klass),
          G_SIGNAL_RUN_LAST,
          0,
          NULL, NULL,
          _tf_marshal_VOID__OBJECT_BOXED,
          G_TYPE_NONE, 2, GST_TYPE_PAD, FS_TYPE_CODEC);
}

/* dummy callback handler for async calling calls with no return values */
static void
async_method_callback (TpMediaStreamHandler *proxy G_GNUC_UNUSED,
    const GError *error,
    gpointer user_data,
    GObject *weak_object)
{
  TfStream *self = TF_STREAM (weak_object);

  if (error != NULL)
    {
      g_warning ("Error calling %s: %s", (gchar *) user_data, error->message);
      g_signal_emit (self, signals[ERROR_SIGNAL], 0);
    }
}


static void
cb_fs_new_local_candidate (TfStream *self,
    FsCandidate *candidate)
{
  DEBUG (self, "called");

  self->priv->local_candidates = g_list_append (self->priv->local_candidates,
      fs_candidate_copy (candidate));
}

static void
cb_fs_local_candidates_prepared (TfStream *self)
{
  DEBUG (self, "called");

  while (self->priv->local_candidates)
    {
      GPtrArray *transports = g_ptr_array_new ();
      FsCandidate *candidate =
          g_list_first (self->priv->local_candidates)->data;
      gchar *foundation = g_strdup (candidate->foundation);

      while (candidate)
        {
          GValue transport = { 0, };
          TpMediaStreamBaseProto proto;
          TpMediaStreamTransportType type;
          GList *item = NULL;

          g_value_init (&transport,
              TP_STRUCT_TYPE_MEDIA_STREAM_HANDLER_TRANSPORT);
          g_value_set_static_boxed (&transport,
              dbus_g_type_specialized_construct (
                  TP_STRUCT_TYPE_MEDIA_STREAM_HANDLER_TRANSPORT));

          switch (candidate->proto) {
          case FS_NETWORK_PROTOCOL_UDP:
            proto = TP_MEDIA_STREAM_BASE_PROTO_UDP;
            break;
          case FS_NETWORK_PROTOCOL_TCP:
            proto = TP_MEDIA_STREAM_BASE_PROTO_TCP;
            break;
          default:
            g_critical ("%s: FarsightTransportInfo.proto has an invalid value",
                G_STRFUNC);
            return;
          }

          switch (candidate->type) {
          case FS_CANDIDATE_TYPE_HOST:
            type = TP_MEDIA_STREAM_TRANSPORT_TYPE_LOCAL;
            break;
          case FS_CANDIDATE_TYPE_SRFLX:
          case FS_CANDIDATE_TYPE_PRFLX:
            type = TP_MEDIA_STREAM_TRANSPORT_TYPE_DERIVED;
            break;
          case FS_CANDIDATE_TYPE_RELAY:
            type = TP_MEDIA_STREAM_TRANSPORT_TYPE_RELAY;
            break;
          default:
            g_critical ("%s: FarsightTransportInfo.proto has an invalid value",
                G_STRFUNC);
            return;
          }

          DEBUG (self, "candidate->ip = '%s'", candidate->ip);

          dbus_g_type_struct_set (&transport,
              0, candidate->component_id,
              1, candidate->ip,
              2, candidate->port,
              3, proto,
              4, "RTP",
              5, "AVP",
              6, (double) candidate->priority / 65536.0,
              7, type,
              8, candidate->username,
              9, candidate->password,
              G_MAXUINT);

          g_ptr_array_add (transports, g_value_dup_boxed (&transport));

          self->priv->local_candidates = g_list_remove (
              self->priv->local_candidates, candidate);

          fs_candidate_destroy (candidate);

          for (item = self->priv->local_candidates;
               item;
               item = g_list_next (item))
            {
              FsCandidate *tmpcand = item->data;
              if (!strcmp (tmpcand->foundation, foundation))
                break;
            }
          if (item)
            candidate = item->data;
          else
            candidate = NULL;
        }

      tp_cli_media_stream_handler_call_new_native_candidate (
          self->priv->stream_handler_proxy, -1, foundation, transports,
          async_method_callback,
          "Media.StreamHandler::NativeCandidatesPrepared",
          NULL, (GObject *) self);

      g_free (foundation);
    }

  tp_cli_media_stream_handler_call_native_candidates_prepared (
    self->priv->stream_handler_proxy, -1, async_method_callback,
    "Media.StreamHandler::NativeCandidatesPrepared",
    NULL, (GObject *) self);
}



/*
 * small helper function to help converting a
 * telepathy dbus candidate to a list of FsCandidate.
 * nothing is copied, so always keep the usage of this within a function
 * Free the result with fs_candidate_list_destroy()
 */
static GList *
tp_transports_to_fs (const gchar* foundation, const GPtrArray *transports)
{
  GList *fs_trans_list = NULL;
  GValueArray *transport;
  FsCandidate *fs_candidate;
  guint i;

  for (i=0; i< transports->len; i++)
    {
      transport = g_ptr_array_index (transports, i);
      FsNetworkProtocol proto;
{'fraction_non_alphanumeric': 0.08834785903751421, 'fraction_numerical': 0.0030314513073133762, 'mean_word_length': 3.114194403305012, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 375, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4411770', 'n_tokens_mistral': 17677, 'n_tokens_neox': 17063, 'n_words': 3569}