Index: tests/Makefile.am
===================================================================
--- tests/Makefile.am	(revision 955)
+++ tests/Makefile.am	(working copy)
@@ -8,6 +8,7 @@
 LIBS = $(top_builddir)/libsoup/libsoup-$(SOUP_API_VERSION).la
 
 noinst_PROGRAMS =	\
+	async-client-test \
 	context-test	\
 	date		\
 	dict		\
@@ -25,6 +26,7 @@
 	$(XMLRPC_TESTS)
 
 auth_test_SOURCES = auth-test.c apache-wrapper.c apache-wrapper.h
+async_client_test_SOURCES = async-client-test.c
 context_test_SOURCES = context-test.c
 date_SOURCES = date.c
 dict_SOURCES = dict.c
@@ -52,7 +54,7 @@
 XMLRPC_TESTS = xmlrpc-test
 endif
 
-TESTS = context-test date header-parsing uri-parsing ntlm-test $(APACHE_TESTS) $(SSL_TESTS) $(XMLRPC_TESTS)
+TESTS = async-client-test context-test date header-parsing uri-parsing ntlm-test $(APACHE_TESTS) $(SSL_TESTS) $(XMLRPC_TESTS)
 
 EXTRA_DIST =		\
 	libsoup.supp	\
--- /dev/null	2007-11-21 09:49:36.004006262 +0100
+++ tests/async-client-test.c	2007-11-21 12:59:51.000000000 +0100
@@ -0,0 +1,718 @@
+/* Copyright (C) <2007> Wouter Cloetens <wouter@mind.be>
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <glib.h>
+#include <string.h>
+#include <stdlib.h>
+#include <libsoup/soup.h>
+#include <libsoup/soup-address.h>
+#include <libsoup/soup-message.h>
+#include <libsoup/soup-misc.h>
+#include <libsoup/soup-server.h>
+#include <libsoup/soup-server-message.h>
+#include <libsoup/soup-session-async.h>
+#include <libsoup/soup-session-sync.h>
+
+/* --- Server ---------------------------------------------------------------- */
+GThread *server_thread;
+char *base_uri;
+
+gboolean content_length;
+#define CHUNKS 4
+#define CHUNKSIZE 4096
+
+static void
+request_failed (SoupMessage *msg, gpointer timeout)
+{
+	if (SOUP_STATUS_IS_TRANSPORT_ERROR (msg->status_code))
+		g_source_destroy (timeout);
+}
+
+static gboolean
+add_body_chunk (gpointer data)
+{
+        int i, j;
+	SoupMessage *msg = data;
+        char buf[CHUNKSIZE];
+
+	for (i = 0; i < CHUNKS; i++) {
+		for (j = 0; j < sizeof(buf); j++) {
+			buf[j] = ' ' + (j % (127 - ' '));
+		}
+		soup_message_add_chunk (msg, SOUP_BUFFER_STATIC,
+					buf, sizeof(buf));
+	}
+
+	soup_message_add_final_chunk (msg);
+	soup_message_io_unpause (msg);
+	g_object_unref (msg);
+
+	return FALSE;
+}
+
+static void
+server_callback (SoupServerContext *context, SoupMessage *msg, gpointer data)
+{
+	GSource *timeout;
+
+	if (soup_method_get_id (msg->method) != SOUP_METHOD_ID_GET) {
+		soup_message_set_status (msg, SOUP_STATUS_NOT_IMPLEMENTED);
+		return;
+	}
+
+	if (!strcmp (context->path, "/shutdown")) {
+		soup_server_quit (context->server);
+		return;
+	}
+
+	soup_message_set_status (msg, SOUP_STATUS_OK);
+	if (!strcmp (context->path, "/fast")) {
+		soup_message_set_response (msg, "text/plain",
+					   SOUP_BUFFER_STATIC, "OK\r\n", 4);
+		return;
+	}
+	if (content_length) {
+		char buf[16];
+
+                snprintf(buf, sizeof(buf), "%d", CHUNKS * CHUNKSIZE);
+		soup_message_add_header (msg->response_headers, "Content-Length", buf);
+	}
+
+	soup_server_message_set_encoding (SOUP_SERVER_MESSAGE (msg),
+					  SOUP_TRANSFER_CHUNKED);
+	g_object_ref (msg);
+	soup_message_io_pause (msg);
+
+	timeout = soup_add_timeout (
+		soup_server_get_async_context (context->server),
+		200, add_body_chunk, msg);
+	g_signal_connect (msg, "finished",
+			  G_CALLBACK (request_failed), timeout);
+}
+
+static gpointer
+run_server_thread (gpointer user_data)
+{
+	SoupServer *server = user_data;
+
+	soup_server_add_handler (server, NULL, NULL,
+				 server_callback, NULL, NULL);
+	soup_server_run (server);
+	g_object_unref (server);
+
+	return NULL;
+}
+
+static guint
+create_server (void)
+{
+	SoupServer *server;
+	GMainContext *async_context;
+	guint port;
+
+	async_context = g_main_context_new ();
+	server = soup_server_new (SOUP_SERVER_PORT, 0,
+				  SOUP_SERVER_ASYNC_CONTEXT, async_context,
+				  NULL);
+	g_main_context_unref (async_context);
+
+	if (!server) {
+		fprintf (stderr, "Unable to bind server\n");
+		exit (1);
+	}
+
+	port = soup_server_get_port (server);
+	server_thread = g_thread_create (run_server_thread, server, TRUE, NULL);
+
+	return port;
+}
+
+static void
+shutdown_server (void)
+{
+	SoupSession *session;
+	char *uri;
+	SoupMessage *msg;
+
+	session = soup_session_sync_new ();
+	uri = g_build_filename (base_uri, "shutdown", NULL);
+	msg = soup_message_new ("GET", uri);
+	soup_session_send_message (session, msg);
+	g_object_unref (msg);
+	g_free (uri);
+
+	soup_session_abort (session);
+	g_object_unref (session);
+
+	g_thread_join (server_thread);
+}
+
+
+/* --- GMainContext structure definition, to be able to check reference count. */
+
+#include "glib.h"
+#include <signal.h>
+#include <sys/types.h>
+#include <time.h>
+#include <stdlib.h>
+#ifdef HAVE_SYS_TIME_H
+#include <sys/time.h>
+#endif /* HAVE_SYS_TIME_H */
+#ifdef GLIB_HAVE_SYS_POLL_H
+#  include <sys/poll.h>
+#  undef events	 /* AIX 4.1.5 & 4.3.2 define this for SVR3,4 compatibility */
+#  undef revents /* AIX 4.1.5 & 4.3.2 define this for SVR3,4 compatibility */
+
+/* The poll() emulation on OS/X doesn't handle fds=NULL, nfds=0,
+ * so we prefer our own poll emulation.
+ */
+#if defined(_POLL_EMUL_H_) || defined(BROKEN_POLL)
+#undef HAVE_POLL
+#endif
+
+#endif /* GLIB_HAVE_SYS_POLL_H */
+#ifdef HAVE_UNISTD_H
+#include <unistd.h>
+#endif /* HAVE_UNISTD_H */
+#include <errno.h>
+
+#ifdef G_OS_WIN32
+#define STRICT
+#include <windows.h>
+#endif /* G_OS_WIN32 */
+
+#ifdef G_OS_BEOS
+#include <sys/socket.h>
+#include <sys/wait.h>
+#endif /* G_OS_BEOS */
+
+#ifdef G_OS_UNIX
+#include <fcntl.h>
+#include <sys/wait.h>
+#endif
+
+//#include "galias.h"
+struct _GMainContext
+{
+#ifdef G_THREADS_ENABLED
+	/* The following lock is used for both the list of sources
+	 * and the list of poll records
+	 */
+	GStaticMutex mutex;
+	GCond *cond;
+	GThread *owner;
+	guint owner_count;
+	GSList *waiters;
+#endif
+
+	gint ref_count;
+
+	GPtrArray *pending_dispatches;
+	gint timeout;			/* Timeout for current iteration */
+
+	guint next_id;
+	GSource *source_list;
+	gint in_check_or_prepare;
+#if 0
+	GPollRec *poll_records;
+	guint n_poll_records;
+	GPollFD *cached_poll_array;
+	guint cached_poll_array_size;
+
+#ifdef G_THREADS_ENABLED
+#ifndef G_OS_WIN32
+	/* this pipe is used to wake up the main loop when a source is added.
+	 */
+	gint wake_up_pipe[2];
+#else /* G_OS_WIN32 */
+	HANDLE wake_up_semaphore;
+#endif /* G_OS_WIN32 */
+
+	GPollFD wake_up_rec;
+	gboolean poll_waiting;
+
+	/* Flag indicating whether the set of fd's changed during a poll */
+	gboolean poll_changed;
+#endif /* G_THREADS_ENABLED */
+
+	GPollFunc poll_func;
+
+	GTimeVal current_time;
+	gboolean time_is_current;
+#endif
+};
+
+/* ---- HTTP source object. -------------------------------------------------- */
+
+enum retcode {
+	RC_OK,
+	RC_ERROR,
+	RC_WAITING,
+};
+
+struct httpsrc {
+	gint uid;                             /* Unique ID identifying this object. */
+	gchar *location;                      /* Full URI. */
+	GMainContext *context;                /* I/O context. */
+	GMainLoop *loop;                      /* Event loop. */
+	SoupSession *session;                 /* Async context. */
+	SoupMessage *msg;                     /* Request message. */
+	enum retcode ret;                     /* Return code from callback. */
+	guint8 **outbuf;                      /* Return buffer allocated by callback. */
+	gboolean interrupted;                 /* Signal unlock(). */
+	gboolean have_size;                   /* Received and parsed Content-Length header. */
+	guint64 content_size;                 /* Value of Content-Length header. */
+	gint64 read_position;                 /* Current position. */
+};
+
+static void dprintf(struct httpsrc *src, const char *format, ...)
+__attribute__ ((format (printf, 2, 3)));
+static void errprintf(struct httpsrc *src, const char *format, ...)
+	__attribute__ ((format (printf, 2, 3)));
+
+static struct httpsrc * src_init (void);
+static gboolean src_set_location (struct httpsrc * src, const gchar * uri);
+static gboolean src_start (struct httpsrc * src);
+static gboolean src_stop (struct httpsrc * src);
+static enum retcode src_create (struct httpsrc * src, guint8 ** outbuf);
+static void src_dispose (struct httpsrc * src);
+static gboolean src_get_size (struct httpsrc * src, guint64 * size);
+static gboolean src_unlock (struct httpsrc * src);
+static gboolean src_unlock_stop (struct httpsrc * src);
+
+static void soup_got_headers (SoupMessage *msg, struct httpsrc * src);
+static void soup_finished (SoupMessage *msg, struct httpsrc * src);
+static void soup_got_body (SoupMessage *msg, struct httpsrc * src);
+static void soup_got_chunk (SoupMessage *msg, struct httpsrc * src);
+static void soup_response (SoupMessage *msg, gpointer user_data);
+static void soup_session_close (struct httpsrc * src);
+
+static gboolean debug;
+static gboolean workaround;
+
+static struct httpsrc *
+src_init (void)
+{
+	static gint uid = 0;
+	struct httpsrc * src;
+
+	src = g_malloc0 (sizeof (struct httpsrc));
+	if (src != NULL)
+		src->uid = uid++;
+
+        return src;
+}
+
+static void
+src_dispose (struct httpsrc * src)
+{
+	int i;
+
+	dprintf (src, "dispose");
+	soup_session_close (src);
+	if (src->loop) {
+		dprintf (src, "context refcount = %u", src->context->ref_count);
+		g_main_loop_unref (src->loop);
+		dprintf (src, "after mainloop unref: context refcount = %u", src->context->ref_count);
+		i = src->context->ref_count - 1;
+		g_main_context_unref (src->context);
+		dprintf (src, "context unref, refcount = %u", i);
+		if (workaround) {
+			for (; i > 0; i--) {
+				g_main_context_unref (src->context);
+				errprintf (src, "context refcount not 0! Unreferenced again, context refcount = %u",
+					   src->context->ref_count);
+			}
+		}
+		else {
+			if (i > 0) {
+				errprintf (src, "context refcount = %u, expected 0!",
+					   src->context->ref_count);
+			}
+		}
+		src->loop = NULL;
+		src->context = NULL;
+	}
+	if (src->location) {
+		g_free (src->location);
+	}
+
+	g_free (src);
+}
+
+static enum retcode
+src_create (struct httpsrc * src, guint8 ** outbuf)
+{
+	src->ret = RC_WAITING;
+	src->outbuf = outbuf;
+	do {
+		if (src->interrupted) {
+			soup_session_close (src);
+			src->ret = RC_WAITING;
+			break;
+		}
+		if (!src->session ||
+		    (src->have_size && (src->read_position >= src->content_size))) {
+			dprintf (src, "EOS reached");
+			soup_session_close (src);
+			src->ret = RC_WAITING;
+			break;
+		}
+
+		switch (src->msg->status)
+		{
+		case SOUP_MESSAGE_STATUS_IDLE:
+			dprintf (src, "Queueing connection request");
+			soup_session_queue_message (src->session, src->msg,
+						    soup_response, src);
+			dprintf (src, "context refcount = %u", src->context->ref_count);
+			dprintf (src, "request queued, session refcount = %u", (G_OBJECT(src->session))->ref_count);
+			break;
+		case SOUP_MESSAGE_STATUS_FINISHED:
+			dprintf (src, "Connection closed");
+			soup_session_close (src);
+			src->ret = RC_WAITING;
+			break;
+		case SOUP_MESSAGE_STATUS_QUEUED:
+		case SOUP_MESSAGE_STATUS_CONNECTING:
+		case SOUP_MESSAGE_STATUS_RUNNING:
+		default:
+			soup_message_io_unpause (src->msg);
+			break;
+		}
+
+		if (src->ret == RC_WAITING)
+			g_main_loop_run (src->loop);
+	} while (src->ret == RC_WAITING);
+
+	return src->ret;
+}
+
+/* create a socket for connecting to remote server */
+static gboolean
+src_start (struct httpsrc * src)
+{
+	if (!src->location) {
+		errprintf (src, "Missing location property");
+		return FALSE;
+	}
+
+	src->context = g_main_context_new ();
+	dprintf (src, "New context, refcount = %u", src->context->ref_count);
+
+	src->loop = g_main_loop_new (src->context, TRUE);
+	if (!src->loop)
+	{
+		errprintf(src, "Failed to start GMainLoop");
+		g_main_context_unref (src->context);
+		return FALSE;
+	}
+	dprintf (src, "New mainloop, context refcount = %u", src->context->ref_count);
+
+	src->session = soup_session_async_new_with_options (
+							    SOUP_SESSION_ASYNC_CONTEXT, src->context, NULL);
+	if (!src->session)
+	{
+		errprintf (src, "Failed to create async session");
+		return FALSE;
+	}
+	dprintf (src, "New session, context refcount = %u", src->context->ref_count);
+	dprintf (src, "new session, refcount = %u", (G_OBJECT(src->session))->ref_count);
+
+	src->msg = soup_message_new (SOUP_METHOD_GET, src->location);
+	if (!src->msg) {
+		errprintf (src, "Error parsing URL");
+		return FALSE;
+	}
+	soup_message_add_header (src->msg->request_headers, "Connection", "close");
+
+	g_signal_connect (src->msg, "got_headers",
+			  G_CALLBACK (soup_got_headers), src);
+	g_signal_connect (src->msg, "got_body",
+			  G_CALLBACK (soup_got_body), src);
+	g_signal_connect (src->msg, "finished",
+			  G_CALLBACK (soup_finished), src);
+	g_signal_connect (src->msg, "got_chunk",
+			  G_CALLBACK (soup_got_chunk), src);
+	soup_message_set_flags (src->msg, SOUP_MESSAGE_OVERWRITE_CHUNKS);
+
+	return TRUE;
+}
+
+/* close the socket and associated resources
+ * used both to recover from errors and go to NULL state */
+static gboolean
+src_stop (struct httpsrc * src)
+{
+	dprintf (src, "stop()");
+	soup_session_close (src);
+	if (src->loop) {
+		dprintf (src, "context refcount = %u", src->context->ref_count);
+	}
+
+	return TRUE;
+}
+
+/* Interrupt a blocking request. */
+static gboolean
+src_unlock (struct httpsrc * src)
+{
+	dprintf (src, "unlock()");
+
+	src->interrupted = TRUE;
+	if (src->loop)
+		g_main_loop_quit (src->loop);
+	return TRUE;
+}
+
+/* Interrupt interrupt. */
+static gboolean
+src_unlock_stop (struct httpsrc * src)
+{
+	dprintf (src, "unlock_stop()");
+
+	src->interrupted = FALSE;
+	return TRUE;
+}
+
+static gboolean
+src_get_size (struct httpsrc * src, guint64 * size)
+{
+	if (src->have_size) {
+		dprintf (src, "get_size() = %llu", src->content_size);
+		*size = src->content_size;
+		return TRUE;
+	}
+	dprintf (src, "get_size() = FALSE");
+	return FALSE;
+}
+
+static gboolean
+src_set_location (struct httpsrc * src, const gchar * uri)
+{
+	if (src->location) {
+		g_free (src->location);
+		src->location = NULL;
+	}
+	src->location = g_strdup (uri);
+
+	return TRUE;
+}
+
+static void
+soup_got_headers (SoupMessage *msg, struct httpsrc * src)
+{
+	const char *value;
+
+	dprintf (src, "got headers");
+
+	value = soup_message_get_header (msg->response_headers, "Content-Length");
+	if (value != NULL) {
+		src->content_size = g_ascii_strtoull (value, NULL, 10);
+		src->have_size = TRUE;
+		dprintf (src, "size = %llu", src->content_size);
+	}
+}
+
+/* Have body. Signal EOS. */
+static void
+soup_got_body (SoupMessage *msg, struct httpsrc * src)
+{
+	dprintf (src, "got body");
+	src->ret = RC_ERROR;
+	g_main_loop_quit (src->loop);
+	soup_message_io_pause (msg);
+}
+
+/* Finished. Signal EOS. */
+static void
+soup_finished (SoupMessage *msg, struct httpsrc * src)
+{
+	dprintf (src, "finished");
+	src->ret = RC_ERROR;
+	if (src->loop)
+		g_main_loop_quit (src->loop);
+}
+
+static void
+soup_got_chunk (SoupMessage *msg, struct httpsrc * src)
+{
+	dprintf (src, "got chunk of %d bytes", msg->response.length);
+
+	/* Create the buffer. */
+	*src->outbuf = g_malloc(msg->response.length);
+	if (*src->outbuf == NULL) {
+		src->ret = RC_ERROR;
+	}
+	else {
+		src->ret = RC_OK;
+		memcpy (*src->outbuf, msg->response.body, msg->response.length);
+		src->read_position += msg->response.length;
+	}
+
+	g_main_loop_quit (src->loop);
+	soup_message_io_pause (msg);
+}
+
+static void
+soup_response (SoupMessage *msg, gpointer user_data)
+{
+	struct httpsrc * src = (struct httpsrc *)user_data;
+
+	dprintf (src, "got response %d: %s", msg->status_code, msg->reason_phrase);
+	g_main_loop_quit (src->loop);
+}
+
+static void
+soup_session_close (struct httpsrc * src)
+{
+	if (src->session) {
+		int i;
+		GObject *obj = G_OBJECT(src->session);
+
+		dprintf (src, "Connection closed, session refcount = %u", obj->ref_count);
+		dprintf (src, "context refcount = %u", src->context->ref_count);
+		soup_session_abort (src->session);  /* This unrefs the message. */
+		dprintf (src, "Session aborted, session refcount = %u", obj->ref_count);
+		dprintf (src, "context refcount = %u", src->context->ref_count);
+		i = obj->ref_count - 1;
+		g_object_unref (src->session);
+		dprintf (src, "Session unreferenced, session refcount = %u", i);
+		dprintf (src, "context refcount = %u", src->context->ref_count);
+		if (workaround) {
+			if (i > 0) {
+				errprintf (src, "session refcount = %u, expected 0!", obj->ref_count);
+			}
+		}
+		else {
+			for (; i > 0; i--) {
+				g_object_unref (src->session);
+				errprintf (src, "WARNING! session refcount not 0! Unreferenced again, session refcount = %u", obj->ref_count);
+				errprintf (src, "context refcount = %u", src->context->ref_count);
+			}
+		}
+		src->session = NULL;
+		src->msg = NULL;
+	}
+}
+
+static void dprintf(struct httpsrc *src, const char *format, ...)
+{
+	va_list args;
+
+	if (!debug)
+		return;
+	fprintf(stderr, "debug: http source #%d: ", src->uid);
+	va_start(args, format);
+	vfprintf(stderr, format, args);
+	va_end(args);
+	fprintf(stderr, "\n");
+}
+
+static void errprintf(struct httpsrc *src, const char *format, ...)
+{
+	va_list args;
+
+	fprintf(stderr, "error: http source #%d: ", src->uid);
+	va_start(args, format);
+	vfprintf(stderr, format, args);
+	va_end(args);
+	fprintf(stderr, "\n");
+}
+
+/* ---- Client test. --------------------------------------------------------- */
+
+int errors = 0;
+
+static void
+do_test (void)
+{
+	struct httpsrc *src;
+	char *uri;
+        guint8 *buf;
+
+	src = src_init ();
+	if (!src) {
+		errors++;
+		return;
+	}
+
+	uri = g_build_filename (base_uri, "slow", NULL);
+	src_set_location (src, uri);
+	g_free (uri);
+
+	if (!src_start (src)) {
+		errors++;
+                goto end;
+	}
+
+	/* This could be run in another thread. */
+	while (src_create (src, &buf) == RC_OK) {
+                g_free (buf);
+	}
+	/* End of client thread. */
+
+	src_stop (src);
+
+end:
+	src_dispose (src);
+}
+
+/* ---- Main. ---------------------------------------------------------------- */
+
+static void
+quit (int sig)
+{
+	/* Exit cleanly on ^C in case we're valgrinding. */
+	exit (0);
+}
+
+int
+main (int argc, char **argv)
+{
+	int opt;
+	guint port;
+
+	g_type_init ();
+	g_thread_init (NULL);
+	signal (SIGINT, quit);
+
+	while ((opt = getopt (argc, argv, "dwc")) != -1) {
+		switch (opt) {
+		case 'd':
+			debug = TRUE;
+			break;
+		case 'w':
+			workaround = TRUE;
+			break;
+		case 'c':
+			content_length = TRUE;
+			break;
+		default:
+			fprintf (stderr, "Usage: %s [-d][-w][-c]\n",
+				 argv[0]);
+			exit (1);
+		}
+	}
+
+	port = create_server ();
+	base_uri = g_strdup_printf ("http://localhost:%u/", port);
+
+	do_test ();
+
+	shutdown_server ();
+	g_free (base_uri);
+	g_main_context_unref (g_main_context_default ());
+
+	printf ("\n");
+	if (errors) {
+		printf ("%s: %d error(s). Run with '-d' for details\n",
+			argv[0], errors);
+	} else
+		printf ("%s: OK\n", argv[0]);
+	return errors != 0;
+}
