gio/giostream.c

   1 /* GIO - GLib Input, Output and Streaming Library
   2  *
   3  * Copyright © 2008 codethink
   4  * Copyright © 2009 Red Hat, Inc.
   5  *
   6  * This library is free software; you can redistribute it and/or
   7  * modify it under the terms of the GNU Lesser General Public
   8  * License as published by the Free Software Foundation; either
   9  * version 2 of the License, or (at your option) any later version.
  10  *
  11  * This library is distributed in the hope that it will be useful,
  12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  14  * Lesser General Public License for more details.
  15  *
  16  * You should have received a copy of the GNU Lesser General
  17  * Public License along with this library; if not, see <http://www.gnu.org/licenses/>.
  18  *
  19  * Authors: Ryan Lortie <desrt@desrt.ca>
  20  *          Alexander Larsson <alexl@redhat.com>
  21  */
  22
  23 #include "config.h"
  24 #include <glib.h>
  25 #include "glibintl.h"
  26
  27 #include "giostream.h"
  28 #include "gasyncresult.h"
  29 #include "gtask.h"
  30
  31 /**
  32  * SECTION:giostream
  33  * @short_description: Base class for implementing read/write streams
  34  * @include: gio/gio.h
  35  * @see_also: #GInputStream, #GOutputStream
  36  *
  37  * GIOStream represents an object that has both read and write streams.
  38  * Generally the two streams acts as separate input and output streams,
  39  * but they share some common resources and state. For instance, for
  40  * seekable streams they may use the same position in both streams.
  41  *
  42  * Examples of #GIOStream objects are #GSocketConnection which represents
  43  * a two-way network connection, and #GFileIOStream which represent a
  44  * file handle opened in read-write mode.
  45  *
  46  * To do the actual reading and writing you need to get the substreams
  47  * with g_io_stream_get_input_stream() and g_io_stream_get_output_stream().
  48  *
  49  * The #GIOStream object owns the input and the output streams, not the other
  50  * way around, so keeping the substreams alive will not keep the #GIOStream
  51  * object alive. If the #GIOStream object is freed it will be closed, thus
  52  * closing the substream, so even if the substreams stay alive they will
  53  * always just return a %G_IO_ERROR_CLOSED for all operations.
  54  *
  55  * To close a stream use g_io_stream_close() which will close the common
  56  * stream object and also the individual substreams. You can also close
  57  * the substreams themselves. In most cases this only marks the
  58  * substream as closed, so further I/O on it fails. However, some streams
  59  * may support "half-closed" states where one direction of the stream
  60  * is actually shut down.
  61  *
  62  * Since: 2.22
  63  */
  64
  65 enum
  66 {
  67   PROP_0,
  68   PROP_INPUT_STREAM,
  69   PROP_OUTPUT_STREAM,
  70   PROP_CLOSED
  71 };
  72
  73 struct _GIOStreamPrivate {
  74   guint closed : 1;
  75   guint pending : 1;
  76   GAsyncReadyCallback outstanding_callback;
  77 };
  78
  79 static gboolean g_io_stream_real_close        (GIOStream            *stream,
  80                                                GCancellable         *cancellable,
  81                                                GError              **error);
  82 static void     g_io_stream_real_close_async  (GIOStream            *stream,
  83                                                int                   io_priority,
  84                                                GCancellable         *cancellable,
  85                                                GAsyncReadyCallback   callback,
  86                                                gpointer              user_data);
  87 static gboolean g_io_stream_real_close_finish (GIOStream            *stream,
  88                                                GAsyncResult         *result,
  89                                                GError              **error);
  90
  91 G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE (GIOStream, g_io_stream, G_TYPE_OBJECT)
  92
  93 static void
  94 g_io_stream_dispose (GObject *object)
  95 {
  96   GIOStream *stream;
  97
  98   stream = G_IO_STREAM (object);
  99
 100   if (!stream->priv->closed)
 101     g_io_stream_close (stream, NULL, NULL);
 102
 103   G_OBJECT_CLASS (g_io_stream_parent_class)->dispose (object);
 104 }
 105
 106 static void
 107 g_io_stream_init (GIOStream *stream)
 108 {
 109   stream->priv = g_io_stream_get_instance_private (stream);
 110 }
 111
 112 static void
 113 g_io_stream_get_property (GObject    *object,
 114                           guint       prop_id,
 115                           GValue     *value,
 116                           GParamSpec *pspec)
 117 {
 118   GIOStream *stream = G_IO_STREAM (object);
 119
 120   switch (prop_id)
 121     {
 122       case PROP_CLOSED:
 123         g_value_set_boolean (value, stream->priv->closed);
 124         break;
 125
 126       case PROP_INPUT_STREAM:
 127         g_value_set_object (value, g_io_stream_get_input_stream (stream));
 128         break;
 129
 130       case PROP_OUTPUT_STREAM:
 131         g_value_set_object (value, g_io_stream_get_output_stream (stream));
 132         break;
 133
 134       default:
 135         G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 136     }
 137 }
 138
 139 static void
 140 g_io_stream_class_init (GIOStreamClass *klass)
 141 {
 142   GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
 143
 144   gobject_class->dispose = g_io_stream_dispose;
 145   gobject_class->get_property = g_io_stream_get_property;
 146
 147   klass->close_fn = g_io_stream_real_close;
 148   klass->close_async = g_io_stream_real_close_async;
 149   klass->close_finish = g_io_stream_real_close_finish;
 150
 151   g_object_class_install_property (gobject_class, PROP_CLOSED,
 152                                    g_param_spec_boolean ("closed",
 153                                                          P_("Closed"),
 154                                                          P_("Is the stream closed"),
 155                                                          FALSE,
 156                                                          G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 157
 158   g_object_class_install_property (gobject_class, PROP_INPUT_STREAM,
 159                                    g_param_spec_object ("input-stream",
 160                                                         P_("Input stream"),
 161                                                         P_("The GInputStream to read from"),
 162                                                         G_TYPE_INPUT_STREAM,
 163                                                         G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 164   g_object_class_install_property (gobject_class, PROP_OUTPUT_STREAM,
 165                                    g_param_spec_object ("output-stream",
 166                                                         P_("Output stream"),
 167                                                         P_("The GOutputStream to write to"),
 168                                                         G_TYPE_OUTPUT_STREAM,
 169                                                         G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
 170 }
 171
 172 /**
 173  * g_io_stream_is_closed:
 174  * @stream: a #GIOStream
 175  *
 176  * Checks if a stream is closed.
 177  *
 178  * Returns: %TRUE if the stream is closed.
 179  *
 180  * Since: 2.22
 181  */
 182 gboolean
 183 g_io_stream_is_closed (GIOStream *stream)
 184 {
 185   g_return_val_if_fail (G_IS_IO_STREAM (stream), TRUE);
 186
 187   return stream->priv->closed;
 188 }
 189
 190 /**
 191  * g_io_stream_get_input_stream:
 192  * @stream: a #GIOStream
 193  *
 194  * Gets the input stream for this object. This is used
 195  * for reading.
 196  *
 197  * Returns: (transfer none): a #GInputStream, owned by the #GIOStream.
 198  * Do not free.
 199  *
 200  * Since: 2.22
 201  */
 202 GInputStream *
 203 g_io_stream_get_input_stream (GIOStream *stream)
 204 {
 205   GIOStreamClass *klass;
 206
 207   klass = G_IO_STREAM_GET_CLASS (stream);
 208
 209   g_assert (klass->get_input_stream != NULL);
 210
 211   return klass->get_input_stream (stream);
 212 }
 213
 214 /**
 215  * g_io_stream_get_output_stream:
 216  * @stream: a #GIOStream
 217  *
 218  * Gets the output stream for this object. This is used for
 219  * writing.
 220  *
 221  * Returns: (transfer none): a #GOutputStream, owned by the #GIOStream.
 222  * Do not free.
 223  *
 224  * Since: 2.22
 225  */
 226 GOutputStream *
 227 g_io_stream_get_output_stream (GIOStream *stream)
 228 {
 229   GIOStreamClass *klass;
 230
 231   klass = G_IO_STREAM_GET_CLASS (stream);
 232
 233   g_assert (klass->get_output_stream != NULL);
 234   return klass->get_output_stream (stream);
 235 }
 236
 237 /**
 238  * g_io_stream_has_pending:
 239  * @stream: a #GIOStream
 240  *
 241  * Checks if a stream has pending actions.
 242  *
 243  * Returns: %TRUE if @stream has pending actions.
 244  *
 245  * Since: 2.22
 246  **/
 247 gboolean
 248 g_io_stream_has_pending (GIOStream *stream)
 249 {
 250   g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
 251
 252   return stream->priv->pending;
 253 }
 254
 255 /**
 256  * g_io_stream_set_pending:
 257  * @stream: a #GIOStream
 258  * @error: a #GError location to store the error occurring, or %NULL to
 259  *     ignore
 260  *
 261  * Sets @stream to have actions pending. If the pending flag is
 262  * already set or @stream is closed, it will return %FALSE and set
 263  * @error.
 264  *
 265  * Return value: %TRUE if pending was previously unset and is now set.
 266  *
 267  * Since: 2.22
 268  */
 269 gboolean
 270 g_io_stream_set_pending (GIOStream  *stream,
 271                          GError    **error)
 272 {
 273   g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
 274
 275   if (stream->priv->closed)
 276     {
 277       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
 278                            _("Stream is already closed"));
 279       return FALSE;
 280     }
 281
 282   if (stream->priv->pending)
 283     {
 284       g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_PENDING,
 285                            /* Translators: This is an error you get if there is
 286                             * already an operation running against this stream when
 287                             * you try to start one */
 288                            _("Stream has outstanding operation"));
 289       return FALSE;
 290     }
 291
 292   stream->priv->pending = TRUE;
 293   return TRUE;
 294 }
 295
 296 /**
 297  * g_io_stream_clear_pending:
 298  * @stream: a #GIOStream
 299  *
 300  * Clears the pending flag on @stream.
 301  *
 302  * Since: 2.22
 303  */
 304 void
 305 g_io_stream_clear_pending (GIOStream *stream)
 306 {
 307   g_return_if_fail (G_IS_IO_STREAM (stream));
 308
 309   stream->priv->pending = FALSE;
 310 }
 311
 312 static gboolean
 313 g_io_stream_real_close (GIOStream     *stream,
 314                         GCancellable  *cancellable,
 315                         GError       **error)
 316 {
 317   gboolean res;
 318
 319   res = g_output_stream_close (g_io_stream_get_output_stream (stream),
 320                                cancellable, error);
 321
 322   /* If this errored out, unset error so that we don't report
 323      further errors, but still do the following ops */
 324   if (error != NULL && *error != NULL)
 325     error = NULL;
 326
 327   res &= g_input_stream_close (g_io_stream_get_input_stream (stream),
 328                                cancellable, error);
 329
 330   return res;
 331 }
 332
 333 /**
 334  * g_io_stream_close:
 335  * @stream: a #GIOStream
 336  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore
 337  * @error: location to store the error occurring, or %NULL to ignore
 338  *
 339  * Closes the stream, releasing resources related to it. This will also
 340  * closes the individual input and output streams, if they are not already
 341  * closed.
 342  *
 343  * Once the stream is closed, all other operations will return
 344  * %G_IO_ERROR_CLOSED. Closing a stream multiple times will not
 345  * return an error.
 346  *
 347  * Closing a stream will automatically flush any outstanding buffers
 348  * in the stream.
 349  *
 350  * Streams will be automatically closed when the last reference
 351  * is dropped, but you might want to call this function to make sure
 352  * resources are released as early as possible.
 353  *
 354  * Some streams might keep the backing store of the stream (e.g. a file
 355  * descriptor) open after the stream is closed. See the documentation for
 356  * the individual stream for details.
 357  *
 358  * On failure the first error that happened will be reported, but the
 359  * close operation will finish as much as possible. A stream that failed
 360  * to close will still return %G_IO_ERROR_CLOSED for all operations.
 361  * Still, it is important to check and report the error to the user,
 362  * otherwise there might be a loss of data as all data might not be written.
 363  *
 364  * If @cancellable is not NULL, then the operation can be cancelled by
 365  * triggering the cancellable object from another thread. If the operation
 366  * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
 367  * Cancelling a close will still leave the stream closed, but some streams
 368  * can use a faster close that doesn't block to e.g. check errors.
 369  *
 370  * The default implementation of this method just calls close on the
 371  * individual input/output streams.
 372  *
 373  * Return value: %TRUE on success, %FALSE on failure
 374  *
 375  * Since: 2.22
 376  */
 377 gboolean
 378 g_io_stream_close (GIOStream     *stream,
 379                    GCancellable  *cancellable,
 380                    GError       **error)
 381 {
 382   GIOStreamClass *class;
 383   gboolean res;
 384
 385   g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
 386
 387   class = G_IO_STREAM_GET_CLASS (stream);
 388
 389   if (stream->priv->closed)
 390     return TRUE;
 391
 392   if (!g_io_stream_set_pending (stream, error))
 393     return FALSE;
 394
 395   if (cancellable)
 396     g_cancellable_push_current (cancellable);
 397
 398   res = TRUE;
 399   if (class->close_fn)
 400     res = class->close_fn (stream, cancellable, error);
 401
 402   if (cancellable)
 403     g_cancellable_pop_current (cancellable);
 404
 405   stream->priv->closed = TRUE;
 406   g_io_stream_clear_pending (stream);
 407
 408   return res;
 409 }
 410
 411 static void
 412 async_ready_close_callback_wrapper (GObject      *source_object,
 413                                     GAsyncResult *res,
 414                                     gpointer      user_data)
 415 {
 416   GIOStream *stream = G_IO_STREAM (source_object);
 417
 418   stream->priv->closed = TRUE;
 419   g_io_stream_clear_pending (stream);
 420   if (stream->priv->outstanding_callback)
 421     (*stream->priv->outstanding_callback) (source_object, res, user_data);
 422   g_object_unref (stream);
 423 }
 424
 425 /**
 426  * g_io_stream_close_async:
 427  * @stream: a #GIOStream
 428  * @io_priority: the io priority of the request
 429  * @cancellable: (allow-none): optional cancellable object
 430  * @callback: (scope async): callback to call when the request is satisfied
 431  * @user_data: (closure): the data to pass to callback function
 432  *
 433  * Requests an asynchronous close of the stream, releasing resources
 434  * related to it. When the operation is finished @callback will be
 435  * called. You can then call g_io_stream_close_finish() to get
 436  * the result of the operation.
 437  *
 438  * For behaviour details see g_io_stream_close().
 439  *
 440  * The asynchronous methods have a default fallback that uses threads
 441  * to implement asynchronicity, so they are optional for inheriting
 442  * classes. However, if you override one you must override all.
 443  *
 444  * Since: 2.22
 445  */
 446 void
 447 g_io_stream_close_async (GIOStream           *stream,
 448                          int                  io_priority,
 449                          GCancellable        *cancellable,
 450                          GAsyncReadyCallback  callback,
 451                          gpointer             user_data)
 452 {
 453   GIOStreamClass *class;
 454   GError *error = NULL;
 455
 456   g_return_if_fail (G_IS_IO_STREAM (stream));
 457
 458   if (stream->priv->closed)
 459     {
 460       GTask *task;
 461
 462       task = g_task_new (stream, cancellable, callback, user_data);
 463       g_task_set_source_tag (task, g_io_stream_close_async);
 464       g_task_return_boolean (task, TRUE);
 465       g_object_unref (task);
 466       return;
 467     }
 468
 469   if (!g_io_stream_set_pending (stream, &error))
 470     {
 471       g_task_report_error (stream, callback, user_data,
 472                            g_io_stream_close_async,
 473                            error);
 474       return;
 475     }
 476
 477   class = G_IO_STREAM_GET_CLASS (stream);
 478   stream->priv->outstanding_callback = callback;
 479   g_object_ref (stream);
 480   class->close_async (stream, io_priority, cancellable,
 481                       async_ready_close_callback_wrapper, user_data);
 482 }
 483
 484 /**
 485  * g_io_stream_close_finish:
 486  * @stream: a #GIOStream
 487  * @result: a #GAsyncResult
 488  * @error: a #GError location to store the error occurring, or %NULL to
 489  *    ignore
 490  *
 491  * Closes a stream.
 492  *
 493  * Returns: %TRUE if stream was successfully closed, %FALSE otherwise.
 494  *
 495  * Since: 2.22
 496  */
 497 gboolean
 498 g_io_stream_close_finish (GIOStream     *stream,
 499                           GAsyncResult  *result,
 500                           GError       **error)
 501 {
 502   GIOStreamClass *class;
 503
 504   g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
 505   g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);
 506
 507   if (g_async_result_legacy_propagate_error (result, error))
 508     return FALSE;
 509   else if (g_async_result_is_tagged (result, g_io_stream_close_async))
 510     return g_task_propagate_boolean (G_TASK (result), error);
 511
 512   class = G_IO_STREAM_GET_CLASS (stream);
 513   return class->close_finish (stream, result, error);
 514 }
 515
 516
 517 static void
 518 close_async_thread (GTask        *task,
 519                     gpointer      source_object,
 520                     gpointer      task_data,
 521                     GCancellable *cancellable)
 522 {
 523   GIOStream *stream = source_object;
 524   GIOStreamClass *class;
 525   GError *error = NULL;
 526   gboolean result;
 527
 528   class = G_IO_STREAM_GET_CLASS (stream);
 529   if (class->close_fn)
 530     {
 531       result = class->close_fn (stream,
 532                                 g_task_get_cancellable (task),
 533                                 &error);
 534       if (!result)
 535         {
 536           g_task_return_error (task, error);
 537           return;
 538         }
 539     }
 540
 541   g_task_return_boolean (task, TRUE);
 542 }
 543
 544 static void
 545 g_io_stream_real_close_async (GIOStream           *stream,
 546                               int                  io_priority,
 547                               GCancellable        *cancellable,
 548                               GAsyncReadyCallback  callback,
 549                               gpointer             user_data)
 550 {
 551   GTask *task;
 552
 553   task = g_task_new (stream, cancellable, callback, user_data);
 554   g_task_set_check_cancellable (task, FALSE);
 555   g_task_set_priority (task, io_priority);
 556
 557   g_task_run_in_thread (task, close_async_thread);
 558   g_object_unref (task);
 559 }
 560
 561 static gboolean
 562 g_io_stream_real_close_finish (GIOStream     *stream,
 563                                GAsyncResult  *result,
 564                                GError       **error)
 565 {
 566   g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
 567
 568   return g_task_propagate_boolean (G_TASK (result), error);
 569 }
 570
 571 typedef struct
 572 {
 573   GIOStream *stream1;
 574   GIOStream *stream2;
 575   GIOStreamSpliceFlags flags;
 576   gint io_priority;
 577   GCancellable *cancellable;
 578   gulong cancelled_id;
 579   GCancellable *op1_cancellable;
 580   GCancellable *op2_cancellable;
 581   guint completed;
 582   GError *error;
 583 } SpliceContext;
 584
 585 static void
 586 splice_context_free (SpliceContext *ctx)
 587 {
 588   g_object_unref (ctx->stream1);
 589   g_object_unref (ctx->stream2);
 590   if (ctx->cancellable != NULL)
 591     g_object_unref (ctx->cancellable);
 592   g_object_unref (ctx->op1_cancellable);
 593   g_object_unref (ctx->op2_cancellable);
 594   g_clear_error (&ctx->error);
 595   g_slice_free (SpliceContext, ctx);
 596 }
 597
 598 static void
 599 splice_complete (GTask         *task,
 600                  SpliceContext *ctx)
 601 {
 602   if (ctx->cancelled_id != 0)
 603     g_cancellable_disconnect (ctx->cancellable, ctx->cancelled_id);
 604   ctx->cancelled_id = 0;
 605
 606   if (ctx->error != NULL)
 607     {
 608       g_task_return_error (task, ctx->error);
 609       ctx->error = NULL;
 610     }
 611   else
 612     g_task_return_boolean (task, TRUE);
 613 }
 614
 615 static void
 616 splice_close_cb (GObject      *iostream,
 617                  GAsyncResult *res,
 618                  gpointer      user_data)
 619 {
 620   GTask *task = user_data;
 621   SpliceContext *ctx = g_task_get_task_data (task);
 622   GError *error = NULL;
 623
 624   g_io_stream_close_finish (G_IO_STREAM (iostream), res, &error);
 625
 626   ctx->completed++;
 627
 628   /* Keep the first error that occurred */
 629   if (error != NULL && ctx->error == NULL)
 630     ctx->error = error;
 631   else
 632     g_clear_error (&error);
 633
 634   /* If all operations are done, complete now */
 635   if (ctx->completed == 4)
 636     splice_complete (task, ctx);
 637
 638   g_object_unref (task);
 639 }
 640
 641 static void
 642 splice_cb (GObject      *ostream,
 643            GAsyncResult *res,
 644            gpointer      user_data)
 645 {
 646   GTask *task = user_data;
 647   SpliceContext *ctx = g_task_get_task_data (task);
 648   GError *error = NULL;
 649
 650   g_output_stream_splice_finish (G_OUTPUT_STREAM (ostream), res, &error);
 651
 652   ctx->completed++;
 653
 654   /* ignore cancellation error if it was not requested by the user */
 655   if (error != NULL &&
 656       g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED) &&
 657       (ctx->cancellable == NULL ||
 658        !g_cancellable_is_cancelled (ctx->cancellable)))
 659     g_clear_error (&error);
 660
 661   /* Keep the first error that occurred */
 662   if (error != NULL && ctx->error == NULL)
 663     ctx->error = error;
 664   else
 665     g_clear_error (&error);
 666
 667    if (ctx->completed == 1 &&
 668        (ctx->flags & G_IO_STREAM_SPLICE_WAIT_FOR_BOTH) == 0)
 669     {
 670       /* We don't want to wait for the 2nd operation to finish, cancel it */
 671       g_cancellable_cancel (ctx->op1_cancellable);
 672       g_cancellable_cancel (ctx->op2_cancellable);
 673     }
 674   else if (ctx->completed == 2)
 675     {
 676       if (ctx->cancellable == NULL ||
 677           !g_cancellable_is_cancelled (ctx->cancellable))
 678         {
 679           g_cancellable_reset (ctx->op1_cancellable);
 680           g_cancellable_reset (ctx->op2_cancellable);
 681         }
 682
 683       /* Close the IO streams if needed */
 684       if ((ctx->flags & G_IO_STREAM_SPLICE_CLOSE_STREAM1) != 0)
 685         {
 686           g_io_stream_close_async (ctx->stream1,
 687                                    g_task_get_priority (task),
 688                                    ctx->op1_cancellable,
 689                                    splice_close_cb, g_object_ref (task));
 690         }
 691       else
 692         ctx->completed++;
 693
 694       if ((ctx->flags & G_IO_STREAM_SPLICE_CLOSE_STREAM2) != 0)
 695         {
 696           g_io_stream_close_async (ctx->stream2,
 697                                    g_task_get_priority (task),
 698                                    ctx->op2_cancellable,
 699                                    splice_close_cb, g_object_ref (task));
 700         }
 701       else
 702         ctx->completed++;
 703
 704       /* If all operations are done, complete now */
 705       if (ctx->completed == 4)
 706         splice_complete (task, ctx);
 707     }
 708
 709   g_object_unref (task);
 710 }
 711
 712 static void
 713 splice_cancelled_cb (GCancellable *cancellable,
 714                      GTask        *task)
 715 {
 716   SpliceContext *ctx;
 717
 718   ctx = g_task_get_task_data (task);
 719   g_cancellable_cancel (ctx->op1_cancellable);
 720   g_cancellable_cancel (ctx->op2_cancellable);
 721 }
 722
 723 /**
 724  * g_io_stream_splice_async:
 725  * @stream1: a #GIOStream.
 726  * @stream2: a #GIOStream.
 727  * @flags: a set of #GIOStreamSpliceFlags.
 728  * @io_priority: the io priority of the request.
 729  * @cancellable: (allow-none): optional #GCancellable object, %NULL to ignore.
 730  * @callback: (scope async): a #GAsyncReadyCallback.
 731  * @user_data: (closure): user data passed to @callback.
 732  *
 733  * Asyncronously splice the output stream of @stream1 to the input stream of
 734  * @stream2, and splice the output stream of @stream2 to the input stream of
 735  * @stream1.
 736  *
 737  * When the operation is finished @callback will be called.
 738  * You can then call g_io_stream_splice_finish() to get the
 739  * result of the operation.
 740  *
 741  * Since: 2.28
 742  **/
 743 void
 744 g_io_stream_splice_async (GIOStream            *stream1,
 745                           GIOStream            *stream2,
 746                           GIOStreamSpliceFlags  flags,
 747                           gint                  io_priority,
 748                           GCancellable         *cancellable,
 749                           GAsyncReadyCallback   callback,
 750                           gpointer              user_data)
 751 {
 752   GTask *task;
 753   SpliceContext *ctx;
 754   GInputStream *istream;
 755   GOutputStream *ostream;
 756
 757   if (cancellable != NULL && g_cancellable_is_cancelled (cancellable))
 758     {
 759       g_task_report_new_error (NULL, callback, user_data,
 760                                g_io_stream_splice_async,
 761                                G_IO_ERROR, G_IO_ERROR_CANCELLED,
 762                                "Operation has been cancelled");
 763       return;
 764     }
 765
 766   ctx = g_slice_new0 (SpliceContext);
 767   ctx->stream1 = g_object_ref (stream1);
 768   ctx->stream2 = g_object_ref (stream2);
 769   ctx->flags = flags;
 770   ctx->op1_cancellable = g_cancellable_new ();
 771   ctx->op2_cancellable = g_cancellable_new ();
 772   ctx->completed = 0;
 773
 774   task = g_task_new (NULL, cancellable, callback, user_data);
 775   g_task_set_task_data (task, ctx, (GDestroyNotify) splice_context_free);
 776
 777   if (cancellable != NULL)
 778     {
 779       ctx->cancellable = g_object_ref (cancellable);
 780       ctx->cancelled_id = g_cancellable_connect (cancellable,
 781           G_CALLBACK (splice_cancelled_cb), g_object_ref (task),
 782           g_object_unref);
 783     }
 784
 785   istream = g_io_stream_get_input_stream (stream1);
 786   ostream = g_io_stream_get_output_stream (stream2);
 787   g_output_stream_splice_async (ostream, istream, G_OUTPUT_STREAM_SPLICE_NONE,
 788       io_priority, ctx->op1_cancellable, splice_cb,
 789       g_object_ref (task));
 790
 791   istream = g_io_stream_get_input_stream (stream2);
 792   ostream = g_io_stream_get_output_stream (stream1);
 793   g_output_stream_splice_async (ostream, istream, G_OUTPUT_STREAM_SPLICE_NONE,
 794       io_priority, ctx->op2_cancellable, splice_cb,
 795       g_object_ref (task));
 796
 797   g_object_unref (task);
 798 }
 799
 800 /**
 801  * g_io_stream_splice_finish:
 802  * @result: a #GAsyncResult.
 803  * @error: a #GError location to store the error occurring, or %NULL to
 804  * ignore.
 805  *
 806  * Finishes an asynchronous io stream splice operation.
 807  *
 808  * Returns: %TRUE on success, %FALSE otherwise.
 809  *
 810  * Since: 2.28
 811  **/
 812 gboolean
 813 g_io_stream_splice_finish (GAsyncResult  *result,
 814                            GError       **error)
 815 {
 816   g_return_val_if_fail (g_task_is_valid (result, NULL), FALSE);
 817
 818   return g_task_propagate_boolean (G_TASK (result), error);
 819 }