1 /* GIO - GLib Input, Output and Streaming Library
2 *
3 * Copyright (C) 2006-2007 Red Hat, Inc.
4 * Copyright (C) 2007 Jürg Billeter
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 * Author: Christian Kellner <gicmo@gnome.org>
20 */
21
22 #include "config.h"
23 #include "gbufferedinputstream.h"
24 #include "ginputstream.h"
25 #include "gcancellable.h"
26 #include "gasyncresult.h"
27 #include "gtask.h"
28 #include "gseekable.h"
29 #include "gioerror.h"
30 #include <string.h>
31 #include "glibintl.h"
32
33
34 /**
35 * SECTION:gbufferedinputstream
36 * @short_description: Buffered Input Stream
37 * @include: gio/gio.h
38 * @see_also: #GFilterInputStream, #GInputStream
39 *
40 * Buffered input stream implements #GFilterInputStream and provides
41 * for buffered reads.
42 *
43 * By default, #GBufferedInputStream's buffer size is set at 4 kilobytes.
44 *
45 * To create a buffered input stream, use g_buffered_input_stream_new(),
46 * or g_buffered_input_stream_new_sized() to specify the buffer's size at
47 * construction.
48 *
49 * To get the size of a buffer within a buffered input stream, use
50 * g_buffered_input_stream_get_buffer_size(). To change the size of a
51 * buffered input stream's buffer, use
52 * g_buffered_input_stream_set_buffer_size(). Note that the buffer's size
53 * cannot be reduced below the size of the data within the buffer.
54 */
55
56
57 #define DEFAULT_BUFFER_SIZE 4096
58
59 struct _GBufferedInputStreamPrivate {
60 guint8 *buffer;
61 gsize len;
62 gsize pos;
63 gsize end;
64 GAsyncReadyCallback outstanding_callback;
65 };
66
67 enum {
68 PROP_0,
69 PROP_BUFSIZE
70 };
71
72 static void g_buffered_input_stream_set_property (GObject *object,
73 guint prop_id,
74 const GValue *value,
75 GParamSpec *pspec);
76
77 static void g_buffered_input_stream_get_property (GObject *object,
78 guint prop_id,
79 GValue *value,
80 GParamSpec *pspec);
81 static void g_buffered_input_stream_finalize (GObject *object);
82
83
84 static gssize g_buffered_input_stream_skip (GInputStream *stream,
85 gsize count,
86 GCancellable *cancellable,
87 GError **error);
88 static void g_buffered_input_stream_skip_async (GInputStream *stream,
89 gsize count,
90 int io_priority,
91 GCancellable *cancellable,
92 GAsyncReadyCallback callback,
93 gpointer user_data);
94 static gssize g_buffered_input_stream_skip_finish (GInputStream *stream,
95 GAsyncResult *result,
96 GError **error);
97 static gssize g_buffered_input_stream_read (GInputStream *stream,
98 void *buffer,
99 gsize count,
100 GCancellable *cancellable,
101 GError **error);
102 static gssize g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
103 gssize count,
104 GCancellable *cancellable,
105 GError **error);
106 static void g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
107 gssize count,
108 int io_priority,
109 GCancellable *cancellable,
110 GAsyncReadyCallback callback,
111 gpointer user_data);
112 static gssize g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
113 GAsyncResult *result,
114 GError **error);
115
116 static void g_buffered_input_stream_seekable_iface_init (GSeekableIface *iface);
117 static goffset g_buffered_input_stream_tell (GSeekable *seekable);
118 static gboolean g_buffered_input_stream_can_seek (GSeekable *seekable);
119 static gboolean g_buffered_input_stream_seek (GSeekable *seekable,
120 goffset offset,
121 GSeekType type,
122 GCancellable *cancellable,
123 GError **error);
124 static gboolean g_buffered_input_stream_can_truncate (GSeekable *seekable);
125 static gboolean g_buffered_input_stream_truncate (GSeekable *seekable,
126 goffset offset,
127 GCancellable *cancellable,
128 GError **error);
129
130 static void compact_buffer (GBufferedInputStream *stream);
131
G_DEFINE_TYPE_WITH_CODE(GBufferedInputStream,g_buffered_input_stream,G_TYPE_FILTER_INPUT_STREAM,G_ADD_PRIVATE (GBufferedInputStream)G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE,g_buffered_input_stream_seekable_iface_init))132 G_DEFINE_TYPE_WITH_CODE (GBufferedInputStream,
133 g_buffered_input_stream,
134 G_TYPE_FILTER_INPUT_STREAM,
135 G_ADD_PRIVATE (GBufferedInputStream)
136 G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE,
137 g_buffered_input_stream_seekable_iface_init))
138
139 static void
140 g_buffered_input_stream_class_init (GBufferedInputStreamClass *klass)
141 {
142 GObjectClass *object_class;
143 GInputStreamClass *istream_class;
144 GBufferedInputStreamClass *bstream_class;
145
146 object_class = G_OBJECT_CLASS (klass);
147 object_class->get_property = g_buffered_input_stream_get_property;
148 object_class->set_property = g_buffered_input_stream_set_property;
149 object_class->finalize = g_buffered_input_stream_finalize;
150
151 istream_class = G_INPUT_STREAM_CLASS (klass);
152 istream_class->skip = g_buffered_input_stream_skip;
153 istream_class->skip_async = g_buffered_input_stream_skip_async;
154 istream_class->skip_finish = g_buffered_input_stream_skip_finish;
155 istream_class->read_fn = g_buffered_input_stream_read;
156
157 bstream_class = G_BUFFERED_INPUT_STREAM_CLASS (klass);
158 bstream_class->fill = g_buffered_input_stream_real_fill;
159 bstream_class->fill_async = g_buffered_input_stream_real_fill_async;
160 bstream_class->fill_finish = g_buffered_input_stream_real_fill_finish;
161
162 g_object_class_install_property (object_class,
163 PROP_BUFSIZE,
164 g_param_spec_uint ("buffer-size",
165 P_("Buffer Size"),
166 P_("The size of the backend buffer"),
167 1,
168 G_MAXUINT,
169 DEFAULT_BUFFER_SIZE,
170 G_PARAM_READWRITE | G_PARAM_CONSTRUCT |
171 G_PARAM_STATIC_NAME|G_PARAM_STATIC_NICK|G_PARAM_STATIC_BLURB));
172
173
174 }
175
176 /**
177 * g_buffered_input_stream_get_buffer_size:
178 * @stream: a #GBufferedInputStream
179 *
180 * Gets the size of the input buffer.
181 *
182 * Returns: the current buffer size.
183 */
184 gsize
g_buffered_input_stream_get_buffer_size(GBufferedInputStream * stream)185 g_buffered_input_stream_get_buffer_size (GBufferedInputStream *stream)
186 {
187 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), 0);
188
189 return stream->priv->len;
190 }
191
192 /**
193 * g_buffered_input_stream_set_buffer_size:
194 * @stream: a #GBufferedInputStream
195 * @size: a #gsize
196 *
197 * Sets the size of the internal buffer of @stream to @size, or to the
198 * size of the contents of the buffer. The buffer can never be resized
199 * smaller than its current contents.
200 */
201 void
g_buffered_input_stream_set_buffer_size(GBufferedInputStream * stream,gsize size)202 g_buffered_input_stream_set_buffer_size (GBufferedInputStream *stream,
203 gsize size)
204 {
205 GBufferedInputStreamPrivate *priv;
206 gsize in_buffer;
207 guint8 *buffer;
208
209 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
210
211 priv = stream->priv;
212
213 if (priv->len == size)
214 return;
215
216 if (priv->buffer)
217 {
218 in_buffer = priv->end - priv->pos;
219
220 /* Never resize smaller than current buffer contents */
221 size = MAX (size, in_buffer);
222
223 buffer = g_malloc (size);
224 memcpy (buffer, priv->buffer + priv->pos, in_buffer);
225 priv->len = size;
226 priv->pos = 0;
227 priv->end = in_buffer;
228 g_free (priv->buffer);
229 priv->buffer = buffer;
230 }
231 else
232 {
233 priv->len = size;
234 priv->pos = 0;
235 priv->end = 0;
236 priv->buffer = g_malloc (size);
237 }
238
239 g_object_notify (G_OBJECT (stream), "buffer-size");
240 }
241
242 static void
g_buffered_input_stream_set_property(GObject * object,guint prop_id,const GValue * value,GParamSpec * pspec)243 g_buffered_input_stream_set_property (GObject *object,
244 guint prop_id,
245 const GValue *value,
246 GParamSpec *pspec)
247 {
248 GBufferedInputStream *bstream;
249
250 bstream = G_BUFFERED_INPUT_STREAM (object);
251
252 switch (prop_id)
253 {
254 case PROP_BUFSIZE:
255 g_buffered_input_stream_set_buffer_size (bstream, g_value_get_uint (value));
256 break;
257
258 default:
259 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
260 break;
261 }
262 }
263
264 static void
g_buffered_input_stream_get_property(GObject * object,guint prop_id,GValue * value,GParamSpec * pspec)265 g_buffered_input_stream_get_property (GObject *object,
266 guint prop_id,
267 GValue *value,
268 GParamSpec *pspec)
269 {
270 GBufferedInputStreamPrivate *priv;
271 GBufferedInputStream *bstream;
272
273 bstream = G_BUFFERED_INPUT_STREAM (object);
274 priv = bstream->priv;
275
276 switch (prop_id)
277 {
278 case PROP_BUFSIZE:
279 g_value_set_uint (value, priv->len);
280 break;
281
282 default:
283 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
284 break;
285 }
286 }
287
288 static void
g_buffered_input_stream_finalize(GObject * object)289 g_buffered_input_stream_finalize (GObject *object)
290 {
291 GBufferedInputStreamPrivate *priv;
292 GBufferedInputStream *stream;
293
294 stream = G_BUFFERED_INPUT_STREAM (object);
295 priv = stream->priv;
296
297 g_free (priv->buffer);
298
299 G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize (object);
300 }
301
302 static void
g_buffered_input_stream_seekable_iface_init(GSeekableIface * iface)303 g_buffered_input_stream_seekable_iface_init (GSeekableIface *iface)
304 {
305 iface->tell = g_buffered_input_stream_tell;
306 iface->can_seek = g_buffered_input_stream_can_seek;
307 iface->seek = g_buffered_input_stream_seek;
308 iface->can_truncate = g_buffered_input_stream_can_truncate;
309 iface->truncate_fn = g_buffered_input_stream_truncate;
310 }
311
312 static void
g_buffered_input_stream_init(GBufferedInputStream * stream)313 g_buffered_input_stream_init (GBufferedInputStream *stream)
314 {
315 stream->priv = g_buffered_input_stream_get_instance_private (stream);
316 }
317
318
319 /**
320 * g_buffered_input_stream_new:
321 * @base_stream: a #GInputStream
322 *
323 * Creates a new #GInputStream from the given @base_stream, with
324 * a buffer set to the default size (4 kilobytes).
325 *
326 * Returns: a #GInputStream for the given @base_stream.
327 */
328 GInputStream *
g_buffered_input_stream_new(GInputStream * base_stream)329 g_buffered_input_stream_new (GInputStream *base_stream)
330 {
331 GInputStream *stream;
332
333 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
334
335 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
336 "base-stream", base_stream,
337 NULL);
338
339 return stream;
340 }
341
342 /**
343 * g_buffered_input_stream_new_sized:
344 * @base_stream: a #GInputStream
345 * @size: a #gsize
346 *
347 * Creates a new #GBufferedInputStream from the given @base_stream,
348 * with a buffer set to @size.
349 *
350 * Returns: a #GInputStream.
351 */
352 GInputStream *
g_buffered_input_stream_new_sized(GInputStream * base_stream,gsize size)353 g_buffered_input_stream_new_sized (GInputStream *base_stream,
354 gsize size)
355 {
356 GInputStream *stream;
357
358 g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);
359
360 stream = g_object_new (G_TYPE_BUFFERED_INPUT_STREAM,
361 "base-stream", base_stream,
362 "buffer-size", (guint)size,
363 NULL);
364
365 return stream;
366 }
367
368 /**
369 * g_buffered_input_stream_fill:
370 * @stream: a #GBufferedInputStream
371 * @count: the number of bytes that will be read from the stream
372 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore
373 * @error: location to store the error occurring, or %NULL to ignore
374 *
375 * Tries to read @count bytes from the stream into the buffer.
376 * Will block during this read.
377 *
378 * If @count is zero, returns zero and does nothing. A value of @count
379 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
380 *
381 * On success, the number of bytes read into the buffer is returned.
382 * It is not an error if this is not the same as the requested size, as it
383 * can happen e.g. near the end of a file. Zero is returned on end of file
384 * (or if @count is zero), but never otherwise.
385 *
386 * If @count is -1 then the attempted read size is equal to the number of
387 * bytes that are required to fill the buffer.
388 *
389 * If @cancellable is not %NULL, then the operation can be cancelled by
390 * triggering the cancellable object from another thread. If the operation
391 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
392 * operation was partially finished when the operation was cancelled the
393 * partial result will be returned, without an error.
394 *
395 * On error -1 is returned and @error is set accordingly.
396 *
397 * For the asynchronous, non-blocking, version of this function, see
398 * g_buffered_input_stream_fill_async().
399 *
400 * Returns: the number of bytes read into @stream's buffer, up to @count,
401 * or -1 on error.
402 */
403 gssize
g_buffered_input_stream_fill(GBufferedInputStream * stream,gssize count,GCancellable * cancellable,GError ** error)404 g_buffered_input_stream_fill (GBufferedInputStream *stream,
405 gssize count,
406 GCancellable *cancellable,
407 GError **error)
408 {
409 GBufferedInputStreamClass *class;
410 GInputStream *input_stream;
411 gssize res;
412
413 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
414
415 input_stream = G_INPUT_STREAM (stream);
416
417 if (count < -1)
418 {
419 g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
420 _("Too large count value passed to %s"), G_STRFUNC);
421 return -1;
422 }
423
424 if (!g_input_stream_set_pending (input_stream, error))
425 return -1;
426
427 if (cancellable)
428 g_cancellable_push_current (cancellable);
429
430 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
431 res = class->fill (stream, count, cancellable, error);
432
433 if (cancellable)
434 g_cancellable_pop_current (cancellable);
435
436 g_input_stream_clear_pending (input_stream);
437
438 return res;
439 }
440
441 static void
async_fill_callback_wrapper(GObject * source_object,GAsyncResult * res,gpointer user_data)442 async_fill_callback_wrapper (GObject *source_object,
443 GAsyncResult *res,
444 gpointer user_data)
445 {
446 GBufferedInputStream *stream = G_BUFFERED_INPUT_STREAM (source_object);
447
448 g_input_stream_clear_pending (G_INPUT_STREAM (stream));
449 (*stream->priv->outstanding_callback) (source_object, res, user_data);
450 g_object_unref (stream);
451 }
452
453 /**
454 * g_buffered_input_stream_fill_async:
455 * @stream: a #GBufferedInputStream
456 * @count: the number of bytes that will be read from the stream
457 * @io_priority: the [I/O priority][io-priority] of the request
458 * @cancellable: (nullable): optional #GCancellable object
459 * @callback: (scope async): a #GAsyncReadyCallback
460 * @user_data: (closure): a #gpointer
461 *
462 * Reads data into @stream's buffer asynchronously, up to @count size.
463 * @io_priority can be used to prioritize reads. For the synchronous
464 * version of this function, see g_buffered_input_stream_fill().
465 *
466 * If @count is -1 then the attempted read size is equal to the number
467 * of bytes that are required to fill the buffer.
468 */
469 void
g_buffered_input_stream_fill_async(GBufferedInputStream * stream,gssize count,int io_priority,GCancellable * cancellable,GAsyncReadyCallback callback,gpointer user_data)470 g_buffered_input_stream_fill_async (GBufferedInputStream *stream,
471 gssize count,
472 int io_priority,
473 GCancellable *cancellable,
474 GAsyncReadyCallback callback,
475 gpointer user_data)
476 {
477 GBufferedInputStreamClass *class;
478 GError *error = NULL;
479
480 g_return_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream));
481
482 if (count == 0)
483 {
484 GTask *task;
485
486 task = g_task_new (stream, cancellable, callback, user_data);
487 g_task_set_source_tag (task, g_buffered_input_stream_fill_async);
488 g_task_return_int (task, 0);
489 g_object_unref (task);
490 return;
491 }
492
493 if (count < -1)
494 {
495 g_task_report_new_error (stream, callback, user_data,
496 g_buffered_input_stream_fill_async,
497 G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
498 _("Too large count value passed to %s"),
499 G_STRFUNC);
500 return;
501 }
502
503 if (!g_input_stream_set_pending (G_INPUT_STREAM (stream), &error))
504 {
505 g_task_report_error (stream, callback, user_data,
506 g_buffered_input_stream_fill_async,
507 error);
508 return;
509 }
510
511 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
512
513 stream->priv->outstanding_callback = callback;
514 g_object_ref (stream);
515 class->fill_async (stream, count, io_priority, cancellable,
516 async_fill_callback_wrapper, user_data);
517 }
518
519 /**
520 * g_buffered_input_stream_fill_finish:
521 * @stream: a #GBufferedInputStream
522 * @result: a #GAsyncResult
523 * @error: a #GError
524 *
525 * Finishes an asynchronous read.
526 *
527 * Returns: a #gssize of the read stream, or `-1` on an error.
528 */
529 gssize
g_buffered_input_stream_fill_finish(GBufferedInputStream * stream,GAsyncResult * result,GError ** error)530 g_buffered_input_stream_fill_finish (GBufferedInputStream *stream,
531 GAsyncResult *result,
532 GError **error)
533 {
534 GBufferedInputStreamClass *class;
535
536 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
537 g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);
538
539 if (g_async_result_legacy_propagate_error (result, error))
540 return -1;
541 else if (g_async_result_is_tagged (result, g_buffered_input_stream_fill_async))
542 return g_task_propagate_int (G_TASK (result), error);
543
544 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
545 return class->fill_finish (stream, result, error);
546 }
547
548 /**
549 * g_buffered_input_stream_get_available:
550 * @stream: #GBufferedInputStream
551 *
552 * Gets the size of the available data within the stream.
553 *
554 * Returns: size of the available stream.
555 */
556 gsize
g_buffered_input_stream_get_available(GBufferedInputStream * stream)557 g_buffered_input_stream_get_available (GBufferedInputStream *stream)
558 {
559 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
560
561 return stream->priv->end - stream->priv->pos;
562 }
563
564 /**
565 * g_buffered_input_stream_peek:
566 * @stream: a #GBufferedInputStream
567 * @buffer: (array length=count) (element-type guint8): a pointer to
568 * an allocated chunk of memory
569 * @offset: a #gsize
570 * @count: a #gsize
571 *
572 * Peeks in the buffer, copying data of size @count into @buffer,
573 * offset @offset bytes.
574 *
575 * Returns: a #gsize of the number of bytes peeked, or -1 on error.
576 */
577 gsize
g_buffered_input_stream_peek(GBufferedInputStream * stream,void * buffer,gsize offset,gsize count)578 g_buffered_input_stream_peek (GBufferedInputStream *stream,
579 void *buffer,
580 gsize offset,
581 gsize count)
582 {
583 gsize available;
584 gsize end;
585
586 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
587 g_return_val_if_fail (buffer != NULL, -1);
588
589 available = g_buffered_input_stream_get_available (stream);
590
591 if (offset > available)
592 return 0;
593
594 end = MIN (offset + count, available);
595 count = end - offset;
596
597 memcpy (buffer, stream->priv->buffer + stream->priv->pos + offset, count);
598 return count;
599 }
600
601 /**
602 * g_buffered_input_stream_peek_buffer:
603 * @stream: a #GBufferedInputStream
604 * @count: (out): a #gsize to get the number of bytes available in the buffer
605 *
606 * Returns the buffer with the currently available bytes. The returned
607 * buffer must not be modified and will become invalid when reading from
608 * the stream or filling the buffer.
609 *
610 * Returns: (array length=count) (element-type guint8) (transfer none):
611 * read-only buffer
612 */
613 const void*
g_buffered_input_stream_peek_buffer(GBufferedInputStream * stream,gsize * count)614 g_buffered_input_stream_peek_buffer (GBufferedInputStream *stream,
615 gsize *count)
616 {
617 GBufferedInputStreamPrivate *priv;
618
619 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), NULL);
620
621 priv = stream->priv;
622
623 if (count)
624 *count = priv->end - priv->pos;
625
626 return priv->buffer + priv->pos;
627 }
628
629 static void
compact_buffer(GBufferedInputStream * stream)630 compact_buffer (GBufferedInputStream *stream)
631 {
632 GBufferedInputStreamPrivate *priv;
633 gsize current_size;
634
635 priv = stream->priv;
636
637 current_size = priv->end - priv->pos;
638
639 memmove (priv->buffer, priv->buffer + priv->pos, current_size);
640
641 priv->pos = 0;
642 priv->end = current_size;
643 }
644
645 static gssize
g_buffered_input_stream_real_fill(GBufferedInputStream * stream,gssize count,GCancellable * cancellable,GError ** error)646 g_buffered_input_stream_real_fill (GBufferedInputStream *stream,
647 gssize count,
648 GCancellable *cancellable,
649 GError **error)
650 {
651 GBufferedInputStreamPrivate *priv;
652 GInputStream *base_stream;
653 gssize nread;
654 gsize in_buffer;
655
656 priv = stream->priv;
657
658 if (count == -1)
659 count = priv->len;
660
661 in_buffer = priv->end - priv->pos;
662
663 /* Never fill more than can fit in the buffer */
664 count = MIN ((gsize) count, priv->len - in_buffer);
665
666 /* If requested length does not fit at end, compact */
667 if (priv->len - priv->end < (gsize) count)
668 compact_buffer (stream);
669
670 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
671 nread = g_input_stream_read (base_stream,
672 priv->buffer + priv->end,
673 count,
674 cancellable,
675 error);
676
677 if (nread > 0)
678 priv->end += nread;
679
680 return nread;
681 }
682
683 static gssize
g_buffered_input_stream_skip(GInputStream * stream,gsize count,GCancellable * cancellable,GError ** error)684 g_buffered_input_stream_skip (GInputStream *stream,
685 gsize count,
686 GCancellable *cancellable,
687 GError **error)
688 {
689 GBufferedInputStream *bstream;
690 GBufferedInputStreamPrivate *priv;
691 GBufferedInputStreamClass *class;
692 GInputStream *base_stream;
693 gsize available, bytes_skipped;
694 gssize nread;
695
696 bstream = G_BUFFERED_INPUT_STREAM (stream);
697 priv = bstream->priv;
698
699 available = priv->end - priv->pos;
700
701 if (count <= available)
702 {
703 priv->pos += count;
704 return count;
705 }
706
707 /* Full request not available, skip all currently available and
708 * request refill for more
709 */
710
711 priv->pos = 0;
712 priv->end = 0;
713 bytes_skipped = available;
714 count -= available;
715
716 if (bytes_skipped > 0)
717 error = NULL; /* Ignore further errors if we already read some data */
718
719 if (count > priv->len)
720 {
721 /* Large request, shortcut buffer */
722
723 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
724
725 nread = g_input_stream_skip (base_stream,
726 count,
727 cancellable,
728 error);
729
730 if (nread < 0 && bytes_skipped == 0)
731 return -1;
732
733 if (nread > 0)
734 bytes_skipped += nread;
735
736 return bytes_skipped;
737 }
738
739 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
740 nread = class->fill (bstream, priv->len, cancellable, error);
741
742 if (nread < 0)
743 {
744 if (bytes_skipped == 0)
745 return -1;
746 else
747 return bytes_skipped;
748 }
749
750 available = priv->end - priv->pos;
751 count = MIN (count, available);
752
753 bytes_skipped += count;
754 priv->pos += count;
755
756 return bytes_skipped;
757 }
758
759 static gssize
g_buffered_input_stream_read(GInputStream * stream,void * buffer,gsize count,GCancellable * cancellable,GError ** error)760 g_buffered_input_stream_read (GInputStream *stream,
761 void *buffer,
762 gsize count,
763 GCancellable *cancellable,
764 GError **error)
765 {
766 GBufferedInputStream *bstream;
767 GBufferedInputStreamPrivate *priv;
768 GBufferedInputStreamClass *class;
769 GInputStream *base_stream;
770 gsize available, bytes_read;
771 gssize nread;
772
773 bstream = G_BUFFERED_INPUT_STREAM (stream);
774 priv = bstream->priv;
775
776 available = priv->end - priv->pos;
777
778 if (count <= available)
779 {
780 memcpy (buffer, priv->buffer + priv->pos, count);
781 priv->pos += count;
782 return count;
783 }
784
785 /* Full request not available, read all currently available and
786 * request refill for more
787 */
788
789 memcpy (buffer, priv->buffer + priv->pos, available);
790 priv->pos = 0;
791 priv->end = 0;
792 bytes_read = available;
793 count -= available;
794
795 if (bytes_read > 0)
796 error = NULL; /* Ignore further errors if we already read some data */
797
798 if (count > priv->len)
799 {
800 /* Large request, shortcut buffer */
801
802 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
803
804 nread = g_input_stream_read (base_stream,
805 (char *)buffer + bytes_read,
806 count,
807 cancellable,
808 error);
809
810 if (nread < 0 && bytes_read == 0)
811 return -1;
812
813 if (nread > 0)
814 bytes_read += nread;
815
816 return bytes_read;
817 }
818
819 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
820 nread = class->fill (bstream, priv->len, cancellable, error);
821 if (nread < 0)
822 {
823 if (bytes_read == 0)
824 return -1;
825 else
826 return bytes_read;
827 }
828
829 available = priv->end - priv->pos;
830 count = MIN (count, available);
831
832 memcpy ((char *)buffer + bytes_read, (char *)priv->buffer + priv->pos, count);
833 bytes_read += count;
834 priv->pos += count;
835
836 return bytes_read;
837 }
838
839 static goffset
g_buffered_input_stream_tell(GSeekable * seekable)840 g_buffered_input_stream_tell (GSeekable *seekable)
841 {
842 GBufferedInputStream *bstream;
843 GBufferedInputStreamPrivate *priv;
844 GInputStream *base_stream;
845 GSeekable *base_stream_seekable;
846 gsize available;
847 goffset base_offset;
848
849 bstream = G_BUFFERED_INPUT_STREAM (seekable);
850 priv = bstream->priv;
851
852 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
853 if (!G_IS_SEEKABLE (base_stream))
854 return 0;
855 base_stream_seekable = G_SEEKABLE (base_stream);
856
857 available = priv->end - priv->pos;
858 base_offset = g_seekable_tell (base_stream_seekable);
859
860 return base_offset - available;
861 }
862
863 static gboolean
g_buffered_input_stream_can_seek(GSeekable * seekable)864 g_buffered_input_stream_can_seek (GSeekable *seekable)
865 {
866 GInputStream *base_stream;
867
868 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
869 return G_IS_SEEKABLE (base_stream) && g_seekable_can_seek (G_SEEKABLE (base_stream));
870 }
871
872 static gboolean
g_buffered_input_stream_seek(GSeekable * seekable,goffset offset,GSeekType type,GCancellable * cancellable,GError ** error)873 g_buffered_input_stream_seek (GSeekable *seekable,
874 goffset offset,
875 GSeekType type,
876 GCancellable *cancellable,
877 GError **error)
878 {
879 GBufferedInputStream *bstream;
880 GBufferedInputStreamPrivate *priv;
881 GInputStream *base_stream;
882 GSeekable *base_stream_seekable;
883
884 bstream = G_BUFFERED_INPUT_STREAM (seekable);
885 priv = bstream->priv;
886
887 base_stream = G_FILTER_INPUT_STREAM (seekable)->base_stream;
888 if (!G_IS_SEEKABLE (base_stream))
889 {
890 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
891 _("Seek not supported on base stream"));
892 return FALSE;
893 }
894
895 base_stream_seekable = G_SEEKABLE (base_stream);
896
897 if (type == G_SEEK_CUR)
898 {
899 if (offset <= (goffset) (priv->end - priv->pos) &&
900 offset >= (goffset) -priv->pos)
901 {
902 priv->pos += offset;
903 return TRUE;
904 }
905 else
906 {
907 offset -= priv->end - priv->pos;
908 }
909 }
910
911 if (g_seekable_seek (base_stream_seekable, offset, type, cancellable, error))
912 {
913 priv->pos = 0;
914 priv->end = 0;
915 return TRUE;
916 }
917 else
918 {
919 return FALSE;
920 }
921 }
922
923 static gboolean
g_buffered_input_stream_can_truncate(GSeekable * seekable)924 g_buffered_input_stream_can_truncate (GSeekable *seekable)
925 {
926 return FALSE;
927 }
928
929 static gboolean
g_buffered_input_stream_truncate(GSeekable * seekable,goffset offset,GCancellable * cancellable,GError ** error)930 g_buffered_input_stream_truncate (GSeekable *seekable,
931 goffset offset,
932 GCancellable *cancellable,
933 GError **error)
934 {
935 g_set_error_literal (error,
936 G_IO_ERROR,
937 G_IO_ERROR_NOT_SUPPORTED,
938 _("Cannot truncate GBufferedInputStream"));
939 return FALSE;
940 }
941
942 /**
943 * g_buffered_input_stream_read_byte:
944 * @stream: a #GBufferedInputStream
945 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore
946 * @error: location to store the error occurring, or %NULL to ignore
947 *
948 * Tries to read a single byte from the stream or the buffer. Will block
949 * during this read.
950 *
951 * On success, the byte read from the stream is returned. On end of stream
952 * -1 is returned but it's not an exceptional error and @error is not set.
953 *
954 * If @cancellable is not %NULL, then the operation can be cancelled by
955 * triggering the cancellable object from another thread. If the operation
956 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
957 * operation was partially finished when the operation was cancelled the
958 * partial result will be returned, without an error.
959 *
960 * On error -1 is returned and @error is set accordingly.
961 *
962 * Returns: the byte read from the @stream, or -1 on end of stream or error.
963 */
964 int
g_buffered_input_stream_read_byte(GBufferedInputStream * stream,GCancellable * cancellable,GError ** error)965 g_buffered_input_stream_read_byte (GBufferedInputStream *stream,
966 GCancellable *cancellable,
967 GError **error)
968 {
969 GBufferedInputStreamPrivate *priv;
970 GBufferedInputStreamClass *class;
971 GInputStream *input_stream;
972 gsize available;
973 gssize nread;
974
975 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
976
977 priv = stream->priv;
978 input_stream = G_INPUT_STREAM (stream);
979
980 if (g_input_stream_is_closed (input_stream))
981 {
982 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
983 _("Stream is already closed"));
984 return -1;
985 }
986
987 if (!g_input_stream_set_pending (input_stream, error))
988 return -1;
989
990 available = priv->end - priv->pos;
991
992 if (available != 0)
993 {
994 g_input_stream_clear_pending (input_stream);
995 return priv->buffer[priv->pos++];
996 }
997
998 /* Byte not available, request refill for more */
999
1000 if (cancellable)
1001 g_cancellable_push_current (cancellable);
1002
1003 priv->pos = 0;
1004 priv->end = 0;
1005
1006 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1007 nread = class->fill (stream, priv->len, cancellable, error);
1008
1009 if (cancellable)
1010 g_cancellable_pop_current (cancellable);
1011
1012 g_input_stream_clear_pending (input_stream);
1013
1014 if (nread <= 0)
1015 return -1; /* error or end of stream */
1016
1017 return priv->buffer[priv->pos++];
1018 }
1019
1020 /* ************************** */
1021 /* Async stuff implementation */
1022 /* ************************** */
1023
1024 static void
fill_async_callback(GObject * source_object,GAsyncResult * result,gpointer user_data)1025 fill_async_callback (GObject *source_object,
1026 GAsyncResult *result,
1027 gpointer user_data)
1028 {
1029 GError *error;
1030 gssize res;
1031 GTask *task = user_data;
1032
1033 error = NULL;
1034 res = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
1035 result, &error);
1036 if (res == -1)
1037 g_task_return_error (task, error);
1038 else
1039 {
1040 GBufferedInputStream *stream;
1041 GBufferedInputStreamPrivate *priv;
1042
1043 stream = g_task_get_source_object (task);
1044 priv = G_BUFFERED_INPUT_STREAM (stream)->priv;
1045
1046 g_assert_cmpint (priv->end + res, <=, priv->len);
1047 priv->end += res;
1048
1049 g_task_return_int (task, res);
1050 }
1051
1052 g_object_unref (task);
1053 }
1054
1055 static void
g_buffered_input_stream_real_fill_async(GBufferedInputStream * stream,gssize count,int io_priority,GCancellable * cancellable,GAsyncReadyCallback callback,gpointer user_data)1056 g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
1057 gssize count,
1058 int io_priority,
1059 GCancellable *cancellable,
1060 GAsyncReadyCallback callback,
1061 gpointer user_data)
1062 {
1063 GBufferedInputStreamPrivate *priv;
1064 GInputStream *base_stream;
1065 GTask *task;
1066 gsize in_buffer;
1067
1068 priv = stream->priv;
1069
1070 if (count == -1)
1071 count = priv->len;
1072
1073 in_buffer = priv->end - priv->pos;
1074
1075 /* Never fill more than can fit in the buffer */
1076 count = MIN ((gsize) count, priv->len - in_buffer);
1077
1078 /* If requested length does not fit at end, compact */
1079 if (priv->len - priv->end < (gsize) count)
1080 compact_buffer (stream);
1081
1082 task = g_task_new (stream, cancellable, callback, user_data);
1083 g_task_set_source_tag (task, g_buffered_input_stream_real_fill_async);
1084
1085 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1086 g_input_stream_read_async (base_stream,
1087 priv->buffer + priv->end,
1088 count,
1089 io_priority,
1090 cancellable,
1091 fill_async_callback,
1092 task);
1093 }
1094
1095 static gssize
g_buffered_input_stream_real_fill_finish(GBufferedInputStream * stream,GAsyncResult * result,GError ** error)1096 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
1097 GAsyncResult *result,
1098 GError **error)
1099 {
1100 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1101
1102 return g_task_propagate_int (G_TASK (result), error);
1103 }
1104
1105 typedef struct
1106 {
1107 gsize bytes_skipped;
1108 gsize count;
1109 } SkipAsyncData;
1110
1111 static void
free_skip_async_data(gpointer _data)1112 free_skip_async_data (gpointer _data)
1113 {
1114 SkipAsyncData *data = _data;
1115 g_slice_free (SkipAsyncData, data);
1116 }
1117
1118 static void
large_skip_callback(GObject * source_object,GAsyncResult * result,gpointer user_data)1119 large_skip_callback (GObject *source_object,
1120 GAsyncResult *result,
1121 gpointer user_data)
1122 {
1123 GTask *task = G_TASK (user_data);
1124 SkipAsyncData *data;
1125 GError *error;
1126 gssize nread;
1127
1128 data = g_task_get_task_data (task);
1129
1130 error = NULL;
1131 nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1132 result, &error);
1133
1134 /* Only report the error if we've not already read some data */
1135 if (nread < 0 && data->bytes_skipped == 0)
1136 g_task_return_error (task, error);
1137 else
1138 {
1139 if (error)
1140 g_error_free (error);
1141
1142 if (nread > 0)
1143 data->bytes_skipped += nread;
1144
1145 g_task_return_int (task, data->bytes_skipped);
1146 }
1147
1148 g_object_unref (task);
1149 }
1150
1151 static void
skip_fill_buffer_callback(GObject * source_object,GAsyncResult * result,gpointer user_data)1152 skip_fill_buffer_callback (GObject *source_object,
1153 GAsyncResult *result,
1154 gpointer user_data)
1155 {
1156 GTask *task = G_TASK (user_data);
1157 GBufferedInputStream *bstream;
1158 GBufferedInputStreamPrivate *priv;
1159 SkipAsyncData *data;
1160 GError *error;
1161 gssize nread;
1162 gsize available;
1163
1164 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1165 priv = bstream->priv;
1166
1167 data = g_task_get_task_data (task);
1168
1169 error = NULL;
1170 nread = g_buffered_input_stream_fill_finish (bstream,
1171 result, &error);
1172
1173 if (nread < 0 && data->bytes_skipped == 0)
1174 g_task_return_error (task, error);
1175 else
1176 {
1177 if (error)
1178 g_error_free (error);
1179
1180 if (nread > 0)
1181 {
1182 available = priv->end - priv->pos;
1183 data->count = MIN (data->count, available);
1184
1185 data->bytes_skipped += data->count;
1186 priv->pos += data->count;
1187 }
1188
1189 g_assert (data->bytes_skipped <= G_MAXSSIZE);
1190 g_task_return_int (task, data->bytes_skipped);
1191 }
1192
1193 g_object_unref (task);
1194 }
1195
1196 static void
g_buffered_input_stream_skip_async(GInputStream * stream,gsize count,int io_priority,GCancellable * cancellable,GAsyncReadyCallback callback,gpointer user_data)1197 g_buffered_input_stream_skip_async (GInputStream *stream,
1198 gsize count,
1199 int io_priority,
1200 GCancellable *cancellable,
1201 GAsyncReadyCallback callback,
1202 gpointer user_data)
1203 {
1204 GBufferedInputStream *bstream;
1205 GBufferedInputStreamPrivate *priv;
1206 GBufferedInputStreamClass *class;
1207 GInputStream *base_stream;
1208 gsize available;
1209 GTask *task;
1210 SkipAsyncData *data;
1211
1212 bstream = G_BUFFERED_INPUT_STREAM (stream);
1213 priv = bstream->priv;
1214
1215 data = g_slice_new (SkipAsyncData);
1216 data->bytes_skipped = 0;
1217 task = g_task_new (stream, cancellable, callback, user_data);
1218 g_task_set_source_tag (task, g_buffered_input_stream_skip_async);
1219 g_task_set_task_data (task, data, free_skip_async_data);
1220
1221 available = priv->end - priv->pos;
1222
1223 if (count <= available)
1224 {
1225 priv->pos += count;
1226
1227 g_task_return_int (task, count);
1228 g_object_unref (task);
1229 return;
1230 }
1231
1232 /* Full request not available, skip all currently available
1233 * and request refill for more
1234 */
1235
1236 priv->pos = 0;
1237 priv->end = 0;
1238
1239 count -= available;
1240
1241 data->bytes_skipped = available;
1242 data->count = count;
1243
1244 if (count > priv->len)
1245 {
1246 /* Large request, shortcut buffer */
1247 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1248
1249 /* If 'count > G_MAXSSIZE then 'g_input_stream_skip_async()'
1250 * will return an error anyway before calling this.
1251 * Assert that this is never called for too big `count` for clarity. */
1252 g_assert ((gssize) count >= 0);
1253 g_input_stream_skip_async (base_stream,
1254 count,
1255 io_priority, cancellable,
1256 large_skip_callback,
1257 task);
1258 }
1259 else
1260 {
1261 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1262 class->fill_async (bstream, priv->len, io_priority, cancellable,
1263 skip_fill_buffer_callback, task);
1264 }
1265 }
1266
1267 static gssize
g_buffered_input_stream_skip_finish(GInputStream * stream,GAsyncResult * result,GError ** error)1268 g_buffered_input_stream_skip_finish (GInputStream *stream,
1269 GAsyncResult *result,
1270 GError **error)
1271 {
1272 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1273
1274 return g_task_propagate_int (G_TASK (result), error);
1275 }
1276