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_ADD_PRIVATE (SoupFilterInputStream)
  33                          G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM,
  34                                                 soup_filter_input_stream_pollable_init))
  35
  36 static void
  37 soup_filter_input_stream_init (SoupFilterInputStream *stream)
  38 {
  39         stream->priv = soup_filter_input_stream_get_instance_private (stream);
  40 }
  41
  42 static void
  43 soup_filter_input_stream_finalize (GObject *object)
  44 {
  45         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (object);
  46
  47         g_clear_pointer (&fstream->priv->buf, g_byte_array_unref);
  48
  49         G_OBJECT_CLASS (soup_filter_input_stream_parent_class)->finalize (object);
  50 }
  51
  52 static gssize
  53 read_from_buf (SoupFilterInputStream *fstream, gpointer buffer, gsize count)
  54 {
  55         GByteArray *buf = fstream->priv->buf;
  56
  57         if (buf->len < count)
  58                 count = buf->len;
  59         memcpy (buffer, buf->data, count);
  60
  61         if (count == buf->len) {
  62                 g_byte_array_free (buf, TRUE);
  63                 fstream->priv->buf = NULL;
  64         } else {
  65                 memmove (buf->data, buf->data + count,
  66                          buf->len - count);
  67                 g_byte_array_set_size (buf, buf->len - count);
  68         }
  69
  70         return count;
  71 }
  72
  73 static gssize
  74 soup_filter_input_stream_read_fn (GInputStream  *stream,
  75                                   void          *buffer,
  76                                   gsize          count,
  77                                   GCancellable  *cancellable,
  78                                   GError       **error)
  79 {
  80         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (stream);
  81
  82         if (!fstream->priv->in_read_until)
  83                 fstream->priv->need_more = FALSE;
  84
  85         if (fstream->priv->buf && !fstream->priv->in_read_until) {
  86                 return read_from_buf (fstream, buffer, count);
  87         } else {
  88                 return g_pollable_stream_read (G_FILTER_INPUT_STREAM (fstream)->base_stream,
  89                                                buffer, count,
  90                                                TRUE, cancellable, error);
  91         }
  92 }
  93
  94 static gboolean
  95 soup_filter_input_stream_is_readable (GPollableInputStream *stream)
  96 {
  97         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (stream);
  98
  99         if (fstream->priv->buf && !fstream->priv->need_more)
 100                 return TRUE;
 101         else
 102                 return g_pollable_input_stream_is_readable (G_POLLABLE_INPUT_STREAM (G_FILTER_INPUT_STREAM (fstream)->base_stream));
 103 }
 104
 105 static gssize
 106 soup_filter_input_stream_read_nonblocking (GPollableInputStream  *stream,
 107                                            void                  *buffer,
 108                                            gsize                  count,
 109                                            GError               **error)
 110 {
 111         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (stream);
 112
 113         if (!fstream->priv->in_read_until)
 114                 fstream->priv->need_more = FALSE;
 115
 116         if (fstream->priv->buf && !fstream->priv->in_read_until) {
 117                 return read_from_buf (fstream, buffer, count);
 118         } else {
 119                 return g_pollable_stream_read (G_FILTER_INPUT_STREAM (fstream)->base_stream,
 120                                                buffer, count,
 121                                                FALSE, NULL, error);
 122         }
 123 }
 124
 125 static GSource *
 126 soup_filter_input_stream_create_source (GPollableInputStream *stream,
 127                                         GCancellable         *cancellable)
 128 {
 129         SoupFilterInputStream *fstream = SOUP_FILTER_INPUT_STREAM (stream);
 130         GSource *base_source, *pollable_source;
 131
 132         if (fstream->priv->buf && !fstream->priv->need_more)
 133                 base_source = g_timeout_source_new (0);
 134         else
 135                 base_source = g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM (G_FILTER_INPUT_STREAM (fstream)->base_stream), cancellable);
 136
 137         g_source_set_dummy_callback (base_source);
 138         pollable_source = g_pollable_source_new (G_OBJECT (stream));
 139         g_source_add_child_source (pollable_source, base_source);
 140         g_source_unref (base_source);
 141
 142         return pollable_source;
 143 }
 144
 145 static void
 146 soup_filter_input_stream_class_init (SoupFilterInputStreamClass *stream_class)
 147 {
 148         GObjectClass *object_class = G_OBJECT_CLASS (stream_class);
 149         GInputStreamClass *input_stream_class = G_INPUT_STREAM_CLASS (stream_class);
 150
 151         object_class->finalize = soup_filter_input_stream_finalize;
 152
 153         input_stream_class->read_fn = soup_filter_input_stream_read_fn;
 154 }
 155
 156 static void
 157 soup_filter_input_stream_pollable_init (GPollableInputStreamInterface *pollable_interface,
 158                                         gpointer                       interface_data)
 159 {
 160         pollable_interface->is_readable = soup_filter_input_stream_is_readable;
 161         pollable_interface->read_nonblocking = soup_filter_input_stream_read_nonblocking;
 162         pollable_interface->create_source = soup_filter_input_stream_create_source;
 163 }
 164
 165 GInputStream *
 166 soup_filter_input_stream_new (GInputStream *base_stream)
 167 {
 168         return g_object_new (SOUP_TYPE_FILTER_INPUT_STREAM,
 169                              "base-stream", base_stream,
 170                              "close-base-stream", FALSE,
 171                              NULL);
 172 }
 173
 174 gssize
 175 soup_filter_input_stream_read_line (SoupFilterInputStream  *fstream,
 176                                     void                   *buffer,
 177                                     gsize                   length,
 178                                     gboolean                blocking,
 179                                     gboolean               *got_line,
 180                                     GCancellable           *cancellable,
 181                                     GError                **error)
 182 {
 183         return soup_filter_input_stream_read_until (fstream, buffer, length,
 184                                                     "\n", 1, blocking,
 185                                                     TRUE, got_line,
 186                                                     cancellable, error);
 187 }
 188
 189 gssize
 190 soup_filter_input_stream_read_until (SoupFilterInputStream  *fstream,
 191                                      void                   *buffer,
 192                                      gsize                   length,
 193                                      const void             *boundary,
 194                                      gsize                   boundary_length,
 195                                      gboolean                blocking,
 196                                      gboolean                include_boundary,
 197                                      gboolean               *got_boundary,
 198                                      GCancellable           *cancellable,
 199                                      GError                **error)
 200 {
 201         gssize nread, read_length;
 202         guint8 *p, *buf, *end;
 203         gboolean eof = FALSE;
 204         GError *my_error = NULL;
 205
 206         g_return_val_if_fail (SOUP_IS_FILTER_INPUT_STREAM (fstream), -1);
 207         g_return_val_if_fail (!include_boundary || (boundary_length < length), -1);
 208
 209         *got_boundary = FALSE;
 210         fstream->priv->need_more = FALSE;
 211
 212         if (!fstream->priv->buf || fstream->priv->buf->len < boundary_length) {
 213                 guint prev_len;
 214
 215         fill_buffer:
 216                 if (!fstream->priv->buf)
 217                         fstream->priv->buf = g_byte_array_new ();
 218                 prev_len = fstream->priv->buf->len;
 219                 g_byte_array_set_size (fstream->priv->buf, length);
 220                 buf = fstream->priv->buf->data;
 221
 222                 fstream->priv->in_read_until = TRUE;
 223                 nread = g_pollable_stream_read (G_INPUT_STREAM (fstream),
 224                                                 buf + prev_len, length - prev_len,
 225                                                 blocking,
 226                                                 cancellable, &my_error);
 227                 fstream->priv->in_read_until = FALSE;
 228                 if (nread <= 0) {
 229                         if (prev_len)
 230                                 fstream->priv->buf->len = prev_len;
 231                         else {
 232                                 g_byte_array_free (fstream->priv->buf, TRUE);
 233                                 fstream->priv->buf = NULL;
 234                         }
 235
 236                         if (nread == 0 && prev_len)
 237                                 eof = TRUE;
 238                         else {
 239                                 if (g_error_matches (my_error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK))
 240                                         fstream->priv->need_more = TRUE;
 241                                 if (my_error)
 242                                         g_propagate_error (error, my_error);
 243
 244                                 return nread;
 245                         }
 246
 247                         if (my_error)
 248                                 g_propagate_error (error, my_error);
 249                 } else
 250                         fstream->priv->buf->len = prev_len + nread;
 251         } else
 252                 buf = fstream->priv->buf->data;
 253
 254         /* Scan for the boundary within the range we can possibly return. */
 255         if (include_boundary)
 256                 end = buf + MIN (fstream->priv->buf->len, length) - boundary_length;
 257         else
 258                 end = buf + MIN (fstream->priv->buf->len - boundary_length, length);
 259         for (p = buf; p <= end; p++) {
 260                 if (*p == *(guint8*)boundary &&
 261                     !memcmp (p, boundary, boundary_length)) {
 262                         if (include_boundary)
 263                                 p += boundary_length;
 264                         *got_boundary = TRUE;
 265                         break;
 266                 }
 267         }
 268
 269         if (!*got_boundary && fstream->priv->buf->len < length && !eof)
 270                 goto fill_buffer;
 271
 272         if (eof && !*got_boundary)
 273                 read_length = MIN (fstream->priv->buf->len, length);
 274         else
 275                 read_length = p - buf;
 276         return read_from_buf (fstream, buffer, read_length);
 277 }