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 (count, priv->len - in_buffer);
665
666 /* If requested length does not fit at end, compact */
667 if (priv->len - priv->end < 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 <= priv->end - priv->pos && offset >= -priv->pos)
900 {
901 priv->pos += offset;
902 return TRUE;
903 }
904 else
905 {
906 offset -= priv->end - priv->pos;
907 }
908 }
909
910 if (g_seekable_seek (base_stream_seekable, offset, type, cancellable, error))
911 {
912 priv->pos = 0;
913 priv->end = 0;
914 return TRUE;
915 }
916 else
917 {
918 return FALSE;
919 }
920 }
921
922 static gboolean
g_buffered_input_stream_can_truncate(GSeekable * seekable)923 g_buffered_input_stream_can_truncate (GSeekable *seekable)
924 {
925 return FALSE;
926 }
927
928 static gboolean
g_buffered_input_stream_truncate(GSeekable * seekable,goffset offset,GCancellable * cancellable,GError ** error)929 g_buffered_input_stream_truncate (GSeekable *seekable,
930 goffset offset,
931 GCancellable *cancellable,
932 GError **error)
933 {
934 g_set_error_literal (error,
935 G_IO_ERROR,
936 G_IO_ERROR_NOT_SUPPORTED,
937 _("Cannot truncate GBufferedInputStream"));
938 return FALSE;
939 }
940
941 /**
942 * g_buffered_input_stream_read_byte:
943 * @stream: a #GBufferedInputStream
944 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore
945 * @error: location to store the error occurring, or %NULL to ignore
946 *
947 * Tries to read a single byte from the stream or the buffer. Will block
948 * during this read.
949 *
950 * On success, the byte read from the stream is returned. On end of stream
951 * -1 is returned but it's not an exceptional error and @error is not set.
952 *
953 * If @cancellable is not %NULL, then the operation can be cancelled by
954 * triggering the cancellable object from another thread. If the operation
955 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
956 * operation was partially finished when the operation was cancelled the
957 * partial result will be returned, without an error.
958 *
959 * On error -1 is returned and @error is set accordingly.
960 *
961 * Returns: the byte read from the @stream, or -1 on end of stream or error.
962 */
963 int
g_buffered_input_stream_read_byte(GBufferedInputStream * stream,GCancellable * cancellable,GError ** error)964 g_buffered_input_stream_read_byte (GBufferedInputStream *stream,
965 GCancellable *cancellable,
966 GError **error)
967 {
968 GBufferedInputStreamPrivate *priv;
969 GBufferedInputStreamClass *class;
970 GInputStream *input_stream;
971 gsize available;
972 gssize nread;
973
974 g_return_val_if_fail (G_IS_BUFFERED_INPUT_STREAM (stream), -1);
975
976 priv = stream->priv;
977 input_stream = G_INPUT_STREAM (stream);
978
979 if (g_input_stream_is_closed (input_stream))
980 {
981 g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
982 _("Stream is already closed"));
983 return -1;
984 }
985
986 if (!g_input_stream_set_pending (input_stream, error))
987 return -1;
988
989 available = priv->end - priv->pos;
990
991 if (available != 0)
992 {
993 g_input_stream_clear_pending (input_stream);
994 return priv->buffer[priv->pos++];
995 }
996
997 /* Byte not available, request refill for more */
998
999 if (cancellable)
1000 g_cancellable_push_current (cancellable);
1001
1002 priv->pos = 0;
1003 priv->end = 0;
1004
1005 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1006 nread = class->fill (stream, priv->len, cancellable, error);
1007
1008 if (cancellable)
1009 g_cancellable_pop_current (cancellable);
1010
1011 g_input_stream_clear_pending (input_stream);
1012
1013 if (nread <= 0)
1014 return -1; /* error or end of stream */
1015
1016 return priv->buffer[priv->pos++];
1017 }
1018
1019 /* ************************** */
1020 /* Async stuff implementation */
1021 /* ************************** */
1022
1023 static void
fill_async_callback(GObject * source_object,GAsyncResult * result,gpointer user_data)1024 fill_async_callback (GObject *source_object,
1025 GAsyncResult *result,
1026 gpointer user_data)
1027 {
1028 GError *error;
1029 gssize res;
1030 GTask *task = user_data;
1031
1032 error = NULL;
1033 res = g_input_stream_read_finish (G_INPUT_STREAM (source_object),
1034 result, &error);
1035 if (res == -1)
1036 g_task_return_error (task, error);
1037 else
1038 {
1039 GBufferedInputStream *stream;
1040 GBufferedInputStreamPrivate *priv;
1041
1042 stream = g_task_get_source_object (task);
1043 priv = G_BUFFERED_INPUT_STREAM (stream)->priv;
1044
1045 g_assert_cmpint (priv->end + res, <=, priv->len);
1046 priv->end += res;
1047
1048 g_task_return_int (task, res);
1049 }
1050
1051 g_object_unref (task);
1052 }
1053
1054 static void
g_buffered_input_stream_real_fill_async(GBufferedInputStream * stream,gssize count,int io_priority,GCancellable * cancellable,GAsyncReadyCallback callback,gpointer user_data)1055 g_buffered_input_stream_real_fill_async (GBufferedInputStream *stream,
1056 gssize count,
1057 int io_priority,
1058 GCancellable *cancellable,
1059 GAsyncReadyCallback callback,
1060 gpointer user_data)
1061 {
1062 GBufferedInputStreamPrivate *priv;
1063 GInputStream *base_stream;
1064 GTask *task;
1065 gsize in_buffer;
1066
1067 priv = stream->priv;
1068
1069 if (count == -1)
1070 count = priv->len;
1071
1072 in_buffer = priv->end - priv->pos;
1073
1074 /* Never fill more than can fit in the buffer */
1075 count = MIN (count, priv->len - in_buffer);
1076
1077 /* If requested length does not fit at end, compact */
1078 if (priv->len - priv->end < count)
1079 compact_buffer (stream);
1080
1081 task = g_task_new (stream, cancellable, callback, user_data);
1082 g_task_set_source_tag (task, g_buffered_input_stream_real_fill_async);
1083
1084 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1085 g_input_stream_read_async (base_stream,
1086 priv->buffer + priv->end,
1087 count,
1088 io_priority,
1089 cancellable,
1090 fill_async_callback,
1091 task);
1092 }
1093
1094 static gssize
g_buffered_input_stream_real_fill_finish(GBufferedInputStream * stream,GAsyncResult * result,GError ** error)1095 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
1096 GAsyncResult *result,
1097 GError **error)
1098 {
1099 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1100
1101 return g_task_propagate_int (G_TASK (result), error);
1102 }
1103
1104 typedef struct
1105 {
1106 gssize bytes_skipped;
1107 gssize count;
1108 } SkipAsyncData;
1109
1110 static void
free_skip_async_data(gpointer _data)1111 free_skip_async_data (gpointer _data)
1112 {
1113 SkipAsyncData *data = _data;
1114 g_slice_free (SkipAsyncData, data);
1115 }
1116
1117 static void
large_skip_callback(GObject * source_object,GAsyncResult * result,gpointer user_data)1118 large_skip_callback (GObject *source_object,
1119 GAsyncResult *result,
1120 gpointer user_data)
1121 {
1122 GTask *task = G_TASK (user_data);
1123 SkipAsyncData *data;
1124 GError *error;
1125 gssize nread;
1126
1127 data = g_task_get_task_data (task);
1128
1129 error = NULL;
1130 nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1131 result, &error);
1132
1133 /* Only report the error if we've not already read some data */
1134 if (nread < 0 && data->bytes_skipped == 0)
1135 g_task_return_error (task, error);
1136 else
1137 {
1138 if (error)
1139 g_error_free (error);
1140
1141 if (nread > 0)
1142 data->bytes_skipped += nread;
1143
1144 g_task_return_int (task, data->bytes_skipped);
1145 }
1146
1147 g_object_unref (task);
1148 }
1149
1150 static void
skip_fill_buffer_callback(GObject * source_object,GAsyncResult * result,gpointer user_data)1151 skip_fill_buffer_callback (GObject *source_object,
1152 GAsyncResult *result,
1153 gpointer user_data)
1154 {
1155 GTask *task = G_TASK (user_data);
1156 GBufferedInputStream *bstream;
1157 GBufferedInputStreamPrivate *priv;
1158 SkipAsyncData *data;
1159 GError *error;
1160 gssize nread;
1161 gsize available;
1162
1163 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1164 priv = bstream->priv;
1165
1166 data = g_task_get_task_data (task);
1167
1168 error = NULL;
1169 nread = g_buffered_input_stream_fill_finish (bstream,
1170 result, &error);
1171
1172 if (nread < 0 && data->bytes_skipped == 0)
1173 g_task_return_error (task, error);
1174 else
1175 {
1176 if (error)
1177 g_error_free (error);
1178
1179 if (nread > 0)
1180 {
1181 available = priv->end - priv->pos;
1182 data->count = MIN (data->count, available);
1183
1184 data->bytes_skipped += data->count;
1185 priv->pos += data->count;
1186 }
1187
1188 g_task_return_int (task, data->bytes_skipped);
1189 }
1190
1191 g_object_unref (task);
1192 }
1193
1194 static void
g_buffered_input_stream_skip_async(GInputStream * stream,gsize count,int io_priority,GCancellable * cancellable,GAsyncReadyCallback callback,gpointer user_data)1195 g_buffered_input_stream_skip_async (GInputStream *stream,
1196 gsize count,
1197 int io_priority,
1198 GCancellable *cancellable,
1199 GAsyncReadyCallback callback,
1200 gpointer user_data)
1201 {
1202 GBufferedInputStream *bstream;
1203 GBufferedInputStreamPrivate *priv;
1204 GBufferedInputStreamClass *class;
1205 GInputStream *base_stream;
1206 gsize available;
1207 GTask *task;
1208 SkipAsyncData *data;
1209
1210 bstream = G_BUFFERED_INPUT_STREAM (stream);
1211 priv = bstream->priv;
1212
1213 data = g_slice_new (SkipAsyncData);
1214 data->bytes_skipped = 0;
1215 task = g_task_new (stream, cancellable, callback, user_data);
1216 g_task_set_source_tag (task, g_buffered_input_stream_skip_async);
1217 g_task_set_task_data (task, data, free_skip_async_data);
1218
1219 available = priv->end - priv->pos;
1220
1221 if (count <= available)
1222 {
1223 priv->pos += count;
1224
1225 g_task_return_int (task, count);
1226 g_object_unref (task);
1227 return;
1228 }
1229
1230 /* Full request not available, skip all currently available
1231 * and request refill for more
1232 */
1233
1234 priv->pos = 0;
1235 priv->end = 0;
1236
1237 count -= available;
1238
1239 data->bytes_skipped = available;
1240 data->count = count;
1241
1242 if (count > priv->len)
1243 {
1244 /* Large request, shortcut buffer */
1245
1246 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1247
1248 g_input_stream_skip_async (base_stream,
1249 count,
1250 io_priority, cancellable,
1251 large_skip_callback,
1252 task);
1253 }
1254 else
1255 {
1256 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1257 class->fill_async (bstream, priv->len, io_priority, cancellable,
1258 skip_fill_buffer_callback, task);
1259 }
1260 }
1261
1262 static gssize
g_buffered_input_stream_skip_finish(GInputStream * stream,GAsyncResult * result,GError ** error)1263 g_buffered_input_stream_skip_finish (GInputStream *stream,
1264 GAsyncResult *result,
1265 GError **error)
1266 {
1267 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1268
1269 return g_task_propagate_int (G_TASK (result), error);
1270 }
1271