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