summaryrefslogtreecommitdiff
path: root/tests/chunk-io-test.c
diff options
context:
space:
mode:
Diffstat (limited to 'tests/chunk-io-test.c')
-rw-r--r--tests/chunk-io-test.c610
1 files changed, 610 insertions, 0 deletions
diff --git a/tests/chunk-io-test.c b/tests/chunk-io-test.c
new file mode 100644
index 00000000..1e53eef1
--- /dev/null
+++ b/tests/chunk-io-test.c
@@ -0,0 +1,610 @@
+/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
+/*
+ * Copyright 2013 Red Hat, Inc.
+ */
+
+#include "test-utils.h"
+
+static void
+force_io_streams_init (void)
+{
+ SoupServer *server;
+ SoupSession *session;
+ guint port;
+ SoupURI *base_uri;
+ SoupMessage *msg;
+
+ /* Poke libsoup enough to cause SoupBodyInputStream and
+ * SoupBodyOutputStream to get defined, so we can find them
+ * via g_type_from_name() later.
+ */
+
+ server = soup_test_server_new (TRUE);
+ port = soup_server_get_port (server);
+
+ base_uri = soup_uri_new ("http://127.0.0.1");
+ soup_uri_set_port (base_uri, port);
+
+ session = soup_test_session_new (SOUP_TYPE_SESSION, NULL);
+ msg = soup_message_new_from_uri ("POST", base_uri);
+ soup_session_send_message (session, msg);
+ g_object_unref (msg);
+ soup_test_session_abort_unref (session);
+
+ soup_uri_free (base_uri);
+ soup_test_server_quit_unref (server);
+}
+
+typedef struct {
+ GFilterInputStream grandparent;
+
+ gpointer *soup_filter_input_stream_private;
+
+ gboolean is_readable;
+} SlowInputStream;
+
+typedef struct {
+ GFilterInputStreamClass grandparent;
+} SlowInputStreamClass;
+
+GType slow_input_stream_get_type (void);
+static void slow_pollable_input_stream_init (GPollableInputStreamInterface *pollable_interface,
+ gpointer interface_data);
+
+G_DEFINE_TYPE_WITH_CODE (SlowInputStream, slow_input_stream,
+ g_type_from_name ("SoupFilterInputStream"),
+ G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_INPUT_STREAM, slow_pollable_input_stream_init);
+ )
+
+static void
+slow_input_stream_init (SlowInputStream *sis)
+{
+}
+
+static gssize
+slow_input_stream_read (GInputStream *stream,
+ void *buffer,
+ gsize count,
+ GCancellable *cancellable,
+ GError **error)
+{
+ return g_input_stream_read (G_FILTER_INPUT_STREAM (stream)->base_stream,
+ buffer, 1, cancellable, error);
+}
+
+static void
+slow_input_stream_class_init (SlowInputStreamClass *sisclass)
+{
+ GInputStreamClass *input_stream_class = G_INPUT_STREAM_CLASS (sisclass);
+
+ input_stream_class->read_fn = slow_input_stream_read;
+}
+
+static gboolean
+slow_input_stream_is_readable (GPollableInputStream *stream)
+{
+ return ((SlowInputStream *)stream)->is_readable;
+}
+
+static gssize
+slow_input_stream_read_nonblocking (GPollableInputStream *stream,
+ void *buffer,
+ gsize count,
+ GError **error)
+{
+ if (((SlowInputStream *)stream)->is_readable) {
+ ((SlowInputStream *)stream)->is_readable = FALSE;
+ return slow_input_stream_read (G_INPUT_STREAM (stream), buffer, count,
+ NULL, error);
+ } else {
+ g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK,
+ "would block");
+ return -1;
+ }
+}
+
+static GSource *
+slow_input_stream_create_source (GPollableInputStream *stream,
+ GCancellable *cancellable)
+{
+ GSource *base_source, *pollable_source;
+
+ ((SlowInputStream *)stream)->is_readable = TRUE;
+ base_source = g_timeout_source_new (0);
+ g_source_set_dummy_callback (base_source);
+
+ pollable_source = g_pollable_source_new (G_OBJECT (stream));
+ g_source_add_child_source (pollable_source, base_source);
+ g_source_unref (base_source);
+
+ return pollable_source;
+}
+
+static void
+slow_pollable_input_stream_init (GPollableInputStreamInterface *pollable_interface,
+ gpointer interface_data)
+{
+ pollable_interface->is_readable = slow_input_stream_is_readable;
+ pollable_interface->read_nonblocking = slow_input_stream_read_nonblocking;
+ pollable_interface->create_source = slow_input_stream_create_source;
+}
+
+typedef struct {
+ GFilterOutputStream parent;
+
+ gboolean is_writable;
+} SlowOutputStream;
+
+typedef struct {
+ GFilterOutputStreamClass parent;
+} SlowOutputStreamClass;
+
+GType slow_output_stream_get_type (void);
+
+static void slow_pollable_output_stream_init (GPollableOutputStreamInterface *pollable_interface,
+ gpointer interface_data);
+
+G_DEFINE_TYPE_WITH_CODE (SlowOutputStream, slow_output_stream,
+ g_type_from_name ("GFilterOutputStream"),
+ G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_OUTPUT_STREAM, slow_pollable_output_stream_init);
+ )
+
+static void
+slow_output_stream_init (SlowOutputStream *sis)
+{
+}
+
+static gssize
+slow_output_stream_write (GOutputStream *stream,
+ const void *buffer,
+ gsize count,
+ GCancellable *cancellable,
+ GError **error)
+{
+ return g_output_stream_write (G_FILTER_OUTPUT_STREAM (stream)->base_stream,
+ buffer, 1, cancellable, error);
+}
+
+static void
+slow_output_stream_class_init (SlowOutputStreamClass *sisclass)
+{
+ GOutputStreamClass *output_stream_class = G_OUTPUT_STREAM_CLASS (sisclass);
+
+ output_stream_class->write_fn = slow_output_stream_write;
+}
+
+static gboolean
+slow_output_stream_is_writable (GPollableOutputStream *stream)
+{
+ return ((SlowOutputStream *)stream)->is_writable;
+}
+
+static gssize
+slow_output_stream_write_nonblocking (GPollableOutputStream *stream,
+ const void *buffer,
+ gsize count,
+ GError **error)
+{
+ if (((SlowOutputStream *)stream)->is_writable) {
+ ((SlowOutputStream *)stream)->is_writable = FALSE;
+ return slow_output_stream_write (G_OUTPUT_STREAM (stream), buffer, count,
+ NULL, error);
+ } else {
+ g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK,
+ "would block");
+ return -1;
+ }
+}
+
+static GSource *
+slow_output_stream_create_source (GPollableOutputStream *stream,
+ GCancellable *cancellable)
+{
+ GSource *base_source, *pollable_source;
+
+ ((SlowOutputStream *)stream)->is_writable = TRUE;
+ base_source = g_timeout_source_new (0);
+ g_source_set_dummy_callback (base_source);
+
+ pollable_source = g_pollable_source_new (G_OBJECT (stream));
+ g_source_add_child_source (pollable_source, base_source);
+ g_source_unref (base_source);
+
+ return pollable_source;
+}
+
+static void
+slow_pollable_output_stream_init (GPollableOutputStreamInterface *pollable_interface,
+ gpointer interface_data)
+{
+ pollable_interface->is_writable = slow_output_stream_is_writable;
+ pollable_interface->write_nonblocking = slow_output_stream_write_nonblocking;
+ pollable_interface->create_source = slow_output_stream_create_source;
+}
+
+typedef struct {
+ GFilterOutputStream parent;
+
+ gboolean is_broken;
+} BreakingOutputStream;
+
+typedef struct {
+ GFilterOutputStreamClass parent;
+} BreakingOutputStreamClass;
+
+GType breaking_output_stream_get_type (void);
+
+static void breaking_pollable_output_stream_init (GPollableOutputStreamInterface *pollable_interface,
+ gpointer interface_data);
+
+G_DEFINE_TYPE_WITH_CODE (BreakingOutputStream, breaking_output_stream,
+ g_type_from_name ("GFilterOutputStream"),
+ G_IMPLEMENT_INTERFACE (G_TYPE_POLLABLE_OUTPUT_STREAM, breaking_pollable_output_stream_init);
+ )
+
+static void
+breaking_output_stream_init (BreakingOutputStream *sis)
+{
+}
+
+static gssize
+breaking_output_stream_write (GOutputStream *stream,
+ const void *buffer,
+ gsize count,
+ GCancellable *cancellable,
+ GError **error)
+{
+ if (((BreakingOutputStream *)stream)->is_broken) {
+ g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_FAILED, "failed");
+ return -1;
+ }
+
+ if (count > 128) {
+ ((BreakingOutputStream *)stream)->is_broken = TRUE;
+ count /= 2;
+ }
+ return g_output_stream_write (G_FILTER_OUTPUT_STREAM (stream)->base_stream,
+ buffer, count, cancellable, error);
+}
+
+static void
+breaking_output_stream_class_init (BreakingOutputStreamClass *sisclass)
+{
+ GOutputStreamClass *output_stream_class = G_OUTPUT_STREAM_CLASS (sisclass);
+
+ output_stream_class->write_fn = breaking_output_stream_write;
+}
+
+static gboolean
+breaking_output_stream_is_writable (GPollableOutputStream *stream)
+{
+ return TRUE;
+}
+
+static gssize
+breaking_output_stream_write_nonblocking (GPollableOutputStream *stream,
+ const void *buffer,
+ gsize count,
+ GError **error)
+{
+ if (((BreakingOutputStream *)stream)->is_broken) {
+ g_set_error_literal (error, G_IO_ERROR, G_IO_ERROR_FAILED, "failed");
+ return -1;
+ }
+
+ if (count > 128) {
+ ((BreakingOutputStream *)stream)->is_broken = TRUE;
+ count /= 2;
+ }
+ return g_pollable_output_stream_write_nonblocking (G_POLLABLE_OUTPUT_STREAM (G_FILTER_OUTPUT_STREAM (stream)->base_stream),
+ buffer, count, NULL, error);
+}
+
+static GSource *
+breaking_output_stream_create_source (GPollableOutputStream *stream,
+ GCancellable *cancellable)
+{
+ GSource *base_source, *pollable_source;
+
+ base_source = g_timeout_source_new (0);
+ g_source_set_dummy_callback (base_source);
+
+ pollable_source = g_pollable_source_new (G_OBJECT (stream));
+ g_source_add_child_source (pollable_source, base_source);
+ g_source_unref (base_source);
+
+ return pollable_source;
+}
+
+static void
+breaking_pollable_output_stream_init (GPollableOutputStreamInterface *pollable_interface,
+ gpointer interface_data)
+{
+ pollable_interface->is_writable = breaking_output_stream_is_writable;
+ pollable_interface->write_nonblocking = breaking_output_stream_write_nonblocking;
+ pollable_interface->create_source = breaking_output_stream_create_source;
+}
+
+#define CHUNK_SIZE 1024
+
+static GString *
+chunkify (const char *str, gsize length)
+{
+ GString *gstr;
+ int i, size;
+
+ gstr = g_string_new (NULL);
+ for (i = 0; i < length; i += CHUNK_SIZE) {
+ size = MIN (CHUNK_SIZE, length - i);
+ g_string_append_printf (gstr, "%x\r\n", size);
+ g_string_append_len (gstr, str + i, size);
+ g_string_append (gstr, "\r\n");
+ }
+ g_string_append (gstr, "0\r\n\r\n");
+
+ return gstr;
+}
+
+static void
+do_io_tests (void)
+{
+ GInputStream *imem, *islow, *in;
+ GOutputStream *omem, *oslow, *out;
+ GMemoryOutputStream *mem;
+ SoupBuffer *raw_contents;
+ char *buf;
+ GString *chunkified;
+ GError *error = NULL;
+ gssize nread, nwrote, total;
+ gssize chunk_length, chunk_total;
+
+ raw_contents = soup_test_get_index ();
+ chunkified = chunkify (raw_contents->data, raw_contents->length);
+
+ debug_printf (1, " sync read\n");
+
+ imem = g_memory_input_stream_new_from_data (chunkified->str, chunkified->len, NULL);
+ islow = g_object_new (slow_input_stream_get_type (),
+ "base-stream", imem,
+ "close-base-stream", TRUE,
+ NULL);
+ in = g_object_new (g_type_from_name ("SoupBodyInputStream"),
+ "base-stream", islow,
+ "close-base-stream", TRUE,
+ "encoding", SOUP_ENCODING_CHUNKED,
+ NULL);
+ g_object_unref (imem);
+ g_object_unref (islow);
+
+ buf = g_malloc (raw_contents->length);
+ total = 0;
+ while (TRUE) {
+ nread = g_input_stream_read (in, buf + total,
+ raw_contents->length - total,
+ NULL, &error);
+ g_assert_no_error (error);
+ g_clear_error (&error);
+ if (nread > 0)
+ total += nread;
+ else
+ break;
+ }
+
+ g_input_stream_close (in, NULL, &error);
+ g_assert_no_error (error);
+ g_clear_error (&error);
+ g_object_unref (in);
+
+ soup_assert_cmpmem (buf, total, raw_contents->data, raw_contents->length);
+ g_free (buf);
+
+ debug_printf (1, " async read\n");
+
+ imem = g_memory_input_stream_new_from_data (chunkified->str, chunkified->len, NULL);
+ islow = g_object_new (slow_input_stream_get_type (),
+ "base-stream", imem,
+ "close-base-stream", TRUE,
+ NULL);
+ in = g_object_new (g_type_from_name ("SoupBodyInputStream"),
+ "base-stream", islow,
+ "close-base-stream", TRUE,
+ "encoding", SOUP_ENCODING_CHUNKED,
+ NULL);
+ g_object_unref (imem);
+ g_object_unref (islow);
+
+ buf = g_malloc (raw_contents->length);
+ total = 0;
+ while (TRUE) {
+ nread = g_pollable_input_stream_read_nonblocking (G_POLLABLE_INPUT_STREAM (in),
+ buf + total,
+ raw_contents->length - total,
+ NULL, &error);
+ if (nread == -1 && g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
+ GSource *source;
+
+ g_clear_error (&error);
+ source = g_pollable_input_stream_create_source (G_POLLABLE_INPUT_STREAM (in), NULL);
+ g_source_set_dummy_callback (source);
+ g_source_attach (source, NULL);
+ while (!g_pollable_input_stream_is_readable (G_POLLABLE_INPUT_STREAM (in)))
+ g_main_context_iteration (NULL, TRUE);
+ g_source_destroy (source);
+ g_source_unref (source);
+ continue;
+ } else if (nread == -1) {
+ g_assert_no_error (error);
+ g_clear_error (&error);
+ break;
+ } else if (nread == 0)
+ break;
+ else
+ total += nread;
+ }
+
+ g_input_stream_close (in, NULL, &error);
+ g_assert_no_error (error);
+ g_clear_error (&error);
+ g_object_unref (in);
+
+ soup_assert_cmpmem (buf, total, raw_contents->data, raw_contents->length);
+ g_free (buf);
+
+ debug_printf (1, " sync write\n");
+
+ buf = g_malloc (chunkified->len);
+ omem = g_memory_output_stream_new (buf, chunkified->len, NULL, NULL);
+ oslow = g_object_new (slow_output_stream_get_type (),
+ "base-stream", omem,
+ "close-base-stream", TRUE,
+ NULL);
+ out = g_object_new (g_type_from_name ("SoupBodyOutputStream"),
+ "base-stream", oslow,
+ "close-base-stream", TRUE,
+ "encoding", SOUP_ENCODING_CHUNKED,
+ NULL);
+ g_object_unref (omem);
+ g_object_unref (oslow);
+
+ total = chunk_length = chunk_total = 0;
+ while (total < raw_contents->length) {
+ if (chunk_total == chunk_length) {
+ chunk_length = MIN (CHUNK_SIZE, raw_contents->length - total);
+ chunk_total = 0;
+ }
+ nwrote = g_output_stream_write (out, raw_contents->data + total,
+ chunk_length - chunk_total, NULL, &error);
+ g_assert_no_error (error);
+ g_clear_error (&error);
+ if (nwrote > 0) {
+ total += nwrote;
+ chunk_total += nwrote;
+ } else
+ break;
+ }
+
+ g_output_stream_close (out, NULL, &error);
+ g_assert_no_error (error);
+ g_clear_error (&error);
+
+ mem = G_MEMORY_OUTPUT_STREAM (omem);
+ soup_assert_cmpmem (g_memory_output_stream_get_data (mem),
+ g_memory_output_stream_get_data_size (mem),
+ chunkified->str, chunkified->len);
+
+ g_object_unref (out);
+ g_free (buf);
+
+ debug_printf (1, " async write\n");
+
+ buf = g_malloc (chunkified->len);
+ omem = g_memory_output_stream_new (buf, chunkified->len, NULL, NULL);
+ oslow = g_object_new (slow_output_stream_get_type (),
+ "base-stream", omem,
+ "close-base-stream", TRUE,
+ NULL);
+ out = g_object_new (g_type_from_name ("SoupBodyOutputStream"),
+ "base-stream", oslow,
+ "close-base-stream", TRUE,
+ "encoding", SOUP_ENCODING_CHUNKED,
+ NULL);
+ g_object_unref (omem);
+ g_object_unref (oslow);
+
+ total = chunk_length = chunk_total = 0;
+ while (total < raw_contents->length) {
+ if (chunk_total == chunk_length) {
+ chunk_length = MIN (CHUNK_SIZE, raw_contents->length - total);
+ chunk_total = 0;
+ }
+ nwrote = g_pollable_output_stream_write_nonblocking (G_POLLABLE_OUTPUT_STREAM (out),
+ raw_contents->data + total,
+ chunk_length - chunk_total,
+ NULL, &error);
+ if (nwrote == -1 && g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
+ GSource *source;
+
+ g_clear_error (&error);
+ source = g_pollable_output_stream_create_source (G_POLLABLE_OUTPUT_STREAM (out), NULL);
+ g_source_set_dummy_callback (source);
+ g_source_attach (source, NULL);
+ while (!g_pollable_output_stream_is_writable (G_POLLABLE_OUTPUT_STREAM (out)))
+ g_main_context_iteration (NULL, TRUE);
+ g_source_destroy (source);
+ g_source_unref (source);
+ continue;
+ } else if (nwrote == -1) {
+ g_assert_no_error (error);
+ g_clear_error (&error);
+ break;
+ } else {
+ total += nwrote;
+ chunk_total += nwrote;
+ }
+ }
+
+ g_output_stream_close (out, NULL, &error);
+ g_assert_no_error (error);
+ g_clear_error (&error);
+
+ mem = G_MEMORY_OUTPUT_STREAM (omem);
+ soup_assert_cmpmem (g_memory_output_stream_get_data (mem),
+ g_memory_output_stream_get_data_size (mem),
+ chunkified->str, chunkified->len);
+
+ g_object_unref (out);
+ g_free (buf);
+
+ debug_printf (1, " failed write\n");
+ /* this succeeds if it doesn't critical */
+
+ buf = g_malloc (chunkified->len);
+ omem = g_memory_output_stream_new (buf, chunkified->len, NULL, NULL);
+ oslow = g_object_new (breaking_output_stream_get_type (),
+ "base-stream", omem,
+ "close-base-stream", TRUE,
+ NULL);
+ out = g_object_new (g_type_from_name ("SoupBodyOutputStream"),
+ "base-stream", oslow,
+ "close-base-stream", TRUE,
+ "encoding", SOUP_ENCODING_CHUNKED,
+ NULL);
+ g_object_unref (omem);
+ g_object_unref (oslow);
+
+ total = 0;
+ while (total < raw_contents->length) {
+ nwrote = g_output_stream_write (out, raw_contents->data + total,
+ raw_contents->length - total, NULL, NULL);
+ if (nwrote == -1)
+ break;
+ else
+ total += nwrote;
+ }
+
+ g_assert_cmpint (total, !=, raw_contents->length);
+
+ g_output_stream_close (out, NULL, NULL);
+ g_object_unref (out);
+
+ g_free (buf);
+
+ g_string_free (chunkified, TRUE);
+}
+
+int
+main (int argc, char **argv)
+{
+ int ret;
+
+ test_init (argc, argv, NULL);
+
+ force_io_streams_init ();
+
+ g_test_add_func ("/chunk-io", do_io_tests);
+
+ ret = g_test_run ();
+
+ test_cleanup ();
+ return ret;
+}
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-24 15:52:24 +0000