session: stop using the old API way to queue messages

Make soup_message_io_run public and use it for tunneling and websockets.

3 files changed, 31 insertions, 21 deletions

diff --git a/libsoup/soup-message-io.c b/libsoup/soup-message-io.c
index 26f6a58a..66119e03 100644
--- a/libsoup/soup-message-io.c
+++ b/libsoup/soup-message-io.c
@@ -98,8 +98,6 @@ typedef struct {
 	gint64 begin_time_nsec;
 #endif
 } SoupMessageIOData;
-	
-static void io_run (SoupMessage *msg, gboolean blocking);
 
 #define RESPONSE_BLOCK_SIZE 8192
 #define HEADER_SIZE_LIMIT (64 * 1024)
@@ -753,10 +751,12 @@ io_read (SoupMessage *msg, gboolean blocking,
 						blocking,
 						cancellable, error);
 		if (nread > 0) {
-                        GBytes *bytes = g_bytes_new (buf, nread);
-			soup_message_body_got_chunk (io->read_body, bytes);
-			soup_message_got_chunk (msg, bytes);
-			g_bytes_unref (bytes);
+			if (io->read_body) {
+				GBytes *bytes = g_bytes_new (buf, nread);
+				soup_message_body_got_chunk (io->read_body, bytes);
+				soup_message_got_chunk (msg, bytes);
+				g_bytes_unref (bytes);
+			}
 			break;
 		}
 
@@ -1058,12 +1058,13 @@ soup_message_io_update_status (SoupMessage  *msg,
 static gboolean
 io_run_ready (SoupMessage *msg, gpointer user_data)
 {
-	io_run (msg, FALSE);
+	soup_message_io_run (msg, FALSE);
 	return FALSE;
 }
 
-static void
-io_run (SoupMessage *msg, gboolean blocking)
+void
+soup_message_io_run (SoupMessage *msg,
+		     gboolean     blocking)
 {
 	SoupMessageIOData *io = soup_message_get_io_data (msg);
 	GError *error = NULL;
@@ -1319,13 +1320,12 @@ soup_message_io_client (SoupMessageQueueItem *item,
 	soup_message_queue_item_ref (item);
 	io->cancellable = item->cancellable;
 
-	io->read_body       = item->msg->response_body;
 	io->write_body      = item->msg->request_body;
 
 	io->write_state     = SOUP_MESSAGE_IO_STATE_HEADERS;
 
 	if (!item->new_api) {
-		io_run (item->msg, !item->async);
+		soup_message_io_run (item->msg, !item->async);
 	}
 }
 
@@ -1349,7 +1349,7 @@ soup_message_io_server (SoupMessage *msg,
 	io->write_body      = msg->response_body;
 
 	io->read_state      = SOUP_MESSAGE_IO_STATE_HEADERS;
-	io_run (msg, FALSE);
+	soup_message_io_run (msg, FALSE);
 }
 
 void  
@@ -1390,7 +1390,7 @@ io_unpause_internal (gpointer msg)
 	if (io->io_source)
 		return FALSE;
 
-	io_run (msg, FALSE);
+	soup_message_io_run (msg, FALSE);
 	return FALSE;
 }
 
diff --git a/libsoup/soup-message-private.h b/libsoup/soup-message-private.h
index adaaf048..06bd1151 100644
--- a/libsoup/soup-message-private.h
+++ b/libsoup/soup-message-private.h
@@ -100,6 +100,8 @@ void           soup_message_set_proxy_auth (SoupMessage *msg,
 SoupAuth      *soup_message_get_proxy_auth (SoupMessage *msg);
 
 /* I/O */
+void       soup_message_io_run         (SoupMessage *msg,
+					gboolean     blocking);
 void       soup_message_io_stop        (SoupMessage *msg);
 void       soup_message_io_finished    (SoupMessage *msg);
 void       soup_message_io_cleanup     (SoupMessage *msg);
diff --git a/libsoup/soup-session.c b/libsoup/soup-session.c
index df61b3bb..0340eb79 100644
--- a/libsoup/soup-session.c
+++ b/libsoup/soup-session.c
@@ -1410,6 +1410,7 @@ tunnel_message_completed (SoupMessage *msg, SoupMessageIOCompletion completion,
 			tunnel_item->state = SOUP_MESSAGE_RUNNING;
 			soup_session_send_queue_item (session, tunnel_item,
 						      tunnel_message_completed);
+			soup_message_io_run (msg, !tunnel_item->async);
 			return;
 		}
 
@@ -1452,9 +1453,8 @@ tunnel_connect (SoupMessageQueueItem *item)
 	soup_message_set_flags (msg, SOUP_MESSAGE_NO_REDIRECT);
 
 	tunnel_item = soup_session_append_queue_item (session, msg,
-						      item->async, FALSE,
+						      item->async, TRUE,
 						      NULL, NULL);
-	g_object_unref (msg);
 	tunnel_item->related = item;
 	soup_message_queue_item_ref (item);
 	soup_session_set_item_connection (session, tunnel_item, item->conn);
@@ -1464,6 +1464,8 @@ tunnel_connect (SoupMessageQueueItem *item)
 
 	soup_session_send_queue_item (session, tunnel_item,
 				      tunnel_message_completed);
+	soup_message_io_run (msg, !item->async);
+	g_object_unref (msg);
 }
 
 static void
@@ -1745,6 +1747,7 @@ soup_session_process_queue_item (SoupSession          *session,
 			break;
 
 		case SOUP_MESSAGE_CACHED:
+		case SOUP_MESSAGE_TUNNELING:
 			/* Will be handled elsewhere */
 			return;
 
@@ -3339,11 +3342,16 @@ run_until_read_done (SoupMessage          *msg,
 static void
 async_send_request_running (SoupSession *session, SoupMessageQueueItem *item)
 {
-	item->io_started = TRUE;
-	soup_message_io_run_until_read_async (item->msg,
-					      item->cancellable,
-					      (GAsyncReadyCallback)run_until_read_done,
-					      item);
+	if (item->task) {
+		item->io_started = TRUE;
+		soup_message_io_run_until_read_async (item->msg,
+						      item->cancellable,
+						      (GAsyncReadyCallback)run_until_read_done,
+						      item);
+		return;
+	}
+
+	soup_message_io_run (item->msg, FALSE);
 }
 
 static void
@@ -4162,7 +4170,7 @@ soup_session_websocket_connect_async (SoupSession          *session,
 	soup_message_set_flags (msg, flags | SOUP_MESSAGE_NEW_CONNECTION);
 
 	task = g_task_new (session, cancellable, callback, user_data);
-	item = soup_session_append_queue_item (session, msg, TRUE, FALSE,
+	item = soup_session_append_queue_item (session, msg, TRUE, TRUE,
 					       websocket_connect_async_complete, task);
 	g_task_set_task_data (task, item, (GDestroyNotify) soup_message_queue_item_unref);