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 #include "soup-marshal.h"
  20
  21 typedef enum {
  22         SOUP_BODY_INPUT_STREAM_STATE_CHUNK_SIZE,
  23         SOUP_BODY_INPUT_STREAM_STATE_CHUNK_END,
  24         SOUP_BODY_INPUT_STREAM_STATE_CHUNK,
  25         SOUP_BODY_INPUT_STREAM_STATE_TRAILERS,
  26         SOUP_BODY_INPUT_STREAM_STATE_DONE
  27 } SoupBodyInputStreamState;
  28
  29 struct _SoupBodyInputStreamPrivate {
  30         GInputStream *base_stream;
  31
  32         SoupEncoding  encoding;
  33         goffset       read_length;
  34         SoupBodyInputStreamState chunked_state;
  35         gboolean      eof;
  36
  37         goffset       pos;
  38 };
  39
  40 enum {
  41         CLOSED,
  42         LAST_SIGNAL
  43 };
  44
  45 static guint signals[LAST_SIGNAL] = { 0 };
  46
  47 enum {
  48         PROP_0,
  49
  50         PROP_ENCODING,
  51         PROP_CONTENT_LENGTH
  52 };
  53
  54 static void soup_body_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface, gpointer interface_data);
  55 static void soup_body_input_stream_seekable_init (GSeekableIface *seekable_interface);
  56
  57 G_DEFINE_TYPE_WITH_CODE (SoupBodyInputStream, soup_body_input_stream, G_TYPE_FILTER_INPUT_STREAM,
  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 = G_TYPE_INSTANCE_GET_PRIVATE (bistream,
  67                                                       SOUP_TYPE_BODY_INPUT_STREAM,
  68                                                       SoupBodyInputStreamPrivate);
  69         bistream->priv->encoding = SOUP_ENCODING_NONE;
  70 }
  71
  72 static void
  73 soup_body_input_stream_constructed (GObject *object)
  74 {
  75         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (object);
  76
  77         bistream->priv->base_stream = g_filter_input_stream_get_base_stream (G_FILTER_INPUT_STREAM (bistream));
  78
  79         if (bistream->priv->encoding == SOUP_ENCODING_NONE ||
  80             (bistream->priv->encoding == SOUP_ENCODING_CONTENT_LENGTH &&
  81              bistream->priv->read_length == 0))
  82                 bistream->priv->eof = TRUE;
  83 }
  84
  85 static void
  86 soup_body_input_stream_set_property (GObject *object, guint prop_id,
  87                                      const GValue *value, GParamSpec *pspec)
  88 {
  89         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (object);
  90
  91         switch (prop_id) {
  92         case PROP_ENCODING:
  93                 bistream->priv->encoding = g_value_get_enum (value);
  94                 if (bistream->priv->encoding == SOUP_ENCODING_CHUNKED)
  95                         bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_CHUNK_SIZE;
  96                 break;
  97         case PROP_CONTENT_LENGTH:
  98                 bistream->priv->read_length = g_value_get_int64 (value);
  99                 break;
 100         default:
 101                 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 102                 break;
 103         }
 104 }
 105
 106 static void
 107 soup_body_input_stream_get_property (GObject *object, guint prop_id,
 108                                      GValue *value, GParamSpec *pspec)
 109 {
 110         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (object);
 111
 112         switch (prop_id) {
 113         case PROP_ENCODING:
 114                 g_value_set_enum (value, bistream->priv->encoding);
 115                 break;
 116         default:
 117                 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 118                 break;
 119         }
 120 }
 121
 122 static gssize
 123 soup_body_input_stream_read_raw (SoupBodyInputStream  *bistream,
 124                                  void                 *buffer,
 125                                  gsize                 count,
 126                                  gboolean              blocking,
 127                                  GCancellable         *cancellable,
 128                                  GError              **error)
 129 {
 130         gssize nread;
 131
 132         nread = g_pollable_stream_read (bistream->priv->base_stream,
 133                                         buffer, count,
 134                                         blocking,
 135                                         cancellable, error);
 136         if (nread == 0) {
 137                 bistream->priv->eof = TRUE;
 138                 if (bistream->priv->encoding != SOUP_ENCODING_EOF) {
 139                         g_set_error_literal (error, G_IO_ERROR,
 140                                              G_IO_ERROR_PARTIAL_INPUT,
 141                                              _("Connection terminated unexpectedly"));
 142                         return -1;
 143                 }
 144         }
 145         return nread;
 146 }
 147
 148 static gssize
 149 soup_body_input_stream_read_chunked (SoupBodyInputStream  *bistream,
 150                                      void                 *buffer,
 151                                      gsize                 count,
 152                                      gboolean              blocking,
 153                                      GCancellable         *cancellable,
 154                                      GError              **error)
 155 {
 156         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (bistream->priv->base_stream);
 157         char metabuf[128];
 158         gssize nread;
 159         gboolean got_line;
 160
 161 again:
 162         switch (bistream->priv->chunked_state) {
 163         case SOUP_BODY_INPUT_STREAM_STATE_CHUNK_SIZE:
 164                 nread = soup_filter_input_stream_read_line (
 165                         fstream, metabuf, sizeof (metabuf), blocking,
 166                         &got_line, cancellable, error);
 167                 if (nread <= 0)
 168                         return nread;
 169                 if (!got_line) {
 170                         g_set_error_literal (error, G_IO_ERROR,
 171                                              G_IO_ERROR_PARTIAL_INPUT,
 172                                              _("Connection terminated unexpectedly"));
 173                         return -1;
 174                 }
 175
 176                 bistream->priv->read_length = strtoul (metabuf, NULL, 16);
 177                 if (bistream->priv->read_length > 0)
 178                         bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_CHUNK;
 179                 else
 180                         bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_TRAILERS;
 181                 break;
 182
 183         case SOUP_BODY_INPUT_STREAM_STATE_CHUNK:
 184                 nread = soup_body_input_stream_read_raw (
 185                         bistream, buffer,
 186                         MIN (count, bistream->priv->read_length),
 187                         blocking, cancellable, error);
 188                 if (nread > 0) {
 189                         bistream->priv->read_length -= nread;
 190                         if (bistream->priv->read_length == 0)
 191                                 bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_CHUNK_END;
 192                 }
 193                 return nread;
 194
 195         case SOUP_BODY_INPUT_STREAM_STATE_CHUNK_END:
 196                 nread = soup_filter_input_stream_read_line (
 197                         SOUP_FILTER_INPUT_STREAM (bistream->priv->base_stream),
 198                         metabuf, sizeof (metabuf), blocking,
 199                         &got_line, cancellable, error);
 200                 if (nread <= 0)
 201                         return nread;
 202                 if (!got_line) {
 203                         g_set_error_literal (error, G_IO_ERROR,
 204                                              G_IO_ERROR_PARTIAL_INPUT,
 205                                              _("Connection terminated unexpectedly"));
 206                         return -1;
 207                 }
 208
 209                 bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_CHUNK_SIZE;
 210                 break;
 211
 212         case SOUP_BODY_INPUT_STREAM_STATE_TRAILERS:
 213                 nread = soup_filter_input_stream_read_line (
 214                         fstream, buffer, count, blocking,
 215                         &got_line, cancellable, error);
 216                 if (nread <= 0)
 217                         return nread;
 218
 219                 if (strncmp (buffer, "\r\n", nread) || strncmp (buffer, "\n", nread)) {
 220                         bistream->priv->chunked_state = SOUP_BODY_INPUT_STREAM_STATE_DONE;
 221                         bistream->priv->eof = TRUE;
 222                 }
 223                 break;
 224
 225         case SOUP_BODY_INPUT_STREAM_STATE_DONE:
 226                 return 0;
 227         }
 228
 229         goto again;
 230 }
 231
 232 static gssize
 233 read_internal (GInputStream  *stream,
 234                void          *buffer,
 235                gsize          count,
 236                gboolean       blocking,
 237                GCancellable  *cancellable,
 238                GError       **error)
 239 {
 240         SoupBodyInputStream *bistream = SOUP_BODY_INPUT_STREAM (stream);
 241         gssize nread;
 242
 243         if (bistream->priv->eof)
 244                 return 0;
 245
 246         switch (bistream->priv->encoding) {
 247         case SOUP_ENCODING_NONE:
 248                 return 0;
 249
 250         case SOUP_ENCODING_CHUNKED:
 251                 return soup_body_input_stream_read_chunked (bistream, buffer, count,
 252                                                             blocking, cancellable, error);
 253
 254         case SOUP_ENCODING_CONTENT_LENGTH:
 255         case SOUP_ENCODING_EOF:
 256                 if (bistream->priv->read_length != -1) {
 257                         count = MIN (count, bistream->priv->read_length);
 258                         if (count == 0)
 259                                 return 0;
 260                 }
 261
 262                 nread = soup_body_input_stream_read_raw (bistream, buffer, count,
 263                                                          blocking, cancellable, error);
 264                 if (bistream->priv->read_length != -1 && nread > 0)
 265                         bistream->priv->read_length -= nread;
 266
 267                 if (bistream->priv->encoding == SOUP_ENCODING_CONTENT_LENGTH)
 268                         bistream->priv->pos += nread;
 269                 return nread;
 270
 271         default:
 272                 g_return_val_if_reached (-1);
 273         }
 274 }
 275
 276 static gssize
 277 soup_body_input_stream_skip (GInputStream *stream,
 278                              gsize         count,
 279                              GCancellable *cancellable,
 280                              GError      **error)
 281 {
 282         SoupBodyInputStreamPrivate *priv = SOUP_BODY_INPUT_STREAM(stream)->priv;
 283         gssize skipped;
 284
 285         skipped = g_input_stream_skip (G_FILTER_INPUT_STREAM (stream)->base_stream,
 286                                        MIN (count, priv->read_length),
 287                                        cancellable, error);
 288
 289         if (skipped != -1)
 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         g_type_class_add_private (stream_class, sizeof (SoupBodyInputStreamPrivate));
 371
 372         object_class->constructed = soup_body_input_stream_constructed;
 373         object_class->set_property = soup_body_input_stream_set_property;
 374         object_class->get_property = soup_body_input_stream_get_property;
 375
 376         input_stream_class->skip = soup_body_input_stream_skip;
 377         input_stream_class->read_fn = soup_body_input_stream_read_fn;
 378         input_stream_class->close_fn = soup_body_input_stream_close_fn;
 379
 380         signals[CLOSED] =
 381                 g_signal_new ("closed",
 382                               G_OBJECT_CLASS_TYPE (object_class),
 383                               G_SIGNAL_RUN_LAST,
 384                               0,
 385                               NULL, NULL,
 386                               _soup_marshal_NONE__NONE,
 387                               G_TYPE_NONE, 0);
 388
 389         g_object_class_install_property (
 390                 object_class, PROP_ENCODING,
 391                 g_param_spec_enum ("encoding",
 392                                    "Encoding",
 393                                    "Message body encoding",
 394                                    SOUP_TYPE_ENCODING,
 395                                    SOUP_ENCODING_NONE,
 396                                    G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY));
 397         g_object_class_install_property (
 398                 object_class, PROP_CONTENT_LENGTH,
 399                 g_param_spec_int64 ("content-length",
 400                                     "Content-Length",
 401                                     "Message body Content-Length",
 402                                     -1, G_MAXINT64, -1,
 403                                     G_PARAM_WRITABLE | G_PARAM_CONSTRUCT_ONLY));
 404 }
 405
 406 static void
 407 soup_body_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface,
 408                                  gpointer interface_data)
 409 {
 410         pollable_interface->can_poll = soup_body_input_stream_can_poll;
 411         pollable_interface->is_readable = soup_body_input_stream_is_readable;
 412         pollable_interface->read_nonblocking = soup_body_input_stream_read_nonblocking;
 413         pollable_interface->create_source = soup_body_input_stream_create_source;
 414 }
 415
 416 static goffset
 417 soup_body_input_stream_tell (GSeekable *seekable)
 418 {
 419         return SOUP_BODY_INPUT_STREAM (seekable)->priv->pos;
 420 }
 421
 422 static gboolean
 423 soup_body_input_stream_can_seek (GSeekable *seekable)
 424 {
 425         SoupBodyInputStreamPrivate *priv = SOUP_BODY_INPUT_STREAM (seekable)->priv;
 426
 427         return priv->encoding == SOUP_ENCODING_CONTENT_LENGTH
 428                 && G_IS_SEEKABLE (priv->base_stream)
 429                 && g_seekable_can_seek (G_SEEKABLE (priv->base_stream));
 430 }
 431
 432 static gboolean
 433 soup_body_input_stream_seek (GSeekable     *seekable,
 434                              goffset        offset,
 435                              GSeekType      type,
 436                              GCancellable  *cancellable,
 437                              GError       **error)
 438 {
 439         SoupBodyInputStreamPrivate *priv = SOUP_BODY_INPUT_STREAM (seekable)->priv;
 440         goffset position, end_position;
 441
 442         end_position = priv->pos + priv->read_length;
 443         switch (type) {
 444         case G_SEEK_CUR:
 445                 position = priv->pos + offset;
 446                 break;
 447         case G_SEEK_SET:
 448                 position = offset;
 449                 break;
 450         case G_SEEK_END:
 451                 position = end_position + offset;
 452                 break;
 453         default:
 454                 g_return_val_if_reached (FALSE);
 455         }
 456
 457         if (position < 0 || position >= end_position) {
 458                 g_set_error_literal (error,
 459                                      G_IO_ERROR,
 460                                      G_IO_ERROR_INVALID_ARGUMENT,
 461                                      _("Invalid seek request"));
 462                 return FALSE;
 463         }
 464
 465         if (!g_seekable_seek (G_SEEKABLE (priv->base_stream), position - priv->pos,
 466                               G_SEEK_CUR, cancellable, error))
 467                 return FALSE;
 468
 469         priv->pos = position;
 470
 471         return TRUE;
 472 }
 473
 474 static gboolean
 475 soup_body_input_stream_can_truncate (GSeekable *seekable)
 476 {
 477         return FALSE;
 478 }
 479
 480 static gboolean
 481 soup_body_input_stream_truncate_fn (GSeekable     *seekable,
 482                                     goffset        offset,
 483                                     GCancellable  *cancellable,
 484                                     GError       **error)
 485 {
 486         g_set_error_literal (error,
 487                              G_IO_ERROR,
 488                              G_IO_ERROR_NOT_SUPPORTED,
 489                              _("Cannot truncate SoupBodyInputStream"));
 490         return FALSE;
 491 }
 492
 493 static void
 494 soup_body_input_stream_seekable_init (GSeekableIface *seekable_interface)
 495 {
 496         seekable_interface->tell         = soup_body_input_stream_tell;
 497         seekable_interface->can_seek     = soup_body_input_stream_can_seek;
 498         seekable_interface->seek         = soup_body_input_stream_seek;
 499         seekable_interface->can_truncate = soup_body_input_stream_can_truncate;
 500         seekable_interface->truncate_fn  = soup_body_input_stream_truncate_fn;
 501 }
 502
 503 GInputStream *
 504 soup_body_input_stream_new (GInputStream *base_stream,
 505                             SoupEncoding  encoding,
 506                             goffset       content_length)
 507 {
 508         if (encoding == SOUP_ENCODING_CHUNKED)
 509                 g_return_val_if_fail (SOUP_IS_FILTER_INPUT_STREAM (base_stream), NULL);
 510
 511         return g_object_new (SOUP_TYPE_BODY_INPUT_STREAM,
 512                              "base-stream", base_stream,
 513                              "close-base-stream", FALSE,
 514                              "encoding", encoding,
 515                              "content-length", content_length,
 516                              NULL);
 517 }