[libsoup] soup-request-http: fix usage with non-default-context

[Dan Winship <danw src gnome org>]

commit 59a25e47cd80ac39c03d74d774121222a4074c87
Author: Dan Winship <danw gnome org>
Date:   Sun Jul 31 11:28:35 2011 -0400

    soup-request-http: fix usage with non-default-context
    
    SoupHTTPInputStream was doing I/O in a thread if the session didn't
    use the global default context. But really, it should have been
    checking the thread-default context instead. And anyway, the threaded
    version doesn't actually work, it turns out. So, fix the check, make
    it into a g_return_if_fail(), and remove the threaded codepath.
    
    Also, fix a handful of places that were adding sources to the global
    default context rather than the thread-default/SoupSession context.
    
    Add tests/requester-test to do some basic
    SoupRequester/SoupRequestHTTP/SoupHTTPInputStream testing.
    
    https://bugzilla.gnome.org/show_bug.cgi?id=653707

 libsoup/soup-cache.c             |   10 +-
 libsoup/soup-http-input-stream.c |   58 +----------
 libsoup/soup-request-http.c      |    5 +-
 tests/Makefile.am                |    3 +
 tests/requester-test.c           |  213 ++++++++++++++++++++++++++++++++++++++
 5 files changed, 227 insertions(+), 62 deletions(-)
---
diff --git a/libsoup/soup-cache.c b/libsoup/soup-cache.c
index d538d68..ad0419f 100644
--- a/libsoup/soup-cache.c
+++ b/libsoup/soup-cache.c
@@ -1452,19 +1452,17 @@ force_flush_timeout (gpointer data)
 /**
  * soup_cache_flush:
  * @cache: a #SoupCache
- * @session: the #SoupSession associated with the @cache
  *
  * This function will force all pending writes in the @cache to be
  * committed to disk. For doing so it will iterate the #GMainContext
- * associated with the @session (which can be the default one) as long
- * as needed.
+ * associated with @cache's session as long as needed.
  **/
 void
 soup_cache_flush (SoupCache *cache)
 {
 	GMainContext *async_context;
 	SoupSession *session;
-	guint timeout_id;
+	GSource *timeout;
 	gboolean forced = FALSE;
 
 	g_return_if_fail (SOUP_IS_CACHE (cache));
@@ -1474,13 +1472,13 @@ soup_cache_flush (SoupCache *cache)
 	async_context = soup_session_get_async_context (session);
 
 	/* We give cache 10 secs to finish */
-	timeout_id = g_timeout_add (10000, force_flush_timeout, &forced);
+	timeout = soup_add_timeout (async_context, 10000, force_flush_timeout, &forced);
 
 	while (!forced && cache->priv->n_pending > 0)
 		g_main_context_iteration (async_context, FALSE);
 
 	if (!forced)
-		g_source_remove (timeout_id);
+		g_source_destroy (timeout);
 	else
 		g_warning ("Cache flush finished despite %d pending requests", cache->priv->n_pending);
 }
diff --git a/libsoup/soup-http-input-stream.c b/libsoup/soup-http-input-stream.c
index 2687133..82e7d46 100644
--- a/libsoup/soup-http-input-stream.c
+++ b/libsoup/soup-http-input-stream.c
@@ -440,7 +440,7 @@ send_sync_finished (GInputStream *stream)
 	priv->finished_cb = NULL;
 
 	/* Wake up the main context iteration */
-	g_source_attach (g_idle_source_new (), NULL);
+	soup_add_completion (priv->async_context, NULL, NULL);
 }
 
 /**
@@ -542,39 +542,6 @@ wrapper_callback (GObject *source_object, GAsyncResult *res,
 }
 
 static void
-send_async_thread (GSimpleAsyncResult *res,
-		   GObject *object,
-		   GCancellable *cancellable)
-{
-	GError *error = NULL;
-	gboolean success;
-
-	success = soup_http_input_stream_send_internal (G_INPUT_STREAM (object),
-							cancellable, &error);
-	g_simple_async_result_set_op_res_gboolean (res, success);
-	if (error) {
-		g_simple_async_result_set_from_error (res, error);
-		g_error_free (error);
-	}
-}
-
-static void
-soup_http_input_stream_send_async_in_thread (GInputStream        *stream,
-					     int                  io_priority,
-					     GCancellable        *cancellable,
-					     GAsyncReadyCallback  callback,
-					     gpointer             user_data)
-{
-	GSimpleAsyncResult *res;
-
-	res = g_simple_async_result_new (G_OBJECT (stream), callback, user_data,
-					 soup_http_input_stream_send_async_in_thread);
-	g_simple_async_result_run_in_thread (res, send_async_thread,
-					     io_priority, cancellable);
-	g_object_unref (res);
-}
-
-static void
 send_async_finished (GInputStream *stream)
 {
 	SoupHTTPInputStreamPrivate *priv = SOUP_HTTP_INPUT_STREAM_GET_PRIVATE (stream);
@@ -609,19 +576,11 @@ soup_http_input_stream_send_async_internal (GInputStream        *stream,
 {
 	SoupHTTPInputStreamPrivate *priv = SOUP_HTTP_INPUT_STREAM_GET_PRIVATE (stream);
 
+	g_return_if_fail (priv->async_context == g_main_context_get_thread_default ());
+
 	g_object_ref (stream);
 	priv->outstanding_callback = callback;
 
-	/* If the session uses the default GMainContext, then we can do
-	 * async I/O directly. But if it has its own main context, it's
-	 * easier to just run it in another thread.
-	 */
-	if (soup_session_get_async_context (priv->session)) {
-		soup_http_input_stream_send_async_in_thread (stream, io_priority, cancellable,
-							     wrapper_callback, user_data);
-		return;
-	}
-
 	priv->got_headers_cb = send_async_finished;
 	priv->finished_cb = send_async_finished;
 
@@ -736,16 +695,7 @@ soup_http_input_stream_read_async (GInputStream        *stream,
 	SoupHTTPInputStreamPrivate *priv = SOUP_HTTP_INPUT_STREAM_GET_PRIVATE (stream);
 	GSimpleAsyncResult *result;
 
-	/* If the session uses the default GMainContext, then we can do
-	 * async I/O directly. But if it has its own main context, we fall
-	 * back to the async-via-sync-in-another-thread implementation.
-	 */
-	if (soup_session_get_async_context (priv->session)) {
-		G_INPUT_STREAM_CLASS (soup_http_input_stream_parent_class)->
-		read_async (stream, buffer, count, io_priority,
-			    cancellable, callback, user_data);
-		return;
-	}
+	g_return_if_fail (priv->async_context == g_main_context_get_thread_default ());
 
 	result = g_simple_async_result_new (G_OBJECT (stream),
 					    callback, user_data,
diff --git a/libsoup/soup-request-http.c b/libsoup/soup-request-http.c
index 555c79f..9cb5e4b 100644
--- a/libsoup/soup-request-http.c
+++ b/libsoup/soup-request-http.c
@@ -253,7 +253,8 @@ soup_request_http_send_async (SoupRequest          *request,
 				helper->callback = callback;
 				helper->user_data = user_data;
 				helper->httpstream = httpstream;
-				g_timeout_add (0, send_async_cb, helper);
+				soup_add_timeout (soup_session_get_async_context (session),
+						  0, send_async_cb, helper);
 				return;
 			}
 		} else if (response == SOUP_CACHE_RESPONSE_NEEDS_VALIDATION) {
@@ -281,7 +282,7 @@ soup_request_http_send_async (SoupRequest          *request,
 	httpstream = soup_http_input_stream_new (soup_request_get_session (request),
 							http->priv->msg);
 	soup_http_input_stream_send_async (httpstream, G_PRIORITY_DEFAULT,
-						  cancellable, sent_async, simple);
+					   cancellable, sent_async, simple);
 }
 
 static GInputStream *
diff --git a/tests/Makefile.am b/tests/Makefile.am
index 8316f77..fcfbdd6 100644
--- a/tests/Makefile.am
+++ b/tests/Makefile.am
@@ -26,6 +26,7 @@ noinst_PROGRAMS =	\
 	misc-test	\
 	ntlm-test	\
 	redirect-test	\
+	requester-test	\
 	simple-httpd	\
 	simple-proxy	\
 	sniffing-test   \
@@ -59,6 +60,7 @@ proxy_test_SOURCES = proxy-test.c $(TEST_SRCS)
 pull_api_SOURCES = pull-api.c $(TEST_SRCS)
 range_test_SOURCES = range-test.c $(TEST_SRCS)
 redirect_test_SOURCES = redirect-test.c $(TEST_SRCS)
+requester_test_SOURCES = requester-test.c $(TEST_SRCS)
 server_auth_test_SOURCES = server-auth-test.c $(TEST_SRCS)
 simple_httpd_SOURCES = simple-httpd.c
 simple_proxy_SOURCES = simple-proxy.c
@@ -90,6 +92,7 @@ TESTS =			\
 	misc-test	\
 	ntlm-test	\
 	redirect-test	\
+	requester-test	\
 	sniffing-test	\
 	streaming-test	\
 	timeout-test	\
diff --git a/tests/requester-test.c b/tests/requester-test.c
new file mode 100644
index 0000000..d303865
--- /dev/null
+++ b/tests/requester-test.c
@@ -0,0 +1,213 @@
+/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
+/*
+ * Copyright (C) 2011 Red Hat, Inc.
+ */
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#define LIBSOUP_USE_UNSTABLE_REQUEST_API
+#include <libsoup/soup.h>
+#include <libsoup/soup-requester.h>
+#include <libsoup/soup-request-http.h>
+
+#include "test-utils.h"
+
+SoupServer *server;
+GMainLoop *loop;
+char buf[1024];
+
+SoupBuffer *response;
+
+static void
+get_index (void)
+{
+	char *contents;
+	gsize length;
+	GError *error = NULL;
+
+	if (!g_file_get_contents (SRCDIR "/index.txt", &contents, &length, &error)) {
+		fprintf (stderr, "Could not read index.txt: %s\n",
+			 error->message);
+		exit (1);
+	}
+
+	response = soup_buffer_new (SOUP_MEMORY_TAKE, contents, length);
+}
+
+static void
+server_callback (SoupServer *server, SoupMessage *msg,
+		 const char *path, GHashTable *query,
+		 SoupClientContext *context, gpointer data)
+{
+	soup_message_set_status (msg, SOUP_STATUS_OK);
+	soup_message_set_response (msg, "text/plain",
+				   SOUP_MEMORY_STATIC, NULL, 0);
+	soup_message_body_append_buffer (msg->response_body, response);
+}
+
+static void
+test_read_ready (GObject *source, GAsyncResult *res, gpointer user_data)
+{
+	GInputStream *stream = G_INPUT_STREAM (source);
+	GString *body = user_data;
+	GError *error = NULL;
+	gsize nread;
+
+	nread = g_input_stream_read_finish (stream, res, &error);
+	if (nread == -1) {
+		debug_printf (1, "  read_async failed: %s", error->message);
+		errors++;
+		g_object_unref (stream);
+		g_main_loop_quit (loop);
+		return;
+	} else if (nread == 0) {
+		g_object_unref (stream);
+		g_main_loop_quit (loop);
+		return;
+	}
+
+	g_string_append_len (body, buf, nread);
+	g_input_stream_read_async (stream, buf, sizeof (buf),
+				   G_PRIORITY_DEFAULT, NULL,
+				   test_read_ready, body);
+}
+
+static void
+test_sent (GObject *source, GAsyncResult *res, gpointer user_data)
+{
+	GString *body = user_data;
+	GInputStream *stream;
+	GError *error = NULL;
+	SoupMessage *msg;
+
+	stream = soup_request_send_finish (SOUP_REQUEST (source), res, &error);
+	if (!stream) {
+		debug_printf (1, "  send_async failed: %s", error->message);
+		errors++;
+		g_main_loop_quit (loop);
+		return;
+	}
+
+	msg = soup_request_http_get_message (SOUP_REQUEST_HTTP (source));
+	if (!SOUP_STATUS_IS_SUCCESSFUL (msg->status_code)) {
+		debug_printf (1, "  GET failed: %d %s", msg->status_code,
+			      msg->reason_phrase);
+		errors++;
+		g_main_loop_quit (loop);
+		return;
+	}
+
+	g_input_stream_read_async (stream, buf, sizeof (buf),
+				   G_PRIORITY_DEFAULT, NULL,
+				   test_read_ready, body);
+}
+
+static void
+do_test_for_thread_and_context (SoupSession *session, const char *uri)
+{
+	SoupRequester *requester;
+	SoupRequest *request;
+	GString *body;
+
+	requester = soup_requester_new ();
+	soup_session_add_feature (session, SOUP_SESSION_FEATURE (requester));
+	g_object_unref (requester);
+
+	body = g_string_new (NULL);
+
+	request = soup_requester_request (requester, uri, NULL);
+	soup_request_send_async (request, NULL, test_sent, body);
+	g_object_unref (request);
+
+	loop = g_main_loop_new (soup_session_get_async_context (session), TRUE);
+	g_main_loop_run (loop);
+	g_main_loop_unref (loop);
+
+	if (body->len != response->length) {
+		debug_printf (1, "  body length mismatch: expected %d, got %d\n",
+			      (int)response->length, (int)body->len);
+		errors++;
+	} else if (memcmp (body->str, response->data, response->length) != 0) {
+		debug_printf (1, "  body data mismatch\n");
+		errors++;
+	}
+
+	g_string_free (body, TRUE);
+}
+
+static void
+do_simple_test (const char *uri)
+{
+	SoupSession *session;
+
+	debug_printf (1, "Simple streaming test\n");
+
+	session = soup_test_session_new (SOUP_TYPE_SESSION_ASYNC, NULL);
+	do_test_for_thread_and_context (session, uri);
+	soup_test_session_abort_unref (session);
+}
+
+static gpointer
+do_test_with_context (const char *uri)
+{
+	GMainContext *async_context;
+	SoupSession *session;
+
+	async_context = g_main_context_new ();
+	g_main_context_push_thread_default (async_context);
+
+	session = soup_test_session_new (SOUP_TYPE_SESSION_ASYNC,
+					 SOUP_SESSION_ASYNC_CONTEXT, async_context,
+					 NULL);
+	g_main_context_unref (async_context);
+
+	do_test_for_thread_and_context (session, uri);
+	soup_test_session_abort_unref (session);
+
+	return NULL;
+}
+
+static void
+do_context_test (const char *uri)
+{
+	debug_printf (1, "Streaming with a non-default-context\n");
+	do_test_with_context (uri);
+}
+
+static void
+do_thread_test (const char *uri)
+{
+	GThread *thread;
+
+	debug_printf (1, "Streaming in another thread\n");
+
+	thread = g_thread_create ((GThreadFunc)do_test_with_context,
+				  (gpointer)uri, TRUE, NULL);
+	g_thread_join (thread);
+}
+
+int
+main (int argc, char **argv)
+{
+	char *uri;
+
+	test_init (argc, argv, NULL);
+	get_index ();
+
+	server = soup_test_server_new (TRUE);
+	soup_server_add_handler (server, NULL, server_callback, NULL, NULL);
+	uri = g_strdup_printf ("http://127.0.0.1:%u/";, soup_server_get_port (server));
+
+	do_simple_test (uri);
+	do_thread_test (uri);
+	do_context_test (uri);
+
+	g_free (uri);
+	soup_buffer_free (response);
+	soup_test_server_quit_unref (server);
+
+	test_cleanup ();
+	return errors != 0;
+}