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   memcpy (data,
 304           buffer->data + buffer->start,
 305           in_buffer);
 306   g_free (buffer->data);
 307   buffer->data = data;
 308   buffer->end -= buffer->start;
 309   buffer->start = 0;
 310   buffer->size = size;
 311 }
 312
 313 /* Ensures that the buffer can fit at_least_size bytes,
 314  * *including* the current in-buffer data */
 315 static void
 316 buffer_ensure_space (Buffer *buffer,
 317                      gsize at_least_size)
 318 {
 319   gsize in_buffer, left_to_fill;
 320
 321   in_buffer = buffer_data_size (buffer);
 322
 323   if (in_buffer >= at_least_size)
 324     return;
 325
 326   left_to_fill = buffer_tailspace (buffer);
 327
 328   if (in_buffer + left_to_fill >= at_least_size)
 329     {
 330       /* We fit in remaining space at end */
 331       /* If the copy is small, compact now anyway so we can fill more */
 332       if (in_buffer < 256)
 333         compact_buffer (buffer);
 334     }
 335   else if (buffer->size >= at_least_size)
 336     {
 337       /* We fit, but only if we compact */
 338       compact_buffer (buffer);
 339     }
 340   else
 341     {
 342       /* Need to grow buffer */
 343       while (buffer->size < at_least_size)
 344         grow_buffer (buffer);
 345     }
 346 }
 347
 348 static void
 349 buffer_append (Buffer *buffer,
 350                const char *data,
 351                gsize data_size)
 352 {
 353   buffer_ensure_space (buffer,
 354                        buffer_data_size (buffer) + data_size);
 355   memcpy (buffer->data + buffer->end, data, data_size);
 356   buffer->end += data_size;
 357 }
 358
 359
 360 static gboolean
 361 flush_buffer (GConverterOutputStream *stream,
 362               gboolean                blocking,
 363               GCancellable           *cancellable,
 364               GError                **error)
 365 {
 366   GConverterOutputStreamPrivate *priv;
 367   GOutputStream *base_stream;
 368   gsize nwritten;
 369   gsize available;
 370   gboolean res;
 371
 372   priv = stream->priv;
 373
 374   base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 375
 376   available = buffer_data_size (&priv->converted_buffer);
 377   if (available > 0)
 378     {
 379       res = g_pollable_stream_write_all (base_stream,
 380                                          buffer_data (&priv->converted_buffer),
 381                                          available,
 382                                          blocking,
 383                                          &nwritten,
 384                                          cancellable,
 385                                          error);
 386       buffer_consumed (&priv->converted_buffer, nwritten);
 387       return res;
 388     }
 389   return TRUE;
 390 }
 391
 392
 393 static gssize
 394 write_internal (GOutputStream  *stream,
 395                 const void     *buffer,
 396                 gsize           count,
 397                 gboolean        blocking,
 398                 GCancellable   *cancellable,
 399                 GError        **error)
 400 {
 401   GConverterOutputStream *cstream;
 402   GConverterOutputStreamPrivate *priv;
 403   gssize retval;
 404   GConverterResult res;
 405   gsize bytes_read;
 406   gsize bytes_written;
 407   GError *my_error;
 408   const char *to_convert;
 409   gsize to_convert_size, converted_bytes;
 410   gboolean converting_from_buffer;
 411
 412   cstream = G_CONVERTER_OUTPUT_STREAM (stream);
 413   priv = cstream->priv;
 414
 415   /* Write out all available pre-converted data and fail if
 416      not possible */
 417   if (!flush_buffer (cstream, blocking, cancellable, error))
 418     return -1;
 419
 420   if (priv->finished)
 421     return 0;
 422
 423   /* Convert as much as possible */
 424   if (buffer_data_size (&priv->output_buffer) > 0)
 425     {
 426       converting_from_buffer = TRUE;
 427       buffer_append (&priv->output_buffer, buffer, count);
 428       to_convert = buffer_data (&priv->output_buffer);
 429       to_convert_size = buffer_data_size (&priv->output_buffer);
 430     }
 431   else
 432     {
 433       converting_from_buffer = FALSE;
 434       to_convert = buffer;
 435       to_convert_size = count;
 436     }
 437
 438   /* Ensure we have *some* initial target space */
 439   buffer_ensure_space (&priv->converted_buffer, to_convert_size);
 440
 441   converted_bytes = 0;
 442   while (!priv->finished && converted_bytes < to_convert_size)
 443     {
 444       /* Ensure we have *some* target space */
 445       if (buffer_tailspace (&priv->converted_buffer) == 0)
 446         grow_buffer (&priv->converted_buffer);
 447
 448       /* Try to convert to our buffer */
 449       my_error = NULL;
 450       res = g_converter_convert (priv->converter,
 451                                  to_convert + converted_bytes,
 452                                  to_convert_size - converted_bytes,
 453                                  buffer_data (&priv->converted_buffer) + buffer_data_size (&priv->converted_buffer),
 454                                  buffer_tailspace (&priv->converted_buffer),
 455                                  0,
 456                                  &bytes_read,
 457                                  &bytes_written,
 458                                  &my_error);
 459
 460       if (res != G_CONVERTER_ERROR)
 461         {
 462           priv->converted_buffer.end += bytes_written;
 463           converted_bytes += bytes_read;
 464
 465           if (res == G_CONVERTER_FINISHED)
 466             priv->finished = TRUE;
 467         }
 468       else
 469         {
 470           /* No-space errors can be handled locally: */
 471           if (g_error_matches (my_error,
 472                                G_IO_ERROR,
 473                                G_IO_ERROR_NO_SPACE))
 474             {
 475               /* Need more destination space, grow it
 476                * Note: if we actually grow the buffer (as opposed to compacting it),
 477                * this will double the size, not just add one byte. */
 478               buffer_ensure_space (&priv->converted_buffer,
 479                                    priv->converted_buffer.size + 1);
 480               g_error_free (my_error);
 481               continue;
 482             }
 483
 484           if (converted_bytes > 0)
 485             {
 486               /* We got an conversion error, but we did convert some bytes before
 487                  that, so handle those before reporting the error */
 488               g_error_free (my_error);
 489               break;
 490             }
 491
 492           if (g_error_matches (my_error,
 493                                G_IO_ERROR,
 494                                G_IO_ERROR_PARTIAL_INPUT))
 495             {
 496               /* Consume everything to buffer that we append to next time
 497                  we write */
 498               if (!converting_from_buffer)
 499                 buffer_append (&priv->output_buffer, buffer, count);
 500               /* in the converting_from_buffer case we already appended this */
 501
 502               g_error_free (my_error);
 503               return count; /* consume everything */
 504             }
 505
 506           /* Converted no data and got an normal error, return it */
 507           g_propagate_error (error, my_error);
 508           return -1;
 509         }
 510     }
 511
 512   if (converting_from_buffer)
 513     {
 514       buffer_consumed (&priv->output_buffer, converted_bytes);
 515       retval = count;
 516     }
 517   else
 518     retval = converted_bytes;
 519
 520   /* We now successfully consumed retval bytes, so we can't return an error,
 521      even if writing this to the base stream fails. If it does we'll just
 522      stop early and report this error when we try again on the next
 523      write call. */
 524   flush_buffer (cstream, blocking, cancellable, NULL);
 525
 526   return retval;
 527 }
 528
 529 static gssize
 530 g_converter_output_stream_write (GOutputStream  *stream,
 531                                  const void     *buffer,
 532                                  gsize           count,
 533                                  GCancellable   *cancellable,
 534                                  GError        **error)
 535 {
 536   return write_internal (stream, buffer, count, TRUE, cancellable, error);
 537 }
 538
 539 static gboolean
 540 g_converter_output_stream_flush (GOutputStream  *stream,
 541                                  GCancellable   *cancellable,
 542                                  GError        **error)
 543 {
 544   GConverterOutputStream *cstream;
 545   GConverterOutputStreamPrivate *priv;
 546   GConverterResult res;
 547   GError *my_error;
 548   gboolean is_closing;
 549   gboolean flushed;
 550   gsize bytes_read;
 551   gsize bytes_written;
 552
 553   cstream = G_CONVERTER_OUTPUT_STREAM (stream);
 554   priv = cstream->priv;
 555
 556   is_closing = g_output_stream_is_closing (stream);
 557
 558   /* Write out all available pre-converted data and fail if
 559      not possible */
 560   if (!flush_buffer (cstream, TRUE, cancellable, error))
 561     return FALSE;
 562
 563   /* Ensure we have *some* initial target space */
 564   buffer_ensure_space (&priv->converted_buffer, 1);
 565
 566   /* Convert whole buffer */
 567   flushed = FALSE;
 568   while (!priv->finished && !flushed)
 569     {
 570       /* Ensure we have *some* target space */
 571       if (buffer_tailspace (&priv->converted_buffer) == 0)
 572         grow_buffer (&priv->converted_buffer);
 573
 574       /* Try to convert to our buffer */
 575       my_error = NULL;
 576       res = g_converter_convert (priv->converter,
 577                                  buffer_data (&priv->output_buffer),
 578                                  buffer_data_size (&priv->output_buffer),
 579                                  buffer_data (&priv->converted_buffer) + buffer_data_size (&priv->converted_buffer),
 580                                  buffer_tailspace (&priv->converted_buffer),
 581                                  is_closing ? G_CONVERTER_INPUT_AT_END : G_CONVERTER_FLUSH,
 582                                  &bytes_read,
 583                                  &bytes_written,
 584                                  &my_error);
 585
 586       if (res != G_CONVERTER_ERROR)
 587         {
 588           priv->converted_buffer.end += bytes_written;
 589           buffer_consumed (&priv->output_buffer, bytes_read);
 590
 591           if (res == G_CONVERTER_FINISHED)
 592             priv->finished = TRUE;
 593           if (!is_closing &&
 594               res == G_CONVERTER_FLUSHED)
 595             {
 596               /* Should not have retured FLUSHED with input left */
 597               g_assert (buffer_data_size (&priv->output_buffer) == 0);
 598               flushed = TRUE;
 599             }
 600         }
 601       else
 602         {
 603           /* No-space errors can be handled locally: */
 604           if (g_error_matches (my_error,
 605                                G_IO_ERROR,
 606                                G_IO_ERROR_NO_SPACE))
 607             {
 608               /* Need more destination space, grow it
 609                * Note: if we actually grow the buffer (as opposed to compacting it),
 610                * this will double the size, not just add one byte. */
 611               buffer_ensure_space (&priv->converted_buffer,
 612                                    priv->converted_buffer.size + 1);
 613               g_error_free (my_error);
 614               continue;
 615             }
 616
 617           /* Any other error, including PARTIAL_INPUT can't be fixed by now
 618              and is an error */
 619           g_propagate_error (error, my_error);
 620           return FALSE;
 621         }
 622     }
 623
 624   /* Now write all converted data to base stream */
 625   if (!flush_buffer (cstream, TRUE, cancellable, error))
 626     return FALSE;
 627
 628   return TRUE;
 629 }
 630
 631 static gboolean
 632 g_converter_output_stream_can_poll (GPollableOutputStream *stream)
 633 {
 634   GOutputStream *base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 635
 636   return (G_IS_POLLABLE_OUTPUT_STREAM (base_stream) &&
 637           g_pollable_output_stream_can_poll (G_POLLABLE_OUTPUT_STREAM (base_stream)));
 638 }
 639
 640 static gboolean
 641 g_converter_output_stream_is_writable (GPollableOutputStream *stream)
 642 {
 643   GOutputStream *base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 644
 645   return g_pollable_output_stream_is_writable (G_POLLABLE_OUTPUT_STREAM (base_stream));
 646 }
 647
 648 static gssize
 649 g_converter_output_stream_write_nonblocking (GPollableOutputStream  *stream,
 650                                              const void             *buffer,
 651                                              gsize                   count,
 652                                              GError                **error)
 653 {
 654   return write_internal (G_OUTPUT_STREAM (stream), buffer, count, FALSE,
 655                          NULL, error);
 656 }
 657
 658 static GSource *
 659 g_converter_output_stream_create_source (GPollableOutputStream *stream,
 660                                          GCancellable          *cancellable)
 661 {
 662   GOutputStream *base_stream = G_FILTER_OUTPUT_STREAM (stream)->base_stream;
 663   GSource *base_source, *pollable_source;
 664
 665   base_source = g_pollable_output_stream_create_source (G_POLLABLE_OUTPUT_STREAM (base_stream), NULL);
 666   pollable_source = g_pollable_source_new_full (stream, base_source,
 667                                                 cancellable);
 668   g_source_unref (base_source);
 669
 670   return pollable_source;
 671 }
 672
 673 /**
 674  * g_converter_output_stream_get_converter:
 675  * @converter_stream: a #GConverterOutputStream
 676  *
 677  * Gets the #GConverter that is used by @converter_stream.
 678  *
 679  * Returns: (transfer none): the converter of the converter output stream
 680  *
 681  * Since: 2.24
 682  */
 683 GConverter *
 684 g_converter_output_stream_get_converter (GConverterOutputStream *converter_stream)
 685 {
 686   return converter_stream->priv->converter;
 687 }