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 | |
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 (oclass: object_class, |
163 | property_id: PROP_BUFSIZE, |
164 | pspec: g_param_spec_uint (name: "buffer-size" , |
165 | P_("Buffer Size" ), |
166 | P_("The size of the backend buffer" ), |
167 | minimum: 1, |
168 | G_MAXUINT, |
169 | DEFAULT_BUFFER_SIZE, |
170 | flags: 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 |
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 |
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 (n_bytes: size); |
224 | memcpy (dest: buffer, src: priv->buffer + priv->pos, n: in_buffer); |
225 | priv->len = size; |
226 | priv->pos = 0; |
227 | priv->end = in_buffer; |
228 | g_free (mem: 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 (n_bytes: size); |
237 | } |
238 | |
239 | g_object_notify (G_OBJECT (stream), property_name: "buffer-size" ); |
240 | } |
241 | |
242 | static void |
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 (stream: bstream, size: 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 |
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, v_uint: 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 |
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 (mem: priv->buffer); |
298 | |
299 | G_OBJECT_CLASS (g_buffered_input_stream_parent_class)->finalize (object); |
300 | } |
301 | |
302 | static void |
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 |
313 | g_buffered_input_stream_init (GBufferedInputStream *stream) |
314 | { |
315 | stream->priv = g_buffered_input_stream_get_instance_private (self: 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 * |
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 | first_property_name: "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 * |
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 | first_property_name: "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 |
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 (err: error, G_IO_ERROR, code: 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 (stream: 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 (stream: input_stream); |
437 | |
438 | return res; |
439 | } |
440 | |
441 | static void |
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 (object: 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 |
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 (source_object: stream, cancellable, callback, callback_data: user_data); |
487 | g_task_set_source_tag (task, g_buffered_input_stream_fill_async); |
488 | g_task_return_int (task, result: 0); |
489 | g_object_unref (object: task); |
490 | return; |
491 | } |
492 | |
493 | if (count < -1) |
494 | { |
495 | g_task_report_new_error (source_object: stream, callback, callback_data: user_data, |
496 | source_tag: g_buffered_input_stream_fill_async, |
497 | G_IO_ERROR, code: 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: &error)) |
504 | { |
505 | g_task_report_error (source_object: stream, callback, callback_data: user_data, |
506 | source_tag: 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 |
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 (res: result, error)) |
540 | return -1; |
541 | else if (g_async_result_is_tagged (res: result, source_tag: 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 |
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 |
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 (dest: buffer, src: stream->priv->buffer + stream->priv->pos + offset, n: 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* |
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 |
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 (dest: priv->buffer, src: priv->buffer + priv->pos, n: current_size); |
640 | |
641 | priv->pos = 0; |
642 | priv->end = current_size; |
643 | } |
644 | |
645 | static gssize |
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 (stream: base_stream, |
672 | buffer: 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 |
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 (stream: 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 |
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 (dest: buffer, src: priv->buffer + priv->pos, n: 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 (dest: buffer, src: priv->buffer + priv->pos, n: 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 (stream: base_stream, |
805 | buffer: (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 (dest: (char *)buffer + bytes_read, src: (char *)priv->buffer + priv->pos, n: count); |
833 | bytes_read += count; |
834 | priv->pos += count; |
835 | |
836 | return bytes_read; |
837 | } |
838 | |
839 | static goffset |
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 (seekable: base_stream_seekable); |
859 | |
860 | return base_offset - available; |
861 | } |
862 | |
863 | static gboolean |
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 |
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 (err: error, G_IO_ERROR, code: 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 (seekable: 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 |
924 | g_buffered_input_stream_can_truncate (GSeekable *seekable) |
925 | { |
926 | return FALSE; |
927 | } |
928 | |
929 | static gboolean |
930 | g_buffered_input_stream_truncate (GSeekable *seekable, |
931 | goffset offset, |
932 | GCancellable *cancellable, |
933 | GError **error) |
934 | { |
935 | g_set_error_literal (err: error, |
936 | G_IO_ERROR, |
937 | code: 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 |
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 (stream: input_stream)) |
981 | { |
982 | g_set_error_literal (err: error, G_IO_ERROR, code: G_IO_ERROR_CLOSED, |
983 | _("Stream is already closed" )); |
984 | return -1; |
985 | } |
986 | |
987 | if (!g_input_stream_set_pending (stream: 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 (stream: 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 (stream: 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 |
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: &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, result: res); |
1050 | } |
1051 | |
1052 | g_object_unref (object: task); |
1053 | } |
1054 | |
1055 | static void |
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 (source_object: stream, cancellable, callback, callback_data: 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 (stream: base_stream, |
1087 | buffer: priv->buffer + priv->end, |
1088 | count, |
1089 | io_priority, |
1090 | cancellable, |
1091 | callback: fill_async_callback, |
1092 | user_data: task); |
1093 | } |
1094 | |
1095 | static gssize |
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 |
1112 | free_skip_async_data (gpointer _data) |
1113 | { |
1114 | SkipAsyncData *data = _data; |
1115 | g_slice_free (SkipAsyncData, data); |
1116 | } |
1117 | |
1118 | static void |
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: &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, result: data->bytes_skipped); |
1146 | } |
1147 | |
1148 | g_object_unref (object: task); |
1149 | } |
1150 | |
1151 | static void |
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 (stream: bstream, |
1171 | result, error: &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, result: data->bytes_skipped); |
1191 | } |
1192 | |
1193 | g_object_unref (object: task); |
1194 | } |
1195 | |
1196 | static void |
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 (source_object: stream, cancellable, callback, callback_data: user_data); |
1218 | g_task_set_source_tag (task, g_buffered_input_stream_skip_async); |
1219 | g_task_set_task_data (task, task_data: data, task_data_destroy: 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, result: count); |
1228 | g_object_unref (object: 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 (stream: base_stream, |
1254 | count, |
1255 | io_priority, cancellable, |
1256 | callback: large_skip_callback, |
1257 | user_data: 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 |
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 | |