Coverage Report

Created: 2025-10-10 07:05

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