SoupSession: continue reorg, remove remaining signal-based flow control

Make the sessions more state-machiney and fix up
soup_session_cancel_message() by having it set the message to the
(new) FINISHING state and letting the session implementation pick it
up from there. Remove the remaining "finished" and "restarted" signal
handlers; move the base session's "finished" handler to
"soup_session_unqueue_item" and call that from the right places in the
subclasses.

5 files changed, 228 insertions, 234 deletions

diff --git a/libsoup/soup-message-queue.h b/libsoup/soup-message-queue.h
index 7211d711..4f655b3c 100644
--- a/libsoup/soup-message-queue.h
+++ b/libsoup/soup-message-queue.h
@@ -31,6 +31,7 @@ typedef enum {
 	SOUP_MESSAGE_READY,
 	SOUP_MESSAGE_RUNNING,
 	SOUP_MESSAGE_RESTARTING,
+	SOUP_MESSAGE_FINISHING,
 	SOUP_MESSAGE_FINISHED
 } SoupMessageQueueItemState;
 
diff --git a/libsoup/soup-session-async.c b/libsoup/soup-session-async.c
index ee7fdf66..ee9ddd72 100644
--- a/libsoup/soup-session-async.c
+++ b/libsoup/soup-session-async.c
@@ -37,6 +37,8 @@ static void do_idle_run_queue (SoupSession *session);
 static void  queue_message   (SoupSession *session, SoupMessage *req,
 			      SoupSessionCallback callback, gpointer user_data);
 static guint send_message    (SoupSession *session, SoupMessage *req);
+static void  cancel_message  (SoupSession *session, SoupMessage *msg,
+			      guint status_code);
 
 static void  auth_required   (SoupSession *session, SoupMessage *msg,
 			      SoupAuth *auth, gboolean retrying);
@@ -76,6 +78,7 @@ soup_session_async_class_init (SoupSessionAsyncClass *soup_session_async_class)
 	/* virtual method override */
 	session_class->queue_message = queue_message;
 	session_class->send_message = send_message;
+	session_class->cancel_message = cancel_message;
 	session_class->auth_required = auth_required;
 
 	object_class->finalize = finalize;
@@ -127,9 +130,9 @@ item_failed (SoupMessageQueueItem *item, guint status)
 	}
 
 	if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
-		item->state = SOUP_MESSAGE_FINISHED;
-		if (status != SOUP_STATUS_CANCELLED)
-			soup_session_cancel_message (item->session, item->msg, status);
+		item->state = SOUP_MESSAGE_FINISHING;
+		if (!item->msg->status_code)
+			soup_message_set_status (item->msg, status);
 		soup_message_queue_item_unref (item);
 		return TRUE;
 	}
@@ -215,37 +218,28 @@ message_completed (SoupMessage *msg, gpointer user_data)
 {
 	SoupMessageQueueItem *item = user_data;
 
-	if (item->state == SOUP_MESSAGE_RESTARTING)
-		soup_message_restarted (msg);
-	else {
-		item->state = SOUP_MESSAGE_FINISHED;
-		soup_message_finished (msg);
-	}
+	if (item->state != SOUP_MESSAGE_RESTARTING)
+		item->state = SOUP_MESSAGE_FINISHING;
+	do_idle_run_queue (item->session);
 }
 
 static void
 tunnel_message_completed (SoupMessage *msg, gpointer user_data)
 {
 	SoupMessageQueueItem *item = user_data;
+	SoupSession *session = item->session;
 
 	if (item->state == SOUP_MESSAGE_RESTARTING) {
 		soup_message_restarted (msg);
 		if (item->conn) {
-			soup_session_send_queue_item (item->session, item, tunnel_message_completed);
+			soup_session_send_queue_item (session, item, tunnel_message_completed);
 			return;
 		}
 
 		soup_message_set_status (msg, SOUP_STATUS_TRY_AGAIN);
-		item->state = SOUP_MESSAGE_FINISHED;
 	}
-	message_completed (msg, item);
-}
 
-static void
-tunnel_connected (SoupMessage *msg, gpointer user_data)
-{
-	SoupMessageQueueItem *item = user_data;
-	SoupSession *session = item->session;
+	item->state = SOUP_MESSAGE_FINISHED;
 
 	if (!SOUP_STATUS_IS_SUCCESSFUL (msg->status_code)) {
 		if (item->conn)
@@ -273,13 +267,13 @@ tunnel_connected (SoupMessage *msg, gpointer user_data)
 	item->related->state = SOUP_MESSAGE_READY;
 
 done:
-	if (item->related->msg->status_code) {
-		item->related->state = SOUP_MESSAGE_FINISHED;
-		message_completed (item->related->msg, item->related);
-	}
+	soup_message_finished (msg);
+	if (item->related->msg->status_code)
+		item->related->state = SOUP_MESSAGE_FINISHING;
 
 	do_idle_run_queue (item->session);
 	soup_message_queue_item_unref (item->related);
+	soup_session_unqueue_item (session, item);
 	soup_message_queue_item_unref (item);
 	g_object_unref (session);
 }
@@ -300,15 +294,11 @@ got_connection (SoupConnection *conn, guint status, gpointer user_data)
 	}
 
 	if (status != SOUP_STATUS_OK) {
-		soup_connection_disconnect (conn);
 		soup_message_set_status (item->msg, status);
-		message_completed (item->msg, item);
+		item->state = SOUP_MESSAGE_FINISHING;
 
-		/* There may have been messages waiting for the
-		 * connection count to go down, so queue a run_queue.
-		 */
+		soup_connection_disconnect (conn);
 		do_idle_run_queue (session);
-
 		soup_message_queue_item_unref (item);
 		g_object_unref (session);
 		return;
@@ -322,8 +312,6 @@ got_connection (SoupConnection *conn, guint status, gpointer user_data)
 
 		tunnel_item = soup_session_make_connect_message (session, conn);
 		tunnel_item->related = item;
-		g_signal_connect (tunnel_item->msg, "finished",
-				  G_CALLBACK (tunnel_connected), tunnel_item);
 		soup_session_send_queue_item (session, tunnel_item, tunnel_message_completed);
 		return;
 	}
@@ -337,15 +325,83 @@ got_connection (SoupConnection *conn, guint status, gpointer user_data)
 }
 
 static void
+process_queue_item (SoupMessageQueueItem *item,
+		    gboolean             *should_prune)
+{
+	SoupSession *session = item->session;
+	SoupProxyURIResolver *proxy_resolver;
+
+	do {
+		switch (item->state) {
+		case SOUP_MESSAGE_STARTING:
+			proxy_resolver = (SoupProxyURIResolver *)soup_session_get_feature_for_message (session, SOUP_TYPE_PROXY_URI_RESOLVER, item->msg);
+			if (!proxy_resolver) {
+				item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
+				break;
+			}
+			resolve_proxy_addr (item, proxy_resolver);
+			return;
+
+		case SOUP_MESSAGE_AWAITING_CONNECTION:
+			if (!soup_session_get_connection (session, item, should_prune))
+				return;
+
+			if (soup_connection_get_state (item->conn) != SOUP_CONNECTION_NEW) {
+				item->state = SOUP_MESSAGE_READY;
+				break;
+			}
+
+			item->state = SOUP_MESSAGE_CONNECTING;
+			soup_message_queue_item_ref (item);
+			g_object_ref (session);
+			soup_connection_connect_async (item->conn, item->cancellable,
+						       got_connection, item);
+			return;
+
+		case SOUP_MESSAGE_READY:
+			item->state = SOUP_MESSAGE_RUNNING;
+			soup_session_send_queue_item (session, item, message_completed);
+			break;
+
+		case SOUP_MESSAGE_RESTARTING:
+			item->state = SOUP_MESSAGE_STARTING;
+			soup_message_restarted (item->msg);
+			break;
+
+		case SOUP_MESSAGE_FINISHING:
+			item->state = SOUP_MESSAGE_FINISHED;
+			soup_message_finished (item->msg);
+			if (item->state != SOUP_MESSAGE_FINISHED)
+				break;
+
+			g_object_ref (session);
+			soup_session_unqueue_item (session, item);
+			if (item->callback)
+				item->callback (session, item->msg, item->callback_data);
+			g_object_unref (item->msg);
+			do_idle_run_queue (session);
+			g_object_unref (session);
+			return;
+
+		default:
+			/* Nothing to do with this message in any
+			 * other state.
+			 */
+			return;
+		}
+	} while (item->state != SOUP_MESSAGE_FINISHED);
+}
+
+static void
 run_queue (SoupSessionAsync *sa)
 {
 	SoupSession *session = SOUP_SESSION (sa);
 	SoupMessageQueue *queue = soup_session_get_queue (session);
 	SoupMessageQueueItem *item;
-	SoupProxyURIResolver *proxy_resolver;
 	SoupMessage *msg;
 	gboolean try_pruning = TRUE, should_prune = FALSE;
 
+	g_object_ref (session);
 	soup_session_cleanup_connections (session, FALSE);
 
  try_again:
@@ -355,39 +411,8 @@ run_queue (SoupSessionAsync *sa)
 		msg = item->msg;
 
 		/* CONNECT messages are handled specially */
-		if (msg->method == SOUP_METHOD_CONNECT)
-			continue;
-
-		if (item->state == SOUP_MESSAGE_STARTING) {
-			proxy_resolver = (SoupProxyURIResolver *)soup_session_get_feature_for_message (session, SOUP_TYPE_PROXY_URI_RESOLVER, msg);
-			if (proxy_resolver) {
-				resolve_proxy_addr (item, proxy_resolver);
-				continue;
-			} else
-				item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
-		}
-
-		if (item->state == SOUP_MESSAGE_AWAITING_CONNECTION) {
-			if (!soup_session_get_connection (session, item,
-							  &should_prune))
-				continue;
-
-			if (soup_connection_get_state (item->conn) == SOUP_CONNECTION_NEW) {
-				item->state = SOUP_MESSAGE_CONNECTING;
-				soup_message_queue_item_ref (item);
-				g_object_ref (item->session);
-				soup_connection_connect_async (item->conn, item->cancellable,
-							       got_connection,
-							       item);
-				continue;
-			} else
-				item->state = SOUP_MESSAGE_READY;
-		}
-
-		if (item->state == SOUP_MESSAGE_READY) {
-			item->state = SOUP_MESSAGE_RUNNING;
-			soup_session_send_queue_item (session, item, message_completed);
-		}
+		if (msg->method != SOUP_METHOD_CONNECT)
+			process_queue_item (item, &should_prune);
 	}
 	if (item)
 		soup_message_queue_item_unref (item);
@@ -402,35 +427,7 @@ run_queue (SoupSessionAsync *sa)
 			goto try_again;
 		}
 	}
-}
 
-static void
-request_restarted (SoupMessage *req, gpointer user_data)
-{
-	SoupMessageQueueItem *item = user_data;
-
-	run_queue ((SoupSessionAsync *)item->session);
-}
-
-static void
-final_finished (SoupMessage *req, gpointer user_data)
-{
-	SoupMessageQueueItem *item = user_data;
-	SoupSession *session = item->session;
-
-	g_object_ref (session);
-
-	if (item->state == SOUP_MESSAGE_FINISHED) {
-		g_signal_handlers_disconnect_by_func (req, final_finished, item);
-		g_signal_handlers_disconnect_by_func (req, request_restarted, item);
-		if (item->callback)
-			item->callback (session, req, item->callback_data);
-
-		g_object_unref (req);
-		soup_message_queue_item_unref (item);
-	}
-
-	do_idle_run_queue (session);
 	g_object_unref (session);
 }
 
@@ -460,18 +457,8 @@ static void
 queue_message (SoupSession *session, SoupMessage *req,
 	       SoupSessionCallback callback, gpointer user_data)
 {
-	SoupMessageQueueItem *item;
-
 	SOUP_SESSION_CLASS (soup_session_async_parent_class)->queue_message (session, req, callback, user_data);
 
-	item = soup_message_queue_lookup (soup_session_get_queue (session), req);
-	g_return_if_fail (item != NULL);
-
-	g_signal_connect (req, "restarted",
-			  G_CALLBACK (request_restarted), item);
-	g_signal_connect_after (req, "finished",
-				G_CALLBACK (final_finished), item);
-
 	do_idle_run_queue (session);
 }
 
@@ -482,7 +469,7 @@ send_message (SoupSession *session, SoupMessage *req)
 	GMainContext *async_context =
 		soup_session_get_async_context (session);
 
-	/* Balance out the unref that final_finished will do */
+	/* Balance out the unref that queuing will eventually do */
 	g_object_ref (req);
 
 	queue_message (session, req, NULL, NULL);
@@ -490,8 +477,7 @@ send_message (SoupSession *session, SoupMessage *req)
 	item = soup_message_queue_lookup (soup_session_get_queue (session), req);
 	g_return_val_if_fail (item != NULL, SOUP_STATUS_MALFORMED);
 
-	while (item->state != SOUP_MESSAGE_FINISHED &&
-	       !SOUP_STATUS_IS_TRANSPORT_ERROR (req->status_code))
+	while (item->state != SOUP_MESSAGE_FINISHED)
 		g_main_context_iteration (async_context, TRUE);
 
 	soup_message_queue_item_unref (item);
@@ -500,6 +486,13 @@ send_message (SoupSession *session, SoupMessage *req)
 }
 
 static void
+cancel_message (SoupSession *session, SoupMessage *msg, guint status_code)
+{
+	SOUP_SESSION_CLASS (soup_session_async_parent_class)->cancel_message (session, msg, status_code);
+	do_idle_run_queue (session);
+}
+
+static void
 got_passwords (SoupPasswordManager *password_manager, SoupMessage *msg,
 	       SoupAuth *auth, gboolean retrying, gpointer session)
 {
diff --git a/libsoup/soup-session-private.h b/libsoup/soup-session-private.h
index 0f9c3ac4..e4938860 100644
--- a/libsoup/soup-session-private.h
+++ b/libsoup/soup-session-private.h
@@ -25,6 +25,8 @@ gboolean              soup_session_cleanup_connections  (SoupSession          *s
 void                  soup_session_send_queue_item      (SoupSession          *session,
 							 SoupMessageQueueItem *item,
 							 SoupMessageCompletionFn completion_cb);
+void                  soup_session_unqueue_item         (SoupSession          *session,
+							 SoupMessageQueueItem *item);
 
 G_END_DECLS
 
diff --git a/libsoup/soup-session-sync.c b/libsoup/soup-session-sync.c
index 0de0b81a..077173aa 100644
--- a/libsoup/soup-session-sync.c
+++ b/libsoup/soup-session-sync.c
@@ -147,23 +147,19 @@ tunnel_connect (SoupSession *session, SoupConnection *conn)
 	item = soup_session_make_connect_message (session, conn);
 	do {
 		soup_session_send_queue_item (session, item, NULL);
-		if (item->state == SOUP_MESSAGE_RESTARTING) {
+		status = item->msg->status_code;
+		if (item->state == SOUP_MESSAGE_RESTARTING &&
+		    soup_connection_get_state (conn) != SOUP_CONNECTION_DISCONNECTED) {
 			item->state = SOUP_MESSAGE_STARTING;
 			soup_message_restarted (item->msg);
 		} else {
+			if (item->state == SOUP_MESSAGE_RESTARTING)
+				status = SOUP_STATUS_TRY_AGAIN;
 			item->state = SOUP_MESSAGE_FINISHED;
 			soup_message_finished (item->msg);
 		}
-	} while (item->state == SOUP_MESSAGE_STARTING &&
-		 soup_connection_get_state (conn) != SOUP_CONNECTION_DISCONNECTED);
-
-	/* If the message was requeued but its connection was closed,
-	 * return TRY_AGAIN to our caller.
-	 */
-	if (item->state == SOUP_MESSAGE_STARTING)
-		status = SOUP_STATUS_TRY_AGAIN;
-	else
-		status = item->msg->status_code;
+	} while (item->state == SOUP_MESSAGE_STARTING);
+	soup_session_unqueue_item (session, item);
 	soup_message_queue_item_unref (item);
 
 	if (SOUP_STATUS_IS_SUCCESSFUL (status)) {
@@ -178,129 +174,137 @@ tunnel_connect (SoupSession *session, SoupConnection *conn)
 	return status;
 }
 
-static gboolean
-wait_for_connection (SoupMessageQueueItem *item)
+static void
+get_connection (SoupMessageQueueItem *item)
 {
 	SoupSession *session = item->session;
 	SoupMessage *msg = item->msg;
-	SoupSessionSyncPrivate *priv = SOUP_SESSION_SYNC_GET_PRIVATE (session);
 	gboolean try_pruning = FALSE;
-	SoupProxyURIResolver *proxy_resolver;
 	guint status;
 
-	proxy_resolver = (SoupProxyURIResolver *)soup_session_get_feature_for_message (session, SOUP_TYPE_PROXY_URI_RESOLVER, msg);
-	if (proxy_resolver && item->state == SOUP_MESSAGE_STARTING) {
-		item->state = SOUP_MESSAGE_RESOLVING_PROXY_URI;
-		status = soup_proxy_uri_resolver_get_proxy_uri_sync (
-			proxy_resolver, soup_message_get_uri (msg),
-			item->cancellable, &item->proxy_uri);
-		if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
-			if (status != SOUP_STATUS_CANCELLED)
-				soup_session_cancel_message (session, msg, status);
-			return FALSE;
-		}
-		item->state = SOUP_MESSAGE_RESOLVED_PROXY_URI;
-
-		if (item->proxy_uri) {
-			item->state = SOUP_MESSAGE_RESOLVING_PROXY_ADDRESS;
-			item->proxy_addr = soup_address_new (
-				item->proxy_uri->host, item->proxy_uri->port);
-			status = soup_address_resolve_sync (item->proxy_addr,
-							    item->cancellable);
-			if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
-				if (status != SOUP_STATUS_CANCELLED)
-					soup_session_cancel_message (session, msg, status);
-				return FALSE;
-			}
-		}
-	}
-
-	g_mutex_lock (priv->lock);
-
+try_again:
 	soup_session_cleanup_connections (session, FALSE);
 
- try_again:
-	item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
-	if (soup_session_get_connection (session, item, &try_pruning)) {
-		if (soup_connection_get_state (item->conn) == SOUP_CONNECTION_NEW) {
-			item->state = SOUP_MESSAGE_CONNECTING;
-			status = soup_connection_connect_sync (item->conn, item->cancellable);
-
-			if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
-				soup_connection_disconnect (item->conn);
-				g_object_unref (item->conn);
-				item->conn = NULL;
-			} else if (item->state == SOUP_MESSAGE_FINISHED) {
-				/* Message was cancelled while we were
-				 * connecting.
-				 */
-				soup_connection_disconnect (item->conn);
-				g_object_unref (item->conn);
-				item->conn = NULL;
-			} else if (soup_connection_get_tunnel_addr (item->conn)) {
-				item->state = SOUP_MESSAGE_TUNNELING;
-				status = tunnel_connect (session, item->conn);
-				if (SOUP_STATUS_IS_SUCCESSFUL (status))
-					item->state = SOUP_MESSAGE_READY;
-				else if (status == SOUP_STATUS_TRY_AGAIN) {
-					soup_connection_disconnect (item->conn);
-					g_object_unref (item->conn);
-					item->conn = NULL;
-					goto try_again;
-				} else {
-					soup_message_set_status (item->msg, status);
-					item->state = SOUP_MESSAGE_FINISHED;
-				}
-                        } else
-				item->state = SOUP_MESSAGE_READY;
-		} else
-			item->state = SOUP_MESSAGE_READY;
-
-		g_mutex_unlock (priv->lock);
-		return item->conn != NULL;
+	if (!soup_session_get_connection (session, item, &try_pruning)) {
+		if (!try_pruning)
+			return;
+		soup_session_cleanup_connections (session, TRUE);
+		if (!soup_session_get_connection (session, item, &try_pruning))
+			return;
+		try_pruning = FALSE;
 	}
 
-	if (try_pruning) {
-		try_pruning = FALSE;
-		if (soup_session_cleanup_connections (session, TRUE))
-			goto try_again;
+	if (soup_connection_get_state (item->conn) != SOUP_CONNECTION_NEW) {
+		item->state = SOUP_MESSAGE_READY;
+		return;
 	}
 
-	/* Wait... */
-	g_cond_wait (priv->cond, priv->lock);
+	status = soup_connection_connect_sync (item->conn, item->cancellable);
+
+	if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
+		if (!msg->status_code)
+			soup_message_set_status (msg, status);
+		item->state = SOUP_MESSAGE_FINISHING;
+		soup_connection_disconnect (item->conn);
+		g_object_unref (item->conn);
+		item->conn = NULL;
+		return;
+	}
 
-	/* See if something bad happened */
-	if (item->state == SOUP_MESSAGE_FINISHED) {
-		g_mutex_unlock (priv->lock);
-		return FALSE;
+	if (soup_connection_get_tunnel_addr (item->conn)) {
+		status = tunnel_connect (session, item->conn);
+		if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
+			soup_connection_disconnect (item->conn);
+			g_object_unref (item->conn);
+			item->conn = NULL;
+			if (status == SOUP_STATUS_TRY_AGAIN)
+				goto try_again;
+			soup_message_set_status (item->msg, status);
+			item->state = SOUP_MESSAGE_FINISHING;
+			return;
+		}
 	}
 
-	goto try_again;
+	item->state = SOUP_MESSAGE_READY;
 }
 
 static void
 process_queue_item (SoupMessageQueueItem *item)
 {
-	SoupSessionSyncPrivate *priv = SOUP_SESSION_SYNC_GET_PRIVATE (item->session);
+	SoupSession *session = item->session;
+	SoupSessionSyncPrivate *priv = SOUP_SESSION_SYNC_GET_PRIVATE (session);
+	SoupMessage *msg = item->msg;
+	SoupProxyURIResolver *proxy_resolver;
+	guint status;
 
 	item->state = SOUP_MESSAGE_STARTING;
 	do {
-		if (!wait_for_connection (item))
+		switch (item->state) {
+		case SOUP_MESSAGE_STARTING:
+			proxy_resolver = (SoupProxyURIResolver *)soup_session_get_feature_for_message (session, SOUP_TYPE_PROXY_URI_RESOLVER, msg);
+			if (!proxy_resolver) {
+				item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
+				break;
+			}
+
+			status = soup_proxy_uri_resolver_get_proxy_uri_sync (
+				proxy_resolver, soup_message_get_uri (msg),
+				item->cancellable, &item->proxy_uri);
+			if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
+				soup_message_set_status (msg, status);
+				item->state = SOUP_MESSAGE_FINISHING;
+				break;
+			}
+			if (!item->proxy_uri) {
+				item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
+				break;
+			}
+
+			item->proxy_addr = soup_address_new (
+				item->proxy_uri->host, item->proxy_uri->port);
+			status = soup_address_resolve_sync (item->proxy_addr,
+							    item->cancellable);
+			if (SOUP_STATUS_IS_SUCCESSFUL (status))
+				item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
+			else {
+				soup_message_set_status (msg, status);
+				item->state = SOUP_MESSAGE_FINISHING;
+			}
 			break;
 
-		if (item->state == SOUP_MESSAGE_READY) {
+		case SOUP_MESSAGE_AWAITING_CONNECTION:
+			g_mutex_lock (priv->lock);
+			do {
+				get_connection (item);
+				if (item->state == SOUP_MESSAGE_AWAITING_CONNECTION)
+					g_cond_wait (priv->cond, priv->lock);
+			} while (item->state == SOUP_MESSAGE_AWAITING_CONNECTION);
+			g_mutex_unlock (priv->lock);
+			break;
+
+		case SOUP_MESSAGE_READY:
 			item->state = SOUP_MESSAGE_RUNNING;
 			soup_session_send_queue_item (item->session, item, NULL);
-		}
+			if (item->state != SOUP_MESSAGE_RESTARTING)
+				item->state = SOUP_MESSAGE_FINISHING;
+			break;
 
-		if (item->state == SOUP_MESSAGE_RESTARTING) {
+		case SOUP_MESSAGE_RESTARTING:
 			item->state = SOUP_MESSAGE_STARTING;
 			soup_message_restarted (item->msg);
-		} else {
+			break;
+
+		case SOUP_MESSAGE_FINISHING:
 			item->state = SOUP_MESSAGE_FINISHED;
 			soup_message_finished (item->msg);
+			soup_session_unqueue_item (session, item);
+			break;
+
+		default:
+			g_warn_if_reached ();
+			item->state = SOUP_MESSAGE_FINISHING;
+			break;
 		}
-		g_cond_broadcast (priv->cond);
 	} while (item->state != SOUP_MESSAGE_FINISHED);
 }
 
diff --git a/libsoup/soup-session.c b/libsoup/soup-session.c
index bd42795b..ff25e33a 100644
--- a/libsoup/soup-session.c
+++ b/libsoup/soup-session.c
@@ -1390,11 +1390,10 @@ soup_session_get_queue (SoupSession *session)
 	return priv->queue;
 }
 
-static void
-message_finished (SoupMessage *msg, gpointer user_data)
+void
+soup_session_unqueue_item (SoupSession          *session,
+			   SoupMessageQueueItem *item)
 {
-	SoupMessageQueueItem *item = user_data;
-	SoupSession *session = item->session;
 	SoupSessionPrivate *priv = SOUP_SESSION_GET_PRIVATE (session);
 	SoupSessionHost *host;
 
@@ -1403,25 +1402,26 @@ message_finished (SoupMessage *msg, gpointer user_data)
 		item->conn = NULL;
 	}
 
-	if (item->state == SOUP_MESSAGE_FINISHED) {
-		soup_message_queue_remove (priv->queue, item);
+	if (item->state != SOUP_MESSAGE_FINISHED) {
+		g_warning ("finished an item with state %d", item->state);
+		return;
+	}
 
-		g_mutex_lock (priv->host_lock);
-		host = get_host_for_message (session, item->msg);
-		host->num_messages--;
-		g_mutex_unlock (priv->host_lock);
+	soup_message_queue_remove (priv->queue, item);
 
-		g_signal_handlers_disconnect_by_func (msg, message_finished, item);
-		/* g_signal_handlers_disconnect_by_func doesn't work if you
-		 * have a metamarshal, meaning it doesn't work with
-		 * soup_message_add_header_handler()
-		 */
-		g_signal_handlers_disconnect_matched (msg, G_SIGNAL_MATCH_DATA,
-						      0, 0, NULL, NULL, session);
-		g_signal_emit (session, signals[REQUEST_UNQUEUED], 0, msg);
-		soup_message_queue_item_unref (item);
-	} else
-		g_warning ("finished an item with state %d", item->state);
+	g_mutex_lock (priv->host_lock);
+	host = get_host_for_message (session, item->msg);
+	host->num_messages--;
+	g_mutex_unlock (priv->host_lock);
+
+	/* g_signal_handlers_disconnect_by_func doesn't work if you
+	 * have a metamarshal, meaning it doesn't work with
+	 * soup_message_add_header_handler()
+	 */
+	g_signal_handlers_disconnect_matched (item->msg, G_SIGNAL_MATCH_DATA,
+					      0, 0, NULL, NULL, session);
+	g_signal_emit (session, signals[REQUEST_UNQUEUED], 0, item->msg);
+	soup_message_queue_item_unref (item);
 }
 
 static void
@@ -1439,9 +1439,6 @@ queue_message (SoupSession *session, SoupMessage *msg,
 	host->num_messages++;
 	g_mutex_unlock (priv->host_lock);
 
-	g_signal_connect_after (msg, "finished",
-				G_CALLBACK (message_finished), item);
-
 	if (!(soup_message_get_flags (msg) & SOUP_MESSAGE_NO_REDIRECT)) {
 		soup_message_add_header_handler (
 			msg, "got_body", "Location",
@@ -1597,11 +1594,8 @@ cancel_message (SoupSession *session, SoupMessage *msg, guint status_code)
 
 	soup_message_io_stop (msg);
 	soup_message_set_status (msg, status_code);
+	item->state = SOUP_MESSAGE_FINISHING;
 
-	if (item->state != SOUP_MESSAGE_FINISHED) {
-		item->state = SOUP_MESSAGE_FINISHED;
-		soup_message_finished (msg);
-	}
 	soup_message_queue_item_unref (item);
 }