/* 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.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/>.
*
* Author: Alexander Larsson <alexl@redhat.com>
*/
#include "config.h"
#include <string.h>
#include "gconverterinputstream.h"
#include "gpollableinputstream.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_ADD_PRIVATE (GConverterInputStream)
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;
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_converter_input_stream_get_instance_private (stream);
}
/**
* 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)
{
if (count != 0)
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);
if (in_buffer != 0)
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;
}