Coverage Report

Created: 2025-07-01 07:09

/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
 * 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
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
0
{
118
0
  GIOStream *stream;
119
120
0
  stream = G_IO_STREAM (object);
121
122
0
  if (!stream->priv->closed)
123
0
    g_io_stream_close (stream, NULL, NULL);
124
125
0
  G_OBJECT_CLASS (g_io_stream_parent_class)->dispose (object);
126
0
}
127
128
static void
129
g_io_stream_init (GIOStream *stream)
130
0
{
131
0
  stream->priv = g_io_stream_get_instance_private (stream);
132
0
}
133
134
static void
135
g_io_stream_get_property (GObject    *object,
136
        guint       prop_id,
137
        GValue     *value,
138
        GParamSpec *pspec)
139
0
{
140
0
  GIOStream *stream = G_IO_STREAM (object);
141
142
0
  switch (prop_id)
143
0
    {
144
0
      case PROP_CLOSED:
145
0
        g_value_set_boolean (value, stream->priv->closed);
146
0
        break;
147
148
0
      case PROP_INPUT_STREAM:
149
0
        g_value_set_object (value, g_io_stream_get_input_stream (stream));
150
0
        break;
151
152
0
      case PROP_OUTPUT_STREAM:
153
0
        g_value_set_object (value, g_io_stream_get_output_stream (stream));
154
0
        break;
155
156
0
      default:
157
0
        G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
158
0
    }
159
0
}
160
161
static void
162
g_io_stream_class_init (GIOStreamClass *klass)
163
0
{
164
0
  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
165
166
0
  gobject_class->dispose = g_io_stream_dispose;
167
0
  gobject_class->get_property = g_io_stream_get_property;
168
169
0
  klass->close_fn = g_io_stream_real_close;
170
0
  klass->close_async = g_io_stream_real_close_async;
171
0
  klass->close_finish = g_io_stream_real_close_finish;
172
173
0
  g_object_class_install_property (gobject_class, PROP_CLOSED,
174
0
                                   g_param_spec_boolean ("closed",
175
0
                                                         P_("Closed"),
176
0
                                                         P_("Is the stream closed"),
177
0
                                                         FALSE,
178
0
                                                         G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
179
180
0
  g_object_class_install_property (gobject_class, PROP_INPUT_STREAM,
181
0
           g_param_spec_object ("input-stream",
182
0
              P_("Input stream"),
183
0
              P_("The GInputStream to read from"),
184
0
              G_TYPE_INPUT_STREAM,
185
0
              G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
186
0
  g_object_class_install_property (gobject_class, PROP_OUTPUT_STREAM,
187
0
           g_param_spec_object ("output-stream",
188
0
              P_("Output stream"),
189
0
              P_("The GOutputStream to write to"),
190
0
              G_TYPE_OUTPUT_STREAM,
191
0
              G_PARAM_READABLE | G_PARAM_STATIC_STRINGS));
192
0
}
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
205
g_io_stream_is_closed (GIOStream *stream)
206
0
{
207
0
  g_return_val_if_fail (G_IS_IO_STREAM (stream), TRUE);
208
209
0
  return stream->priv->closed;
210
0
}
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 *
225
g_io_stream_get_input_stream (GIOStream *stream)
226
0
{
227
0
  GIOStreamClass *klass;
228
229
0
  klass = G_IO_STREAM_GET_CLASS (stream);
230
231
0
  g_assert (klass->get_input_stream != NULL);
232
233
0
  return klass->get_input_stream (stream);
234
0
}
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 *
249
g_io_stream_get_output_stream (GIOStream *stream)
250
0
{
251
0
  GIOStreamClass *klass;
252
253
0
  klass = G_IO_STREAM_GET_CLASS (stream);
254
255
0
  g_assert (klass->get_output_stream != NULL);
256
0
  return klass->get_output_stream (stream);
257
0
}
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
270
g_io_stream_has_pending (GIOStream *stream)
271
0
{
272
0
  g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
273
274
0
  return stream->priv->pending;
275
0
}
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
292
g_io_stream_set_pending (GIOStream  *stream,
293
       GError    **error)
294
0
{
295
0
  g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
296
297
0
  if (stream->priv->closed)
298
0
    {
299
0
      g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
300
0
                           _("Stream is already closed"));
301
0
      return FALSE;
302
0
    }
303
304
0
  if (stream->priv->pending)
305
0
    {
306
0
      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
0
                           _("Stream has outstanding operation"));
311
0
      return FALSE;
312
0
    }
313
314
0
  stream->priv->pending = TRUE;
315
0
  return TRUE;
316
0
}
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
327
g_io_stream_clear_pending (GIOStream *stream)
328
0
{
329
0
  g_return_if_fail (G_IS_IO_STREAM (stream));
330
331
0
  stream->priv->pending = FALSE;
332
0
}
333
334
static gboolean
335
g_io_stream_real_close (GIOStream     *stream,
336
      GCancellable  *cancellable,
337
      GError       **error)
338
0
{
339
0
  gboolean res;
340
341
0
  res = g_output_stream_close (g_io_stream_get_output_stream (stream),
342
0
             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
0
  if (error != NULL && *error != NULL)
347
0
    error = NULL;
348
349
0
  res &= g_input_stream_close (g_io_stream_get_input_stream (stream),
350
0
             cancellable, error);
351
352
0
  return res;
353
0
}
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
400
g_io_stream_close (GIOStream     *stream,
401
       GCancellable  *cancellable,
402
       GError       **error)
403
0
{
404
0
  GIOStreamClass *class;
405
0
  gboolean res;
406
407
0
  g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
408
409
0
  class = G_IO_STREAM_GET_CLASS (stream);
410
411
0
  if (stream->priv->closed)
412
0
    return TRUE;
413
414
0
  if (!g_io_stream_set_pending (stream, error))
415
0
    return FALSE;
416
417
0
  if (cancellable)
418
0
    g_cancellable_push_current (cancellable);
419
420
0
  res = TRUE;
421
0
  if (class->close_fn)
422
0
    res = class->close_fn (stream, cancellable, error);
423
424
0
  if (cancellable)
425
0
    g_cancellable_pop_current (cancellable);
426
427
0
  stream->priv->closed = TRUE;
428
0
  g_io_stream_clear_pending (stream);
429
430
0
  return res;
431
0
}
432
433
static void
434
async_ready_close_callback_wrapper (GObject      *source_object,
435
                                    GAsyncResult *res,
436
                                    gpointer      user_data)
437
0
{
438
0
  GIOStream *stream = G_IO_STREAM (source_object);
439
0
  GIOStreamClass *klass = G_IO_STREAM_GET_CLASS (stream);
440
0
  GTask *task = user_data;
441
0
  GError *error = NULL;
442
0
  gboolean success;
443
444
0
  stream->priv->closed = TRUE;
445
0
  g_io_stream_clear_pending (stream);
446
447
0
  if (g_async_result_legacy_propagate_error (res, &error))
448
0
    success = FALSE;
449
0
  else
450
0
    success = klass->close_finish (stream, res, &error);
451
452
0
  if (error)
453
0
    g_task_return_error (task, error);
454
0
  else
455
0
    g_task_return_boolean (task, success);
456
457
0
  g_object_unref (task);
458
0
}
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
482
g_io_stream_close_async (GIOStream           *stream,
483
       int                  io_priority,
484
       GCancellable        *cancellable,
485
       GAsyncReadyCallback  callback,
486
       gpointer             user_data)
487
0
{
488
0
  GIOStreamClass *class;
489
0
  GError *error = NULL;
490
0
  GTask *task;
491
492
0
  g_return_if_fail (G_IS_IO_STREAM (stream));
493
494
0
  task = g_task_new (stream, cancellable, callback, user_data);
495
0
  g_task_set_source_tag (task, g_io_stream_close_async);
496
497
0
  if (stream->priv->closed)
498
0
    {
499
0
      g_task_return_boolean (task, TRUE);
500
0
      g_object_unref (task);
501
0
      return;
502
0
    }
503
504
0
  if (!g_io_stream_set_pending (stream, &error))
505
0
    {
506
0
      g_task_return_error (task, error);
507
0
      g_object_unref (task);
508
0
      return;
509
0
    }
510
511
0
  class = G_IO_STREAM_GET_CLASS (stream);
512
513
0
  class->close_async (stream, io_priority, cancellable,
514
0
          async_ready_close_callback_wrapper, task);
515
0
}
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
531
g_io_stream_close_finish (GIOStream     *stream,
532
        GAsyncResult  *result,
533
        GError       **error)
534
0
{
535
0
  g_return_val_if_fail (G_IS_IO_STREAM (stream), FALSE);
536
0
  g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
537
538
0
  return g_task_propagate_boolean (G_TASK (result), error);
539
0
}
540
541
542
static void
543
close_async_thread (GTask        *task,
544
                    gpointer      source_object,
545
                    gpointer      task_data,
546
                    GCancellable *cancellable)
547
0
{
548
0
  GIOStream *stream = source_object;
549
0
  GIOStreamClass *class;
550
0
  GError *error = NULL;
551
0
  gboolean result;
552
553
0
  class = G_IO_STREAM_GET_CLASS (stream);
554
0
  if (class->close_fn)
555
0
    {
556
0
      result = class->close_fn (stream,
557
0
                                g_task_get_cancellable (task),
558
0
                                &error);
559
0
      if (!result)
560
0
        {
561
0
          g_task_return_error (task, error);
562
0
          return;
563
0
        }
564
0
    }
565
566
0
  g_task_return_boolean (task, TRUE);
567
0
}
568
569
typedef struct
570
{
571
  GError *error;
572
  gint pending;
573
} CloseAsyncData;
574
575
static void
576
stream_close_complete (GObject      *source,
577
                       GAsyncResult *result,
578
                       gpointer      user_data)
579
0
{
580
0
  GTask *task = user_data;
581
0
  CloseAsyncData *data;
582
583
0
  data = g_task_get_task_data (task);
584
0
  data->pending--;
585
586
0
  if (G_IS_OUTPUT_STREAM (source))
587
0
    {
588
0
      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
0
      g_output_stream_close_finish (G_OUTPUT_STREAM (source), result, &error);
594
0
      if (error)
595
0
        {
596
0
          if (data->error)
597
0
            g_error_free (data->error);
598
0
          data->error = error;
599
0
        }
600
0
    }
601
0
  else
602
0
    g_input_stream_close_finish (G_INPUT_STREAM (source), result, data->error ? NULL : &data->error);
603
604
0
  if (data->pending == 0)
605
0
    {
606
0
      if (data->error)
607
0
        g_task_return_error (task, data->error);
608
0
      else
609
0
        g_task_return_boolean (task, TRUE);
610
611
0
      g_slice_free (CloseAsyncData, data);
612
0
      g_object_unref (task);
613
0
    }
614
0
}
615
616
static void
617
g_io_stream_real_close_async (GIOStream           *stream,
618
            int                  io_priority,
619
            GCancellable        *cancellable,
620
            GAsyncReadyCallback  callback,
621
            gpointer             user_data)
622
0
{
623
0
  GInputStream *input;
624
0
  GOutputStream *output;
625
0
  GTask *task;
626
627
0
  task = g_task_new (stream, cancellable, callback, user_data);
628
0
  g_task_set_source_tag (task, g_io_stream_real_close_async);
629
0
  g_task_set_check_cancellable (task, FALSE);
630
0
  g_task_set_priority (task, io_priority);
631
632
0
  input = g_io_stream_get_input_stream (stream);
633
0
  output = g_io_stream_get_output_stream (stream);
634
635
0
  if (g_input_stream_async_close_is_via_threads (input) && g_output_stream_async_close_is_via_threads (output))
636
0
    {
637
      /* No sense in dispatching to the thread twice -- just do it all
638
       * in one go.
639
       */
640
0
      g_task_run_in_thread (task, close_async_thread);
641
0
      g_object_unref (task);
642
0
    }
643
0
  else
644
0
    {
645
0
      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
0
      data = g_slice_new (CloseAsyncData);
652
0
      data->error = NULL;
653
0
      data->pending = 2;
654
655
0
      g_task_set_task_data (task, data, NULL);
656
0
      g_input_stream_close_async (input, io_priority, cancellable, stream_close_complete, task);
657
0
      g_output_stream_close_async (output, io_priority, cancellable, stream_close_complete, task);
658
0
    }
659
0
}
660
661
static gboolean
662
g_io_stream_real_close_finish (GIOStream     *stream,
663
             GAsyncResult  *result,
664
             GError       **error)
665
0
{
666
0
  g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
667
668
0
  return g_task_propagate_boolean (G_TASK (result), error);
669
0
}
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
686
splice_context_free (SpliceContext *ctx)
687
0
{
688
0
  g_object_unref (ctx->stream1);
689
0
  g_object_unref (ctx->stream2);
690
0
  if (ctx->cancellable != NULL)
691
0
    g_object_unref (ctx->cancellable);
692
0
  g_object_unref (ctx->op1_cancellable);
693
0
  g_object_unref (ctx->op2_cancellable);
694
0
  g_clear_error (&ctx->error);
695
0
  g_slice_free (SpliceContext, ctx);
696
0
}
697
698
static void
699
splice_complete (GTask         *task,
700
                 SpliceContext *ctx)
701
0
{
702
0
  if (ctx->cancelled_id != 0)
703
0
    g_cancellable_disconnect (ctx->cancellable, ctx->cancelled_id);
704
0
  ctx->cancelled_id = 0;
705
706
0
  if (ctx->error != NULL)
707
0
    {
708
0
      g_task_return_error (task, ctx->error);
709
0
      ctx->error = NULL;
710
0
    }
711
0
  else
712
0
    g_task_return_boolean (task, TRUE);
713
0
}
714
715
static void
716
splice_close_cb (GObject      *iostream,
717
                 GAsyncResult *res,
718
                 gpointer      user_data)
719
0
{
720
0
  GTask *task = user_data;
721
0
  SpliceContext *ctx = g_task_get_task_data (task);
722
0
  GError *error = NULL;
723
724
0
  g_io_stream_close_finish (G_IO_STREAM (iostream), res, &error);
725
726
0
  ctx->completed++;
727
728
  /* Keep the first error that occurred */
729
0
  if (error != NULL && ctx->error == NULL)
730
0
    ctx->error = error;
731
0
  else
732
0
    g_clear_error (&error);
733
734
  /* If all operations are done, complete now */
735
0
  if (ctx->completed == 4)
736
0
    splice_complete (task, ctx);
737
738
0
  g_object_unref (task);
739
0
}
740
741
static void
742
splice_cb (GObject      *ostream,
743
           GAsyncResult *res,
744
           gpointer      user_data)
745
0
{
746
0
  GTask *task = user_data;
747
0
  SpliceContext *ctx = g_task_get_task_data (task);
748
0
  GError *error = NULL;
749
750
0
  g_output_stream_splice_finish (G_OUTPUT_STREAM (ostream), res, &error);
751
752
0
  ctx->completed++;
753
754
  /* ignore cancellation error if it was not requested by the user */
755
0
  if (error != NULL &&
756
0
      g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED) &&
757
0
      (ctx->cancellable == NULL ||
758
0
       !g_cancellable_is_cancelled (ctx->cancellable)))
759
0
    g_clear_error (&error);
760
761
  /* Keep the first error that occurred */
762
0
  if (error != NULL && ctx->error == NULL)
763
0
    ctx->error = error;
764
0
  else
765
0
    g_clear_error (&error);
766
767
0
   if (ctx->completed == 1 &&
768
0
       (ctx->flags & G_IO_STREAM_SPLICE_WAIT_FOR_BOTH) == 0)
769
0
    {
770
      /* We don't want to wait for the 2nd operation to finish, cancel it */
771
0
      g_cancellable_cancel (ctx->op1_cancellable);
772
0
      g_cancellable_cancel (ctx->op2_cancellable);
773
0
    }
774
0
  else if (ctx->completed == 2)
775
0
    {
776
0
      if (ctx->cancellable == NULL ||
777
0
          !g_cancellable_is_cancelled (ctx->cancellable))
778
0
        {
779
0
          g_cancellable_reset (ctx->op1_cancellable);
780
0
          g_cancellable_reset (ctx->op2_cancellable);
781
0
        }
782
783
      /* Close the IO streams if needed */
784
0
      if ((ctx->flags & G_IO_STREAM_SPLICE_CLOSE_STREAM1) != 0)
785
0
  {
786
0
    g_io_stream_close_async (ctx->stream1,
787
0
                                   g_task_get_priority (task),
788
0
                                   ctx->op1_cancellable,
789
0
                                   splice_close_cb, g_object_ref (task));
790
0
  }
791
0
      else
792
0
        ctx->completed++;
793
794
0
      if ((ctx->flags & G_IO_STREAM_SPLICE_CLOSE_STREAM2) != 0)
795
0
  {
796
0
    g_io_stream_close_async (ctx->stream2,
797
0
                                   g_task_get_priority (task),
798
0
                                   ctx->op2_cancellable,
799
0
                                   splice_close_cb, g_object_ref (task));
800
0
  }
801
0
      else
802
0
        ctx->completed++;
803
804
      /* If all operations are done, complete now */
805
0
      if (ctx->completed == 4)
806
0
        splice_complete (task, ctx);
807
0
    }
808
809
0
  g_object_unref (task);
810
0
}
811
812
static void
813
splice_cancelled_cb (GCancellable *cancellable,
814
                     GTask        *task)
815
0
{
816
0
  SpliceContext *ctx;
817
818
0
  ctx = g_task_get_task_data (task);
819
0
  g_cancellable_cancel (ctx->op1_cancellable);
820
0
  g_cancellable_cancel (ctx->op2_cancellable);
821
0
}
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
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
0
{
852
0
  GTask *task;
853
0
  SpliceContext *ctx;
854
0
  GInputStream *istream;
855
0
  GOutputStream *ostream;
856
857
0
  if (cancellable != NULL && g_cancellable_is_cancelled (cancellable))
858
0
    {
859
0
      g_task_report_new_error (NULL, callback, user_data,
860
0
                               g_io_stream_splice_async,
861
0
                               G_IO_ERROR, G_IO_ERROR_CANCELLED,
862
0
                               "Operation has been cancelled");
863
0
      return;
864
0
    }
865
866
0
  ctx = g_slice_new0 (SpliceContext);
867
0
  ctx->stream1 = g_object_ref (stream1);
868
0
  ctx->stream2 = g_object_ref (stream2);
869
0
  ctx->flags = flags;
870
0
  ctx->op1_cancellable = g_cancellable_new ();
871
0
  ctx->op2_cancellable = g_cancellable_new ();
872
0
  ctx->completed = 0;
873
874
0
  task = g_task_new (NULL, cancellable, callback, user_data);
875
0
  g_task_set_source_tag (task, g_io_stream_splice_async);
876
0
  g_task_set_task_data (task, ctx, (GDestroyNotify) splice_context_free);
877
878
0
  if (cancellable != NULL)
879
0
    {
880
0
      ctx->cancellable = g_object_ref (cancellable);
881
0
      ctx->cancelled_id = g_cancellable_connect (cancellable,
882
0
          G_CALLBACK (splice_cancelled_cb), g_object_ref (task),
883
0
          g_object_unref);
884
0
    }
885
886
0
  istream = g_io_stream_get_input_stream (stream1);
887
0
  ostream = g_io_stream_get_output_stream (stream2);
888
0
  g_output_stream_splice_async (ostream, istream, G_OUTPUT_STREAM_SPLICE_NONE,
889
0
      io_priority, ctx->op1_cancellable, splice_cb,
890
0
      g_object_ref (task));
891
892
0
  istream = g_io_stream_get_input_stream (stream2);
893
0
  ostream = g_io_stream_get_output_stream (stream1);
894
0
  g_output_stream_splice_async (ostream, istream, G_OUTPUT_STREAM_SPLICE_NONE,
895
0
      io_priority, ctx->op2_cancellable, splice_cb,
896
0
      g_object_ref (task));
897
898
0
  g_object_unref (task);
899
0
}
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
914
g_io_stream_splice_finish (GAsyncResult  *result,
915
                           GError       **error)
916
0
{
917
0
  g_return_val_if_fail (g_task_is_valid (result, NULL), FALSE);
918
919
0
  return g_task_propagate_boolean (G_TASK (result), error);
920
0
}