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