Coverage Report

Created: 2025-06-13 06:55

/src/glib/gio/giostream.c
Line
Count
Source (jump to first uncovered line)
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
G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE (GIOStream, g_io_stream, G_TYPE_OBJECT)
116
117
static void
118
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): a #GAsyncReadyCallback
468
 *   to call when the request is satisfied
469
 * @user_data: the data to pass to callback function
470
 *
471
 * Requests an asynchronous close of the stream, releasing resources
472
 * related to it. When the operation is finished @callback will be
473
 * called. You can then call g_io_stream_close_finish() to get
474
 * the result of the operation.
475
 *
476
 * For behaviour details see g_io_stream_close().
477
 *
478
 * The asynchronous methods have a default fallback that uses threads
479
 * to implement asynchronicity, so they are optional for inheriting
480
 * classes. However, if you override one you must override all.
481
 *
482
 * Since: 2.22
483
 */
484
void
485
g_io_stream_close_async (GIOStream           *stream,
486
       int                  io_priority,
487
       GCancellable        *cancellable,
488
       GAsyncReadyCallback  callback,
489
       gpointer             user_data)
490
0
{
491
0
  GIOStreamClass *class;
492
0
  GError *error = NULL;
493
0
  GTask *task;
494
495
0
  g_return_if_fail (G_IS_IO_STREAM (stream));
496
497
0
  task = g_task_new (stream, cancellable, callback, user_data);
498
0
  g_task_set_source_tag (task, g_io_stream_close_async);
499
500
0
  if (stream->priv->closed)
501
0
    {
502
0
      g_task_return_boolean (task, TRUE);
503
0
      g_object_unref (task);
504
0
      return;
505
0
    }
506
507
0
  if (!g_io_stream_set_pending (stream, &error))
508
0
    {
509
0
      g_task_return_error (task, error);
510
0
      g_object_unref (task);
511
0
      return;
512
0
    }
513
514
0
  class = G_IO_STREAM_GET_CLASS (stream);
515
516
0
  class->close_async (stream, io_priority, cancellable,
517
0
          async_ready_close_callback_wrapper, task);
518
0
}
519
520
/**
521
 * g_io_stream_close_finish:
522
 * @stream: a #GIOStream
523
 * @result: a #GAsyncResult
524
 * @error: a #GError location to store the error occurring, or %NULL to
525
 *    ignore
526
 *
527
 * Closes a stream.
528
 *
529
 * Returns: %TRUE if stream was successfully closed, %FALSE otherwise.
530
 *
531
 * Since: 2.22
532
 */
533
gboolean
534
g_io_stream_close_finish (GIOStream     *stream,
535
        GAsyncResult  *result,
536
        GError       **error)
537
0
{
538
0
  g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
539
0
  g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
540
541
0
  return g_task_propagate_boolean (G_TASK (result), error);
542
0
}
543
544
545
static void
546
close_async_thread (GTask        *task,
547
                    gpointer      source_object,
548
                    gpointer      task_data,
549
                    GCancellable *cancellable)
550
0
{
551
0
  GIOStream *stream = source_object;
552
0
  GIOStreamClass *class;
553
0
  GError *error = NULL;
554
0
  gboolean result;
555
556
0
  class = G_IO_STREAM_GET_CLASS (stream);
557
0
  if (class->close_fn)
558
0
    {
559
0
      result = class->close_fn (stream,
560
0
                                g_task_get_cancellable (task),
561
0
                                &error);
562
0
      if (!result)
563
0
        {
564
0
          g_task_return_error (task, error);
565
0
          return;
566
0
        }
567
0
    }
568
569
0
  g_task_return_boolean (task, TRUE);
570
0
}
571
572
typedef struct
573
{
574
  GError *error;
575
  gint pending;
576
} CloseAsyncData;
577
578
static void
579
stream_close_complete (GObject      *source,
580
                       GAsyncResult *result,
581
                       gpointer      user_data)
582
0
{
583
0
  GTask *task = user_data;
584
0
  CloseAsyncData *data;
585
586
0
  data = g_task_get_task_data (task);
587
0
  data->pending--;
588
589
0
  if (G_IS_OUTPUT_STREAM (source))
590
0
    {
591
0
      GError *error = NULL;
592
593
      /* Match behaviour with the sync route and give precedent to the
594
       * error returned from closing the output stream.
595
       */
596
0
      g_output_stream_close_finish (G_OUTPUT_STREAM (source), result, &error);
597
0
      if (error)
598
0
        {
599
0
          if (data->error)
600
0
            g_error_free (data->error);
601
0
          data->error = error;
602
0
        }
603
0
    }
604
0
  else
605
0
    g_input_stream_close_finish (G_INPUT_STREAM (source), result, data->error ? NULL : &data->error);
606
607
0
  if (data->pending == 0)
608
0
    {
609
0
      if (data->error)
610
0
        g_task_return_error (task, data->error);
611
0
      else
612
0
        g_task_return_boolean (task, TRUE);
613
614
0
      g_slice_free (CloseAsyncData, data);
615
0
      g_object_unref (task);
616
0
    }
617
0
}
618
619
static void
620
g_io_stream_real_close_async (GIOStream           *stream,
621
            int                  io_priority,
622
            GCancellable        *cancellable,
623
            GAsyncReadyCallback  callback,
624
            gpointer             user_data)
625
0
{
626
0
  GInputStream *input;
627
0
  GOutputStream *output;
628
0
  GTask *task;
629
630
0
  task = g_task_new (stream, cancellable, callback, user_data);
631
0
  g_task_set_source_tag (task, g_io_stream_real_close_async);
632
0
  g_task_set_check_cancellable (task, FALSE);
633
0
  g_task_set_priority (task, io_priority);
634
635
0
  input = g_io_stream_get_input_stream (stream);
636
0
  output = g_io_stream_get_output_stream (stream);
637
638
0
  if (g_input_stream_async_close_is_via_threads (input) && g_output_stream_async_close_is_via_threads (output))
639
0
    {
640
      /* No sense in dispatching to the thread twice -- just do it all
641
       * in one go.
642
       */
643
0
      g_task_run_in_thread (task, close_async_thread);
644
0
      g_object_unref (task);
645
0
    }
646
0
  else
647
0
    {
648
0
      CloseAsyncData *data;
649
650
      /* We should avoid dispatching to another thread in case either
651
       * object that would not do it for itself because it may not be
652
       * threadsafe.
653
       */
654
0
      data = g_slice_new (CloseAsyncData);
655
0
      data->error = NULL;
656
0
      data->pending = 2;
657
658
0
      g_task_set_task_data (task, data, NULL);
659
0
      g_input_stream_close_async (input, io_priority, cancellable, stream_close_complete, task);
660
0
      g_output_stream_close_async (output, io_priority, cancellable, stream_close_complete, task);
661
0
    }
662
0
}
663
664
static gboolean
665
g_io_stream_real_close_finish (GIOStream     *stream,
666
             GAsyncResult  *result,
667
             GError       **error)
668
0
{
669
0
  g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
670
671
0
  return g_task_propagate_boolean (G_TASK (result), error);
672
0
}
673
674
typedef struct
675
{
676
  GIOStream *stream1;
677
  GIOStream *stream2;
678
  GIOStreamSpliceFlags flags;
679
  gint io_priority;
680
  GCancellable *cancellable;
681
  gulong cancelled_id;
682
  GCancellable *op1_cancellable;
683
  GCancellable *op2_cancellable;
684
  guint completed;
685
  GError *error;
686
} SpliceContext;
687
688
static void
689
splice_context_free (SpliceContext *ctx)
690
0
{
691
0
  g_object_unref (ctx->stream1);
692
0
  g_object_unref (ctx->stream2);
693
0
  if (ctx->cancellable != NULL)
694
0
    g_object_unref (ctx->cancellable);
695
0
  g_object_unref (ctx->op1_cancellable);
696
0
  g_object_unref (ctx->op2_cancellable);
697
0
  g_clear_error (&ctx->error);
698
0
  g_slice_free (SpliceContext, ctx);
699
0
}
700
701
static void
702
splice_complete (GTask         *task,
703
                 SpliceContext *ctx)
704
0
{
705
0
  if (ctx->cancelled_id != 0)
706
0
    g_cancellable_disconnect (ctx->cancellable, ctx->cancelled_id);
707
0
  ctx->cancelled_id = 0;
708
709
0
  if (ctx->error != NULL)
710
0
    {
711
0
      g_task_return_error (task, ctx->error);
712
0
      ctx->error = NULL;
713
0
    }
714
0
  else
715
0
    g_task_return_boolean (task, TRUE);
716
0
}
717
718
static void
719
splice_close_cb (GObject      *iostream,
720
                 GAsyncResult *res,
721
                 gpointer      user_data)
722
0
{
723
0
  GTask *task = user_data;
724
0
  SpliceContext *ctx = g_task_get_task_data (task);
725
0
  GError *error = NULL;
726
727
0
  g_io_stream_close_finish (G_IO_STREAM (iostream), res, &error);
728
729
0
  ctx->completed++;
730
731
  /* Keep the first error that occurred */
732
0
  if (error != NULL && ctx->error == NULL)
733
0
    ctx->error = error;
734
0
  else
735
0
    g_clear_error (&error);
736
737
  /* If all operations are done, complete now */
738
0
  if (ctx->completed == 4)
739
0
    splice_complete (task, ctx);
740
741
0
  g_object_unref (task);
742
0
}
743
744
static void
745
splice_cb (GObject      *ostream,
746
           GAsyncResult *res,
747
           gpointer      user_data)
748
0
{
749
0
  GTask *task = user_data;
750
0
  SpliceContext *ctx = g_task_get_task_data (task);
751
0
  GError *error = NULL;
752
753
0
  g_output_stream_splice_finish (G_OUTPUT_STREAM (ostream), res, &error);
754
755
0
  ctx->completed++;
756
757
  /* ignore cancellation error if it was not requested by the user */
758
0
  if (error != NULL &&
759
0
      g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED) &&
760
0
      (ctx->cancellable == NULL ||
761
0
       !g_cancellable_is_cancelled (ctx->cancellable)))
762
0
    g_clear_error (&error);
763
764
  /* Keep the first error that occurred */
765
0
  if (error != NULL && ctx->error == NULL)
766
0
    ctx->error = error;
767
0
  else
768
0
    g_clear_error (&error);
769
770
0
   if (ctx->completed == 1 &&
771
0
       (ctx->flags & G_IO_STREAM_SPLICE_WAIT_FOR_BOTH) == 0)
772
0
    {
773
      /* We don't want to wait for the 2nd operation to finish, cancel it */
774
0
      g_cancellable_cancel (ctx->op1_cancellable);
775
0
      g_cancellable_cancel (ctx->op2_cancellable);
776
0
    }
777
0
  else if (ctx->completed == 2)
778
0
    {
779
0
      if (ctx->cancellable == NULL ||
780
0
          !g_cancellable_is_cancelled (ctx->cancellable))
781
0
        {
782
0
          g_cancellable_reset (ctx->op1_cancellable);
783
0
          g_cancellable_reset (ctx->op2_cancellable);
784
0
        }
785
786
      /* Close the IO streams if needed */
787
0
      if ((ctx->flags & G_IO_STREAM_SPLICE_CLOSE_STREAM1) != 0)
788
0
  {
789
0
    g_io_stream_close_async (ctx->stream1,
790
0
                                   g_task_get_priority (task),
791
0
                                   ctx->op1_cancellable,
792
0
                                   splice_close_cb, g_object_ref (task));
793
0
  }
794
0
      else
795
0
        ctx->completed++;
796
797
0
      if ((ctx->flags & G_IO_STREAM_SPLICE_CLOSE_STREAM2) != 0)
798
0
  {
799
0
    g_io_stream_close_async (ctx->stream2,
800
0
                                   g_task_get_priority (task),
801
0
                                   ctx->op2_cancellable,
802
0
                                   splice_close_cb, g_object_ref (task));
803
0
  }
804
0
      else
805
0
        ctx->completed++;
806
807
      /* If all operations are done, complete now */
808
0
      if (ctx->completed == 4)
809
0
        splice_complete (task, ctx);
810
0
    }
811
812
0
  g_object_unref (task);
813
0
}
814
815
static void
816
splice_cancelled_cb (GCancellable *cancellable,
817
                     GTask        *task)
818
0
{
819
0
  SpliceContext *ctx;
820
821
0
  ctx = g_task_get_task_data (task);
822
0
  g_cancellable_cancel (ctx->op1_cancellable);
823
0
  g_cancellable_cancel (ctx->op2_cancellable);
824
0
}
825
826
/**
827
 * g_io_stream_splice_async:
828
 * @stream1: a #GIOStream.
829
 * @stream2: a #GIOStream.
830
 * @flags: a set of #GIOStreamSpliceFlags.
831
 * @io_priority: the io priority of the request.
832
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.
833
 * @callback: (scope async): a #GAsyncReadyCallback
834
 *   to call when the request is satisfied
835
 * @user_data: the data to pass to callback function
836
 *
837
 * Asynchronously splice the output stream of @stream1 to the input stream of
838
 * @stream2, and splice the output stream of @stream2 to the input stream of
839
 * @stream1.
840
 *
841
 * When the operation is finished @callback will be called.
842
 * You can then call g_io_stream_splice_finish() to get the
843
 * result of the operation.
844
 *
845
 * Since: 2.28
846
 **/
847
void
848
g_io_stream_splice_async (GIOStream            *stream1,
849
                          GIOStream            *stream2,
850
                          GIOStreamSpliceFlags  flags,
851
                          gint                  io_priority,
852
                          GCancellable         *cancellable,
853
                          GAsyncReadyCallback   callback,
854
                          gpointer              user_data)
855
0
{
856
0
  GTask *task;
857
0
  SpliceContext *ctx;
858
0
  GInputStream *istream;
859
0
  GOutputStream *ostream;
860
861
0
  if (cancellable != NULL && g_cancellable_is_cancelled (cancellable))
862
0
    {
863
0
      g_task_report_new_error (NULL, callback, user_data,
864
0
                               g_io_stream_splice_async,
865
0
                               G_IO_ERROR, G_IO_ERROR_CANCELLED,
866
0
                               "Operation has been cancelled");
867
0
      return;
868
0
    }
869
870
0
  ctx = g_slice_new0 (SpliceContext);
871
0
  ctx->stream1 = g_object_ref (stream1);
872
0
  ctx->stream2 = g_object_ref (stream2);
873
0
  ctx->flags = flags;
874
0
  ctx->op1_cancellable = g_cancellable_new ();
875
0
  ctx->op2_cancellable = g_cancellable_new ();
876
0
  ctx->completed = 0;
877
878
0
  task = g_task_new (NULL, cancellable, callback, user_data);
879
0
  g_task_set_source_tag (task, g_io_stream_splice_async);
880
0
  g_task_set_task_data (task, ctx, (GDestroyNotify) splice_context_free);
881
882
0
  if (cancellable != NULL)
883
0
    {
884
0
      ctx->cancellable = g_object_ref (cancellable);
885
0
      ctx->cancelled_id = g_cancellable_connect (cancellable,
886
0
          G_CALLBACK (splice_cancelled_cb), g_object_ref (task),
887
0
          g_object_unref);
888
0
    }
889
890
0
  istream = g_io_stream_get_input_stream (stream1);
891
0
  ostream = g_io_stream_get_output_stream (stream2);
892
0
  g_output_stream_splice_async (ostream, istream, G_OUTPUT_STREAM_SPLICE_NONE,
893
0
      io_priority, ctx->op1_cancellable, splice_cb,
894
0
      g_object_ref (task));
895
896
0
  istream = g_io_stream_get_input_stream (stream2);
897
0
  ostream = g_io_stream_get_output_stream (stream1);
898
0
  g_output_stream_splice_async (ostream, istream, G_OUTPUT_STREAM_SPLICE_NONE,
899
0
      io_priority, ctx->op2_cancellable, splice_cb,
900
0
      g_object_ref (task));
901
902
0
  g_object_unref (task);
903
0
}
904
905
/**
906
 * g_io_stream_splice_finish:
907
 * @result: a #GAsyncResult.
908
 * @error: a #GError location to store the error occurring, or %NULL to
909
 * ignore.
910
 *
911
 * Finishes an asynchronous io stream splice operation.
912
 *
913
 * Returns: %TRUE on success, %FALSE otherwise.
914
 *
915
 * Since: 2.28
916
 **/
917
gboolean
918
g_io_stream_splice_finish (GAsyncResult  *result,
919
                           GError       **error)
920
0
{
921
0
  g_return_val_if_fail (g_task_is_valid (result, NULL), FALSE);
922
923
0
  return g_task_propagate_boolean (G_TASK (result), error);
924
0
}