gio/gconverteroutputstream.c

   1 /* GIO - GLib Input, Output and Streaming Library
   2  *
   3  * Copyright (C) 2009 Red Hat, Inc.
   4  *
   5  * This library is free software; you can redistribute it and/or
   6  * modify it under the terms of the GNU Lesser General Public
   7  * License as published by the Free Software Foundation; either
   8  * version 2 of the License, or (at your option) any later version.
   9  *
  10  * This library is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * Lesser General Public License for more details.
  14  *
  15  * 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  * Author: Alexander Larsson <alexl@redhat.com>
  19  */
  20
  21 #include "config.h"
  22
  23 #include <string.h>
  24
  25 #include "gconverteroutputstream.h"
  26 #include "gpollableoutputstream.h"
  27 #include "gsimpleasyncresult.h"
  28 #include "gcancellable.h"
  29 #include "gioenumtypes.h"
  30 #include "gioerror.h"
  31 #include "glibintl.h"
  32
  33
  34 /**
  35  * SECTION:gconverteroutputstream
  36  * @short_description: Converter Output Stream
  37  * @include: gio/gio.h
  38  * @see_also: #GOutputStream, #GConverter
  39  *
  40  * Converter output stream implements #GOutputStream and allows
  41  * conversion of data of various types during reading.
  42  *
  43  * As of GLib 2.34, #GConverterOutputStream implements
  44  * #GPollableOutputStream.
  45  **/
  46
  47 #define INITIAL_BUFFER_SIZE 4096
  48
  49 typedef struct {
  50   char *data;
  51   gsize start;
  52   gsize end;
  53   gsize size;
  54 } Buffer;
  55
  56 struct _GConverterOutputStreamPrivate {
  57   gboolean at_output_end;
  58   gboolean finished;
  59   GConverter *converter;
  60   Buffer output_buffer; /* To be converted and written */
  61   Buffer converted_buffer; /* Already converted */
  62 };
  63
  64 /* Buffering strategy:
  65  *
  66  * Each time we write we must at least consume some input, or
  67  * return an error. Thus we start with writing all already
  68  * converted data and *then* we start converting (reporting
  69  * an error at any point in this).
  70  *
  71  * Its possible that what the user wrote is not enough data
  72  * for the converter, so we must then buffer it in output_buffer
  73  * and ask for more data, but we want to avoid this as much as
  74  * possible, converting directly from the users buffer.
  75  */
  76
  77 enum {
  78   PROP_0,
  79   PROP_CONVERTER
  80 };
  81
  82 static void   g_converter_output_stream_set_property (GObject        *object,
  83                                                       guint           prop_id,
  84                                                       const GValue   *value,
  85                                                       GParamSpec     *pspec);
  86 static void   g_converter_output_stream_get_property (GObject        *object,
  87                                                       guint           prop_id,
  88                                                       GValue         *value,
  89                                                       GParamSpec     *pspec);
  90 static void   g_converter_output_stream_finalize     (GObject        *object);
  91 static gssize g_converter_output_stream_write        (GOutputStream  *stream,
  92                                                       const void     *buffer,
  93                                                       gsize           count,
  94                                                       GCancellable   *cancellable,
  95                                                       GError        **error);
  96 static gboolean g_converter_output_stream_flush      (GOutputStream  *stream,
  97                                                       GCancellable   *cancellable,
  98                                                       GError        **error);
  99
 100 static gboolean g_converter_output_stream_can_poll          (GPollableOutputStream *stream);
 101 static gboolean g_converter_output_stream_is_writable       (GPollableOutputStream *stream);
 102 static gssize   g_converter_output_stream_write_nonblocking (GPollableOutputStream  *stream,
 103                                                              const void             *buffer,
 104                                                              gsize                  size,
 105                                                              GError               **error);
 106
 107 static GSource *g_converter_output_stream_create_source     (GPollableOutputStream *stream,
 108                                                              GCancellable          *cancellable);
 109
 110 static void g_converter_output_stream_pollable_iface_init (GPollableOutputStreamInterface *iface);
 111
 112 G_DEFINE_TYPE_WITH_CODE (GConverterOutputStream,
 113                          g_converter_output_stream,
 114                          G_TYPE_FILTER_OUTPUT_STREAM,
 115                          G_ADD_PRIVATE (GConverterOutputStream)
 116                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_OUTPUT_STREAM,
 117                                                 g_converter_output_stream_pollable_iface_init))
 118
 119 static void
 120 g_converter_output_stream_class_init (GConverterOutputStreamClass *klass)
 121 {
 122   GObjectClass *object_class;
 123   GOutputStreamClass *istream_class;
 124
 125   object_class = G_OBJECT_CLASS (klass);
 126   object_class->get_property = g_converter_output_stream_get_property;
 127   object_class->set_property = g_converter_output_stream_set_property;
 128   object_class->finalize     = g_converter_output_stream_finalize;
 129
 130   istream_class = G_OUTPUT_STREAM_CLASS (klass);
 131   istream_class->write_fn = g_converter_output_stream_write;
 132   istream_class->flush = g_converter_output_stream_flush;
 133
 134   g_object_class_install_property (object_class,
 135                                    PROP_CONVERTER,
 136                                    g_param_spec_object ("converter",
 137                                                         P_("Converter"),
 138                                                         P_("The converter object"),
 139                                                         G_TYPE_CONVERTER,
 140                                                         G_PARAM_READWRITE|
 141                                                         G_PARAM_CONSTRUCT_ONLY|
 142                                                         G_PARAM_STATIC_STRINGS));
 143
 144 }
 145
 146 static void
 147 g_converter_output_stream_pollable_iface_init (GPollableOutputStreamInterface *iface)
 148 {
 149   iface->can_poll = g_converter_output_stream_can_poll;
 150   iface->is_writable = g_converter_output_stream_is_writable;
 151   iface->write_nonblocking = g_converter_output_stream_write_nonblocking;
 152   iface->create_source = g_converter_output_stream_create_source;
 153 }
 154
 155 static void
 156 g_converter_output_stream_finalize (GObject *object)
 157 {
 158   GConverterOutputStreamPrivate *priv;
 159   GConverterOutputStream        *stream;
 160
 161   stream = G_CONVERTER_OUTPUT_STREAM (object);
 162   priv = stream->priv;
 163
 164   g_free (priv->output_buffer.data);
 165   g_free (priv->converted_buffer.data);
 166   if (priv->converter)
 167     g_object_unref (priv->converter);
 168
 169   G_OBJECT_CLASS (g_converter_output_stream_parent_class)->finalize (object);
 170 }
 171
 172 static void
 173 g_converter_output_stream_set_property (GObject      *object,
 174                                        guint         prop_id,
 175                                        const GValue *value,
 176                                        GParamSpec   *pspec)
 177 {
 178   GConverterOutputStream *cstream;
 179
 180   cstream = G_CONVERTER_OUTPUT_STREAM (object);
 181
 182    switch (prop_id)
 183     {
 184     case PROP_CONVERTER:
 185       cstream->priv->converter = g_value_dup_object (value);
 186       break;
 187
 188     default:
 189       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 190       break;
 191     }
 192
 193 }
 194
 195 static void
 196 g_converter_output_stream_get_property (GObject    *object,
 197                                        guint       prop_id,
 198                                        GValue     *value,
 199                                        GParamSpec *pspec)
 200 {
 201   GConverterOutputStreamPrivate *priv;
 202   GConverterOutputStream        *cstream;
 203
 204   cstream = G_CONVERTER_OUTPUT_STREAM (object);
 205   priv = cstream->priv;
 206
 207   switch (prop_id)
 208     {
 209     case PROP_CONVERTER:
 210       g_value_set_object (value, priv->converter);
 211       break;
 212
 213     default:
 214       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 215       break;
 216     }
 217 }
 218
 219 static void
 220 g_converter_output_stream_init (GConverterOutputStream *stream)
 221 {
 222   stream->priv = g_converter_output_stream_get_instance_private (stream);
 223 }
 224
 225 /**
 226  * g_converter_output_stream_new:
 227  * @base_stream: a #GOutputStream
 228  * @converter: a #GConverter
 229  *
 230  * Creates a new converter output stream for the @base_stream.
 231  *
 232  * Returns: a new #GOutputStream.
 233  **/
 234 GOutputStream *
 235 g_converter_output_stream_new (GOutputStream *base_stream,
 236                                GConverter    *converter)
 237 {
 238   GOutputStream *stream;
 239
 240   g_return_val_if_fail (G_IS_OUTPUT_STREAM (base_stream), NULL);
 241
 242   stream = g_object_new (G_TYPE_CONVERTER_OUTPUT_STREAM,
 243                          "base-stream", base_stream,
 244                          "converter", converter,
 245                          NULL);
 246
 247   return stream;
 248 }
 249
 250 static gsize
 251 buffer_data_size (Buffer *buffer)
 252 {
 253   return buffer->end - buffer->start;
 254 }
 255
 256 static gsize
 257 buffer_tailspace (Buffer *buffer)
 258 {
 259   return buffer->size - buffer->end;
 260 }
 261
 262 static char *
 263 buffer_data (Buffer *buffer)
 264 {
 265   return buffer->data + buffer->start;
 266 }
 267
 268 static void
 269 buffer_consumed (Buffer *buffer,
 270                  gsize count)
 271 {
 272   buffer->start += count;
 273   if (buffer->start == buffer->end)
 274     buffer->start = buffer->end = 0;
 275 }
 276
 277 static void
 278 compact_buffer (Buffer *buffer)
 279 {
 280   gsize in_buffer;
 281
 282   in_buffer = buffer_data_size (buffer);
 283   memmove (buffer->data,
 284            buffer->data + buffer->start,
 285            in_buffer);
 286   buffer->end -= buffer->start;
 287   buffer->start = 0;
 288 }
 289
 290 static void
 291 grow_buffer (Buffer *buffer)
 292 {
 293   char *data;
 294   gsize size, in_buffer;
 295
 296   if (buffer->size == 0)
 297     size = INITIAL_BUFFER_SIZE;
 298   else
 299     size = buffer->size * 2;
 300
 301   data = g_malloc (size);
 302   in_buffer = buffer_data_size (buffer);
 303
 304   memcpy (data,
 305           buffer->data + buffer->start,
 306           in_buffer);
 307   g_free (buffer->data);
 308   buffer->data = data;
 309   buffer->end -= buffer->start;
 310   buffer->start = 0;
 311   buffer->size = size;
 312 }
 313
 314 /* Ensures that the buffer can fit at_least_size bytes,
 315  * *including* the current in-buffer data */
 316 static void
 317 buffer_ensure_space (Buffer *buffer,
 318                      gsize at_least_size)
 319 {
 320   gsize in_buffer, left_to_fill;
 321
 322   in_buffer = buffer_data_size (buffer);
 323
 324   if (in_buffer >= at_least_size)
 325     return;
 326
 327   left_to_fill = buffer_tailspace (buffer);
 328
 329   if (in_buffer + left_to_fill >= at_least_size)
 330     {
 331       /* We fit in remaining space at end */
 332       /* If the copy is small, compact now anyway so we can fill more */
 333       if (in_buffer < 256)
 334         compact_buffer (buffer);
 335     }
 336   else if (buffer->size >= at_least_size)
 337     {
 338       /* We fit, but only if we compact */
 339       compact_buffer (buffer);
 340     }
 341   else
 342     {
 343       /* Need to grow buffer */
 344       while (buffer->size < at_least_size)
 345         grow_buffer (buffer);
 346     }
 347 }
 348
 349 static void
 350 buffer_append (Buffer *buffer,
 351                const char *data,
 352                gsize data_size)
 353 {
 354   buffer_ensure_space (buffer,
 355                        buffer_data_size (buffer) + data_size);
 356   memcpy (buffer->data + buffer->end, data, data_size);
 357   buffer->end += data_size;
 358 }
 359
 360
 361 static gboolean
 362 flush_buffer (GConverterOutputStream *stream,
 363               gboolean                blocking,
 364               GCancellable           *cancellable,
 365               GError                **error)
 366 {
 367   GConverterOutputStreamPrivate *priv;
 368   GOutputStream *base_stream;
 369   gsize nwritten;
 370   gsize available;
 371   gboolean res;
 372
 373   priv = stream->priv;
 374
 375   base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 376
 377   available = buffer_data_size (&priv->converted_buffer);
 378   if (available > 0)
 379     {
 380       res = g_pollable_stream_write_all (base_stream,
 381                                          buffer_data (&priv->converted_buffer),
 382                                          available,
 383                                          blocking,
 384                                          &nwritten,
 385                                          cancellable,
 386                                          error);
 387       buffer_consumed (&priv->converted_buffer, nwritten);
 388       return res;
 389     }
 390   return TRUE;
 391 }
 392
 393
 394 static gssize
 395 write_internal (GOutputStream  *stream,
 396                 const void     *buffer,
 397                 gsize           count,
 398                 gboolean        blocking,
 399                 GCancellable   *cancellable,
 400                 GError        **error)
 401 {
 402   GConverterOutputStream *cstream;
 403   GConverterOutputStreamPrivate *priv;
 404   gssize retval;
 405   GConverterResult res;
 406   gsize bytes_read;
 407   gsize bytes_written;
 408   GError *my_error;
 409   const char *to_convert;
 410   gsize to_convert_size, converted_bytes;
 411   gboolean converting_from_buffer;
 412
 413   cstream = G_CONVERTER_OUTPUT_STREAM (stream);
 414   priv = cstream->priv;
 415
 416   /* Write out all available pre-converted data and fail if
 417      not possible */
 418   if (!flush_buffer (cstream, blocking, cancellable, error))
 419     return -1;
 420
 421   if (priv->finished)
 422     return 0;
 423
 424   /* Convert as much as possible */
 425   if (buffer_data_size (&priv->output_buffer) > 0)
 426     {
 427       converting_from_buffer = TRUE;
 428       buffer_append (&priv->output_buffer, buffer, count);
 429       to_convert = buffer_data (&priv->output_buffer);
 430       to_convert_size = buffer_data_size (&priv->output_buffer);
 431     }
 432   else
 433     {
 434       converting_from_buffer = FALSE;
 435       to_convert = buffer;
 436       to_convert_size = count;
 437     }
 438
 439   /* Ensure we have *some* initial target space */
 440   buffer_ensure_space (&priv->converted_buffer, to_convert_size);
 441
 442   converted_bytes = 0;
 443   while (!priv->finished && converted_bytes < to_convert_size)
 444     {
 445       /* Ensure we have *some* target space */
 446       if (buffer_tailspace (&priv->converted_buffer) == 0)
 447         grow_buffer (&priv->converted_buffer);
 448
 449       /* Try to convert to our buffer */
 450       my_error = NULL;
 451       res = g_converter_convert (priv->converter,
 452                                  to_convert + converted_bytes,
 453                                  to_convert_size - converted_bytes,
 454                                  buffer_data (&priv->converted_buffer) + buffer_data_size (&priv->converted_buffer),
 455                                  buffer_tailspace (&priv->converted_buffer),
 456                                  0,
 457                                  &bytes_read,
 458                                  &bytes_written,
 459                                  &my_error);
 460
 461       if (res != G_CONVERTER_ERROR)
 462         {
 463           priv->converted_buffer.end += bytes_written;
 464           converted_bytes += bytes_read;
 465
 466           if (res == G_CONVERTER_FINISHED)
 467             priv->finished = TRUE;
 468         }
 469       else
 470         {
 471           /* No-space errors can be handled locally: */
 472           if (g_error_matches (my_error,
 473                                G_IO_ERROR,
 474                                G_IO_ERROR_NO_SPACE))
 475             {
 476               /* Need more destination space, grow it
 477                * Note: if we actually grow the buffer (as opposed to compacting it),
 478                * this will double the size, not just add one byte. */
 479               buffer_ensure_space (&priv->converted_buffer,
 480                                    priv->converted_buffer.size + 1);
 481               g_error_free (my_error);
 482               continue;
 483             }
 484
 485           if (converted_bytes > 0)
 486             {
 487               /* We got an conversion error, but we did convert some bytes before
 488                  that, so handle those before reporting the error */
 489               g_error_free (my_error);
 490               break;
 491             }
 492
 493           if (g_error_matches (my_error,
 494                                G_IO_ERROR,
 495                                G_IO_ERROR_PARTIAL_INPUT))
 496             {
 497               /* Consume everything to buffer that we append to next time
 498                  we write */
 499               if (!converting_from_buffer)
 500                 buffer_append (&priv->output_buffer, buffer, count);
 501               /* in the converting_from_buffer case we already appended this */
 502
 503               g_error_free (my_error);
 504               return count; /* consume everything */
 505             }
 506
 507           /* Converted no data and got an normal error, return it */
 508           g_propagate_error (error, my_error);
 509           return -1;
 510         }
 511     }
 512
 513   if (converting_from_buffer)
 514     {
 515       buffer_consumed (&priv->output_buffer, converted_bytes);
 516       retval = count;
 517     }
 518   else
 519     retval = converted_bytes;
 520
 521   /* We now successfully consumed retval bytes, so we can't return an error,
 522      even if writing this to the base stream fails. If it does we'll just
 523      stop early and report this error when we try again on the next
 524      write call. */
 525   flush_buffer (cstream, blocking, cancellable, NULL);
 526
 527   return retval;
 528 }
 529
 530 static gssize
 531 g_converter_output_stream_write (GOutputStream  *stream,
 532                                  const void     *buffer,
 533                                  gsize           count,
 534                                  GCancellable   *cancellable,
 535                                  GError        **error)
 536 {
 537   return write_internal (stream, buffer, count, TRUE, cancellable, error);
 538 }
 539
 540 static gboolean
 541 g_converter_output_stream_flush (GOutputStream  *stream,
 542                                  GCancellable   *cancellable,
 543                                  GError        **error)
 544 {
 545   GConverterOutputStream *cstream;
 546   GConverterOutputStreamPrivate *priv;
 547   GConverterResult res;
 548   GError *my_error;
 549   gboolean is_closing;
 550   gboolean flushed;
 551   gsize bytes_read;
 552   gsize bytes_written;
 553
 554   cstream = G_CONVERTER_OUTPUT_STREAM (stream);
 555   priv = cstream->priv;
 556
 557   is_closing = g_output_stream_is_closing (stream);
 558
 559   /* Write out all available pre-converted data and fail if
 560      not possible */
 561   if (!flush_buffer (cstream, TRUE, cancellable, error))
 562     return FALSE;
 563
 564   /* Ensure we have *some* initial target space */
 565   buffer_ensure_space (&priv->converted_buffer, 1);
 566
 567   /* Convert whole buffer */
 568   flushed = FALSE;
 569   while (!priv->finished && !flushed)
 570     {
 571       /* Ensure we have *some* target space */
 572       if (buffer_tailspace (&priv->converted_buffer) == 0)
 573         grow_buffer (&priv->converted_buffer);
 574
 575       /* Try to convert to our buffer */
 576       my_error = NULL;
 577       res = g_converter_convert (priv->converter,
 578                                  buffer_data (&priv->output_buffer),
 579                                  buffer_data_size (&priv->output_buffer),
 580                                  buffer_data (&priv->converted_buffer) + buffer_data_size (&priv->converted_buffer),
 581                                  buffer_tailspace (&priv->converted_buffer),
 582                                  is_closing ? G_CONVERTER_INPUT_AT_END : G_CONVERTER_FLUSH,
 583                                  &bytes_read,
 584                                  &bytes_written,
 585                                  &my_error);
 586
 587       if (res != G_CONVERTER_ERROR)
 588         {
 589           priv->converted_buffer.end += bytes_written;
 590           buffer_consumed (&priv->output_buffer, bytes_read);
 591
 592           if (res == G_CONVERTER_FINISHED)
 593             priv->finished = TRUE;
 594           if (!is_closing &&
 595               res == G_CONVERTER_FLUSHED)
 596             {
 597               /* Should not have retured FLUSHED with input left */
 598               g_assert (buffer_data_size (&priv->output_buffer) == 0);
 599               flushed = TRUE;
 600             }
 601         }
 602       else
 603         {
 604           /* No-space errors can be handled locally: */
 605           if (g_error_matches (my_error,
 606                                G_IO_ERROR,
 607                                G_IO_ERROR_NO_SPACE))
 608             {
 609               /* Need more destination space, grow it
 610                * Note: if we actually grow the buffer (as opposed to compacting it),
 611                * this will double the size, not just add one byte. */
 612               buffer_ensure_space (&priv->converted_buffer,
 613                                    priv->converted_buffer.size + 1);
 614               g_error_free (my_error);
 615               continue;
 616             }
 617
 618           /* Any other error, including PARTIAL_INPUT can't be fixed by now
 619              and is an error */
 620           g_propagate_error (error, my_error);
 621           return FALSE;
 622         }
 623     }
 624
 625   /* Now write all converted data to base stream */
 626   if (!flush_buffer (cstream, TRUE, cancellable, error))
 627     return FALSE;
 628
 629   return TRUE;
 630 }
 631
 632 static gboolean
 633 g_converter_output_stream_can_poll (GPollableOutputStream *stream)
 634 {
 635   GOutputStream *base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 636
 637   return (G_IS_POLLABLE_OUTPUT_STREAM (base_stream) &&
 638           g_pollable_output_stream_can_poll (G_POLLABLE_OUTPUT_STREAM (base_stream)));
 639 }
 640
 641 static gboolean
 642 g_converter_output_stream_is_writable (GPollableOutputStream *stream)
 643 {
 644   GOutputStream *base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 645
 646   return g_pollable_output_stream_is_writable (G_POLLABLE_OUTPUT_STREAM (base_stream));
 647 }
 648
 649 static gssize
 650 g_converter_output_stream_write_nonblocking (GPollableOutputStream  *stream,
 651                                              const void             *buffer,
 652                                              gsize                   count,
 653                                              GError                **error)
 654 {
 655   return write_internal (G_OUTPUT_STREAM (stream), buffer, count, FALSE,
 656                          NULL, error);
 657 }
 658
 659 static GSource *
 660 g_converter_output_stream_create_source (GPollableOutputStream *stream,
 661                                          GCancellable          *cancellable)
 662 {
 663   GOutputStream *base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 664   GSource *base_source, *pollable_source;
 665
 666   base_source = g_pollable_output_stream_create_source (G_POLLABLE_OUTPUT_STREAM (base_stream), NULL);
 667   pollable_source = g_pollable_source_new_full (stream, base_source,
 668                                                 cancellable);
 669   g_source_unref (base_source);
 670
 671   return pollable_source;
 672 }
 673
 674 /**
 675  * g_converter_output_stream_get_converter:
 676  * @converter_stream: a #GConverterOutputStream
 677  *
 678  * Gets the #GConverter that is used by @converter_stream.
 679  *
 680  * Returns: (transfer none): the converter of the converter output stream
 681  *
 682  * Since: 2.24
 683  */
 684 GConverter *
 685 g_converter_output_stream_get_converter (GConverterOutputStream *converter_stream)
 686 {
 687   return converter_stream->priv->converter;
 688 }