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