/* GIO - GLib Input, Output and Streaming Library

 *

 * Copyright (C) 2010 Red Hat, Inc.

 *

 * SPDX-License-Identifier: LGPL-2.1-or-later

 *

 * 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

 * version 2.1 of the License, or (at your option) any later version.

 *

 * 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

 * Public License along with this library; if not, see <http://www.gnu.org/licenses/>.

 */

#include "config.h"

#include <errno.h>

#include "gpollableinputstream.h"

#include "gasynchelper.h"

#include "glibintl.h"

typedef struct {

  GSource       source;

  GObject      *stream;

} GPollableSource;

static gboolean

pollable_source_dispatch (GSource     *source,

        GSourceFunc  callback,

        gpointer     user_data)

{

  GPollableSourceFunc func = (GPollableSourceFunc)callback;

  GPollableSource *pollable_source = (GPollableSource *)source;

  return (*func) (pollable_source->stream, user_data);

}

static void

pollable_source_finalize (GSource *source)

{

  GPollableSource *pollable_source = (GPollableSource *)source;

  g_object_unref (pollable_source->stream);

}

static gboolean

pollable_source_closure_callback (GObject  *stream,

          gpointer  data)

{

  GClosure *closure = data;

  GValue param = G_VALUE_INIT;

  GValue result_value = G_VALUE_INIT;

  gboolean result;

  g_value_init (&result_value, G_TYPE_BOOLEAN);

  g_value_init (&param, G_TYPE_OBJECT);

  g_value_set_object (&param, stream);

  g_closure_invoke (closure, &result_value, 1, &param, NULL);

  result = g_value_get_boolean (&result_value);

  g_value_unset (&result_value);

  g_value_unset (&param);

  return result;

}

static GSourceFuncs pollable_source_funcs =

{

  NULL,

  NULL,

  pollable_source_dispatch,

  pollable_source_finalize,

  (GSourceFunc)pollable_source_closure_callback,

  NULL,

};

/**

 * g_pollable_source_new:

 * @pollable_stream: the stream associated with the new source

 *

 * Utility method for #GPollableInputStream and #GPollableOutputStream

 * implementations. Creates a new #GSource that expects a callback of

 * type #GPollableSourceFunc. The new source does not actually do

 * anything on its own; use g_source_add_child_source() to add other

 * sources to it to cause it to trigger.

 *

 * Returns: (transfer full): the new #GSource.

 *

 * Since: 2.28

 */

GSource *

g_pollable_source_new (GObject *pollable_stream)

{

  GSource *source;

  GPollableSource *pollable_source;

  g_return_val_if_fail (G_IS_POLLABLE_INPUT_STREAM (pollable_stream) ||

      G_IS_POLLABLE_OUTPUT_STREAM (pollable_stream), NULL);

  source = g_source_new (&pollable_source_funcs, sizeof (GPollableSource));

  g_source_set_static_name (source, "GPollableSource");

  pollable_source = (GPollableSource *)source;

  pollable_source->stream = g_object_ref (pollable_stream);

  return source;

}

/**

 * g_pollable_source_new_full:

 * @pollable_stream: (type GObject): the stream associated with the

 *   new source

 * @child_source: (nullable): optional child source to attach

 * @cancellable: (nullable): optional #GCancellable to attach

 *

 * Utility method for #GPollableInputStream and #GPollableOutputStream

 * implementations. Creates a new #GSource, as with

 * g_pollable_source_new(), but also attaching @child_source (with a

 * dummy callback), and @cancellable, if they are non-%NULL.

 *

 * Returns: (transfer full): the new #GSource.

 *

 * Since: 2.34

 */

GSource *

g_pollable_source_new_full (gpointer      pollable_stream,

          GSource      *child_source,

          GCancellable *cancellable)

{

  GSource *source;

  g_return_val_if_fail (G_IS_POLLABLE_INPUT_STREAM (pollable_stream) ||

      G_IS_POLLABLE_OUTPUT_STREAM (pollable_stream), NULL);

  source = g_pollable_source_new (pollable_stream);

  if (child_source)

    {

      g_source_set_dummy_callback (child_source);

      g_source_add_child_source (source, child_source);

    }

  if (cancellable)

    {

      GSource *cancellable_source = g_cancellable_source_new (cancellable);

      g_source_set_dummy_callback (cancellable_source);

      g_source_add_child_source (source, cancellable_source);

      g_source_unref (cancellable_source);

    }

  return source;

}

/**

 * g_pollable_stream_read:

 * @stream: a #GInputStream

 * @buffer: (array length=count) (element-type guint8): a buffer to

 *   read data into

 * @count: the number of bytes to read

 * @blocking: whether to do blocking I/O

 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.

 * @error: location to store the error occurring, or %NULL to ignore

 *

 * Tries to read from @stream, as with g_input_stream_read() (if

 * @blocking is %TRUE) or g_pollable_input_stream_read_nonblocking()

 * (if @blocking is %FALSE). This can be used to more easily share

 * code between blocking and non-blocking implementations of a method.

 *

 * If @blocking is %FALSE, then @stream must be a

 * #GPollableInputStream for which g_pollable_input_stream_can_poll()

 * returns %TRUE, or else the behavior is undefined. If @blocking is

 * %TRUE, then @stream does not need to be a #GPollableInputStream.

 *

 * Returns: the number of bytes read, or -1 on error.

 *

 * Since: 2.34

 */

gssize

g_pollable_stream_read (GInputStream   *stream,

      void           *buffer,

      gsize           count,

      gboolean        blocking,

      GCancellable   *cancellable,

      GError        **error)

{

  if (blocking)

    {

      return g_input_stream_read (stream,

          buffer, count,

          cancellable, error);

    }

  else

    {

      return g_pollable_input_stream_read_nonblocking (G_POLLABLE_INPUT_STREAM (stream),

                   buffer, count,

                   cancellable, error);

    }

}

/**

 * g_pollable_stream_write:

 * @stream: a #GOutputStream.

 * @buffer: (array length=count) (element-type guint8): the buffer

 *   containing the data to write.

 * @count: the number of bytes to write

 * @blocking: whether to do blocking I/O

 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.

 * @error: location to store the error occurring, or %NULL to ignore

 *

 * Tries to write to @stream, as with g_output_stream_write() (if

 * @blocking is %TRUE) or g_pollable_output_stream_write_nonblocking()

 * (if @blocking is %FALSE). This can be used to more easily share

 * code between blocking and non-blocking implementations of a method.

 *

 * If @blocking is %FALSE, then @stream must be a

 * #GPollableOutputStream for which

 * g_pollable_output_stream_can_poll() returns %TRUE or else the

 * behavior is undefined. If @blocking is %TRUE, then @stream does not

 * need to be a #GPollableOutputStream.

 *

 * Returns: the number of bytes written, or -1 on error.

 *

 * Since: 2.34

 */

gssize

g_pollable_stream_write (GOutputStream   *stream,

       const void      *buffer,

       gsize            count,

       gboolean         blocking,

       GCancellable    *cancellable,

       GError         **error)

{

  if (blocking)

    {

      return g_output_stream_write (stream,

            buffer, count,

            cancellable, error);

    }

  else

    {

      return g_pollable_output_stream_write_nonblocking (G_POLLABLE_OUTPUT_STREAM (stream),

               buffer, count,

               cancellable, error);

    }

}

/**

 * g_pollable_stream_write_all:

 * @stream: a #GOutputStream.

 * @buffer: (array length=count) (element-type guint8): the buffer

 *   containing the data to write.

 * @count: the number of bytes to write

 * @blocking: whether to do blocking I/O

 * @bytes_written: (out): location to store the number of bytes that was 

 *   written to the stream

 * @cancellable: (nullable): optional #GCancellable object, %NULL to ignore.

 * @error: location to store the error occurring, or %NULL to ignore

 *

 * Tries to write @count bytes to @stream, as with

 * g_output_stream_write_all(), but using g_pollable_stream_write()

 * rather than g_output_stream_write().

 *

 * On a successful write of @count bytes, %TRUE is returned, and

 * @bytes_written is set to @count.

 * 

 * If there is an error during the operation (including

 * %G_IO_ERROR_WOULD_BLOCK in the non-blocking case), %FALSE is

 * returned and @error is set to indicate the error status,

 * @bytes_written is updated to contain the number of bytes written

 * into the stream before the error occurred.

 *

 * As with g_pollable_stream_write(), if @blocking is %FALSE, then

 * @stream must be a #GPollableOutputStream for which

 * g_pollable_output_stream_can_poll() returns %TRUE or else the

 * behavior is undefined. If @blocking is %TRUE, then @stream does not

 * need to be a #GPollableOutputStream.

 *

 * Returns: %TRUE on success, %FALSE if there was an error

 *

 * Since: 2.34

 */

gboolean

g_pollable_stream_write_all (GOutputStream  *stream,

           const void     *buffer,

           gsize           count,

           gboolean        blocking,

           gsize          *bytes_written,

           GCancellable   *cancellable,

           GError        **error)

{

  gsize _bytes_written;

  gssize res;

  _bytes_written = 0;

  while (_bytes_written < count)

    {

      res = g_pollable_stream_write (stream,

             (char *)buffer + _bytes_written,

             count - _bytes_written,

             blocking,

             cancellable, error);

      if (res == -1)

  {

    if (bytes_written)

      *bytes_written = _bytes_written;

    return FALSE;

  }

      if (res == 0)

  g_warning ("Write returned zero without error");

      _bytes_written += res;

    }

  if (bytes_written)

    *bytes_written = _bytes_written;

  return TRUE;

}