libsoup/soup-multipart-input-stream.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-multipart-input-stream.c
   4  *
   5  * Copyright (C) 2012 Collabora Ltd.
   6  */
   7
   8 #ifdef HAVE_CONFIG_H
   9 #include <config.h>
  10 #endif
  11
  12 #include <string.h>
  13
  14 #include "soup-body-input-stream.h"
  15 #include "soup-filter-input-stream.h"
  16 #include "soup-enum-types.h"
  17 #include "soup-message.h"
  18 #include "soup-message-private.h"
  19 #include "soup-multipart-input-stream.h"
  20
  21 #define RESPONSE_BLOCK_SIZE 8192
  22
  23 /**
  24  * SECTION:soup-multipart-input-stream
  25  * @short_description: Multipart input handling stream
  26  *
  27  * This adds support for the multipart responses. For handling the
  28  * multiple parts the user needs to wrap the #GInputStream obtained by
  29  * sending the request with a #SoupMultipartInputStream and use
  30  * soup_multipart_input_stream_next_part() before reading. Responses
  31  * which are not wrapped will be treated like non-multipart responses.
  32  *
  33  * Note that although #SoupMultipartInputStream is a #GInputStream,
  34  * you should not read directly from it, and the results are undefined
  35  * if you do.
  36  *
  37  * Since: 2.40
  38  **/
  39
  40 static void soup_multipart_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface, gpointer interface_data);
  41
  42 G_DEFINE_TYPE_WITH_CODE (SoupMultipartInputStream, soup_multipart_input_stream, G_TYPE_FILTER_INPUT_STREAM,
  43                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM,
  44                                                 soup_multipart_input_stream_pollable_init))
  45
  46 enum {
  47         PROP_0,
  48
  49         PROP_MESSAGE,
  50 };
  51
  52 struct _SoupMultipartInputStreamPrivate {
  53         SoupMessage             *msg;
  54
  55         gboolean                 done_with_part;
  56
  57         GByteArray              *meta_buf;
  58         SoupMessageHeaders      *current_headers;
  59
  60         SoupFilterInputStream   *base_stream;
  61
  62         char                    *boundary;
  63         gsize                    boundary_size;
  64
  65         goffset                 remaining_bytes;
  66 };
  67
  68 static void
  69 soup_multipart_input_stream_dispose (GObject *object)
  70 {
  71         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (object);
  72
  73         g_clear_object (&multipart->priv->msg);
  74         g_clear_object (&multipart->priv->base_stream);
  75
  76         G_OBJECT_CLASS (soup_multipart_input_stream_parent_class)->dispose (object);
  77 }
  78
  79 static void
  80 soup_multipart_input_stream_finalize (GObject *object)
  81 {
  82         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (object);
  83
  84         g_free (multipart->priv->boundary);
  85
  86         if (multipart->priv->meta_buf)
  87                 g_clear_pointer (&multipart->priv->meta_buf, g_byte_array_unref);
  88
  89         G_OBJECT_CLASS (soup_multipart_input_stream_parent_class)->finalize (object);
  90 }
  91
  92 static void
  93 soup_multipart_input_stream_set_property (GObject *object, guint prop_id,
  94                                           const GValue *value, GParamSpec *pspec)
  95 {
  96         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (object);
  97
  98         switch (prop_id) {
  99         case PROP_MESSAGE:
 100                 multipart->priv->msg = g_value_dup_object (value);
 101                 break;
 102         default:
 103                 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 104                 break;
 105         }
 106 }
 107
 108 static void
 109 soup_multipart_input_stream_get_property (GObject *object, guint prop_id,
 110                                           GValue *value, GParamSpec *pspec)
 111 {
 112         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (object);
 113
 114         switch (prop_id) {
 115         case PROP_MESSAGE:
 116                 g_value_set_object (value, multipart->priv->msg);
 117                 break;
 118         default:
 119                 G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
 120                 break;
 121         }
 122 }
 123
 124 static gssize
 125 soup_multipart_input_stream_read_real (GInputStream     *stream,
 126                                        void             *buffer,
 127                                        gsize             count,
 128                                        gboolean          blocking,
 129                                        GCancellable     *cancellable,
 130                                        GError          **error)
 131 {
 132         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (stream);
 133         SoupMultipartInputStreamPrivate *priv = multipart->priv;
 134         gboolean got_boundary = FALSE;
 135         gssize nread = 0;
 136         guint8 *buf;
 137
 138         g_return_val_if_fail (priv->boundary != NULL, -1);
 139
 140         /* If we have received a Content-Length, and are not yet close to the end of
 141          * the part, let's not look for the boundary for now. This optimization is
 142          * necessary for keeping CPU usage civil.
 143          */
 144         if (priv->remaining_bytes > priv->boundary_size) {
 145                 goffset bytes_to_read = MIN (count, priv->remaining_bytes - priv->boundary_size);
 146
 147                 nread = g_pollable_stream_read (G_INPUT_STREAM (priv->base_stream),
 148                                                 buffer, bytes_to_read, blocking,
 149                                                 cancellable, error);
 150
 151                 if (nread > 0)
 152                         priv->remaining_bytes -= nread;
 153
 154                 return nread;
 155         }
 156
 157         if (priv->done_with_part)
 158                 return 0;
 159
 160         nread = soup_filter_input_stream_read_until (priv->base_stream, buffer, count,
 161                                                      priv->boundary, priv->boundary_size,
 162                                                      blocking, FALSE, &got_boundary,
 163                                                      cancellable, error);
 164
 165         if (nread <= 0)
 166                 return nread;
 167
 168         if (!got_boundary)
 169                 return nread;
 170
 171         priv->done_with_part = TRUE;
 172
 173         /* Ignore the newline that preceded the boundary. */
 174         if (nread == 1) {
 175                 buf = ((guint8*)buffer);
 176                 if (!memcmp (buf, "\n", 1))
 177                         nread -= 1;
 178         } else {
 179                 buf = ((guint8*)buffer) + nread - 2;
 180                 if (!memcmp (buf, "\r\n", 2))
 181                         nread -= 2;
 182                 else if (!memcmp (buf, "\n", 1))
 183                         nread -= 1;
 184         }
 185
 186         return nread;
 187 }
 188
 189 static gssize
 190 soup_multipart_input_stream_read (GInputStream  *stream,
 191                                   void          *buffer,
 192                                   gsize          count,
 193                                   GCancellable  *cancellable,
 194                                   GError       **error)
 195 {
 196         return soup_multipart_input_stream_read_real (stream, buffer, count,
 197                                                       TRUE, cancellable, error);
 198 }
 199
 200 static void
 201 soup_multipart_input_stream_init (SoupMultipartInputStream *multipart)
 202 {
 203         SoupMultipartInputStreamPrivate *priv;
 204         priv = multipart->priv = G_TYPE_INSTANCE_GET_PRIVATE (multipart,
 205                                                               SOUP_TYPE_MULTIPART_INPUT_STREAM,
 206                                                               SoupMultipartInputStreamPrivate);
 207
 208         priv->meta_buf = g_byte_array_sized_new (RESPONSE_BLOCK_SIZE);
 209         priv->done_with_part = FALSE;
 210 }
 211
 212 static void
 213 soup_multipart_input_stream_constructed (GObject *object)
 214 {
 215         SoupMultipartInputStream *multipart;
 216         SoupMultipartInputStreamPrivate *priv;
 217         GInputStream *base_stream;
 218         const char* boundary;
 219         GHashTable *params = NULL;
 220
 221         multipart = SOUP_MULTIPART_INPUT_STREAM (object);
 222         priv = multipart->priv;
 223
 224         base_stream = G_FILTER_INPUT_STREAM (multipart)->base_stream;
 225         priv->base_stream = SOUP_FILTER_INPUT_STREAM (soup_filter_input_stream_new (base_stream));
 226
 227         soup_message_headers_get_content_type (priv->msg->response_headers,
 228                                                &params);
 229
 230         boundary = g_hash_table_lookup (params, "boundary");
 231         if (boundary) {
 232                 if (g_str_has_prefix (boundary, "--"))
 233                         priv->boundary = g_strdup (boundary);
 234                 else
 235                         priv->boundary = g_strdup_printf ("--%s", boundary);
 236
 237                 priv->boundary_size = strlen (priv->boundary);
 238         } else {
 239                 g_warning ("No boundary found in message tagged as multipart.");
 240         }
 241
 242         g_hash_table_destroy (params);
 243
 244         if (G_OBJECT_CLASS (soup_multipart_input_stream_parent_class)->constructed)
 245                 G_OBJECT_CLASS (soup_multipart_input_stream_parent_class)->constructed (object);
 246 }
 247
 248 static gboolean
 249 soup_multipart_input_stream_is_readable (GPollableInputStream *stream)
 250 {
 251         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (stream);
 252         SoupMultipartInputStreamPrivate *priv = multipart->priv;
 253
 254         return g_pollable_input_stream_is_readable (G_POLLABLE_INPUT_STREAM (priv->base_stream));
 255 }
 256
 257 static gssize
 258 soup_multipart_input_stream_read_nonblocking (GPollableInputStream  *stream,
 259                                               void                  *buffer,
 260                                               gsize                  count,
 261                                               GError               **error)
 262 {
 263         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (stream);
 264
 265         return soup_multipart_input_stream_read_real (G_INPUT_STREAM (multipart),
 266                                                       buffer, count,
 267                                                       FALSE, NULL, error);
 268 }
 269
 270 static GSource *
 271 soup_multipart_input_stream_create_source (GPollableInputStream *stream,
 272                                            GCancellable         *cancellable)
 273 {
 274         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (stream);
 275         SoupMultipartInputStreamPrivate *priv = multipart->priv;
 276         GSource *base_source, *pollable_source;
 277
 278         base_source = g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM (priv->base_stream), cancellable);
 279
 280         pollable_source = g_pollable_source_new_full (stream, base_source, cancellable);
 281         g_source_unref (base_source);
 282
 283         return pollable_source;
 284 }
 285
 286 static void
 287 soup_multipart_input_stream_class_init (SoupMultipartInputStreamClass *multipart_class)
 288 {
 289         GObjectClass *object_class = G_OBJECT_CLASS (multipart_class);
 290         GInputStreamClass *input_stream_class =
 291                 G_INPUT_STREAM_CLASS (multipart_class);
 292
 293         g_type_class_add_private (multipart_class, sizeof (SoupMultipartInputStreamPrivate));
 294
 295         object_class->dispose = soup_multipart_input_stream_dispose;
 296         object_class->finalize = soup_multipart_input_stream_finalize;
 297         object_class->constructed = soup_multipart_input_stream_constructed;
 298         object_class->set_property = soup_multipart_input_stream_set_property;
 299         object_class->get_property = soup_multipart_input_stream_get_property;
 300
 301         input_stream_class->read_fn = soup_multipart_input_stream_read;
 302
 303         g_object_class_install_property (
 304                 object_class, PROP_MESSAGE,
 305                 g_param_spec_object ("message",
 306                                      "Message",
 307                                      "The SoupMessage",
 308                                      SOUP_TYPE_MESSAGE,
 309                                      G_PARAM_READWRITE | G_PARAM_CONSTRUCT_ONLY));
 310
 311 }
 312
 313 static void
 314 soup_multipart_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface,
 315                                            gpointer                       interface_data)
 316 {
 317         pollable_interface->is_readable = soup_multipart_input_stream_is_readable;
 318         pollable_interface->read_nonblocking = soup_multipart_input_stream_read_nonblocking;
 319         pollable_interface->create_source = soup_multipart_input_stream_create_source;
 320 }
 321
 322 static void
 323 soup_multipart_input_stream_parse_headers (SoupMultipartInputStream *multipart)
 324 {
 325         SoupMultipartInputStreamPrivate *priv = multipart->priv;
 326         gboolean success;
 327
 328         priv->current_headers = soup_message_headers_new (SOUP_MESSAGE_HEADERS_MULTIPART);
 329
 330         /* The part lacks headers, but is there. */
 331         if (!priv->meta_buf->len)
 332                 return;
 333
 334         success = soup_headers_parse ((const char*) priv->meta_buf->data,
 335                                       (int) priv->meta_buf->len,
 336                                       priv->current_headers);
 337
 338         if (success)
 339                 priv->remaining_bytes = soup_message_headers_get_content_length (priv->current_headers);
 340         else
 341                 g_clear_pointer (&priv->current_headers, soup_message_headers_free);
 342
 343         g_byte_array_remove_range (priv->meta_buf, 0, priv->meta_buf->len);
 344 }
 345
 346 static gboolean
 347 soup_multipart_input_stream_read_headers (SoupMultipartInputStream  *multipart,
 348                                           GCancellable              *cancellable,
 349                                           GError                   **error)
 350 {
 351         SoupMultipartInputStreamPrivate *priv = multipart->priv;
 352         guchar read_buf[RESPONSE_BLOCK_SIZE];
 353         guchar *buf;
 354         gboolean got_boundary = FALSE;
 355         gboolean got_lf = FALSE;
 356         gssize nread = 0;
 357
 358         g_return_val_if_fail (priv->boundary != NULL, TRUE);
 359
 360         g_clear_pointer (&priv->current_headers, soup_message_headers_free);
 361
 362         while (1) {
 363                 nread = soup_filter_input_stream_read_line (priv->base_stream, read_buf, sizeof (read_buf),
 364                                                             /* blocking */ TRUE, &got_lf, cancellable, error);
 365
 366                 if (nread <= 0)
 367                         break;
 368
 369                 g_byte_array_append (priv->meta_buf, read_buf, nread);
 370
 371                 /* Need to do this boundary check before checking for the line feed, since we
 372                  * may get the multipart end indicator without getting a new line.
 373                  */
 374                 if (!got_boundary &&
 375                     !strncmp ((char *)priv->meta_buf->data,
 376                               priv->boundary,
 377                               priv->boundary_size)) {
 378                         got_boundary = TRUE;
 379
 380                         /* Now check for possible multipart termination. */
 381                         buf = &read_buf[nread - 4];
 382                         if ((nread >= 4 && !memcmp (buf, "--\r\n", 4)) ||
 383                             (nread >= 3 && !memcmp (buf + 1, "--\n", 3)) ||
 384                             (nread >= 3 && !memcmp (buf + 2, "--", 2))) {
 385                                 g_byte_array_set_size (priv->meta_buf, 0);
 386                                 return FALSE;
 387                         }
 388                 }
 389
 390                 g_return_val_if_fail (got_lf, FALSE);
 391
 392                 /* Discard pre-boundary lines. */
 393                 if (!got_boundary) {
 394                         g_byte_array_set_size (priv->meta_buf, 0);
 395                         continue;
 396                 }
 397
 398                 if (nread == 1 &&
 399                     priv->meta_buf->len >= 2 &&
 400                     !strncmp ((char *)priv->meta_buf->data +
 401                               priv->meta_buf->len - 2,
 402                               "\n\n", 2))
 403                         break;
 404                 else if (nread == 2 &&
 405                          priv->meta_buf->len >= 3 &&
 406                          !strncmp ((char *)priv->meta_buf->data +
 407                                    priv->meta_buf->len - 3,
 408                                    "\n\r\n", 3))
 409                         break;
 410         }
 411
 412         return TRUE;
 413 }
 414
 415 /* Public APIs */
 416
 417 /**
 418  * soup_multipart_input_stream_new:
 419  * @msg: the #SoupMessage the response is related to.
 420  * @base_stream: the #GInputStream returned by sending the request.
 421  *
 422  * Creates a new #SoupMultipartInputStream that wraps the
 423  * #GInputStream obtained by sending the #SoupRequest. Reads should
 424  * not be done directly through this object, use the input streams
 425  * returned by soup_multipart_input_stream_next_part() or its async
 426  * counterpart instead.
 427  *
 428  * Returns: a new #SoupMultipartInputStream
 429  *
 430  * Since: 2.40
 431  **/
 432 SoupMultipartInputStream *
 433 soup_multipart_input_stream_new (SoupMessage  *msg,
 434                                  GInputStream *base_stream)
 435 {
 436         return g_object_new (SOUP_TYPE_MULTIPART_INPUT_STREAM,
 437                              "message", msg,
 438                              "base-stream", base_stream,
 439                              NULL);
 440 }
 441
 442 /**
 443  * soup_multipart_input_stream_next_part:
 444  * @multipart: the #SoupMultipartInputStream
 445  * @cancellable: a #GCancellable
 446  * @error: a #GError
 447  *
 448  * Obtains an input stream for the next part. When dealing with a
 449  * multipart response the input stream needs to be wrapped in a
 450  * #SoupMultipartInputStream and this function or its async
 451  * counterpart need to be called to obtain the first part for
 452  * reading.
 453  *
 454  * After calling this function,
 455  * soup_multipart_input_stream_get_headers() can be used to obtain the
 456  * headers for the first part. A read of 0 bytes indicates the end of
 457  * the part; a new call to this function should be done at that point,
 458  * to obtain the next part.
 459  *
 460  * Return value: (transfer full): a new #GInputStream, or %NULL if
 461  * there are no more parts
 462  *
 463  * Since: 2.40
 464  */
 465 GInputStream *
 466 soup_multipart_input_stream_next_part (SoupMultipartInputStream  *multipart,
 467                                        GCancellable              *cancellable,
 468                                        GError                   **error)
 469 {
 470         if (!soup_multipart_input_stream_read_headers (multipart, cancellable, error))
 471                 return NULL;
 472
 473         soup_multipart_input_stream_parse_headers (multipart);
 474
 475         multipart->priv->done_with_part = FALSE;
 476
 477         return G_INPUT_STREAM (g_object_new (SOUP_TYPE_BODY_INPUT_STREAM,
 478                                              "base-stream", G_INPUT_STREAM (multipart),
 479                                              "close-base-stream", FALSE,
 480                                              "encoding", SOUP_ENCODING_EOF,
 481                                              NULL));
 482
 483 }
 484
 485 static void
 486 soup_multipart_input_stream_next_part_thread (GTask        *task,
 487                                               gpointer      object,
 488                                               gpointer      task_data,
 489                                               GCancellable *cancellable)
 490 {
 491         SoupMultipartInputStream *multipart = SOUP_MULTIPART_INPUT_STREAM (object);
 492         GError *error = NULL;
 493         GInputStream *new_stream;
 494
 495         new_stream = soup_multipart_input_stream_next_part (multipart, cancellable, &error);
 496
 497         g_input_stream_clear_pending (G_INPUT_STREAM (multipart));
 498
 499         if (error)
 500                 g_task_return_error (task, error);
 501         else
 502                 g_task_return_pointer (task, new_stream, g_object_unref);
 503 }
 504
 505 /**
 506  * soup_multipart_input_stream_next_part_async:
 507  * @multipart: the #SoupMultipartInputStream.
 508  * @io_priority: the I/O priority for the request.
 509  * @cancellable: a #GCancellable.
 510  * @callback: callback to call when request is satisfied.
 511  * @data: data for @callback
 512  *
 513  * Obtains a #GInputStream for the next request. See
 514  * soup_multipart_input_stream_next_part() for details on the
 515  * workflow.
 516  *
 517  * Since: 2.40
 518  */
 519 void
 520 soup_multipart_input_stream_next_part_async (SoupMultipartInputStream *multipart,
 521                                              int                       io_priority,
 522                                              GCancellable             *cancellable,
 523                                              GAsyncReadyCallback       callback,
 524                                              gpointer                  data)
 525 {
 526         GInputStream *stream = G_INPUT_STREAM (multipart);
 527         GTask *task;
 528         GError *error = NULL;
 529
 530         g_return_if_fail (SOUP_IS_MULTIPART_INPUT_STREAM (multipart));
 531
 532         task = g_task_new (multipart, cancellable, callback, data);
 533         g_task_set_priority (task, io_priority);
 534
 535         if (!g_input_stream_set_pending (stream, &error)) {
 536                 g_task_return_error (task, error);
 537                 g_object_unref (task);
 538                 return;
 539         }
 540
 541         g_task_run_in_thread (task, soup_multipart_input_stream_next_part_thread);
 542         g_object_unref (task);
 543 }
 544
 545 /**
 546  * soup_multipart_input_stream_next_part_finish:
 547  * @multipart: a #SoupMultipartInputStream.
 548  * @result: a #GAsyncResult.
 549  * @error: a #GError location to store any error, or NULL to ignore.
 550  *
 551  * Finishes an asynchronous request for the next part.
 552  *
 553  * Return value: (transfer full): a newly created #GInputStream for
 554  * reading the next part or %NULL if there are no more parts.
 555  *
 556  * Since: 2.40
 557  */
 558 GInputStream *
 559 soup_multipart_input_stream_next_part_finish (SoupMultipartInputStream  *multipart,
 560                                               GAsyncResult              *result,
 561                                               GError                   **error)
 562 {
 563         g_return_val_if_fail (g_task_is_valid (result, multipart), FALSE);
 564
 565         return g_task_propagate_pointer (G_TASK (result), error);
 566 }
 567
 568 /**
 569  * soup_multipart_input_stream_get_headers:
 570  * @multipart: a #SoupMultipartInputStream.
 571  *
 572  * Obtains the headers for the part currently being processed. Note
 573  * that the #SoupMessageHeaders that are returned are owned by the
 574  * #SoupMultipartInputStream and will be replaced when a call is made
 575  * to soup_multipart_input_stream_next_part() or its async
 576  * counterpart, so if keeping the headers is required, a copy must be
 577  * made.
 578  *
 579  * Note that if a part had no headers at all an empty #SoupMessageHeaders
 580  * will be returned.
 581  *
 582  * Return value: (transfer none): a #SoupMessageHeaders containing the headers
 583  * for the part currently being processed or %NULL if the headers failed to
 584  * parse.
 585  *
 586  * Since: 2.40
 587  */
 588 SoupMessageHeaders *
 589 soup_multipart_input_stream_get_headers (SoupMultipartInputStream *multipart)
 590 {
 591         return multipart->priv->current_headers;
 592 }