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