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