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