libempathy/empathy-tp-file.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * Copyright (C) 2007 Collabora Ltd.
   4  *   @author: Xavier Claessens <xclaesse@gmail.com>
   5  * Copyright (C) 2007 Marco Barisione <marco@barisione.org>
   6  *
   7  * This library is free software; you can redistribute it and/or
   8  * modify it under the terms of the GNU Lesser General Public
   9  * License as published by the Free Software Foundation; either
  10  * version 2.1 of the License, or (at your option) any later version.
  11  *
  12  * This library is distributed in the hope that it will be useful,
  13  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * Lesser General Public License for more details.
  16  *
  17  * You should have received a copy of the GNU Lesser General Public
  18  * License along with this library; if not, write to the Free Software
  19  * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  20  */
  21
  22 #include <config.h>
  23
  24 #include <string.h>
  25 #include <unistd.h>
  26 #include <sys/types.h>
  27 #include <sys/socket.h>
  28 #include <sys/un.h>
  29
  30 #include <glib/gi18n.h>
  31
  32 #include <gio/gio.h>
  33 #include <gio/gunixinputstream.h>
  34 #include <gio/gunixoutputstream.h>
  35
  36 #include <libtelepathy/tp-conn.h>
  37 #include <libtelepathy/tp-helpers.h>
  38 #include <libtelepathy/tp-props-iface.h>
  39
  40 #include <telepathy-glib/proxy-subclass.h>
  41
  42 #include "empathy-tp-file.h"
  43 #include "empathy-contact-factory.h"
  44 #include "empathy-marshal.h"
  45 #include "empathy-time.h"
  46 #include "empathy-utils.h"
  47
  48 #define DEBUG_FLAG EMPATHY_DEBUG_FT
  49 #include "empathy-debug.h"
  50
  51 /**
  52  * SECTION:empathy-tp-file
  53  * @short_description: File channel
  54  * @see_also: #EmpathyTpFile, #EmpathyContact, empathy_send_file()
  55  * @include: libempthy/empathy-tp-file.h
  56  *
  57  * The #EmpathyTpFile object represents a Telepathy file channel.
  58  */
  59
  60 /**
  61  * EMPATHY_TP_FILE_UNKNOWN_SIZE:
  62  *
  63  * Value used for the "size" or "estimated-size" properties when the size of
  64  * the transferred file is unknown.
  65  */
  66
  67 static void empathy_tp_file_class_init (EmpathyTpFileClass *klass);
  68 static void empathy_tp_file_init (EmpathyTpFile *tp_file);
  69 static void tp_file_finalize (GObject *object);
  70 static GObject *tp_file_constructor (GType type, guint n_props,
  71     GObjectConstructParam *props);
  72 static void tp_file_get_property (GObject *object, guint param_id,
  73     GValue *value, GParamSpec *pspec);
  74 static void tp_file_set_property (GObject *object, guint param_id,
  75     const GValue *value, GParamSpec *pspec);
  76 static void tp_file_destroy_cb (TpChannel *file_chan, EmpathyTpFile *tp_file);
  77 static void tp_file_closed_cb (TpChannel *file_chan, EmpathyTpFile *tp_file,
  78     GObject *weak_object);
  79 static void tp_file_state_changed_cb (DBusGProxy *file_iface, guint state,
  80     guint reason, EmpathyTpFile *tp_file);
  81 static void tp_file_transferred_bytes_changed_cb (TpProxy *proxy,
  82     guint64 count, EmpathyTpFile *tp_file, GObject *weak_object);
  83 static void copy_stream (GInputStream *in, GOutputStream *out,
  84     GCancellable *cancellable);
  85
  86 /* EmpathyTpFile object */
  87
  88 #define GET_PRIV(obj) (G_TYPE_INSTANCE_GET_PRIVATE ((obj), \
  89            EMPATHY_TYPE_TP_FILE, EmpathyTpFilePriv))
  90
  91 typedef struct _EmpathyTpFilePriv  EmpathyTpFilePriv;
  92
  93 struct _EmpathyTpFilePriv {
  94   EmpathyContactFactory *factory;
  95   McAccount *account;
  96   gchar *id;
  97   MissionControl *mc;
  98   TpChannel *channel;
  99
 100   EmpathyTpFile *cached_empathy_file;
 101   EmpathyContact *contact;
 102   GInputStream *in_stream;
 103   GOutputStream *out_stream;
 104   gboolean incoming;
 105   gchar *filename;
 106   EmpFileTransferState state;
 107   EmpFileTransferStateChangeReason state_change_reason;
 108   guint64 size;
 109   guint64 transferred_bytes;
 110   gint64 start_time;
 111   gchar *unix_socket_path;
 112   gchar *content_hash;
 113   EmpFileHashType content_hash_type;
 114   gchar *content_type;
 115   gchar *description;
 116   GCancellable *cancellable;
 117 };
 118
 119 enum {
 120   PROP_0,
 121   PROP_ACCOUNT,
 122   PROP_CHANNEL,
 123   PROP_STATE,
 124   PROP_INCOMING,
 125   PROP_FILENAME,
 126   PROP_SIZE,
 127   PROP_CONTENT_TYPE,
 128   PROP_TRANSFERRED_BYTES,
 129   PROP_CONTENT_HASH_TYPE,
 130   PROP_CONTENT_HASH,
 131   PROP_IN_STREAM,
 132 };
 133
 134 G_DEFINE_TYPE (EmpathyTpFile, empathy_tp_file, G_TYPE_OBJECT);
 135
 136 static void
 137 empathy_tp_file_class_init (EmpathyTpFileClass *klass)
 138 {
 139   GObjectClass *object_class = G_OBJECT_CLASS (klass);
 140
 141   object_class->finalize = tp_file_finalize;
 142   object_class->constructor = tp_file_constructor;
 143   object_class->get_property = tp_file_get_property;
 144   object_class->set_property = tp_file_set_property;
 145
 146   /* Construct-only properties */
 147   g_object_class_install_property (object_class,
 148       PROP_ACCOUNT,
 149       g_param_spec_object ("account",
 150           "channel Account",
 151           "The account associated with the channel",
 152           MC_TYPE_ACCOUNT,
 153           G_PARAM_READWRITE |
 154           G_PARAM_CONSTRUCT_ONLY));
 155
 156   g_object_class_install_property (object_class,
 157       PROP_CHANNEL,
 158       g_param_spec_object ("channel",
 159           "telepathy channel",
 160           "The file transfer channel",
 161           TP_TYPE_CHANNEL,
 162           G_PARAM_READWRITE |
 163           G_PARAM_CONSTRUCT_ONLY));
 164
 165   g_object_class_install_property (object_class,
 166       PROP_STATE,
 167       g_param_spec_uint ("state",
 168           "state of the transfer",
 169           "The file transfer state",
 170           0,
 171           G_MAXUINT,
 172           G_MAXUINT,
 173           G_PARAM_READWRITE |
 174           G_PARAM_CONSTRUCT));
 175
 176   g_object_class_install_property (object_class,
 177       PROP_INCOMING,
 178       g_param_spec_boolean ("incoming",
 179           "incoming",
 180           "Whether the transfer is incoming",
 181           FALSE,
 182           G_PARAM_READWRITE |
 183           G_PARAM_CONSTRUCT));
 184
 185   g_object_class_install_property (object_class,
 186       PROP_FILENAME,
 187       g_param_spec_string ("filename",
 188           "name of the transfer",
 189           "The file transfer filename",
 190           "",
 191           G_PARAM_READWRITE));
 192
 193   g_object_class_install_property (object_class,
 194       PROP_SIZE,
 195       g_param_spec_uint64 ("size",
 196           "size of the file",
 197           "The file transfer size",
 198           0,
 199           G_MAXUINT64,
 200           G_MAXUINT64,
 201           G_PARAM_READWRITE));
 202
 203   g_object_class_install_property (object_class,
 204       PROP_CONTENT_TYPE,
 205       g_param_spec_string ("content-type",
 206           "file transfer content-type",
 207           "The file transfer content-type",
 208           "",
 209           G_PARAM_READWRITE));
 210
 211   g_object_class_install_property (object_class,
 212       PROP_CONTENT_HASH_TYPE,
 213       g_param_spec_uint ("content-hash-type",
 214           "file transfer hash type",
 215           "The type of the file transfer hash",
 216           0,
 217           G_MAXUINT,
 218           0,
 219           G_PARAM_READWRITE));
 220
 221   g_object_class_install_property (object_class,
 222       PROP_CONTENT_HASH,
 223       g_param_spec_string ("content-hash",
 224           "file transfer hash",
 225           "The hash of the transfer's contents",
 226           "",
 227           G_PARAM_READWRITE));
 228
 229   g_object_class_install_property (object_class,
 230       PROP_TRANSFERRED_BYTES,
 231       g_param_spec_uint64 ("transferred-bytes",
 232           "bytes transferred",
 233           "The number of bytes transferred",
 234           0,
 235           G_MAXUINT64,
 236           0,
 237           G_PARAM_READWRITE));
 238
 239   g_object_class_install_property (object_class,
 240       PROP_IN_STREAM,
 241       g_param_spec_object ("in-stream",
 242           "transfer input stream",
 243           "The input stream for file transfer",
 244           G_TYPE_INPUT_STREAM,
 245           G_PARAM_READWRITE));
 246
 247   g_type_class_add_private (object_class, sizeof (EmpathyTpFilePriv));
 248 }
 249
 250 static void
 251 empathy_tp_file_init (EmpathyTpFile *tp_file)
 252 {
 253 }
 254
 255 static void
 256 tp_file_finalize (GObject *object)
 257 {
 258   EmpathyTpFilePriv *priv;
 259   EmpathyTpFile *tp_file;
 260
 261   tp_file = EMPATHY_TP_FILE (object);
 262   priv = GET_PRIV (tp_file);
 263
 264   if (priv->channel)
 265     {
 266       DEBUG ("Closing channel..");
 267       g_signal_handlers_disconnect_by_func (priv->channel,
 268           tp_file_destroy_cb, object);
 269       tp_cli_channel_run_close (priv->channel, -1, NULL, NULL);
 270       if (G_IS_OBJECT (priv->channel))
 271         g_object_unref (priv->channel);
 272     }
 273
 274   if (priv->factory)
 275     {
 276       g_object_unref (priv->factory);
 277     }
 278   if (priv->account)
 279     {
 280       g_object_unref (priv->account);
 281     }
 282   if (priv->mc)
 283     {
 284       g_object_unref (priv->mc);
 285     }
 286
 287   g_free (priv->id);
 288   g_free (priv->filename);
 289   g_free (priv->unix_socket_path);
 290   g_free (priv->description);
 291   g_free (priv->content_hash);
 292   g_free (priv->content_type);
 293
 294   if (priv->in_stream)
 295     g_object_unref (priv->in_stream);
 296
 297   if (priv->out_stream)
 298     g_object_unref (priv->out_stream);
 299
 300   if (priv->contact)
 301     g_object_unref (priv->contact);
 302
 303   if (priv->cancellable)
 304     g_object_unref (priv->cancellable);
 305
 306   G_OBJECT_CLASS (empathy_tp_file_parent_class)->finalize (object);
 307 }
 308
 309 static GObject *
 310 tp_file_constructor (GType type,
 311                      guint n_props,
 312                      GObjectConstructParam *props)
 313 {
 314   GObject *tp_file;
 315   EmpathyTpFilePriv *priv;
 316   GError *error = NULL;
 317   GHashTable *properties;
 318   TpHandle handle;
 319
 320   tp_file = G_OBJECT_CLASS (empathy_tp_file_parent_class)->constructor (type,
 321       n_props, props);
 322
 323   priv = GET_PRIV (tp_file);
 324
 325   priv->factory = empathy_contact_factory_new ();
 326   priv->mc = empathy_mission_control_new ();
 327
 328   tp_cli_channel_connect_to_closed (priv->channel,
 329       (tp_cli_channel_signal_callback_closed) tp_file_closed_cb,
 330       tp_file,
 331       NULL, NULL, NULL);
 332
 333   emp_cli_channel_type_file_connect_to_file_transfer_state_changed (
 334       TP_PROXY (priv->channel),
 335       (emp_cli_channel_type_file_signal_callback_file_transfer_state_changed)
 336           tp_file_state_changed_cb,
 337       tp_file,
 338       NULL, NULL, NULL);
 339
 340   emp_cli_channel_type_file_connect_to_transferred_bytes_changed (
 341       TP_PROXY (priv->channel),
 342       (emp_cli_channel_type_file_signal_callback_transferred_bytes_changed)
 343           tp_file_transferred_bytes_changed_cb,
 344       tp_file,
 345       NULL, NULL, NULL);
 346
 347
 348   handle = tp_channel_get_handle (priv->channel, NULL);
 349   priv->contact = empathy_contact_factory_get_from_handle (priv->factory,
 350       priv->account,
 351       (guint) handle);
 352
 353   if (!tp_cli_dbus_properties_run_get_all (priv->channel,
 354       -1, EMP_IFACE_CHANNEL_TYPE_FILE, &properties, &error, NULL))
 355     {
 356       DEBUG ("Failed to get properties: %s",
 357           error ? error->message : "No error given");
 358       g_clear_error (&error);
 359       return NULL;
 360     }
 361
 362   priv->size = g_value_get_uint64 (g_hash_table_lookup (properties, "Size"));
 363
 364   priv->state = g_value_get_uint (g_hash_table_lookup (properties, "State"));
 365
 366   /* Invalid reason, so empathy_file_get_state_change_reason() can give
 367    * a warning if called for a not closed file transfer. */
 368   priv->state_change_reason = -1;
 369
 370   priv->transferred_bytes = g_value_get_uint64 (g_hash_table_lookup (
 371       properties, "TransferredBytes"));
 372
 373   priv->filename = g_value_dup_string (g_hash_table_lookup (properties,
 374       "Filename"));
 375
 376   priv->content_hash = g_value_dup_string (g_hash_table_lookup (properties,
 377       "ContentHash"));
 378
 379   priv->description = g_value_dup_string (g_hash_table_lookup (properties,
 380       "Description"));
 381
 382   if (priv->state == EMP_FILE_TRANSFER_STATE_LOCAL_PENDING)
 383     priv->incoming = TRUE;
 384
 385   g_hash_table_destroy (properties);
 386
 387   return tp_file;
 388 }
 389
 390 static void
 391 tp_file_get_property (GObject *object,
 392                       guint param_id,
 393                       GValue *value,
 394                       GParamSpec *pspec)
 395 {
 396   EmpathyTpFilePriv *priv;
 397   EmpathyTpFile *tp_file;
 398
 399   priv = GET_PRIV (object);
 400   tp_file = EMPATHY_TP_FILE (object);
 401
 402   switch (param_id)
 403     {
 404       case PROP_ACCOUNT:
 405         g_value_set_object (value, priv->account);
 406         break;
 407       case PROP_CHANNEL:
 408         g_value_set_object (value, priv->channel);
 409         break;
 410       default:
 411         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, param_id, pspec);
 412         break;
 413     };
 414 }
 415
 416 static void
 417 tp_file_channel_set_dbus_property (gpointer proxy,
 418                                    const gchar *property,
 419                                    const GValue *value)
 420 {
 421         DEBUG ("Setting %s property", property);
 422         tp_cli_dbus_properties_run_set (TP_PROXY (proxy),
 423             -1,
 424             EMP_IFACE_CHANNEL_TYPE_FILE,
 425             property,
 426             value,
 427             NULL, NULL);
 428         DEBUG ("done");
 429 }
 430
 431
 432 static void
 433 tp_file_set_property (GObject *object,
 434                       guint param_id,
 435                       const GValue *value,
 436                       GParamSpec *pspec)
 437 {
 438   EmpathyTpFilePriv *priv;
 439
 440   priv = GET_PRIV (object);
 441
 442   switch (param_id)
 443     {
 444       case PROP_ACCOUNT:
 445         priv->account = g_object_ref (g_value_get_object (value));
 446         break;
 447       case PROP_CHANNEL:
 448         priv->channel = g_object_ref (g_value_get_object (value));
 449         break;
 450       case PROP_STATE:
 451         priv->state = g_value_get_uint (value);
 452         break;
 453       case PROP_INCOMING:
 454         priv->incoming = g_value_get_boolean (value);
 455         break;
 456       case PROP_FILENAME:
 457         g_free (priv->filename);
 458         priv->filename = g_value_dup_string (value);
 459         tp_file_channel_set_dbus_property (priv->channel, "Filename", value);
 460         break;
 461       case PROP_SIZE:
 462         priv->size = g_value_get_uint64 (value);
 463         tp_file_channel_set_dbus_property (priv->channel, "Size", value);
 464         break;
 465       case PROP_CONTENT_TYPE:
 466         tp_file_channel_set_dbus_property (priv->channel, "ContentType", value);
 467         g_free (priv->content_type);
 468         priv->content_type = g_value_dup_string (value);
 469         break;
 470       case PROP_CONTENT_HASH:
 471         tp_file_channel_set_dbus_property (priv->channel, "ContentHash", value);
 472         g_free (priv->content_hash);
 473         priv->content_hash = g_value_dup_string (value);
 474         break;
 475       case PROP_IN_STREAM:
 476         if (priv->in_stream)
 477           g_object_unref (priv->in_stream);
 478         priv->in_stream = g_object_ref (g_value_get_object (value));
 479         break;
 480       default:
 481         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, param_id, pspec);
 482         break;
 483     };
 484 }
 485
 486 /**
 487  * empathy_tp_file_new:
 488  * @account: the #McAccount for the channel
 489  * @channel: a Telepathy channel
 490  *
 491  * Creates a new #EmpathyTpFile wrapping @channel.
 492  *
 493  * Returns: a new #EmpathyTpFile
 494  */
 495 EmpathyTpFile *
 496 empathy_tp_file_new (McAccount *account,
 497                   TpChannel *channel)
 498 {
 499   return g_object_new (EMPATHY_TYPE_TP_FILE,
 500       "account", account,
 501       "channel", channel,
 502       NULL);
 503 }
 504
 505 /**
 506  * empathy_tp_file_get_id:
 507  * @tp_file: an #EmpathyTpFile
 508  *
 509  * Returns the ID of @tp_file.
 510  *
 511  * Returns: the ID
 512  */
 513 const gchar *
 514 empathy_tp_file_get_id (EmpathyTpFile *tp_file)
 515 {
 516   EmpathyTpFilePriv *priv;
 517
 518   g_return_val_if_fail (EMPATHY_IS_TP_FILE (tp_file), NULL);
 519
 520   priv = GET_PRIV (tp_file);
 521
 522   return priv->id;
 523 }
 524
 525 /**
 526  * empathy_tp_file_get_channel
 527  * @tp_file: an #EmpathyTpFile
 528  *
 529  * Returns the Telepathy file transfer channel
 530  *
 531  * Returns: the #TpChannel
 532  */
 533 TpChannel *
 534 empathy_tp_file_get_channel (EmpathyTpFile *tp_file)
 535 {
 536   EmpathyTpFilePriv *priv;
 537
 538   g_return_val_if_fail (EMPATHY_IS_TP_FILE (tp_file), NULL);
 539
 540   priv = GET_PRIV (tp_file);
 541
 542   return priv->channel;
 543 }
 544
 545 static void
 546 tp_file_destroy_cb (TpChannel *file_channel,
 547                     EmpathyTpFile *tp_file)
 548 {
 549   EmpathyTpFilePriv *priv;
 550
 551   priv = GET_PRIV (tp_file);
 552
 553   DEBUG ("Channel Closed or CM crashed");
 554
 555   g_object_unref (priv->channel);
 556   priv->channel = NULL;
 557 }
 558
 559 static void
 560 tp_file_closed_cb (TpChannel *file_channel,
 561                    EmpathyTpFile *tp_file,
 562                    GObject *weak_object)
 563 {
 564   EmpathyTpFilePriv *priv;
 565
 566   priv = GET_PRIV (tp_file);
 567
 568   /* The channel is closed, do just like if the proxy was destroyed */
 569   g_signal_handlers_disconnect_by_func (priv->channel,
 570       tp_file_destroy_cb,
 571       tp_file);
 572   tp_file_destroy_cb (file_channel, tp_file);
 573 }
 574
 575 static gint64
 576 get_time_msec (void)
 577 {
 578   GTimeVal tv;
 579
 580   g_get_current_time (&tv);
 581   return ((gint64) tv.tv_sec) * 1000 + tv.tv_usec / 1000;
 582 }
 583
 584 static gint
 585 _get_local_socket (EmpathyTpFile *tp_file)
 586 {
 587   gint fd;
 588   size_t path_len;
 589   struct sockaddr_un addr;
 590   EmpathyTpFilePriv *priv;
 591
 592   priv = GET_PRIV (tp_file);
 593
 594   if (G_STR_EMPTY (priv->unix_socket_path))
 595     return -1;
 596
 597   fd = socket (PF_UNIX, SOCK_STREAM, 0);
 598   if (fd < 0)
 599     return -1;
 600
 601   memset (&addr, 0, sizeof (addr));
 602   addr.sun_family = AF_UNIX;
 603   path_len = strlen (priv->unix_socket_path);
 604   strncpy (addr.sun_path, priv->unix_socket_path, path_len);
 605
 606   if (connect (fd, (struct sockaddr*) &addr,
 607       sizeof (addr)) < 0)
 608     {
 609       close (fd);
 610       return -1;
 611     }
 612
 613   return fd;
 614 }
 615
 616 /**
 617  * empathy_tp_file_accept:
 618  * @tp_file: an #EmpathyTpFile
 619  *
 620  * Accepts a file transfer that's in the "local pending" state (i.e.
 621  * EMP_FILE_TRANSFER_STATE_LOCAL_PENDING).
 622  */
 623 void
 624 empathy_tp_file_accept (EmpathyTpFile *tp_file,
 625                         guint64 offset)
 626 {
 627   EmpathyTpFilePriv *priv;
 628   GValue *address;
 629   GValue nothing = { 0 };
 630   GError *error = NULL;
 631
 632   g_return_if_fail (EMPATHY_IS_TP_FILE (tp_file));
 633
 634   priv = GET_PRIV (tp_file);
 635
 636   g_return_if_fail (priv->out_stream != NULL);
 637
 638   DEBUG ("Accepting file: filename=%s", priv->filename);
 639
 640   g_value_init (&nothing, G_TYPE_STRING);
 641   g_value_set_string (&nothing, "");
 642
 643   if (!emp_cli_channel_type_file_run_accept_file (TP_PROXY (priv->channel),
 644       -1, TP_SOCKET_ADDRESS_TYPE_UNIX, TP_SOCKET_ACCESS_CONTROL_LOCALHOST,
 645       &nothing, offset, &address, &error, NULL))
 646     {
 647       DEBUG ("Accept error: %s",
 648           error ? error->message : "No message given");
 649       g_clear_error (&error);
 650       return;
 651     }
 652
 653   if (priv->unix_socket_path)
 654     g_free (priv->unix_socket_path);
 655
 656   priv->unix_socket_path = g_value_dup_string (address);
 657   g_value_unset (address);
 658
 659   DEBUG ("Got unix socket path: %s", priv->unix_socket_path);
 660 }
 661
 662 void
 663 empathy_tp_file_offer (EmpathyTpFile *tp_file)
 664 {
 665   EmpathyTpFilePriv *priv;
 666   GValue *address;
 667   GError *error = NULL;
 668   GValue nothing = { 0 };
 669
 670   g_return_if_fail (EMPATHY_IS_TP_FILE (tp_file));
 671
 672   priv = GET_PRIV (tp_file);
 673
 674   g_value_init (&nothing, G_TYPE_STRING);
 675   g_value_set_string (&nothing, "");
 676
 677   if (!emp_cli_channel_type_file_run_offer_file (TP_PROXY (priv->channel),
 678       -1, TP_SOCKET_ADDRESS_TYPE_UNIX, TP_SOCKET_ACCESS_CONTROL_LOCALHOST,
 679       &nothing, &address, &error, NULL))
 680     {
 681       DEBUG ("OfferFile error: %s",
 682           error ? error->message : "No message given");
 683       g_clear_error (&error);
 684       return;
 685     }
 686
 687   if (priv->unix_socket_path)
 688     g_free (priv->unix_socket_path);
 689
 690   priv->unix_socket_path = g_value_dup_string (address);
 691   g_value_unset (address);
 692
 693   DEBUG ("Got unix socket path: %s", priv->unix_socket_path);
 694 }
 695
 696 static void
 697 receive_tp_file (EmpathyTpFile *tp_file)
 698 {
 699   EmpathyTpFilePriv *priv;
 700   GInputStream *socket_stream;
 701   gint socket_fd;
 702
 703   priv = GET_PRIV (tp_file);
 704
 705   socket_fd = _get_local_socket (tp_file);
 706
 707   if (socket_fd < 0)
 708     return;
 709
 710   socket_stream = g_unix_input_stream_new (socket_fd, TRUE);
 711
 712   priv->cancellable = g_cancellable_new ();
 713
 714   copy_stream (socket_stream, priv->out_stream, priv->cancellable);
 715
 716   g_object_unref (socket_stream);
 717 }
 718
 719 static void
 720 send_tp_file (EmpathyTpFile *tp_file)
 721 {
 722   gint socket_fd;
 723   GOutputStream *socket_stream;
 724   EmpathyTpFilePriv *priv;
 725
 726   priv = GET_PRIV (tp_file);
 727
 728   DEBUG ("Sending file content: filename=%s",
 729            priv->filename);
 730
 731   g_return_if_fail (priv->in_stream);
 732
 733   socket_fd = _get_local_socket (tp_file);
 734   if (socket_fd < 0)
 735     {
 736       DEBUG ("failed to get local socket fd");
 737       return;
 738     }
 739   DEBUG ("got local socket fd");
 740   socket_stream = g_unix_output_stream_new (socket_fd, TRUE);
 741
 742   priv->cancellable = g_cancellable_new ();
 743
 744   copy_stream (priv->in_stream, socket_stream, priv->cancellable);
 745
 746   g_object_unref (socket_stream);
 747 }
 748
 749 static void
 750 tp_file_state_changed_cb (DBusGProxy *tp_file_iface,
 751                           EmpFileTransferState state,
 752                           EmpFileTransferStateChangeReason reason,
 753                           EmpathyTpFile *tp_file)
 754 {
 755   EmpathyTpFilePriv *priv;
 756
 757   priv = GET_PRIV (tp_file);
 758
 759   DEBUG ("File transfer state changed: filename=%s, "
 760       "old state=%u, state=%u, reason=%u",
 761       priv->filename, priv->state, state, reason);
 762
 763   if (state == EMP_FILE_TRANSFER_STATE_OPEN)
 764     priv->start_time = get_time_msec ();
 765
 766   DEBUG ("state = %u, incoming = %s, in_stream = %s, out_stream = %s",
 767       state, priv->incoming ? "yes" : "no",
 768       priv->in_stream ? "present" : "not present",
 769       priv->out_stream ? "present" : "not present");
 770
 771   if (state == EMP_FILE_TRANSFER_STATE_OPEN && !priv->incoming &&
 772       priv->in_stream)
 773     send_tp_file (tp_file);
 774   else if (state == EMP_FILE_TRANSFER_STATE_OPEN && priv->incoming &&
 775       priv->out_stream)
 776     receive_tp_file (tp_file);
 777
 778   priv->state = state;
 779   priv->state_change_reason = reason;
 780
 781   g_object_notify (G_OBJECT (tp_file), "state");
 782 }
 783
 784 static void
 785 tp_file_transferred_bytes_changed_cb (TpProxy *proxy,
 786                                       guint64 count,
 787                                       EmpathyTpFile *tp_file,
 788                                       GObject *weak_object)
 789 {
 790   EmpathyTpFilePriv *priv;
 791
 792   priv = GET_PRIV (tp_file);
 793
 794   if (priv->transferred_bytes == count)
 795     return;
 796
 797   priv->transferred_bytes = count;
 798
 799   g_object_notify (G_OBJECT (tp_file), "transferred-bytes");
 800 }
 801
 802 EmpathyContact *
 803 empathy_tp_file_get_contact (EmpathyTpFile *tp_file)
 804 {
 805   EmpathyTpFilePriv *priv;
 806
 807   priv = GET_PRIV (tp_file);
 808
 809   return priv->contact;
 810 }
 811
 812 GInputStream *
 813 empathy_tp_file_get_input_stream (EmpathyTpFile *tp_file)
 814 {
 815   EmpathyTpFilePriv *priv;
 816
 817   priv = GET_PRIV (tp_file);
 818
 819   return priv->in_stream;
 820 }
 821
 822 GOutputStream *
 823 empathy_tp_file_get_output_stream (EmpathyTpFile *tp_file)
 824 {
 825   EmpathyTpFilePriv *priv;
 826
 827   priv = GET_PRIV (tp_file);
 828
 829   return priv->out_stream;
 830 }
 831
 832 const gchar *
 833 empathy_tp_file_get_filename (EmpathyTpFile *tp_file)
 834 {
 835   EmpathyTpFilePriv *priv;
 836
 837   priv = GET_PRIV (tp_file);
 838
 839   return priv->filename;
 840 }
 841
 842 gboolean
 843 empathy_tp_file_get_incoming (EmpathyTpFile *tp_file)
 844 {
 845   EmpathyTpFilePriv *priv;
 846
 847   priv = GET_PRIV (tp_file);
 848
 849   return priv->incoming;
 850 }
 851
 852 EmpFileTransferState
 853 empathy_tp_file_get_state (EmpathyTpFile *tp_file)
 854 {
 855   EmpathyTpFilePriv *priv;
 856
 857   priv = GET_PRIV (tp_file);
 858
 859   return priv->state;
 860 }
 861
 862 EmpFileTransferStateChangeReason
 863 empathy_tp_file_get_state_change_reason (EmpathyTpFile *tp_file)
 864 {
 865   EmpathyTpFilePriv *priv;
 866
 867   priv = GET_PRIV (tp_file);
 868
 869   g_return_val_if_fail (priv->state_change_reason >= 0,
 870       EMP_FILE_TRANSFER_STATE_CHANGE_REASON_NONE);
 871
 872   return priv->state_change_reason;
 873 }
 874
 875 guint64
 876 empathy_tp_file_get_size (EmpathyTpFile *tp_file)
 877 {
 878   EmpathyTpFilePriv *priv;
 879
 880   priv = GET_PRIV (tp_file);
 881
 882   return priv->size;
 883 }
 884
 885 guint64
 886 empathy_tp_file_get_transferred_bytes (EmpathyTpFile *tp_file)
 887 {
 888   EmpathyTpFilePriv *priv;
 889
 890   priv = GET_PRIV (tp_file);
 891
 892   return priv->transferred_bytes;
 893 }
 894
 895 gint
 896 empathy_tp_file_get_remaining_time (EmpathyTpFile *tp_file)
 897 {
 898   EmpathyTpFilePriv *priv;
 899   gint64 curr_time, elapsed_time;
 900   gdouble time_per_byte;
 901   gdouble remaining_time;
 902
 903   priv = GET_PRIV (tp_file);
 904
 905   if (priv->size == EMPATHY_TP_FILE_UNKNOWN_SIZE)
 906     return -1;
 907
 908   if (priv->transferred_bytes == priv->size)
 909     return 0;
 910
 911   curr_time = get_time_msec ();
 912   elapsed_time = curr_time - priv->start_time;
 913   time_per_byte = (gdouble) elapsed_time / (gdouble) priv->transferred_bytes;
 914   remaining_time = (time_per_byte * (priv->size - priv->transferred_bytes)) / 1000;
 915
 916   return (gint) (remaining_time + 0.5);
 917 }
 918
 919 void
 920 empathy_tp_file_cancel (EmpathyTpFile *tp_file)
 921 {
 922   EmpathyTpFilePriv *priv;
 923
 924   priv = GET_PRIV (tp_file);
 925
 926   tp_cli_channel_run_close (priv->channel, -1, NULL, NULL);
 927
 928   g_cancellable_cancel (priv->cancellable);
 929 }
 930
 931 void
 932 empathy_tp_file_set_input_stream (EmpathyTpFile *tp_file,
 933                                   GInputStream *in_stream)
 934 {
 935   EmpathyTpFilePriv *priv;
 936
 937   priv = GET_PRIV (tp_file);
 938
 939   if (priv->in_stream == in_stream)
 940     return;
 941
 942   if (priv->incoming)
 943     g_warning ("Setting an input stream for incoming file "
 944          "transfers is useless");
 945
 946   if (priv->in_stream)
 947     g_object_unref (priv->in_stream);
 948
 949   if (in_stream)
 950     g_object_ref (in_stream);
 951
 952   priv->in_stream = in_stream;
 953
 954   g_object_notify (G_OBJECT (tp_file), "in-stream");
 955 }
 956
 957 void
 958 empathy_tp_file_set_output_stream (EmpathyTpFile *tp_file,
 959                                    GOutputStream *out_stream)
 960 {
 961   EmpathyTpFilePriv *priv;
 962
 963   priv = GET_PRIV (tp_file);
 964
 965   if (priv->out_stream == out_stream)
 966     return;
 967
 968   if (!priv->incoming)
 969     g_warning ("Setting an output stream for outgoing file "
 970          "transfers is useless");
 971
 972   if (priv->out_stream)
 973     g_object_unref (priv->out_stream);
 974
 975   if (out_stream)
 976     g_object_ref (out_stream);
 977
 978   priv->out_stream = out_stream;
 979 }
 980
 981 void
 982 empathy_tp_file_set_filename (EmpathyTpFile *tp_file,
 983                               const gchar *filename)
 984 {
 985   EmpathyTpFilePriv *priv;
 986
 987   priv = GET_PRIV (tp_file);
 988
 989   g_return_if_fail (filename != NULL);
 990
 991   if (priv->filename && strcmp (filename, priv->filename) == 0)
 992     return;
 993
 994   g_free (priv->filename);
 995   priv->filename = g_strdup (filename);
 996
 997   g_object_notify (G_OBJECT (tp_file), "filename");
 998 }
 999
1000 /* Functions to copy the content of a GInputStream to a GOutputStream */
1001
1002 #define N_BUFFERS 2
1003 #define BUFFER_SIZE 4096
1004
1005 typedef struct {
1006   GInputStream *in;
1007   GOutputStream *out;
1008   GCancellable  *cancellable;
1009   char *buff[N_BUFFERS]; /* the temporary buffers */
1010   gsize count[N_BUFFERS]; /* how many bytes are used in the buffers */
1011   gboolean is_full[N_BUFFERS]; /* whether the buffers contain data */
1012   gint curr_read; /* index of the buffer used for reading */
1013   gint curr_write; /* index of the buffer used for writing */
1014   gboolean is_reading; /* we are reading */
1015   gboolean is_writing; /* we are writing */
1016   guint n_closed; /* number of streams that have been closed */
1017 } CopyData;
1018
1019 static void schedule_next (CopyData *copy);
1020
1021 static void
1022 free_copy_data_if_closed (CopyData *copy)
1023 {
1024   gint i;
1025
1026   /* Free the data only if both the input and output streams have
1027    * been closed. */
1028   copy->n_closed++;
1029   if (copy->n_closed < 2)
1030     return;
1031
1032   if (copy->in != NULL)
1033     g_object_unref (copy->in);
1034
1035   if (copy->out != NULL)
1036     g_object_unref (copy->out);
1037
1038   for (i = 0; i < N_BUFFERS; i++)
1039     g_free (copy->buff[i]);
1040
1041   g_object_unref (copy->cancellable);
1042   g_free (copy);
1043 }
1044
1045 static void
1046 io_error (CopyData *copy,
1047           GError *error)
1048 {
1049   g_cancellable_cancel (copy->cancellable);
1050
1051   if (error == NULL)
1052     g_warning ("I/O error");
1053   else if (error->domain == G_IO_ERROR && error->code == G_IO_ERROR_CANCELLED)
1054     ; /* Ignore cancellations */
1055   else
1056     g_warning ("I/O error: %d: %s\n", error->code, error->message);
1057
1058   if (copy->in != NULL)
1059     g_input_stream_close (copy->in, NULL, NULL);
1060
1061   if (copy->out != NULL)
1062     g_output_stream_close (copy->out, NULL, NULL);
1063
1064   free_copy_data_if_closed (copy);
1065 }
1066
1067 static void
1068 close_done (GObject *source_object,
1069             GAsyncResult *res,
1070             gpointer user_data)
1071 {
1072   CopyData *copy = user_data;
1073
1074   g_object_unref (source_object);
1075   free_copy_data_if_closed (copy);
1076 }
1077
1078 static void
1079 write_done_cb (GObject *source_object,
1080                GAsyncResult *res,
1081                gpointer user_data)
1082 {
1083   CopyData *copy = user_data;
1084   gssize count_write;
1085   GError *error = NULL;
1086
1087   count_write = g_output_stream_write_finish (copy->out, res, &error);
1088
1089   if (count_write <= 0)
1090     {
1091       io_error (copy, error);
1092       g_error_free (error);
1093       return;
1094     }
1095
1096   copy->is_full[copy->curr_write] = FALSE;
1097   copy->curr_write = (copy->curr_write + 1) % N_BUFFERS;
1098   copy->is_writing = FALSE;
1099
1100   schedule_next (copy);
1101 }
1102
1103 static void
1104 read_done_cb (GObject *source_object,
1105               GAsyncResult *res,
1106               gpointer user_data)
1107 {
1108   CopyData *copy = user_data;
1109   gssize count_read;
1110   GError *error = NULL;
1111
1112   count_read = g_input_stream_read_finish (copy->in, res, &error);
1113
1114   if (count_read == 0)
1115     {
1116       g_input_stream_close_async (copy->in, 0, copy->cancellable,
1117           close_done, copy);
1118       copy->in = NULL;
1119     }
1120   else if (count_read < 0)
1121     {
1122       io_error (copy, error);
1123       g_error_free (error);
1124       return;
1125     }
1126
1127   copy->count[copy->curr_read] = count_read;
1128   copy->is_full[copy->curr_read] = TRUE;
1129   copy->curr_read = (copy->curr_read + 1) % N_BUFFERS;
1130   copy->is_reading = FALSE;
1131
1132   schedule_next (copy);
1133 }
1134
1135 static void
1136 schedule_next (CopyData *copy)
1137 {
1138   if (copy->in != NULL &&
1139       !copy->is_reading &&
1140       !copy->is_full[copy->curr_read])
1141     {
1142       /* We are not reading and the current buffer is empty, so
1143        * start an async read. */
1144       copy->is_reading = TRUE;
1145       g_input_stream_read_async (copy->in,
1146           copy->buff[copy->curr_read],
1147           BUFFER_SIZE, 0, copy->cancellable,
1148           read_done_cb, copy);
1149     }
1150
1151   if (!copy->is_writing &&
1152       copy->is_full[copy->curr_write])
1153     {
1154       if (copy->count[copy->curr_write] == 0)
1155         {
1156           /* The last read on the buffer read 0 bytes, this
1157            * means that we got an EOF, so we can close
1158            * the output channel. */
1159           g_output_stream_close_async (copy->out, 0,
1160               copy->cancellable,
1161               close_done, copy);
1162       copy->out = NULL;
1163         }
1164       else
1165         {
1166           /* We are not writing and the current buffer contains
1167            * data, so start an async write. */
1168           copy->is_writing = TRUE;
1169           g_output_stream_write_async (copy->out,
1170               copy->buff[copy->curr_write],
1171               copy->count[copy->curr_write],
1172               0, copy->cancellable,
1173               write_done_cb, copy);
1174         }
1175     }
1176 }
1177
1178 static void
1179 copy_stream (GInputStream *in,
1180              GOutputStream *out,
1181              GCancellable *cancellable)
1182 {
1183   CopyData *copy;
1184   gint i;
1185
1186   g_return_if_fail (in != NULL);
1187   g_return_if_fail (out != NULL);
1188
1189   copy = g_new0 (CopyData, 1);
1190   copy->in = g_object_ref (in);
1191   copy->out = g_object_ref (out);
1192
1193   if (cancellable != NULL)
1194     copy->cancellable = g_object_ref (cancellable);
1195   else
1196     copy->cancellable = g_cancellable_new ();
1197
1198   for (i = 0; i < N_BUFFERS; i++)
1199     copy->buff[i] = g_malloc (BUFFER_SIZE);
1200
1201   schedule_next (copy);
1202 }