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