nexmon – Blame information for rev 1

Subversion Repositories:
Rev:
Rev Author Line No. Line
1 office 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 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 (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
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 (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
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
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
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
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 (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 "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 "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: (allow-none): 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 (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
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: (allow-none): 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 (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
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
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 (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*
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 (priv->buffer, priv->buffer + priv->pos, 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 (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
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
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
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
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 (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
923 g_buffered_input_stream_can_truncate (GSeekable *seekable)
924 {
925 return FALSE;
926 }
927  
928 static gboolean
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: (allow-none): 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
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
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
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  
1083 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1084 g_input_stream_read_async (base_stream,
1085 priv->buffer + priv->end,
1086 count,
1087 io_priority,
1088 cancellable,
1089 fill_async_callback,
1090 task);
1091 }
1092  
1093 static gssize
1094 g_buffered_input_stream_real_fill_finish (GBufferedInputStream *stream,
1095 GAsyncResult *result,
1096 GError **error)
1097 {
1098 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1099  
1100 return g_task_propagate_int (G_TASK (result), error);
1101 }
1102  
1103 typedef struct
1104 {
1105 gssize bytes_skipped;
1106 gssize count;
1107 } SkipAsyncData;
1108  
1109 static void
1110 free_skip_async_data (gpointer _data)
1111 {
1112 SkipAsyncData *data = _data;
1113 g_slice_free (SkipAsyncData, data);
1114 }
1115  
1116 static void
1117 large_skip_callback (GObject *source_object,
1118 GAsyncResult *result,
1119 gpointer user_data)
1120 {
1121 GTask *task = G_TASK (user_data);
1122 SkipAsyncData *data;
1123 GError *error;
1124 gssize nread;
1125  
1126 data = g_task_get_task_data (task);
1127  
1128 error = NULL;
1129 nread = g_input_stream_skip_finish (G_INPUT_STREAM (source_object),
1130 result, &error);
1131  
1132 /* Only report the error if we've not already read some data */
1133 if (nread < 0 && data->bytes_skipped == 0)
1134 g_task_return_error (task, error);
1135 else
1136 {
1137 if (error)
1138 g_error_free (error);
1139  
1140 if (nread > 0)
1141 data->bytes_skipped += nread;
1142  
1143 g_task_return_int (task, data->bytes_skipped);
1144 }
1145  
1146 g_object_unref (task);
1147 }
1148  
1149 static void
1150 skip_fill_buffer_callback (GObject *source_object,
1151 GAsyncResult *result,
1152 gpointer user_data)
1153 {
1154 GTask *task = G_TASK (user_data);
1155 GBufferedInputStream *bstream;
1156 GBufferedInputStreamPrivate *priv;
1157 SkipAsyncData *data;
1158 GError *error;
1159 gssize nread;
1160 gsize available;
1161  
1162 bstream = G_BUFFERED_INPUT_STREAM (source_object);
1163 priv = bstream->priv;
1164  
1165 data = g_task_get_task_data (task);
1166  
1167 error = NULL;
1168 nread = g_buffered_input_stream_fill_finish (bstream,
1169 result, &error);
1170  
1171 if (nread < 0 && data->bytes_skipped == 0)
1172 g_task_return_error (task, error);
1173 else
1174 {
1175 if (error)
1176 g_error_free (error);
1177  
1178 if (nread > 0)
1179 {
1180 available = priv->end - priv->pos;
1181 data->count = MIN (data->count, available);
1182  
1183 data->bytes_skipped += data->count;
1184 priv->pos += data->count;
1185 }
1186  
1187 g_task_return_int (task, data->bytes_skipped);
1188 }
1189  
1190 g_object_unref (task);
1191 }
1192  
1193 static void
1194 g_buffered_input_stream_skip_async (GInputStream *stream,
1195 gsize count,
1196 int io_priority,
1197 GCancellable *cancellable,
1198 GAsyncReadyCallback callback,
1199 gpointer user_data)
1200 {
1201 GBufferedInputStream *bstream;
1202 GBufferedInputStreamPrivate *priv;
1203 GBufferedInputStreamClass *class;
1204 GInputStream *base_stream;
1205 gsize available;
1206 GTask *task;
1207 SkipAsyncData *data;
1208  
1209 bstream = G_BUFFERED_INPUT_STREAM (stream);
1210 priv = bstream->priv;
1211  
1212 data = g_slice_new (SkipAsyncData);
1213 data->bytes_skipped = 0;
1214 task = g_task_new (stream, cancellable, callback, user_data);
1215 g_task_set_task_data (task, data, free_skip_async_data);
1216  
1217 available = priv->end - priv->pos;
1218  
1219 if (count <= available)
1220 {
1221 priv->pos += count;
1222  
1223 g_task_return_int (task, count);
1224 g_object_unref (task);
1225 return;
1226 }
1227  
1228 /* Full request not available, skip all currently available
1229 * and request refill for more
1230 */
1231  
1232 priv->pos = 0;
1233 priv->end = 0;
1234  
1235 count -= available;
1236  
1237 data->bytes_skipped = available;
1238 data->count = count;
1239  
1240 if (count > priv->len)
1241 {
1242 /* Large request, shortcut buffer */
1243  
1244 base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
1245  
1246 g_input_stream_skip_async (base_stream,
1247 count,
1248 io_priority, cancellable,
1249 large_skip_callback,
1250 task);
1251 }
1252 else
1253 {
1254 class = G_BUFFERED_INPUT_STREAM_GET_CLASS (stream);
1255 class->fill_async (bstream, priv->len, io_priority, cancellable,
1256 skip_fill_buffer_callback, task);
1257 }
1258 }
1259  
1260 static gssize
1261 g_buffered_input_stream_skip_finish (GInputStream *stream,
1262 GAsyncResult *result,
1263 GError **error)
1264 {
1265 g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1266  
1267 return g_task_propagate_int (G_TASK (result), error);
1268 }