ginputstream.c 47.8 KB
Newer Older
1 2 3 4 5 6 7
/* GIO - GLib Input, Output and Streaming Library
 * 
 * Copyright (C) 2006-2007 Red Hat, Inc.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
8
 * version 2.1 of the License, or (at your option) any later version.
9 10 11 12 13 14 15
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General
16
 * Public License along with this library; if not, see <http://www.gnu.org/licenses/>.
17 18 19 20
 *
 * Author: Alexander Larsson <alexl@redhat.com>
 */

21
#include "config.h"
22 23 24 25
#include <glib.h>
#include "glibintl.h"

#include "ginputstream.h"
26
#include "gioprivate.h"
27
#include "gseekable.h"
28 29 30
#include "gcancellable.h"
#include "gasyncresult.h"
#include "gioerror.h"
31
#include "gpollableinputstream.h"
32

33 34
/**
 * SECTION:ginputstream
Matthias Clasen's avatar
Matthias Clasen committed
35
 * @short_description: Base class for implementing streaming input
Matthias Clasen's avatar
Matthias Clasen committed
36
 * @include: gio/gio.h
37
 *
Dan Winship's avatar
Dan Winship committed
38
 * #GInputStream has functions to read from a stream (g_input_stream_read()),
Matthias Clasen's avatar
Matthias Clasen committed
39 40 41 42
 * to close a stream (g_input_stream_close()) and to skip some content
 * (g_input_stream_skip()). 
 *
 * To copy the content of an input stream to an output stream without 
43 44 45 46
 * manually handling the reads and writes, use g_output_stream_splice().
 *
 * See the documentation for #GIOStream for details of thread safety of
 * streaming APIs.
Matthias Clasen's avatar
Matthias Clasen committed
47 48
 *
 * All of these functions have async variants too.
49 50
 **/

51 52 53 54 55 56
struct _GInputStreamPrivate {
  guint closed : 1;
  guint pending : 1;
  GAsyncReadyCallback outstanding_callback;
};

57 58
G_DEFINE_ABSTRACT_TYPE_WITH_PRIVATE (GInputStream, g_input_stream, G_TYPE_OBJECT)

59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99
static gssize   g_input_stream_real_skip         (GInputStream         *stream,
						  gsize                 count,
						  GCancellable         *cancellable,
						  GError              **error);
static void     g_input_stream_real_read_async   (GInputStream         *stream,
						  void                 *buffer,
						  gsize                 count,
						  int                   io_priority,
						  GCancellable         *cancellable,
						  GAsyncReadyCallback   callback,
						  gpointer              user_data);
static gssize   g_input_stream_real_read_finish  (GInputStream         *stream,
						  GAsyncResult         *result,
						  GError              **error);
static void     g_input_stream_real_skip_async   (GInputStream         *stream,
						  gsize                 count,
						  int                   io_priority,
						  GCancellable         *cancellable,
						  GAsyncReadyCallback   callback,
						  gpointer              data);
static gssize   g_input_stream_real_skip_finish  (GInputStream         *stream,
						  GAsyncResult         *result,
						  GError              **error);
static void     g_input_stream_real_close_async  (GInputStream         *stream,
						  int                   io_priority,
						  GCancellable         *cancellable,
						  GAsyncReadyCallback   callback,
						  gpointer              data);
static gboolean g_input_stream_real_close_finish (GInputStream         *stream,
						  GAsyncResult         *result,
						  GError              **error);

static void
g_input_stream_dispose (GObject *object)
{
  GInputStream *stream;

  stream = G_INPUT_STREAM (object);
  
  if (!stream->priv->closed)
    g_input_stream_close (stream, NULL, NULL);
100 101

  G_OBJECT_CLASS (g_input_stream_parent_class)->dispose (object);
102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123
}


static void
g_input_stream_class_init (GInputStreamClass *klass)
{
  GObjectClass *gobject_class = G_OBJECT_CLASS (klass);
  
  gobject_class->dispose = g_input_stream_dispose;
  
  klass->skip = g_input_stream_real_skip;
  klass->read_async = g_input_stream_real_read_async;
  klass->read_finish = g_input_stream_real_read_finish;
  klass->skip_async = g_input_stream_real_skip_async;
  klass->skip_finish = g_input_stream_real_skip_finish;
  klass->close_async = g_input_stream_real_close_async;
  klass->close_finish = g_input_stream_real_close_finish;
}

static void
g_input_stream_init (GInputStream *stream)
{
124
  stream->priv = g_input_stream_get_instance_private (stream);
125 126 127 128 129
}

/**
 * g_input_stream_read:
 * @stream: a #GInputStream.
130 131
 * @buffer: (array length=count) (element-type guint8): a buffer to
 *     read data into (which should be at least count bytes long).
132
 * @count: the number of bytes that will be read from the stream
133
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.
Matthias Clasen's avatar
Matthias Clasen committed
134
 * @error: location to store the error occurring, or %NULL to ignore
135 136 137 138 139 140 141 142 143 144 145 146
 *
 * Tries to read @count bytes from the stream into the buffer starting at
 * @buffer. Will block during this read.
 * 
 * If count is zero returns zero and does nothing. A value of @count
 * larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 *
 * On success, the number of bytes read into the buffer is returned.
 * It is not an error if this is not the same as the requested size, as it
 * can happen e.g. near the end of a file. Zero is returned on end of file
 * (or if @count is zero),  but never otherwise.
 *
147 148 149
 * The returned @buffer is not a nul-terminated string, it can contain nul bytes
 * at any position, and this function doesn't nul-terminate the @buffer.
 *
Dan Winship's avatar
Dan Winship committed
150
 * If @cancellable is not %NULL, then the operation can be cancelled by
151
 * triggering the cancellable object from another thread. If the operation
Dan Winship's avatar
Dan Winship committed
152
 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
153 154 155 156 157
 * operation was partially finished when the operation was cancelled the
 * partial result will be returned, without an error.
 *
 * On error -1 is returned and @error is set accordingly.
 * 
158
 * Returns: Number of bytes read, or -1 on error, or 0 on end of file.
159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178
 **/
gssize
g_input_stream_read  (GInputStream  *stream,
		      void          *buffer,
		      gsize          count,
		      GCancellable  *cancellable,
		      GError       **error)
{
  GInputStreamClass *class;
  gssize res;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
  g_return_val_if_fail (buffer != NULL, 0);

  if (count == 0)
    return 0;
  
  if (((gssize) count) < 0)
    {
      g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
179
		   _("Too large count value passed to %s"), G_STRFUNC);
180 181 182 183 184
      return -1;
    }

  class = G_INPUT_STREAM_GET_CLASS (stream);

185
  if (class->read_fn == NULL) 
186
    {
187
      g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_NOT_SUPPORTED,
188
                           _("Input stream doesn’t implement read"));
189 190 191
      return -1;
    }

192 193 194
  if (!g_input_stream_set_pending (stream, error))
    return -1;

195
  if (cancellable)
196
    g_cancellable_push_current (cancellable);
197
  
198
  res = class->read_fn (stream, buffer, count, cancellable, error);
199 200

  if (cancellable)
201
    g_cancellable_pop_current (cancellable);
202
  
203 204
  g_input_stream_clear_pending (stream);

205 206 207 208 209 210
  return res;
}

/**
 * g_input_stream_read_all:
 * @stream: a #GInputStream.
211 212
 * @buffer: (array length=count) (element-type guint8): a buffer to
 *     read data into (which should be at least count bytes long).
213
 * @count: the number of bytes that will be read from the stream
214
 * @bytes_read: (out): location to store the number of bytes that was read from the stream
215
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.
Matthias Clasen's avatar
Matthias Clasen committed
216
 * @error: location to store the error occurring, or %NULL to ignore
217 218 219 220 221 222 223 224
 *
 * Tries to read @count bytes from the stream into the buffer starting at
 * @buffer. Will block during this read.
 *
 * This function is similar to g_input_stream_read(), except it tries to
 * read as many bytes as requested, only stopping on an error or end of stream.
 *
 * On a successful read of @count bytes, or if we reached the end of the
225
 * stream,  %TRUE is returned, and @bytes_read is set to the number of bytes
226 227
 * read into @buffer.
 * 
228
 * If there is an error during the operation %FALSE is returned and @error
229 230 231 232 233 234 235 236
 * is set to indicate the error status.
 *
 * As a special exception to the normal conventions for functions that
 * use #GError, if this function returns %FALSE (and sets @error) then
 * @bytes_read will be set to the number of bytes that were successfully
 * read before the error was encountered.  This functionality is only
 * available from C.  If you need it from another language then you must
 * write your own loop around g_input_stream_read().
237
 *
238
 * Returns: %TRUE on success, %FALSE if there was an error
239 240
 **/
gboolean
Matthias Clasen's avatar
Matthias Clasen committed
241 242 243 244 245 246
g_input_stream_read_all (GInputStream  *stream,
			 void          *buffer,
			 gsize          count,
			 gsize         *bytes_read,
			 GCancellable  *cancellable,
			 GError       **error)
247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276
{
  gsize _bytes_read;
  gssize res;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
  g_return_val_if_fail (buffer != NULL, FALSE);

  _bytes_read = 0;
  while (_bytes_read < count)
    {
      res = g_input_stream_read (stream, (char *)buffer + _bytes_read, count - _bytes_read,
				 cancellable, error);
      if (res == -1)
	{
	  if (bytes_read)
	    *bytes_read = _bytes_read;
	  return FALSE;
	}
      
      if (res == 0)
	break;

      _bytes_read += res;
    }

  if (bytes_read)
    *bytes_read = _bytes_read;
  return TRUE;
}

277 278 279 280 281
/**
 * g_input_stream_read_bytes:
 * @stream: a #GInputStream.
 * @count: maximum number of bytes that will be read from the stream. Common
 * values include 4096 and 8192.
282
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.
283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308
 * @error: location to store the error occurring, or %NULL to ignore
 *
 * Like g_input_stream_read(), this tries to read @count bytes from
 * the stream in a blocking fashion. However, rather than reading into
 * a user-supplied buffer, this will create a new #GBytes containing
 * the data that was read. This may be easier to use from language
 * bindings.
 *
 * If count is zero, returns a zero-length #GBytes and does nothing. A
 * value of @count larger than %G_MAXSSIZE will cause a
 * %G_IO_ERROR_INVALID_ARGUMENT error.
 *
 * On success, a new #GBytes is returned. It is not an error if the
 * size of this object is not the same as the requested size, as it
 * can happen e.g. near the end of a file. A zero-length #GBytes is
 * returned on end of file (or if @count is zero), but never
 * otherwise.
 *
 * If @cancellable is not %NULL, then the operation can be cancelled by
 * triggering the cancellable object from another thread. If the operation
 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
 * operation was partially finished when the operation was cancelled the
 * partial result will be returned, without an error.
 *
 * On error %NULL is returned and @error is set accordingly.
 *
309
 * Returns: (transfer full): a new #GBytes, or %NULL on error
310 311
 *
 * Since: 2.34
312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337
 **/
GBytes *
g_input_stream_read_bytes (GInputStream  *stream,
			   gsize          count,
			   GCancellable  *cancellable,
			   GError       **error)
{
  guchar *buf;
  gssize nread;

  buf = g_malloc (count);
  nread = g_input_stream_read (stream, buf, count, cancellable, error);
  if (nread == -1)
    {
      g_free (buf);
      return NULL;
    }
  else if (nread == 0)
    {
      g_free (buf);
      return g_bytes_new_static ("", 0);
    }
  else
    return g_bytes_new_take (buf, nread);
}

338 339 340 341
/**
 * g_input_stream_skip:
 * @stream: a #GInputStream.
 * @count: the number of bytes that will be skipped from the stream
342
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore. 
Matthias Clasen's avatar
Matthias Clasen committed
343
 * @error: location to store the error occurring, or %NULL to ignore
344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359
 *
 * Tries to skip @count bytes from the stream. Will block during the operation.
 *
 * This is identical to g_input_stream_read(), from a behaviour standpoint,
 * but the bytes that are skipped are not returned to the user. Some
 * streams have an implementation that is more efficient than reading the data.
 *
 * This function is optional for inherited classes, as the default implementation
 * emulates it using read.
 *
 * If @cancellable is not %NULL, then the operation can be cancelled by
 * triggering the cancellable object from another thread. If the operation
 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned. If an
 * operation was partially finished when the operation was cancelled the
 * partial result will be returned, without an error.
 *
360
 * Returns: Number of bytes skipped, or -1 on error
361 362
 **/
gssize
Matthias Clasen's avatar
Matthias Clasen committed
363 364 365 366
g_input_stream_skip (GInputStream  *stream,
		     gsize          count,
		     GCancellable  *cancellable,
		     GError       **error)
367 368 369 370 371 372 373 374 375 376 377 378
{
  GInputStreamClass *class;
  gssize res;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);

  if (count == 0)
    return 0;

  if (((gssize) count) < 0)
    {
      g_set_error (error, G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
379
		   _("Too large count value passed to %s"), G_STRFUNC);
380 381 382 383 384
      return -1;
    }
  
  class = G_INPUT_STREAM_GET_CLASS (stream);

385 386 387
  if (!g_input_stream_set_pending (stream, error))
    return -1;

388
  if (cancellable)
389
    g_cancellable_push_current (cancellable);
390 391 392 393
  
  res = class->skip (stream, count, cancellable, error);

  if (cancellable)
394
    g_cancellable_pop_current (cancellable);
395
  
396 397
  g_input_stream_clear_pending (stream);

398 399 400 401
  return res;
}

static gssize
Matthias Clasen's avatar
Matthias Clasen committed
402 403 404 405
g_input_stream_real_skip (GInputStream  *stream,
			  gsize          count,
			  GCancellable  *cancellable,
			  GError       **error)
406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424
{
  GInputStreamClass *class;
  gssize ret, read_bytes;
  char buffer[8192];
  GError *my_error;

  if (G_IS_SEEKABLE (stream) && g_seekable_can_seek (G_SEEKABLE (stream)))
    {
      if (g_seekable_seek (G_SEEKABLE (stream),
			   count,
			   G_SEEK_CUR,
			   cancellable,
			   NULL))
	return count;
    }

  /* If not seekable, or seek failed, fall back to reading data: */

  class = G_INPUT_STREAM_GET_CLASS (stream);
425

426 427 428 429 430
  read_bytes = 0;
  while (1)
    {
      my_error = NULL;

431
      ret = class->read_fn (stream, buffer, MIN (sizeof (buffer), count),
432
                            cancellable, &my_error);
433 434 435 436 437 438 439 440 441
      if (ret == -1)
	{
	  if (read_bytes > 0 &&
	      my_error->domain == G_IO_ERROR &&
	      my_error->code == G_IO_ERROR_CANCELLED)
	    {
	      g_error_free (my_error);
	      return read_bytes;
	    }
442

443 444 445 446 447 448
	  g_propagate_error (error, my_error);
	  return -1;
	}

      count -= ret;
      read_bytes += ret;
449

450
      if (ret == 0 || count == 0)
451
        return read_bytes;
452 453 454 455 456 457
    }
}

/**
 * g_input_stream_close:
 * @stream: A #GInputStream.
458
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.
Matthias Clasen's avatar
Matthias Clasen committed
459
 * @error: location to store the error occurring, or %NULL to ignore
460 461 462 463 464 465 466
 *
 * Closes the stream, releasing resources related to it.
 *
 * Once the stream is closed, all other operations will return %G_IO_ERROR_CLOSED.
 * Closing a stream multiple times will not return an error.
 *
 * Streams will be automatically closed when the last reference
467 468
 * is dropped, but you might want to call this function to make sure 
 * resources are released as early as possible.
469 470 471 472 473 474 475
 *
 * Some streams might keep the backing store of the stream (e.g. a file descriptor)
 * open after the stream is closed. See the documentation for the individual
 * stream for details.
 *
 * On failure the first error that happened will be reported, but the close
 * operation will finish as much as possible. A stream that failed to
476
 * close will still return %G_IO_ERROR_CLOSED for all operations. Still, it
477 478
 * is important to check and report the error to the user.
 *
Dan Winship's avatar
Dan Winship committed
479
 * If @cancellable is not %NULL, then the operation can be cancelled by
480 481 482 483 484
 * triggering the cancellable object from another thread. If the operation
 * was cancelled, the error %G_IO_ERROR_CANCELLED will be returned.
 * Cancelling a close will still leave the stream closed, but some streams
 * can use a faster close that doesn't block to e.g. check errors. 
 *
485
 * Returns: %TRUE on success, %FALSE on failure
486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503
 **/
gboolean
g_input_stream_close (GInputStream  *stream,
		      GCancellable  *cancellable,
		      GError       **error)
{
  GInputStreamClass *class;
  gboolean res;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);

  class = G_INPUT_STREAM_GET_CLASS (stream);

  if (stream->priv->closed)
    return TRUE;

  res = TRUE;

504 505
  if (!g_input_stream_set_pending (stream, error))
    return FALSE;
506 507

  if (cancellable)
508
    g_cancellable_push_current (cancellable);
509

510 511
  if (class->close_fn)
    res = class->close_fn (stream, cancellable, error);
512 513

  if (cancellable)
514
    g_cancellable_pop_current (cancellable);
515 516

  g_input_stream_clear_pending (stream);
517 518 519 520 521 522 523
  
  stream->priv->closed = TRUE;
  
  return res;
}

static void
Matthias Clasen's avatar
Matthias Clasen committed
524
async_ready_callback_wrapper (GObject      *source_object,
525 526 527 528 529
			      GAsyncResult *res,
			      gpointer      user_data)
{
  GInputStream *stream = G_INPUT_STREAM (source_object);

530
  g_input_stream_clear_pending (stream);
531 532 533 534 535 536
  if (stream->priv->outstanding_callback)
    (*stream->priv->outstanding_callback) (source_object, res, user_data);
  g_object_unref (stream);
}

static void
Matthias Clasen's avatar
Matthias Clasen committed
537
async_ready_close_callback_wrapper (GObject      *source_object,
538 539 540 541 542
				    GAsyncResult *res,
				    gpointer      user_data)
{
  GInputStream *stream = G_INPUT_STREAM (source_object);

543
  g_input_stream_clear_pending (stream);
544 545 546 547 548 549 550 551 552
  stream->priv->closed = TRUE;
  if (stream->priv->outstanding_callback)
    (*stream->priv->outstanding_callback) (source_object, res, user_data);
  g_object_unref (stream);
}

/**
 * g_input_stream_read_async:
 * @stream: A #GInputStream.
553 554
 * @buffer: (array length=count) (element-type guint8): a buffer to
 *     read data into (which should be at least count bytes long).
555
 * @count: the number of bytes that will be read from the stream
556
 * @io_priority: the [I/O priority][io-priority]
557
 * of the request. 
558
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.
559 560
 * @callback: (scope async): callback to call when the request is satisfied
 * @user_data: (closure): the data to pass to callback function
561 562
 *
 * Request an asynchronous read of @count bytes from the stream into the buffer
563 564 565
 * starting at @buffer. When the operation is finished @callback will be called. 
 * You can then call g_input_stream_read_finish() to get the result of the 
 * operation.
566
 *
567
 * During an async request no other sync and async calls are allowed on @stream, and will
568 569 570 571 572 573 574 575 576 577 578 579 580 581
 * result in %G_IO_ERROR_PENDING errors. 
 *
 * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 *
 * On success, the number of bytes read into the buffer will be passed to the
 * callback. It is not an error if this is not the same as the requested size, as it
 * can happen e.g. near the end of a file, but generally we try to read
 * as many bytes as requested. Zero is returned on end of file
 * (or if @count is zero),  but never otherwise.
 *
 * Any outstanding i/o request with higher priority (lower numerical value) will
 * be executed before an outstanding request with lower priority. Default
 * priority is %G_PRIORITY_DEFAULT.
 *
582
 * The asynchronous methods have a default fallback that uses threads to implement
583 584 585 586 587 588 589 590 591 592 593 594 595
 * asynchronicity, so they are optional for inheriting classes. However, if you
 * override one you must override all.
 **/
void
g_input_stream_read_async (GInputStream        *stream,
			   void                *buffer,
			   gsize                count,
			   int                  io_priority,
			   GCancellable        *cancellable,
			   GAsyncReadyCallback  callback,
			   gpointer             user_data)
{
  GInputStreamClass *class;
596
  GError *error = NULL;
597 598 599 600 601 602

  g_return_if_fail (G_IS_INPUT_STREAM (stream));
  g_return_if_fail (buffer != NULL);

  if (count == 0)
    {
603 604 605 606 607 608
      GTask *task;

      task = g_task_new (stream, cancellable, callback, user_data);
      g_task_set_source_tag (task, g_input_stream_read_async);
      g_task_return_int (task, 0);
      g_object_unref (task);
609 610 611 612 613
      return;
    }
  
  if (((gssize) count) < 0)
    {
614 615 616 617 618
      g_task_report_new_error (stream, callback, user_data,
                               g_input_stream_read_async,
                               G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
                               _("Too large count value passed to %s"),
                               G_STRFUNC);
619 620 621
      return;
    }

622
  if (!g_input_stream_set_pending (stream, &error))
623
    {
624 625 626
      g_task_report_error (stream, callback, user_data,
                           g_input_stream_read_async,
                           error);
627 628 629 630 631 632 633 634 635 636 637 638
      return;
    }

  class = G_INPUT_STREAM_GET_CLASS (stream);
  stream->priv->outstanding_callback = callback;
  g_object_ref (stream);
  class->read_async (stream, buffer, count, io_priority, cancellable,
		     async_ready_callback_wrapper, user_data);
}

/**
 * g_input_stream_read_finish:
639 640
 * @stream: a #GInputStream.
 * @result: a #GAsyncResult.
Matthias Clasen's avatar
Matthias Clasen committed
641
 * @error: a #GError location to store the error occurring, or %NULL to 
642
 * ignore.
643 644 645
 * 
 * Finishes an asynchronous stream read operation. 
 * 
646
 * Returns: number of bytes read in, or -1 on error, or 0 on end of file.
647 648
 **/
gssize
Matthias Clasen's avatar
Matthias Clasen committed
649 650 651
g_input_stream_read_finish (GInputStream  *stream,
			    GAsyncResult  *result,
			    GError       **error)
652 653 654 655 656 657
{
  GInputStreamClass *class;
  
  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
  g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);

658 659
  if (g_async_result_legacy_propagate_error (result, error))
    return -1;
660
  else if (g_async_result_is_tagged (result, g_input_stream_read_async))
661
    return g_task_propagate_int (G_TASK (result), error);
662 663 664 665 666

  class = G_INPUT_STREAM_GET_CLASS (stream);
  return class->read_finish (stream, result, error);
}

667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744
typedef struct
{
  gchar *buffer;
  gsize to_read;
  gsize bytes_read;
} AsyncReadAll;

static void
free_async_read_all (gpointer data)
{
  g_slice_free (AsyncReadAll, data);
}

static void
read_all_callback (GObject      *stream,
                   GAsyncResult *result,
                   gpointer      user_data)
{
  GTask *task = user_data;
  AsyncReadAll *data = g_task_get_task_data (task);
  gboolean got_eof = FALSE;

  if (result)
    {
      GError *error = NULL;
      gssize nread;

      nread = g_input_stream_read_finish (G_INPUT_STREAM (stream), result, &error);

      if (nread == -1)
        {
          g_task_return_error (task, error);
          g_object_unref (task);
          return;
        }

      g_assert_cmpint (nread, <=, data->to_read);
      data->to_read -= nread;
      data->bytes_read += nread;
      got_eof = (nread == 0);
    }

  if (got_eof || data->to_read == 0)
    {
      g_task_return_boolean (task, TRUE);
      g_object_unref (task);
    }

  else
    g_input_stream_read_async (G_INPUT_STREAM (stream),
                               data->buffer + data->bytes_read,
                               data->to_read,
                               g_task_get_priority (task),
                               g_task_get_cancellable (task),
                               read_all_callback, task);
}


static void
read_all_async_thread (GTask        *task,
                       gpointer      source_object,
                       gpointer      task_data,
                       GCancellable *cancellable)
{
  GInputStream *stream = source_object;
  AsyncReadAll *data = task_data;
  GError *error = NULL;

  if (g_input_stream_read_all (stream, data->buffer, data->to_read, &data->bytes_read,
                               g_task_get_cancellable (task), &error))
    g_task_return_boolean (task, TRUE);
  else
    g_task_return_error (task, error);
}

/**
 * g_input_stream_read_all_async:
 * @stream: A #GInputStream
745 746
 * @buffer: (array length=count) (element-type guint8): a buffer to
 *     read data into (which should be at least count bytes long)
747 748
 * @count: the number of bytes that will be read from the stream
 * @io_priority: the [I/O priority][io-priority] of the request
749
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore
750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785
 * @callback: (scope async): callback to call when the request is satisfied
 * @user_data: (closure): the data to pass to callback function
 *
 * Request an asynchronous read of @count bytes from the stream into the
 * buffer starting at @buffer.
 *
 * This is the asynchronous equivalent of g_input_stream_read_all().
 *
 * Call g_input_stream_read_all_finish() to collect the result.
 *
 * Any outstanding I/O request with higher priority (lower numerical
 * value) will be executed before an outstanding request with lower
 * priority. Default priority is %G_PRIORITY_DEFAULT.
 *
 * Since: 2.44
 **/
void
g_input_stream_read_all_async (GInputStream        *stream,
                               void                *buffer,
                               gsize                count,
                               int                  io_priority,
                               GCancellable        *cancellable,
                               GAsyncReadyCallback  callback,
                               gpointer             user_data)
{
  AsyncReadAll *data;
  GTask *task;

  g_return_if_fail (G_IS_INPUT_STREAM (stream));
  g_return_if_fail (buffer != NULL || count == 0);

  task = g_task_new (stream, cancellable, callback, user_data);
  data = g_slice_new0 (AsyncReadAll);
  data->buffer = buffer;
  data->to_read = count;

786
  g_task_set_source_tag (task, g_input_stream_read_all_async);
787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846
  g_task_set_task_data (task, data, free_async_read_all);
  g_task_set_priority (task, io_priority);

  /* If async reads are going to be handled via the threadpool anyway
   * then we may as well do it with a single dispatch instead of
   * bouncing in and out.
   */
  if (g_input_stream_async_read_is_via_threads (stream))
    {
      g_task_run_in_thread (task, read_all_async_thread);
      g_object_unref (task);
    }
  else
    read_all_callback (G_OBJECT (stream), NULL, task);
}

/**
 * g_input_stream_read_all_finish:
 * @stream: a #GInputStream
 * @result: a #GAsyncResult
 * @bytes_read: (out): location to store the number of bytes that was read from the stream
 * @error: a #GError location to store the error occurring, or %NULL to ignore
 *
 * Finishes an asynchronous stream read operation started with
 * g_input_stream_read_all_async().
 *
 * As a special exception to the normal conventions for functions that
 * use #GError, if this function returns %FALSE (and sets @error) then
 * @bytes_read will be set to the number of bytes that were successfully
 * read before the error was encountered.  This functionality is only
 * available from C.  If you need it from another language then you must
 * write your own loop around g_input_stream_read_async().
 *
 * Returns: %TRUE on success, %FALSE if there was an error
 *
 * Since: 2.44
 **/
gboolean
g_input_stream_read_all_finish (GInputStream  *stream,
                                GAsyncResult  *result,
                                gsize         *bytes_read,
                                GError       **error)
{
  GTask *task;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
  g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);

  task = G_TASK (result);

  if (bytes_read)
    {
      AsyncReadAll *data = g_task_get_task_data (task);

      *bytes_read = data->bytes_read;
    }

  return g_task_propagate_boolean (task, error);
}

847 848 849 850 851
static void
read_bytes_callback (GObject      *stream,
		     GAsyncResult *result,
		     gpointer      user_data)
{
852 853
  GTask *task = user_data;
  guchar *buf = g_task_get_task_data (task);
854 855 856 857 858 859 860 861 862
  GError *error = NULL;
  gssize nread;
  GBytes *bytes = NULL;

  nread = g_input_stream_read_finish (G_INPUT_STREAM (stream),
				      result, &error);
  if (nread == -1)
    {
      g_free (buf);
863
      g_task_return_error (task, error);
864 865 866 867 868 869 870 871 872 873
    }
  else if (nread == 0)
    {
      g_free (buf);
      bytes = g_bytes_new_static ("", 0);
    }
  else
    bytes = g_bytes_new_take (buf, nread);

  if (bytes)
874 875 876
    g_task_return_pointer (task, bytes, (GDestroyNotify)g_bytes_unref);

  g_object_unref (task);
877 878 879 880 881 882
}

/**
 * g_input_stream_read_bytes_async:
 * @stream: A #GInputStream.
 * @count: the number of bytes that will be read from the stream
883
 * @io_priority: the [I/O priority][io-priority] of the request
884
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.
885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907
 * @callback: (scope async): callback to call when the request is satisfied
 * @user_data: (closure): the data to pass to callback function
 *
 * Request an asynchronous read of @count bytes from the stream into a
 * new #GBytes. When the operation is finished @callback will be
 * called. You can then call g_input_stream_read_bytes_finish() to get the
 * result of the operation.
 *
 * During an async request no other sync and async calls are allowed
 * on @stream, and will result in %G_IO_ERROR_PENDING errors.
 *
 * A value of @count larger than %G_MAXSSIZE will cause a
 * %G_IO_ERROR_INVALID_ARGUMENT error.
 *
 * On success, the new #GBytes will be passed to the callback. It is
 * not an error if this is smaller than the requested size, as it can
 * happen e.g. near the end of a file, but generally we try to read as
 * many bytes as requested. Zero is returned on end of file (or if
 * @count is zero), but never otherwise.
 *
 * Any outstanding I/O request with higher priority (lower numerical
 * value) will be executed before an outstanding request with lower
 * priority. Default priority is %G_PRIORITY_DEFAULT.
908 909
 *
 * Since: 2.34
910 911 912 913 914 915 916 917 918
 **/
void
g_input_stream_read_bytes_async (GInputStream          *stream,
				 gsize                  count,
				 int                    io_priority,
				 GCancellable          *cancellable,
				 GAsyncReadyCallback    callback,
				 gpointer               user_data)
{
919
  GTask *task;
920 921
  guchar *buf;

922
  task = g_task_new (stream, cancellable, callback, user_data);
923 924
  g_task_set_source_tag (task, g_input_stream_read_bytes_async);

925
  buf = g_malloc (count);
926
  g_task_set_task_data (task, buf, NULL);
927 928

  g_input_stream_read_async (stream, buf, count,
929 930
                             io_priority, cancellable,
                             read_bytes_callback, task);
931 932 933 934 935 936 937 938 939 940 941
}

/**
 * g_input_stream_read_bytes_finish:
 * @stream: a #GInputStream.
 * @result: a #GAsyncResult.
 * @error: a #GError location to store the error occurring, or %NULL to
 *   ignore.
 *
 * Finishes an asynchronous stream read-into-#GBytes operation.
 *
942
 * Returns: (transfer full): the newly-allocated #GBytes, or %NULL on error
943 944
 *
 * Since: 2.34
945 946 947 948 949 950 951
 **/
GBytes *
g_input_stream_read_bytes_finish (GInputStream  *stream,
				  GAsyncResult  *result,
				  GError       **error)
{
  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), NULL);
952
  g_return_val_if_fail (g_task_is_valid (result, stream), NULL);
953

954
  return g_task_propagate_pointer (G_TASK (result), error);
955 956
}

957 958 959 960
/**
 * g_input_stream_skip_async:
 * @stream: A #GInputStream.
 * @count: the number of bytes that will be skipped from the stream
961
 * @io_priority: the [I/O priority][io-priority] of the request
962
 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.
963 964
 * @callback: (scope async): callback to call when the request is satisfied
 * @user_data: (closure): the data to pass to callback function
965
 *
966
 * Request an asynchronous skip of @count bytes from the stream.
967 968 969
 * When the operation is finished @callback will be called.
 * You can then call g_input_stream_skip_finish() to get the result
 * of the operation.
970
 *
971 972
 * During an async request no other sync and async calls are allowed,
 * and will result in %G_IO_ERROR_PENDING errors.
973 974 975
 *
 * A value of @count larger than %G_MAXSSIZE will cause a %G_IO_ERROR_INVALID_ARGUMENT error.
 *
976 977
 * On success, the number of bytes skipped will be passed to the callback.
 * It is not an error if this is not the same as the requested size, as it
978 979 980 981
 * can happen e.g. near the end of a file, but generally we try to skip
 * as many bytes as requested. Zero is returned on end of file
 * (or if @count is zero), but never otherwise.
 *
982 983 984
 * Any outstanding i/o request with higher priority (lower numerical value)
 * will be executed before an outstanding request with lower priority.
 * Default priority is %G_PRIORITY_DEFAULT.
985
 *
986 987 988
 * The asynchronous methods have a default fallback that uses threads to
 * implement asynchronicity, so they are optional for inheriting classes.
 * However, if you override one, you must override all.
989 990 991 992 993 994 995 996 997 998
 **/
void
g_input_stream_skip_async (GInputStream        *stream,
			   gsize                count,
			   int                  io_priority,
			   GCancellable        *cancellable,
			   GAsyncReadyCallback  callback,
			   gpointer             user_data)
{
  GInputStreamClass *class;
999
  GError *error = NULL;
1000 1001 1002 1003 1004

  g_return_if_fail (G_IS_INPUT_STREAM (stream));

  if (count == 0)
    {
1005
      GTask *task;
1006

1007 1008 1009 1010
      task = g_task_new (stream, cancellable, callback, user_data);
      g_task_set_source_tag (task, g_input_stream_skip_async);
      g_task_return_int (task, 0);
      g_object_unref (task);
1011 1012 1013 1014 1015
      return;
    }
  
  if (((gssize) count) < 0)
    {
1016 1017 1018 1019 1020
      g_task_report_new_error (stream, callback, user_data,
                               g_input_stream_skip_async,
                               G_IO_ERROR, G_IO_ERROR_INVALID_ARGUMENT,
                               _("Too large count value passed to %s"),
                               G_STRFUNC);
1021 1022 1023
      return;
    }

1024
  if (!g_input_stream_set_pending (stream, &error))
1025
    {
1026 1027 1028
      g_task_report_error (stream, callback, user_data,
                           g_input_stream_skip_async,
                           error);
1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040
      return;
    }

  class = G_INPUT_STREAM_GET_CLASS (stream);
  stream->priv->outstanding_callback = callback;
  g_object_ref (stream);
  class->skip_async (stream, count, io_priority, cancellable,
		     async_ready_callback_wrapper, user_data);
}

/**
 * g_input_stream_skip_finish:
1041 1042
 * @stream: a #GInputStream.
 * @result: a #GAsyncResult.
Matthias Clasen's avatar
Matthias Clasen committed
1043
 * @error: a #GError location to store the error occurring, or %NULL to 
1044
 * ignore.
1045 1046 1047 1048
 * 
 * Finishes a stream skip operation.
 * 
 * Returns: the size of the bytes skipped, or %-1 on error.
1049 1050
 **/
gssize
Matthias Clasen's avatar
Matthias Clasen committed
1051 1052 1053
g_input_stream_skip_finish (GInputStream  *stream,
			    GAsyncResult  *result,
			    GError       **error)
1054 1055 1056 1057 1058 1059
{
  GInputStreamClass *class;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), -1);
  g_return_val_if_fail (G_IS_ASYNC_RESULT (result), -1);

1060 1061
  if (g_async_result_legacy_propagate_error (result, error))
    return -1;
1062
  else if (g_async_result_is_tagged (result, g_input_stream_skip_async))
1063
    return g_task_propagate_int (G_TASK (result), error);
1064 1065 1066 1067 1068 1069 1070 1071

  class = G_INPUT_STREAM_GET_CLASS (stream);
  return class->skip_finish (stream, result, error);
}

/**
 * g_input_stream_close_async:
 * @stream: A #GInputStream.
1072
 * @io_priority: the [I/O priority][io-priority] of the request
1073
 * @cancellable: (nullable): optional cancellable object
1074 1075
 * @callback: (scope async): callback to call when the request is satisfied
 * @user_data: (closure): the data to pass to callback function
1076 1077
 *
 * Requests an asynchronous closes of the stream, releasing resources related to it.
1078 1079 1080
 * When the operation is finished @callback will be called. 
 * You can then call g_input_stream_close_finish() to get the result of the 
 * operation.
1081 1082 1083
 *
 * For behaviour details see g_input_stream_close().
 *
1084
 * The asynchronous methods have a default fallback that uses threads to implement
1085 1086 1087 1088
 * asynchronicity, so they are optional for inheriting classes. However, if you
 * override one you must override all.
 **/
void
Matthias Clasen's avatar
Matthias Clasen committed
1089 1090 1091 1092 1093
g_input_stream_close_async (GInputStream        *stream,
			    int                  io_priority,
			    GCancellable        *cancellable,
			    GAsyncReadyCallback  callback,
			    gpointer             user_data)
1094 1095
{
  GInputStreamClass *class;
1096
  GError *error = NULL;
1097 1098 1099 1100 1101

  g_return_if_fail (G_IS_INPUT_STREAM (stream));

  if (stream->priv->closed)
    {
1102
      GTask *task;
1103

1104 1105 1106 1107
      task = g_task_new (stream, cancellable, callback, user_data);
      g_task_set_source_tag (task, g_input_stream_close_async);
      g_task_return_boolean (task, TRUE);
      g_object_unref (task);
1108 1109 1110
      return;
    }

1111
  if (!g_input_stream_set_pending (stream, &error))
1112
    {
1113 1114 1115
      g_task_report_error (stream, callback, user_data,
                           g_input_stream_close_async,
                           error);
1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127
      return;
    }
  
  class = G_INPUT_STREAM_GET_CLASS (stream);
  stream->priv->outstanding_callback = callback;
  g_object_ref (stream);
  class->close_async (stream, io_priority, cancellable,
		      async_ready_close_callback_wrapper, user_data);
}

/**
 * g_input_stream_close_finish:
1128 1129
 * @stream: a #GInputStream.
 * @result: a #GAsyncResult.
Matthias Clasen's avatar
Matthias Clasen committed
1130
 * @error: a #GError location to store the error occurring, or %NULL to 
1131
 * ignore.
1132 1133 1134 1135
 * 
 * Finishes closing a stream asynchronously, started from g_input_stream_close_async().
 * 
 * Returns: %TRUE if the stream was closed successfully.
1136 1137
 **/
gboolean
Matthias Clasen's avatar
Matthias Clasen committed
1138 1139 1140
g_input_stream_close_finish (GInputStream  *stream,
			     GAsyncResult  *result,
			     GError       **error)
1141 1142 1143 1144 1145 1146
{
  GInputStreamClass *class;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
  g_return_val_if_fail (G_IS_ASYNC_RESULT (result), FALSE);

1147 1148
  if (g_async_result_legacy_propagate_error (result, error))
    return FALSE;
1149
  else if (g_async_result_is_tagged (result, g_input_stream_close_async))
1150
    return g_task_propagate_boolean (G_TASK (result), error);
1151 1152 1153 1154 1155 1156 1157 1158 1159

  class = G_INPUT_STREAM_GET_CLASS (stream);
  return class->close_finish (stream, result, error);
}

/**
 * g_input_stream_is_closed:
 * @stream: input stream.
 * 
1160 1161
 * Checks if an input stream is closed.
 * 
1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175
 * Returns: %TRUE if the stream is closed.
 **/
gboolean
g_input_stream_is_closed (GInputStream *stream)
{
  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
  
  return stream->priv->closed;
}
 
/**
 * g_input_stream_has_pending:
 * @stream: input stream.
 * 
1176 1177
 * Checks if an input stream has pending actions.
 * 
1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190
 * Returns: %TRUE if @stream has pending actions.
 **/  
gboolean
g_input_stream_has_pending (GInputStream *stream)
{
  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), TRUE);
  
  return stream->priv->pending;
}

/**
 * g_input_stream_set_pending:
 * @stream: input stream
Matthias Clasen's avatar
Matthias Clasen committed
1191
 * @error: a #GError location to store the error occurring, or %NULL to 
1192
 * ignore.
1193
 * 
1194 1195 1196 1197
 * Sets @stream to have actions pending. If the pending flag is
 * already set or @stream is closed, it will return %FALSE and set
 * @error.
 *
1198
 * Returns: %TRUE if pending was previously unset and is now set.
1199 1200 1201 1202 1203 1204 1205 1206
 **/
gboolean
g_input_stream_set_pending (GInputStream *stream, GError **error)
{
  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);
  
  if (stream->priv->closed)
    {
1207 1208
      g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_CLOSED,
                           _("Stream is already closed"));
1209 1210 1211 1212 1213
      return FALSE;
    }
  
  if (stream->priv->pending)
    {
1214
      g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_PENDING,
1215 1216 1217 1218
		/* Translators: This is an error you get if there is already an
		 * operation running against this stream when you try to start
		 * one */
		 _("Stream has outstanding operation"));
1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230
      return FALSE;
    }
  
  stream->priv->pending = TRUE;
  return TRUE;
}

/**
 * g_input_stream_clear_pending:
 * @stream: input stream
 * 
 * Clears the pending flag on @stream.
1231 1232
 **/
void
1233
g_input_stream_clear_pending (GInputStream *stream)
1234 1235 1236
{
  g_return_if_fail (G_IS_INPUT_STREAM (stream));
  
1237
  stream->priv->pending = FALSE;
1238 1239
}

1240
/*< internal >
1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261
 * g_input_stream_async_read_is_via_threads:
 * @stream: input stream
 *
 * Checks if an input stream's read_async function uses threads.
 *
 * Returns: %TRUE if @stream's read_async function uses threads.
 **/
gboolean
g_input_stream_async_read_is_via_threads (GInputStream *stream)
{
  GInputStreamClass *class;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);

  class = G_INPUT_STREAM_GET_CLASS (stream);

  return (class->read_async == g_input_stream_real_read_async &&
      !(G_IS_POLLABLE_INPUT_STREAM (stream) &&
        g_pollable_input_stream_can_poll (G_POLLABLE_INPUT_STREAM (stream))));
}

1262 1263 1264 1265 1266 1267 1268 1269 1270 1271 1272 1273 1274 1275 1276 1277 1278 1279 1280 1281
/*< internal >
 * g_input_stream_async_close_is_via_threads:
 * @stream: input stream
 *
 * Checks if an input stream's close_async function uses threads.
 *
 * Returns: %TRUE if @stream's close_async function uses threads.
 **/
gboolean
g_input_stream_async_close_is_via_threads (GInputStream *stream)
{
  GInputStreamClass *class;

  g_return_val_if_fail (G_IS_INPUT_STREAM (stream), FALSE);

  class = G_INPUT_STREAM_GET_CLASS (stream);

  return class->close_async == g_input_stream_real_close_async;
}

1282 1283 1284 1285 1286
/********************************************
 *   Default implementation of async ops    *
 ********************************************/

typedef struct {
1287 1288
  void   *buffer;
  gsize   count;
1289 1290
} ReadData;

1291 1292 1293 1294 1295 1296
static void
free_read_data (ReadData *op)
{
  g_slice_free (ReadData, op);
}

1297
static void
1298 1299 1300 1301
read_async_thread (GTask        *task,
                   gpointer      source_object,
                   gpointer      task_data,
                   GCancellable *cancellable)
1302
{
1303 1304
  GInputStream *stream = source_object;
  ReadData *op = task_data;
1305 1306
  GInputStreamClass *class;
  GError *error = NULL;
1307
  gssize nread;
1308
 
1309
  class = G_INPUT_STREAM_GET_CLASS (stream);
1310

1311 1312 1313 1314 1315 1316 1317 1318
  nread = class->read_fn (stream,
                          op->buffer, op->count,
                          g_task_get_cancellable (task),
                          &error);
  if (nread == -1)
    g_task_return_error (task, error);
  else
    g_task_return_int (task, nread);
1319 1320
}

1321
static void read_async_pollable (GPollableInputStream *stream,
1322
                                 GTask                *task);
1323 1324 1325 1326 1327

static gboolean
read_async_pollable_ready (GPollableInputStream *stream,
			   gpointer              user_data)
{
1328
  GTask *task = user_data;
1329

1330
  read_async_pollable (stream, task);
1331 1332 1333 1334 1335
  return FALSE;
}

static void
read_async_pollable (GPollableInputStream *stream,
1336
                     GTask                *task)
1337
{
1338
  ReadData *op = g_task_get_task_data (task);
1339
  GError *error = NULL;
1340
  gssize nread;
1341

1342 1343 1344 1345 1346
  if (g_task_return_error_if_cancelled (task))
    return;

  nread = G_POLLABLE_INPUT_STREAM_GET_INTERFACE (stream)->
    read_nonblocking (stream, op->buffer, op->count, &error);
1347 1348 1349 1350 1351 1352

  if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
    {
      GSource *source;

      g_error_free (error);
1353 1354 1355 1356 1357

      source = g_pollable_input_stream_create_source (stream,
                                                      g_task_get_cancellable (task));
      g_task_attach_source (task, source,
                            (GSourceFunc) read_async_pollable_ready);
1358 1359 1360 1361
      g_source_unref (source);
      return;
    }

1362 1363
  if (nread == -1)
    g_task_return_error (task, error);
1364
  else
1365 1366
    g_task_return_int (task, nread);
  /* g_input_stream_real_read_async() unrefs task */
1367 1368
}

1369

1370
static void
Matthias Clasen's avatar
Matthias Clasen committed
1371 1372 1373 1374 1375 1376 1377
g_input_stream_real_read_async (GInputStream        *stream,
				void                *buffer,
				gsize                count,
				int                  io_priority,
				GCancellable        *cancellable,
				GAsyncReadyCallback  callback,
				gpointer             user_data)
1378
{
1379
  GTask *task;
1380 1381
  ReadData *op;
  
1382
  op = g_slice_new0 (ReadData);
1383
  task = g_task_new (stream, cancellable, callback, user_data);
1384
  g_task_set_source_tag (task, g_input_stream_real_read_async);
1385 1386
  g_task_set_task_data (task, op, (GDestroyNotify) free_read_data);
  g_task_set_priority (task, io_priority);
1387
  op->buffer = buffer;
1388
  op->count = count;
1389

1390
  if (!g_input_stream_async_read_is_via_threads (stream))
1391
    read_async_pollable (G_POLLABLE_INPUT_STREAM (stream), task);
1392
  else
1393 1394
    g_task_run_in_thread (task, read_async_thread);
  g_object_unref (task);
1395 1396 1397
}

static gssize
Matthias Clasen's avatar
Matthias Clasen committed
1398 1399 1400
g_input_stream_real_read_finish (GInputStream  *stream,
				 GAsyncResult  *result,
				 GError       **error)
1401
{
1402
  g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1403

1404
  return g_task_propagate_int (G_TASK (result), error);
1405 1406 1407 1408
}


static void
1409 1410 1411 1412
skip_async_thread (GTask        *task,
                   gpointer      source_object,
                   gpointer      task_data,
                   GCancellable *cancellable)
1413
{
1414 1415
  GInputStream *stream = source_object;
  gsize count = GPOINTER_TO_SIZE (task_data);
1416 1417
  GInputStreamClass *class;
  GError *error = NULL;
1418 1419 1420 1421 1422 1423 1424 1425 1426 1427
  gssize ret;

  class = G_INPUT_STREAM_GET_CLASS (stream);
  ret = class->skip (stream, count,
                     g_task_get_cancellable (task),
                     &error);
  if (ret == -1)
    g_task_return_error (task, error);
  else
    g_task_return_int (task, ret);
1428 1429 1430 1431 1432 1433 1434 1435 1436
}

typedef struct {
  char buffer[8192];
  gsize count;
  gsize count_skipped;
} SkipFallbackAsyncData;

static void
Matthias Clasen's avatar
Matthias Clasen committed
1437
skip_callback_wrapper (GObject      *source_object,
1438
		       GAsyncResult *res,
Matthias Clasen's avatar
Matthias Clasen committed
1439
		       gpointer      user_data)
1440 1441
{
  GInputStreamClass *class;
1442 1443
  GTask *task = user_data;
  SkipFallbackAsyncData *data = g_task_get_task_data (task);
1444 1445 1446 1447 1448 1449 1450 1451 1452 1453 1454 1455 1456
  GError *error = NULL;
  gssize ret;

  ret = g_input_stream_read_finish (G_INPUT_STREAM (source_object), res, &error);

  if (ret > 0)
    {
      data->count -= ret;
      data->count_skipped += ret;

      if (data->count > 0)
	{
	  class = G_INPUT_STREAM_GET_CLASS (source_object);
1457 1458 1459 1460
	  class->read_async (G_INPUT_STREAM (source_object),
                             data->buffer, MIN (8192, data->count),
                             g_task_get_priority (task),
                             g_task_get_cancellable (task),
1461
                             skip_callback_wrapper, task);
1462 1463 1464 1465
	  return;
	}
    }

1466 1467 1468
  if (ret == -1 &&
      g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED) &&
      data->count_skipped)
1469
    {
1470 1471
      /* No error, return partial read */
      g_clear_error (&error);
1472 1473
    }

1474 1475 1476 1477 1478
  if (error)
    g_task_return_error (task, error);
  else
    g_task_return_int (task, data->count_skipped);
  g_object_unref (task);
1479 1480 1481 1482 1483 1484 1485 1486 1487 1488 1489 1490
 }

static void
g_input_stream_real_skip_async (GInputStream        *stream,
				gsize                count,
				int                  io_priority,
				GCancellable        *cancellable,
				GAsyncReadyCallback  callback,
				gpointer             user_data)
{
  GInputStreamClass *class;
  SkipFallbackAsyncData *data;
1491
  GTask *task;
1492 1493 1494

  class = G_INPUT_STREAM_GET_CLASS (stream);

1495
  task = g_task_new (stream, cancellable, callback, user_data);
1496
  g_task_set_source_tag (task, g_input_stream_real_skip_async);
1497 1498
  g_task_set_priority (task, io_priority);

1499
  if (g_input_stream_async_read_is_via_threads (stream))
1500 1501 1502 1503
    {
      /* Read is thread-using async fallback.
       * Make skip use threads too, so that we can use a possible sync skip
       * implementation. */
1504
      g_task_set_task_data (task, GSIZE_TO_POINTER (count), NULL);
1505

1506 1507
      g_task_run_in_thread (task, skip_async_thread);
      g_object_unref (task);
1508 1509 1510 1511 1512 1513 1514 1515 1516
    }
  else
    {
      /* TODO: Skip fallback uses too much memory, should do multiple read calls */
      
      /* There is a custom async read function, lets use that. */
      data = g_new (SkipFallbackAsyncData, 1);
      data->count = count;
      data->count_skipped = 0;
1517 1518
      g_task_set_task_data (task, data, g_free);
      g_task_set_check_cancellable (task, FALSE);
1519
      class->read_async (stream, data->buffer, MIN (8192, count), io_priority, cancellable,
1520
			 skip_callback_wrapper, task);
1521 1522 1523 1524 1525
    }

}

static gssize
Matthias Clasen's avatar
Matthias Clasen committed
1526 1527 1528
g_input_stream_real_skip_finish (GInputStream  *stream,
				 GAsyncResult  *result,
				 GError       **error)
1529
{
1530
  g_return_val_if_fail (g_task_is_valid (result, stream), -1);
1531

1532
  return g_task_propagate_int (G_TASK (result), error);
1533 1534 1535
}

static void
1536 1537 1538 1539
close_async_thread (GTask        *task,
                    gpointer      source_object,
                    gpointer      task_data,
                    GCancellable *cancellable)
1540
{
1541
  GInputStream *stream = source_object;
1542 1543 1544 1545
  GInputStreamClass *class;
  GError *error = NULL;
  gboolean result;

1546
  class = G_INPUT_STREAM_GET_CLASS (stream);
1547
  if (class->close_fn)
1548
    {
1549 1550 1551
      result = class->close_fn (stream,
                                g_task_get_cancellable (task),
                                &error);
1552
      if (!result)
1553 1554 1555 1556
        {
          g_task_return_error (task, error);
          return;
        }
1557
    }
1558 1559

  g_task_return_boolean (task, TRUE);
1560 1561 1562 1563 1564 1565 1566 1567 1568
}

static void
g_input_stream_real_close_async (GInputStream        *stream,
				 int                  io_priority,
				 GCancellable        *cancellable,
				 GAsyncReadyCallback  callback,
				 gpointer             user_data)
{
1569
  GTask *task;
1570

1571
  task = g_task_new (stream, cancellable, callback, user_data);
1572
  g_task_set_source_tag (task, g_input_stream_real_close_async);
1573 1574
  g_task_set_check_cancellable (task, FALSE);
  g_task_set_priority (task, io_priority);
1575
  
1576 1577
  g_task_run_in_thread (task, close_async_thread);
  g_object_unref (task);
1578 1579 1580
}

static gboolean
Matthias Clasen's avatar
Matthias Clasen committed
1581 1582 1583
g_input_stream_real_close_finish (GInputStream  *stream,
				  GAsyncResult  *result,
				  GError       **error)
1584
{
1585
  g_return_val_if_fail (g_task_is_valid (result, stream), FALSE);
1586

1587
  return g_task_propagate_boolean (G_TASK (result), error);
1588
}