glib/gio/gconverterinputstream.c

/* GIO - GLib Input, Output and Streaming Library
 *
 * Copyright (C) 2009 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
 * version 2 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, write to the
 * Free Software Foundation, Inc., 59 Temple Place, Suite 330,
 * Boston, MA 02111-1307, USA.
 *
 * Author: Alexander Larsson <alexl@redhat.com>
 */

#include "config.h"

#include <string.h>

#include "gconverterinputstream.h"
#include "gpollableinputstream.h"
#include "gsimpleasyncresult.h"
#include "gcancellable.h"
#include "gioenumtypes.h"
#include "gioerror.h"
#include "glibintl.h"


/**
 * SECTION:gconverterinputstream
 * @short_description: Converter Input Stream
 * @include: gio/gio.h
 * @see_also: #GInputStream, #GConverter
 *
 * Converter input stream implements #GInputStream and allows
 * conversion of data of various types during reading.
 *
 * As of GLib 2.34, #GConverterInputStream implements
 * #GPollableInputStream.
 **/

#define INITIAL_BUFFER_SIZE 4096

typedef struct {
  char *data;
  gsize start;
  gsize end;
  gsize size;
} Buffer;

struct _GConverterInputStreamPrivate {
  gboolean at_input_end;
  gboolean finished;
  gboolean need_input;
  GConverter *converter;
  Buffer input_buffer;
  Buffer converted_buffer;
};

enum {
  PROP_0,
  PROP_CONVERTER
};

static void   g_converter_input_stream_set_property (GObject       *object,
						     guint          prop_id,
						     const GValue  *value,
						     GParamSpec    *pspec);
static void   g_converter_input_stream_get_property (GObject       *object,
						     guint          prop_id,
						     GValue        *value,
						     GParamSpec    *pspec);
static void   g_converter_input_stream_finalize     (GObject       *object);
static gssize g_converter_input_stream_read         (GInputStream  *stream,
						     void          *buffer,
						     gsize          count,
						     GCancellable  *cancellable,
						     GError       **error);

static gboolean g_converter_input_stream_can_poll         (GPollableInputStream *stream);
static gboolean g_converter_input_stream_is_readable      (GPollableInputStream *stream);
static gssize   g_converter_input_stream_read_nonblocking (GPollableInputStream  *stream,
							   void                  *buffer,
							   gsize                  size,
							   GError               **error);

static GSource *g_converter_input_stream_create_source    (GPollableInputStream *stream,
							   GCancellable          *cancellable);

static void g_converter_input_stream_pollable_iface_init  (GPollableInputStreamInterface *iface);

G_DEFINE_TYPE_WITH_CODE (GConverterInputStream,
			 g_converter_input_stream,
			 G_TYPE_FILTER_INPUT_STREAM,
			 G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM,
						g_converter_input_stream_pollable_iface_init);
			 )

static void
g_converter_input_stream_class_init (GConverterInputStreamClass *klass)
{
  GObjectClass *object_class;
  GInputStreamClass *istream_class;

  g_type_class_add_private (klass, sizeof (GConverterInputStreamPrivate));

  object_class = G_OBJECT_CLASS (klass);
  object_class->get_property = g_converter_input_stream_get_property;
  object_class->set_property = g_converter_input_stream_set_property;
  object_class->finalize     = g_converter_input_stream_finalize;

  istream_class = G_INPUT_STREAM_CLASS (klass);
  istream_class->read_fn = g_converter_input_stream_read;

  g_object_class_install_property (object_class,
				   PROP_CONVERTER,
				   g_param_spec_object ("converter",
							P_("Converter"),
							P_("The converter object"),
							G_TYPE_CONVERTER,
							G_PARAM_READWRITE|
							G_PARAM_CONSTRUCT_ONLY|
							G_PARAM_STATIC_STRINGS));

}

static void
g_converter_input_stream_pollable_iface_init (GPollableInputStreamInterface *iface)
{
  iface->can_poll = g_converter_input_stream_can_poll;
  iface->is_readable = g_converter_input_stream_is_readable;
  iface->read_nonblocking = g_converter_input_stream_read_nonblocking;
  iface->create_source = g_converter_input_stream_create_source;
}

static void
g_converter_input_stream_finalize (GObject *object)
{
  GConverterInputStreamPrivate *priv;
  GConverterInputStream        *stream;

  stream = G_CONVERTER_INPUT_STREAM (object);
  priv = stream->priv;

  g_free (priv->input_buffer.data);
  g_free (priv->converted_buffer.data);
  if (priv->converter)
    g_object_unref (priv->converter);

  G_OBJECT_CLASS (g_converter_input_stream_parent_class)->finalize (object);
}

static void
g_converter_input_stream_set_property (GObject      *object,
				       guint         prop_id,
				       const GValue *value,
				       GParamSpec   *pspec)
{
  GConverterInputStream *cstream;

  cstream = G_CONVERTER_INPUT_STREAM (object);

   switch (prop_id)
    {
    case PROP_CONVERTER:
      cstream->priv->converter = g_value_dup_object (value);
      break;

    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
    }

}

static void
g_converter_input_stream_get_property (GObject    *object,
				       guint       prop_id,
				       GValue     *value,
				       GParamSpec *pspec)
{
  GConverterInputStreamPrivate *priv;
  GConverterInputStream        *cstream;

  cstream = G_CONVERTER_INPUT_STREAM (object);
  priv = cstream->priv;

  switch (prop_id)
    {
    case PROP_CONVERTER:
      g_value_set_object (value, priv->converter);
      break;

    default:
      G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
      break;
    }

}
static void
g_converter_input_stream_init (GConverterInputStream *stream)
{
  stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
					      G_TYPE_CONVERTER_INPUT_STREAM,
					      GConverterInputStreamPrivate);
}

/**
 * g_converter_input_stream_new:
 * @base_stream: a #GInputStream
 * @converter: a #GConverter
 *
 * Creates a new converter input stream for the @base_stream.
 *
 * Returns: a new #GInputStream.
 **/
GInputStream *
g_converter_input_stream_new (GInputStream *base_stream,
                              GConverter   *converter)
{
  GInputStream *stream;

  g_return_val_if_fail (G_IS_INPUT_STREAM (base_stream), NULL);

  stream = g_object_new (G_TYPE_CONVERTER_INPUT_STREAM,
                         "base-stream", base_stream,
			 "converter", converter,
			 NULL);

  return stream;
}

static gsize
buffer_data_size (Buffer *buffer)
{
  return buffer->end - buffer->start;
}

static gsize
buffer_tailspace (Buffer *buffer)
{
  return buffer->size - buffer->end;
}

static char *
buffer_data (Buffer *buffer)
{
  return buffer->data + buffer->start;
}

static void
buffer_consumed (Buffer *buffer,
		 gsize count)
{
  buffer->start += count;
  if (buffer->start == buffer->end)
    buffer->start = buffer->end = 0;
}

static void
buffer_read (Buffer *buffer,
	     char *dest,
	     gsize count)
{
  memcpy (dest, buffer->data + buffer->start, count);
  buffer_consumed (buffer, count);
}

static void
compact_buffer (Buffer *buffer)
{
  gsize in_buffer;

  in_buffer = buffer_data_size (buffer);
  memmove (buffer->data,
	   buffer->data + buffer->start,
	   in_buffer);
  buffer->end -= buffer->start;
  buffer->start = 0;
}

static void
grow_buffer (Buffer *buffer)
{
  char *data;
  gsize size, in_buffer;

  if (buffer->size == 0)
    size = INITIAL_BUFFER_SIZE;
  else
    size = buffer->size * 2;

  data = g_malloc (size);
  in_buffer = buffer_data_size (buffer);

  memcpy (data,
	  buffer->data + buffer->start,
	  in_buffer);
  g_free (buffer->data);
  buffer->data = data;
  buffer->end -= buffer->start;
  buffer->start = 0;
  buffer->size = size;
}

/* Ensures that the buffer can fit at_least_size bytes,
 * *including* the current in-buffer data */
static void
buffer_ensure_space (Buffer *buffer,
		     gsize at_least_size)
{
  gsize in_buffer, left_to_fill;

  in_buffer = buffer_data_size (buffer);

  if (in_buffer >= at_least_size)
    return;

  left_to_fill = buffer_tailspace (buffer);

  if (in_buffer + left_to_fill >= at_least_size)
    {
      /* We fit in remaining space at end */
      /* If the copy is small, compact now anyway so we can fill more */
      if (in_buffer < 256)
	compact_buffer (buffer);
    }
  else if (buffer->size >= at_least_size)
    {
      /* We fit, but only if we compact */
      compact_buffer (buffer);
    }
  else
    {
      /* Need to grow buffer */
      while (buffer->size < at_least_size)
	grow_buffer (buffer);
    }
}

static gssize
fill_input_buffer (GConverterInputStream  *stream,
		   gsize                   at_least_size,
		   gboolean                blocking,
		   GCancellable           *cancellable,
		   GError                **error)
{
  GConverterInputStreamPrivate *priv;
  GInputStream *base_stream;
  gssize nread;

  priv = stream->priv;

  buffer_ensure_space (&priv->input_buffer, at_least_size);

  base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
  nread = g_pollable_stream_read (base_stream,
				  priv->input_buffer.data + priv->input_buffer.end,
				  buffer_tailspace (&priv->input_buffer),
				  blocking,
				  cancellable,
				  error);

  if (nread > 0)
    {
      priv->input_buffer.end += nread;
      priv->need_input = FALSE;
    }

  return nread;
}


static gssize
read_internal (GInputStream *stream,
	       void         *buffer,
	       gsize         count,
	       gboolean      blocking,
	       GCancellable *cancellable,
	       GError      **error)
{
  GConverterInputStream *cstream;
  GConverterInputStreamPrivate *priv;
  gsize available, total_bytes_read;
  gssize nread;
  GConverterResult res;
  gsize bytes_read;
  gsize bytes_written;
  GError *my_error;
  GError *my_error2;

  cstream = G_CONVERTER_INPUT_STREAM (stream);
  priv = cstream->priv;

  available = buffer_data_size (&priv->converted_buffer);

  if (available > 0 &&
      count <= available)
    {
      /* Converted data available, return that */
      buffer_read (&priv->converted_buffer, buffer, count);
      return count;
    }

  /* Full request not available, read all currently available and request
     refill/conversion for more */

  buffer_read (&priv->converted_buffer, buffer, available);

  total_bytes_read = available;
  buffer = (char *) buffer + available;
  count -= available;

  /* If there is no data to convert, and no pre-converted data,
     do some i/o for more input */
  if (buffer_data_size (&priv->input_buffer) == 0 &&
      total_bytes_read == 0 &&
      !priv->at_input_end)
    {
      nread = fill_input_buffer (cstream, count, blocking, cancellable, error);
      if (nread < 0)
	return -1;
      if (nread == 0)
	priv->at_input_end = TRUE;
    }

  /* First try to convert any available data (or state) directly to the user buffer: */
  if (!priv->finished)
    {
      my_error = NULL;
      res = g_converter_convert (priv->converter,
				 buffer_data (&priv->input_buffer),
				 buffer_data_size (&priv->input_buffer),
				 buffer, count,
				 priv->at_input_end ? G_CONVERTER_INPUT_AT_END : 0,
				 &bytes_read,
				 &bytes_written,
				 &my_error);
      if (res != G_CONVERTER_ERROR)
	{
	  total_bytes_read += bytes_written;
	  buffer_consumed (&priv->input_buffer, bytes_read);
	  if (res == G_CONVERTER_FINISHED)
	    priv->finished = TRUE; /* We're done converting */
	}
      else if (total_bytes_read == 0 &&
	       !g_error_matches (my_error,
				 G_IO_ERROR,
				 G_IO_ERROR_PARTIAL_INPUT) &&
	       !g_error_matches (my_error,
				 G_IO_ERROR,
				 G_IO_ERROR_NO_SPACE))
	{
	  /* No previously read data and no "special" error, return error */
	  g_propagate_error (error, my_error);
	  return -1;
	}
      else
	g_error_free (my_error);
    }

  /* We had some pre-converted data and/or we converted directly to the
     user buffer */
  if (total_bytes_read > 0)
    return total_bytes_read;

  /* If there is no more to convert, return EOF */
  if (priv->finished)
    {
      g_assert (buffer_data_size (&priv->converted_buffer) == 0);
      return 0;
    }

  /* There was "complexity" in the straight-to-buffer conversion,
   * convert to our own buffer and write from that.
   * At this point we didn't produce any data into @buffer.
   */

  /* Ensure we have *some* initial target space */
  buffer_ensure_space (&priv->converted_buffer, count);

  while (TRUE)
    {
      g_assert (!priv->finished);

      /* Try to convert to our buffer */
      my_error = NULL;
      res = g_converter_convert (priv->converter,
				 buffer_data (&priv->input_buffer),
				 buffer_data_size (&priv->input_buffer),
				 buffer_data (&priv->converted_buffer),
				 buffer_tailspace (&priv->converted_buffer),
				 priv->at_input_end ? G_CONVERTER_INPUT_AT_END : 0,
				 &bytes_read,
				 &bytes_written,
				 &my_error);
      if (res != G_CONVERTER_ERROR)
	{
	  priv->converted_buffer.end += bytes_written;
	  buffer_consumed (&priv->input_buffer, bytes_read);

	  /* Maybe we consumed without producing any output */
	  if (buffer_data_size (&priv->converted_buffer) == 0 && res != G_CONVERTER_FINISHED)
	    continue; /* Convert more */

	  if (res == G_CONVERTER_FINISHED)
	    priv->finished = TRUE;

	  total_bytes_read = MIN (count, buffer_data_size (&priv->converted_buffer));
	  buffer_read (&priv->converted_buffer, buffer, total_bytes_read);

	  g_assert (priv->finished || total_bytes_read > 0);

	  return total_bytes_read;
	}

      /* There was some kind of error filling our buffer */

      if (g_error_matches (my_error,
			   G_IO_ERROR,
			   G_IO_ERROR_PARTIAL_INPUT) &&
	  !priv->at_input_end)
	{
	  /* Need more data */
	  my_error2 = NULL;
	  nread = fill_input_buffer (cstream,
				     buffer_data_size (&priv->input_buffer) + 4096,
				     blocking,
				     cancellable,
				     &my_error2);
	  if (nread < 0)
	    {
	      /* Can't read any more data, return that error */
	      g_error_free (my_error);
	      g_propagate_error (error, my_error2);
	      priv->need_input = TRUE;
	      return -1;
	    }
	  else if (nread == 0)
	    {
	      /* End of file, try INPUT_AT_END */
	      priv->at_input_end = TRUE;
	    }
	  g_error_free (my_error);
	  continue;
	}

      if (g_error_matches (my_error,
			   G_IO_ERROR,
			   G_IO_ERROR_NO_SPACE))
	{
	  /* Need more destination space, grow it
	   * Note: if we actually grow the buffer (as opposed to compacting it),
	   * this will double the size, not just add one byte. */
	  buffer_ensure_space (&priv->converted_buffer,
			       priv->converted_buffer.size + 1);
	  g_error_free (my_error);
	  continue;
	}

      /* Any other random error, return it */
      g_propagate_error (error, my_error);
      return -1;
    }

  g_assert_not_reached ();
}

static gssize
g_converter_input_stream_read (GInputStream *stream,
			       void         *buffer,
			       gsize         count,
			       GCancellable *cancellable,
			       GError      **error)
{
  return read_internal (stream, buffer, count, TRUE, cancellable, error);
}

static gboolean
g_converter_input_stream_can_poll (GPollableInputStream *stream)
{
  GInputStream *base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;

  return (G_IS_POLLABLE_INPUT_STREAM (base_stream) &&
	  g_pollable_input_stream_can_poll (G_POLLABLE_INPUT_STREAM (base_stream)));
}

static gboolean
g_converter_input_stream_is_readable (GPollableInputStream *stream)
{
  GInputStream *base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
  GConverterInputStream *cstream = G_CONVERTER_INPUT_STREAM (stream);

  if (buffer_data_size (&cstream->priv->converted_buffer))
    return TRUE;
  else if (buffer_data_size (&cstream->priv->input_buffer) &&
	   !cstream->priv->need_input)
    return TRUE;
  else
    return g_pollable_input_stream_is_readable (G_POLLABLE_INPUT_STREAM (base_stream));
}

static gssize
g_converter_input_stream_read_nonblocking (GPollableInputStream  *stream,
					   void                  *buffer,
					   gsize                  count,
					   GError               **error)
{
  return read_internal (G_INPUT_STREAM (stream), buffer, count,
			FALSE, NULL, error);
}

static GSource *
g_converter_input_stream_create_source (GPollableInputStream *stream,
					GCancellable         *cancellable)
{
  GInputStream *base_stream = G_FILTER_INPUT_STREAM (stream)->base_stream;
  GSource *base_source, *pollable_source;

  if (g_pollable_input_stream_is_readable (stream))
    base_source = g_timeout_source_new (0);
  else
    base_source = g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM (base_stream), NULL);

  pollable_source = g_pollable_source_new_full (stream, base_source,
						cancellable);
  g_source_unref (base_source);

  return pollable_source;
}


/**
 * g_converter_input_stream_get_converter:
 * @converter_stream: a #GConverterInputStream
 *
 * Gets the #GConverter that is used by @converter_stream.
 *
 * Returns: (transfer none): the converter of the converter input stream
 *
 * Since: 2.24
 */
GConverter *
g_converter_input_stream_get_converter (GConverterInputStream *converter_stream)
{
  return converter_stream->priv->converter;
}