libsoup/soup-filter-input-stream.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-filter-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 <string.h>
  13
  14 #include "soup-filter-input-stream.h"
  15 #include "soup.h"
  16
  17 /* This is essentially a subset of GDataInputStream, except that we
  18  * can do the equivalent of "fill_nonblocking()" on it. (We could use
  19  * an actual GDataInputStream, and implement the nonblocking semantics
  20  * via fill_async(), but that would be more work...)
  21  */
  22
  23 struct _SoupFilterInputStreamPrivate {
  24         GByteArray *buf;
  25         gboolean need_more;
  26         gboolean in_read_until;
  27 };
  28
  29 static void soup_filter_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface, gpointer interface_data);
  30
  31 G_DEFINE_TYPE_WITH_CODE (SoupFilterInputStream, soup_filter_input_stream, G_TYPE_FILTER_INPUT_STREAM,
  32                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM,
  33                                                 soup_filter_input_stream_pollable_init))
  34
  35 static void
  36 soup_filter_input_stream_init (SoupFilterInputStream *stream)
  37 {
  38         stream->priv = G_TYPE_INSTANCE_GET_PRIVATE (stream,
  39                                                     SOUP_TYPE_FILTER_INPUT_STREAM,
  40                                                     SoupFilterInputStreamPrivate);
  41 }
  42
  43 static void
  44 soup_filter_input_stream_finalize (GObject *object)
  45 {
  46         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (object);
  47
  48         g_clear_pointer (&fstream->priv->buf, g_byte_array_unref);
  49
  50         G_OBJECT_CLASS (soup_filter_input_stream_parent_class)->finalize (object);
  51 }
  52
  53 static gssize
  54 read_from_buf (SoupFilterInputStream *fstream, gpointer buffer, gsize count)
  55 {
  56         GByteArray *buf = fstream->priv->buf;
  57
  58         if (buf->len < count)
  59                 count = buf->len;
  60         memcpy (buffer, buf->data, count);
  61
  62         if (count == buf->len) {
  63                 g_byte_array_free (buf, TRUE);
  64                 fstream->priv->buf = NULL;
  65         } else {
  66                 memmove (buf->data, buf->data + count,
  67                          buf->len - count);
  68                 g_byte_array_set_size (buf, buf->len - count);
  69         }
  70
  71         return count;
  72 }
  73
  74 static gssize
  75 soup_filter_input_stream_read_fn (GInputStream  *stream,
  76                                   void          *buffer,
  77                                   gsize          count,
  78                                   GCancellable  *cancellable,
  79                                   GError       **error)
  80 {
  81         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (stream);
  82
  83         if (!fstream->priv->in_read_until)
  84                 fstream->priv->need_more = FALSE;
  85
  86         if (fstream->priv->buf && !fstream->priv->in_read_until) {
  87                 return read_from_buf (fstream, buffer, count);
  88         } else {
  89                 return g_pollable_stream_read (G_FILTER_INPUT_STREAM (fstream)->base_stream,
  90                                                buffer, count,
  91                                                TRUE, cancellable, error);
  92         }
  93 }
  94
  95 static gboolean
  96 soup_filter_input_stream_is_readable (GPollableInputStream *stream)
  97 {
  98         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (stream);
  99
 100         if (fstream->priv->buf && !fstream->priv->need_more)
 101                 return TRUE;
 102         else
 103                 return g_pollable_input_stream_is_readable (G_POLLABLE_INPUT_STREAM (G_FILTER_INPUT_STREAM (fstream)->base_stream));
 104 }
 105
 106 static gssize
 107 soup_filter_input_stream_read_nonblocking (GPollableInputStream  *stream,
 108                                            void                  *buffer,
 109                                            gsize                  count,
 110                                            GError               **error)
 111 {
 112         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (stream);
 113
 114         if (!fstream->priv->in_read_until)
 115                 fstream->priv->need_more = FALSE;
 116
 117         if (fstream->priv->buf && !fstream->priv->in_read_until) {
 118                 return read_from_buf (fstream, buffer, count);
 119         } else {
 120                 return g_pollable_stream_read (G_FILTER_INPUT_STREAM (fstream)->base_stream,
 121                                                buffer, count,
 122                                                FALSE, NULL, error);
 123         }
 124 }
 125
 126 static GSource *
 127 soup_filter_input_stream_create_source (GPollableInputStream *stream,
 128                                         GCancellable         *cancellable)
 129 {
 130         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (stream);
 131         GSource *base_source, *pollable_source;
 132
 133         if (fstream->priv->buf && !fstream->priv->need_more)
 134                 base_source = g_timeout_source_new (0);
 135         else
 136                 base_source = g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM (G_FILTER_INPUT_STREAM (fstream)->base_stream), cancellable);
 137
 138         g_source_set_dummy_callback (base_source);
 139         pollable_source = g_pollable_source_new (G_OBJECT (stream));
 140         g_source_add_child_source (pollable_source, base_source);
 141         g_source_unref (base_source);
 142
 143         return pollable_source;
 144 }
 145
 146 static void
 147 soup_filter_input_stream_class_init (SoupFilterInputStreamClass *stream_class)
 148 {
 149         GObjectClass *object_class = G_OBJECT_CLASS (stream_class);
 150         GInputStreamClass *input_stream_class = G_INPUT_STREAM_CLASS (stream_class);
 151
 152         g_type_class_add_private (stream_class, sizeof (SoupFilterInputStreamPrivate));
 153
 154         object_class->finalize = soup_filter_input_stream_finalize;
 155
 156         input_stream_class->read_fn = soup_filter_input_stream_read_fn;
 157 }
 158
 159 static void
 160 soup_filter_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface,
 161                                         gpointer                       interface_data)
 162 {
 163         pollable_interface->is_readable = soup_filter_input_stream_is_readable;
 164         pollable_interface->read_nonblocking = soup_filter_input_stream_read_nonblocking;
 165         pollable_interface->create_source = soup_filter_input_stream_create_source;
 166 }
 167
 168 GInputStream *
 169 soup_filter_input_stream_new (GInputStream *base_stream)
 170 {
 171         return g_object_new (SOUP_TYPE_FILTER_INPUT_STREAM,
 172                              "base-stream", base_stream,
 173                              "close-base-stream", FALSE,
 174                              NULL);
 175 }
 176
 177 gssize
 178 soup_filter_input_stream_read_line (SoupFilterInputStream  *fstream,
 179                                     void                   *buffer,
 180                                     gsize                   length,
 181                                     gboolean                blocking,
 182                                     gboolean               *got_line,
 183                                     GCancellable           *cancellable,
 184                                     GError                **error)
 185 {
 186         return soup_filter_input_stream_read_until (fstream, buffer, length,
 187                                                     "\n", 1, blocking,
 188                                                     TRUE, got_line,
 189                                                     cancellable, error);
 190 }
 191
 192 gssize
 193 soup_filter_input_stream_read_until (SoupFilterInputStream  *fstream,
 194                                      void                   *buffer,
 195                                      gsize                   length,
 196                                      const void             *boundary,
 197                                      gsize                   boundary_length,
 198                                      gboolean                blocking,
 199                                      gboolean                include_boundary,
 200                                      gboolean               *got_boundary,
 201                                      GCancellable           *cancellable,
 202                                      GError                **error)
 203 {
 204         gssize nread;
 205         guint8 *p, *buf, *end;
 206         gboolean eof = FALSE;
 207         GError *my_error = NULL;
 208
 209         g_return_val_if_fail (SOUP_IS_FILTER_INPUT_STREAM (fstream), -1);
 210         g_return_val_if_fail (!include_boundary || (boundary_length < length), -1);
 211
 212         *got_boundary = FALSE;
 213         fstream->priv->need_more = FALSE;
 214
 215         if (!fstream->priv->buf || fstream->priv->buf->len < boundary_length) {
 216                 guint prev_len;
 217
 218         fill_buffer:
 219                 if (!fstream->priv->buf)
 220                         fstream->priv->buf = g_byte_array_new ();
 221                 prev_len = fstream->priv->buf->len;
 222                 g_byte_array_set_size (fstream->priv->buf, length);
 223                 buf = fstream->priv->buf->data;
 224
 225                 fstream->priv->in_read_until = TRUE;
 226                 nread = g_pollable_stream_read (G_INPUT_STREAM (fstream),
 227                                                 buf + prev_len, length - prev_len,
 228                                                 blocking,
 229                                                 cancellable, &my_error);
 230                 fstream->priv->in_read_until = FALSE;
 231                 if (nread <= 0) {
 232                         if (prev_len)
 233                                 fstream->priv->buf->len = prev_len;
 234                         else {
 235                                 g_byte_array_free (fstream->priv->buf, TRUE);
 236                                 fstream->priv->buf = NULL;
 237                         }
 238
 239                         if (nread == 0 && prev_len)
 240                                 eof = TRUE;
 241                         else {
 242                                 if (g_error_matches (my_error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
 243                                         fstream->priv->need_more = TRUE;
 244                                 if (my_error)
 245                                         g_propagate_error (error, my_error);
 246
 247                                 return nread;
 248                         }
 249
 250                         if (my_error)
 251                                 g_propagate_error (error, my_error);
 252                 } else
 253                         fstream->priv->buf->len = prev_len + nread;
 254         } else
 255                 buf = fstream->priv->buf->data;
 256
 257         /* Scan for the boundary */
 258         end = buf + fstream->priv->buf->len;
 259         if (!eof)
 260                 end -= boundary_length;
 261         for (p = buf; p <= end; p++) {
 262                 if (*p == *(guint8*)boundary &&
 263                     !memcmp (p, boundary, boundary_length)) {
 264                         if (include_boundary)
 265                                 p += boundary_length;
 266                         *got_boundary = TRUE;
 267                         break;
 268                 }
 269         }
 270
 271         if (!*got_boundary && fstream->priv->buf->len < length && !eof)
 272                 goto fill_buffer;
 273
 274         /* Return everything up to 'p' (which is either just after the boundary if
 275          * include_boundary is TRUE, just before the boundary if include_boundary is
 276          * FALSE, @boundary_len - 1 bytes before the end of the buffer, or end-of-
 277          * file).
 278          */
 279         return read_from_buf (fstream, buffer, p - buf);
 280 }