libsoup/soup-body-input-stream.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-body-input-stream.c
   4  *
   5  * Copyright 2012 Red Hat, Inc.
   6  */
   7
   8 #ifdef HAVE_CONFIG_H
   9 #include <config.h>
  10 #endif
  11
  12 #include <stdlib.h>
  13
  14 #include <glib/gi18n-lib.h>
  15
  16 #include "soup-body-input-stream.h"
  17 #include "soup.h"
  18 #include "soup-filter-input-stream.h"
  19
  20 typedef enum {
  21         SOUP_BODY_INPUT_STREAM_STATE_CHUNK_SIZE,
  22         SOUP_BODY_INPUT_STREAM_STATE_CHUNK_END,
  23         SOUP_BODY_INPUT_STREAM_STATE_CHUNK,
  24         SOUP_BODY_INPUT_STREAM_STATE_TRAILERS,
  25         SOUP_BODY_INPUT_STREAM_STATE_DONE
  26 } SoupBodyInputStreamState;
  27
  28 struct _SoupBodyInputStreamPrivate {
  29         GInputStream *base_stream;
  30
  31         SoupEncoding  encoding;
  32         goffset       read_length;
  33         SoupBodyInputStreamState chunked_state;
  34         gboolean      eof;
  35
  36         goffset       pos;
  37 };
  38
  39 enum {
  40         CLOSED,
  41         LAST_SIGNAL
  42 };
  43
  44 static guint signals[LAST_SIGNAL] = { 0 };
  45
  46 enum {
  47         PROP_0,
  48
  49         PROP_ENCODING,
  50         PROP_CONTENT_LENGTH
  51 };
  52
  53 static void soup_body_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface, gpointer interface_data);
  54 static void soup_body_input_stream_seekable_init (GSeekableIface *seekable_interface);
  55
  56 G_DEFINE_TYPE_WITH_CODE (SoupBodyInputStream, soup_body_input_stream, G_TYPE_FILTER_INPUT_STREAM,
  57                          G_ADD_PRIVATE (SoupBodyInputStream)
  58                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM,
  59                                                 soup_body_input_stream_pollable_init)
  60                          G_IMPLEMENT_INTERFACE (G_TYPE_SEEKABLE,
  61                                                 soup_body_input_stream_seekable_init))
  62
  63 static void
  64 soup_body_input_stream_init (SoupBodyInputStream *bistream)
  65 {
  66         bistream->priv = soup_body_input_stream_get_instance_private (bistream);
  67         bistream->priv->encoding = SOUP_ENCODING_NONE;
  68 }
  69
  70 static void
  71 soup_body_input_stream_constructed (GObject *object)
  72 {
  73         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (object);
  74
  75         bistream->priv->base_stream = g_filter_input_stream_get_base_stream (G_FILTER_INPUT_STREAM (bistream));
  76
  77         if (bistream->priv->encoding == SOUP_ENCODING_NONE ||
  78             (bistream->priv->encoding == SOUP_ENCODING_CONTENT_LENGTH &&
  79              bistream->priv->read_length == 0))
  80                 bistream->priv->eof = TRUE;
  81 }
  82
  83 static void
  84 soup_body_input_stream_set_property (GObject *object, guint prop_id,
  85                                      const GValue *value, GParamSpec *pspec)
  86 {
  87         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (object);
  88
  89         switch (prop_id) {
  90         case PROP_ENCODING:
  91                 bistream->priv->encoding = g_value_get_enum (value);
  92                 if (bistream->priv->encoding == SOUP_ENCODING_CHUNKED)
  93                         bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_CHUNK_SIZE;
  94                 break;
  95         case PROP_CONTENT_LENGTH:
  96                 bistream->priv->read_length = g_value_get_int64 (value);
  97                 break;
  98         default:
  99                 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 100                 break;
 101         }
 102 }
 103
 104 static void
 105 soup_body_input_stream_get_property (GObject *object, guint prop_id,
 106                                      GValue *value, GParamSpec *pspec)
 107 {
 108         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (object);
 109
 110         switch (prop_id) {
 111         case PROP_ENCODING:
 112                 g_value_set_enum (value, bistream->priv->encoding);
 113                 break;
 114         default:
 115                 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 116                 break;
 117         }
 118 }
 119
 120 static gssize
 121 soup_body_input_stream_read_raw (SoupBodyInputStream  *bistream,
 122                                  void                 *buffer,
 123                                  gsize                 count,
 124                                  gboolean              blocking,
 125                                  GCancellable         *cancellable,
 126                                  GError              **error)
 127 {
 128         gssize nread;
 129
 130         nread = g_pollable_stream_read (bistream->priv->base_stream,
 131                                         buffer, count,
 132                                         blocking,
 133                                         cancellable, error);
 134         if (nread == 0) {
 135                 bistream->priv->eof = TRUE;
 136                 if (bistream->priv->encoding != SOUP_ENCODING_EOF) {
 137                         g_set_error_literal (error, G_IO_ERROR,
 138                                              G_IO_ERROR_PARTIAL_INPUT,
 139                                              _("Connection terminated unexpectedly"));
 140                         return -1;
 141                 }
 142         }
 143         return nread;
 144 }
 145
 146 static gssize
 147 soup_body_input_stream_read_chunked (SoupBodyInputStream  *bistream,
 148                                      void                 *buffer,
 149                                      gsize                 count,
 150                                      gboolean              blocking,
 151                                      GCancellable         *cancellable,
 152                                      GError              **error)
 153 {
 154         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (bistream->priv->base_stream);
 155         char metabuf[128];
 156         gssize nread;
 157         gboolean got_line;
 158
 159 again:
 160         switch (bistream->priv->chunked_state) {
 161         case SOUP_BODY_INPUT_STREAM_STATE_CHUNK_SIZE:
 162                 nread = soup_filter_input_stream_read_line (
 163                         fstream, metabuf, sizeof (metabuf), blocking,
 164                         &got_line, cancellable, error);
 165                 if (nread <= 0)
 166                         return nread;
 167                 if (!got_line) {
 168                         g_set_error_literal (error, G_IO_ERROR,
 169                                              G_IO_ERROR_PARTIAL_INPUT,
 170                                              _("Connection terminated unexpectedly"));
 171                         return -1;
 172                 }
 173
 174                 bistream->priv->read_length = strtoul (metabuf, NULL, 16);
 175                 if (bistream->priv->read_length > 0)
 176                         bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_CHUNK;
 177                 else
 178                         bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_TRAILERS;
 179                 break;
 180
 181         case SOUP_BODY_INPUT_STREAM_STATE_CHUNK:
 182                 nread = soup_body_input_stream_read_raw (
 183                         bistream, buffer,
 184                         MIN (count, bistream->priv->read_length),
 185                         blocking, cancellable, error);
 186                 if (nread > 0) {
 187                         bistream->priv->read_length -= nread;
 188                         if (bistream->priv->read_length == 0)
 189                                 bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_CHUNK_END;
 190                 }
 191                 return nread;
 192
 193         case SOUP_BODY_INPUT_STREAM_STATE_CHUNK_END:
 194                 nread = soup_filter_input_stream_read_line (
 195                         SOUP_FILTER_INPUT_STREAM (bistream->priv->base_stream),
 196                         metabuf, sizeof (metabuf), blocking,
 197                         &got_line, cancellable, error);
 198                 if (nread <= 0)
 199                         return nread;
 200                 if (!got_line) {
 201                         g_set_error_literal (error, G_IO_ERROR,
 202                                              G_IO_ERROR_PARTIAL_INPUT,
 203                                              _("Connection terminated unexpectedly"));
 204                         return -1;
 205                 }
 206
 207                 bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_CHUNK_SIZE;
 208                 break;
 209
 210         case SOUP_BODY_INPUT_STREAM_STATE_TRAILERS:
 211                 nread = soup_filter_input_stream_read_line (
 212                         fstream, buffer, count, blocking,
 213                         &got_line, cancellable, error);
 214                 if (nread <= 0)
 215                         return nread;
 216
 217                 if (strncmp (buffer, "\r\n", nread) || strncmp (buffer, "\n", nread)) {
 218                         bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_DONE;
 219                         bistream->priv->eof = TRUE;
 220                 }
 221                 break;
 222
 223         case SOUP_BODY_INPUT_STREAM_STATE_DONE:
 224                 return 0;
 225         }
 226
 227         goto again;
 228 }
 229
 230 static gssize
 231 read_internal (GInputStream  *stream,
 232                void          *buffer,
 233                gsize          count,
 234                gboolean       blocking,
 235                GCancellable  *cancellable,
 236                GError       **error)
 237 {
 238         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (stream);
 239         gssize nread;
 240
 241         if (bistream->priv->eof)
 242                 return 0;
 243
 244         switch (bistream->priv->encoding) {
 245         case SOUP_ENCODING_NONE:
 246                 return 0;
 247
 248         case SOUP_ENCODING_CHUNKED:
 249                 return soup_body_input_stream_read_chunked (bistream, buffer, count,
 250                                                             blocking, cancellable, error);
 251
 252         case SOUP_ENCODING_CONTENT_LENGTH:
 253         case SOUP_ENCODING_EOF:
 254                 if (bistream->priv->read_length != -1) {
 255                         count = MIN (count, bistream->priv->read_length);
 256                         if (count == 0)
 257                                 return 0;
 258                 }
 259
 260                 nread = soup_body_input_stream_read_raw (bistream, buffer, count,
 261                                                          blocking, cancellable, error);
 262                 if (bistream->priv->read_length != -1 && nread > 0)
 263                         bistream->priv->read_length -= nread;
 264
 265                 if (bistream->priv->encoding == SOUP_ENCODING_CONTENT_LENGTH)
 266                         bistream->priv->pos += nread;
 267                 return nread;
 268
 269         default:
 270                 g_return_val_if_reached (-1);
 271         }
 272 }
 273
 274 static gssize
 275 soup_body_input_stream_skip (GInputStream *stream,
 276                              gsize         count,
 277                              GCancellable *cancellable,
 278                              GError      **error)
 279 {
 280         SoupBodyInputStreamPrivate *priv = SOUP_BODY_INPUT_STREAM(stream)->priv;
 281         gssize skipped;
 282
 283         skipped = g_input_stream_skip (G_FILTER_INPUT_STREAM (stream)->base_stream,
 284                                        MIN (count, priv->read_length),
 285                                        cancellable, error);
 286
 287         if (skipped == 0)
 288                 priv->eof = TRUE;
 289         else if (skipped > 0)
 290                 priv->pos += skipped;
 291
 292         return skipped;
 293 }
 294
 295 static gssize
 296 soup_body_input_stream_read_fn (GInputStream  *stream,
 297                                 void          *buffer,
 298                                 gsize          count,
 299                                 GCancellable  *cancellable,
 300                                 GError       **error)
 301 {
 302         return read_internal (stream, buffer, count, TRUE,
 303                               cancellable, error);
 304 }
 305
 306 static gboolean
 307 soup_body_input_stream_close_fn (GInputStream  *stream,
 308                                  GCancellable  *cancellable,
 309                                  GError       **error)
 310 {
 311         g_signal_emit (stream, signals[CLOSED], 0);
 312
 313         return G_INPUT_STREAM_CLASS (soup_body_input_stream_parent_class)->close_fn (stream, cancellable, error);
 314 }
 315
 316 static gboolean
 317 soup_body_input_stream_is_readable (GPollableInputStream *stream)
 318 {
 319         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (stream);
 320
 321         return bistream->priv->eof ||
 322                 g_pollable_input_stream_is_readable (G_POLLABLE_INPUT_STREAM (bistream->priv->base_stream));
 323 }
 324
 325 static gboolean
 326 soup_body_input_stream_can_poll (GPollableInputStream *pollable)
 327 {
 328         GInputStream *base_stream = SOUP_BODY_INPUT_STREAM (pollable)->priv->base_stream;
 329
 330         return G_IS_POLLABLE_INPUT_STREAM (base_stream) &&
 331                 g_pollable_input_stream_can_poll (G_POLLABLE_INPUT_STREAM (base_stream));
 332 }
 333
 334 static gssize
 335 soup_body_input_stream_read_nonblocking (GPollableInputStream  *stream,
 336                                          void                  *buffer,
 337                                          gsize                  count,
 338                                          GError               **error)
 339 {
 340         return read_internal (G_INPUT_STREAM (stream), buffer, count, FALSE,
 341                               NULL, error);
 342 }
 343
 344 static GSource *
 345 soup_body_input_stream_create_source (GPollableInputStream *stream,
 346                                       GCancellable *cancellable)
 347 {
 348         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (stream);
 349         GSource *base_source, *pollable_source;
 350
 351         if (bistream->priv->eof)
 352                 base_source = g_timeout_source_new (0);
 353         else
 354                 base_source = g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM (bistream->priv->base_stream), cancellable);
 355         g_source_set_dummy_callback (base_source);
 356
 357         pollable_source = g_pollable_source_new (G_OBJECT (stream));
 358         g_source_add_child_source (pollable_source, base_source);
 359         g_source_unref (base_source);
 360
 361         return pollable_source;
 362 }
 363
 364 static void
 365 soup_body_input_stream_class_init (SoupBodyInputStreamClass *stream_class)
 366 {
 367         GObjectClass *object_class = G_OBJECT_CLASS (stream_class);
 368         GInputStreamClass *input_stream_class = G_INPUT_STREAM_CLASS (stream_class);
 369
 370         object_class->constructed = soup_body_input_stream_constructed;
 371         object_class->set_property = soup_body_input_stream_set_property;
 372         object_class->get_property = soup_body_input_stream_get_property;
 373
 374         input_stream_class->skip = soup_body_input_stream_skip;
 375         input_stream_class->read_fn = soup_body_input_stream_read_fn;
 376         input_stream_class->close_fn = soup_body_input_stream_close_fn;
 377
 378         signals[CLOSED] =
 379                 g_signal_new ("closed",
 380                               G_OBJECT_CLASS_TYPE (object_class),
 381                               G_SIGNAL_RUN_LAST,
 382                               0,
 383                               NULL, NULL,
 384                               NULL,
 385                               G_TYPE_NONE, 0);
 386
 387         g_object_class_install_property (
 388                 object_class, PROP_ENCODING,
 389                 g_param_spec_enum ("encoding",
 390                                    "Encoding",
 391                                    "Message body encoding",
 392                                    SOUP_TYPE_ENCODING,
 393                                    SOUP_ENCODING_NONE,
 394                                    G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 395         g_object_class_install_property (
 396                 object_class, PROP_CONTENT_LENGTH,
 397                 g_param_spec_int64 ("content-length",
 398                                     "Content-Length",
 399                                     "Message body Content-Length",
 400                                     -1, G_MAXINT64, -1,
 401                                     G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY | G_PARAM_STATIC_STRINGS));
 402 }
 403
 404 static void
 405 soup_body_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface,
 406                                  gpointer interface_data)
 407 {
 408         pollable_interface->can_poll = soup_body_input_stream_can_poll;
 409         pollable_interface->is_readable = soup_body_input_stream_is_readable;
 410         pollable_interface->read_nonblocking = soup_body_input_stream_read_nonblocking;
 411         pollable_interface->create_source = soup_body_input_stream_create_source;
 412 }
 413
 414 static goffset
 415 soup_body_input_stream_tell (GSeekable *seekable)
 416 {
 417         return SOUP_BODY_INPUT_STREAM (seekable)->priv->pos;
 418 }
 419
 420 static gboolean
 421 soup_body_input_stream_can_seek (GSeekable *seekable)
 422 {
 423         SoupBodyInputStreamPrivate *priv = SOUP_BODY_INPUT_STREAM (seekable)->priv;
 424
 425         return priv->encoding == SOUP_ENCODING_CONTENT_LENGTH
 426                 && G_IS_SEEKABLE (priv->base_stream)
 427                 && g_seekable_can_seek (G_SEEKABLE (priv->base_stream));
 428 }
 429
 430 static gboolean
 431 soup_body_input_stream_seek (GSeekable     *seekable,
 432                              goffset        offset,
 433                              GSeekType      type,
 434                              GCancellable  *cancellable,
 435                              GError       **error)
 436 {
 437         SoupBodyInputStreamPrivate *priv = SOUP_BODY_INPUT_STREAM (seekable)->priv;
 438         goffset position, end_position;
 439
 440         end_position = priv->pos + priv->read_length;
 441         switch (type) {
 442         case G_SEEK_CUR:
 443                 position = priv->pos + offset;
 444                 break;
 445         case G_SEEK_SET:
 446                 position = offset;
 447                 break;
 448         case G_SEEK_END:
 449                 position = end_position + offset;
 450                 break;
 451         default:
 452                 g_return_val_if_reached (FALSE);
 453         }
 454
 455         if (position < 0 || position >= end_position) {
 456                 g_set_error_literal (error,
 457                                      G_IO_ERROR,
 458                                      G_IO_ERROR_INVALID_ARGUMENT,
 459                                      _("Invalid seek request"));
 460                 return FALSE;
 461         }
 462
 463         if (!g_seekable_seek (G_SEEKABLE (priv->base_stream), position - priv->pos,
 464                               G_SEEK_CUR, cancellable, error))
 465                 return FALSE;
 466
 467         priv->pos = position;
 468
 469         return TRUE;
 470 }
 471
 472 static gboolean
 473 soup_body_input_stream_can_truncate (GSeekable *seekable)
 474 {
 475         return FALSE;
 476 }
 477
 478 static gboolean
 479 soup_body_input_stream_truncate_fn (GSeekable     *seekable,
 480                                     goffset        offset,
 481                                     GCancellable  *cancellable,
 482                                     GError       **error)
 483 {
 484         g_set_error_literal (error,
 485                              G_IO_ERROR,
 486                              G_IO_ERROR_NOT_SUPPORTED,
 487                              _("Cannot truncate SoupBodyInputStream"));
 488         return FALSE;
 489 }
 490
 491 static void
 492 soup_body_input_stream_seekable_init (GSeekableIface *seekable_interface)
 493 {
 494         seekable_interface->tell         = soup_body_input_stream_tell;
 495         seekable_interface->can_seek     = soup_body_input_stream_can_seek;
 496         seekable_interface->seek         = soup_body_input_stream_seek;
 497         seekable_interface->can_truncate = soup_body_input_stream_can_truncate;
 498         seekable_interface->truncate_fn  = soup_body_input_stream_truncate_fn;
 499 }
 500
 501 GInputStream *
 502 soup_body_input_stream_new (GInputStream *base_stream,
 503                             SoupEncoding  encoding,
 504                             goffset       content_length)
 505 {
 506         if (encoding == SOUP_ENCODING_CHUNKED)
 507                 g_return_val_if_fail (SOUP_IS_FILTER_INPUT_STREAM (base_stream), NULL);
 508
 509         return g_object_new (SOUP_TYPE_BODY_INPUT_STREAM,
 510                              "base-stream", base_stream,
 511                              "close-base-stream", FALSE,
 512                              "encoding", encoding,
 513                              "content-length", content_length,
 514                              NULL);
 515 }