Add new API to create a new connection from a SoupSession

Add soup_session_connect_async() to create a new dedicated connection to
the given SoupURI. The operation finishes when the connection has been
completed, including any TLS handshacking or proxy negotiation,
returning a GIOStream that can be used to communicate with the server.

This is based on a patch by Dirkjan Ochtman <dirkjan@ochtman.nl>.

https://bugzilla.gnome.org/show_bug.cgi?id=792212

6 files changed, 416 insertions, 28 deletions

diff --git a/docs/reference/libsoup-2.4-sections.txt b/docs/reference/libsoup-2.4-sections.txt
index b3e15c9f..a9b21384 100644
--- a/docs/reference/libsoup-2.4-sections.txt
+++ b/docs/reference/libsoup-2.4-sections.txt
@@ -475,6 +475,10 @@ soup_session_has_feature
 <SUBSECTION>
 soup_session_steal_connection
 <SUBSECTION>
+SoupSessionConnectProgressCallback
+soup_session_connect_async
+soup_session_connect_finish
+<SUBSECTION>
 SOUP_SESSION_PROXY_URI
 SOUP_SESSION_PROXY_RESOLVER
 SOUP_SESSION_MAX_CONNS
diff --git a/libsoup/soup-message-queue.h b/libsoup/soup-message-queue.h
index cfb8d8ad..275ea38d 100644
--- a/libsoup/soup-message-queue.h
+++ b/libsoup/soup-message-queue.h
@@ -50,6 +50,7 @@ struct _SoupMessageQueueItem {
 	guint async             : 1;
 	guint async_pending     : 1;
 	guint conn_is_dedicated : 1;
+	guint connect_only      : 1;
 	guint priority          : 3;
 	guint resend_count      : 5;
 
diff --git a/libsoup/soup-session.c b/libsoup/soup-session.c
index 3387f68a..567aefe2 100644
--- a/libsoup/soup-session.c
+++ b/libsoup/soup-session.c
@@ -1976,6 +1976,11 @@ soup_session_process_queue_item (SoupSession          *session,
 			break;
 
 		case SOUP_MESSAGE_READY:
+			if (item->connect_only) {
+				item->state = SOUP_MESSAGE_FINISHING;
+				break;
+			}
+
 			if (item->msg->status_code) {
 				if (item->msg->status_code == SOUP_STATUS_TRY_AGAIN) {
 					soup_message_cleanup_response (item->msg);
@@ -4683,6 +4688,42 @@ soup_request_error_quark (void)
 	return error;
 }
 
+static GIOStream *
+steal_connection (SoupSession          *session,
+                  SoupMessageQueueItem *item)
+{
+        SoupSessionPrivate *priv = soup_session_get_instance_private (session);
+        SoupConnection *conn;
+        SoupSocket *sock;
+        SoupSessionHost *host;
+        GIOStream *stream;
+
+        conn = g_object_ref (item->conn);
+        soup_session_set_item_connection (session, item, NULL);
+
+        g_mutex_lock (&priv->conn_lock);
+        host = get_host_for_message (session, item->msg);
+        g_hash_table_remove (priv->conns, conn);
+        drop_connection (session, host, conn);
+        g_mutex_unlock (&priv->conn_lock);
+
+        sock = soup_connection_get_socket (conn);
+        g_object_set (sock,
+                      SOUP_SOCKET_TIMEOUT, 0,
+                      NULL);
+
+	if (item->connect_only)
+		stream = g_object_ref (soup_socket_get_connection (sock));
+	else
+		stream = soup_message_io_steal (item->msg);
+        g_object_set_data_full (G_OBJECT (stream), "GSocket",
+                                soup_socket_steal_gsocket (sock),
+                                g_object_unref);
+        g_object_unref (conn);
+
+	return stream;
+}
+
 /**
  * soup_session_steal_connection:
  * @session: a #SoupSession
@@ -4710,41 +4751,17 @@ soup_session_steal_connection (SoupSession *session,
 {
 	SoupSessionPrivate *priv = soup_session_get_instance_private (session);
 	SoupMessageQueueItem *item;
-	SoupConnection *conn;
-	SoupSocket *sock;
-	SoupSessionHost *host;
-	GIOStream *stream;
+	GIOStream *stream = NULL;
 
 	item = soup_message_queue_lookup (priv->queue, msg);
 	if (!item)
 		return NULL;
-	if (!item->conn ||
-	    soup_connection_get_state (item->conn) != SOUP_CONNECTION_IN_USE) {
-		soup_message_queue_item_unref (item);
-		return NULL;
-	}
 
-	conn = g_object_ref (item->conn);
-	soup_session_set_item_connection (session, item, NULL);
-
-	g_mutex_lock (&priv->conn_lock);
-	host = get_host_for_message (session, item->msg);
-	g_hash_table_remove (priv->conns, conn);
-	drop_connection (session, host, conn);
-	g_mutex_unlock (&priv->conn_lock);
-
-	sock = soup_connection_get_socket (conn);
-	g_object_set (sock,
-		      SOUP_SOCKET_TIMEOUT, 0,
-		      NULL);
-
-	stream = soup_message_io_steal (item->msg);
-	g_object_set_data_full (G_OBJECT (stream), "GSocket",
-				soup_socket_steal_gsocket (sock),
-				g_object_unref);
-	g_object_unref (conn);
+	if (item->conn && soup_connection_get_state (item->conn) == SOUP_CONNECTION_IN_USE)
+		stream = steal_connection (session, item);
 
 	soup_message_queue_item_unref (item);
+
 	return stream;
 }
 
@@ -4881,3 +4898,155 @@ soup_session_websocket_connect_finish (SoupSession      *session,
 
 	return g_task_propagate_pointer (G_TASK (result), error);
 }
+
+/**
+ * SoupSessionConnectProgressCallback:
+ * @session: the #SoupSession
+ * @event: a #GSocketClientEvent
+ * @connection: the current state of the network connection
+ * @user_data: the data passed to soup_session_connect_async().
+ *
+ * Prototype for the progress callback passed to soup_session_connect_async().
+ *
+ * Since: 2.62
+ */
+
+typedef struct {
+        SoupMessageQueueItem *item;
+        SoupSessionConnectProgressCallback progress_callback;
+        gpointer user_data;
+} ConnectAsyncData;
+
+static ConnectAsyncData *
+connect_async_data_new (SoupMessageQueueItem              *item,
+                        SoupSessionConnectProgressCallback progress_callback,
+                        gpointer                           user_data)
+{
+        ConnectAsyncData *data;
+
+        data = g_slice_new (ConnectAsyncData);
+        data->item = soup_message_queue_item_ref (item);
+        data->progress_callback = progress_callback;
+        data->user_data = user_data;
+
+        return data;
+}
+
+static void
+connect_async_data_free (ConnectAsyncData *data)
+{
+        soup_message_queue_item_unref (data->item);
+
+        g_slice_free (ConnectAsyncData, data);
+}
+
+static void
+connect_async_message_network_event (SoupMessage        *msg,
+                                     GSocketClientEvent  event,
+                                     GIOStream          *connection,
+                                     GTask              *task)
+{
+        ConnectAsyncData *data = g_task_get_task_data (task);
+
+        if (data->progress_callback)
+                data->progress_callback (data->item->session, event, connection, data->user_data);
+}
+
+static void
+connect_async_message_finished (SoupMessage *msg,
+                                GTask       *task)
+{
+        ConnectAsyncData *data = g_task_get_task_data (task);
+        SoupMessageQueueItem *item = data->item;
+
+        if (!item->conn || item->error) {
+                g_task_return_error (task, g_error_copy (item->error));
+        } else {
+                g_task_return_pointer (task,
+                                       steal_connection (item->session, item),
+                                       g_object_unref);
+        }
+        g_object_unref (task);
+}
+
+/**
+ * soup_session_connect_async:
+ * @session: a #SoupSession
+ * @uri: a #SoupURI to connect to
+ * @cancellable: a #GCancellable
+ * @progress_callback: (allow-none) (scope async): a #SoupSessionConnectProgressCallback which
+ * will be called for every network event that occurs during the connection.
+ * @callback: (allow-none) (scope async): the callback to invoke when the operation finishes
+ * @user_data: data for @progress_callback and @callback
+ *
+ * Start a connection to @uri. The operation can be monitored by providing a @progress_callback
+ * and finishes when the connection is done or an error ocurred.
+ *
+ * Call soup_session_connect_finish() to get the #GIOStream to communicate with the server.
+ *
+ * Since: 2.62
+ */
+void
+soup_session_connect_async (SoupSession                       *session,
+                            SoupURI                           *uri,
+                            GCancellable                      *cancellable,
+                            SoupSessionConnectProgressCallback progress_callback,
+                            GAsyncReadyCallback                callback,
+                            gpointer                           user_data)
+{
+        SoupSessionPrivate *priv;
+        SoupMessage *msg;
+        SoupMessageQueueItem *item;
+        ConnectAsyncData *data;
+        GTask *task;
+
+        g_return_if_fail (SOUP_IS_SESSION (session));
+        g_return_if_fail (!SOUP_IS_SESSION_SYNC (session));
+        priv = soup_session_get_instance_private (session);
+        g_return_if_fail (priv->use_thread_context);
+        g_return_if_fail (uri != NULL);
+
+        task = g_task_new (session, cancellable, callback, user_data);
+
+        msg = soup_message_new_from_uri (SOUP_METHOD_HEAD, uri);
+        soup_message_set_flags (msg, SOUP_MESSAGE_NEW_CONNECTION);
+        g_signal_connect_object (msg, "finished",
+                                 G_CALLBACK (connect_async_message_finished),
+                                 task, 0);
+        if (progress_callback) {
+                g_signal_connect_object (msg, "network-event",
+                                         G_CALLBACK (connect_async_message_network_event),
+                                         task, 0);
+        }
+
+        item = soup_session_append_queue_item (session, msg, TRUE, FALSE, NULL, NULL);
+        item->connect_only = TRUE;
+        data = connect_async_data_new (item, progress_callback, user_data);
+        g_task_set_task_data (task, data, (GDestroyNotify) connect_async_data_free);
+        soup_session_kick_queue (session);
+        soup_message_queue_item_unref (item);
+        g_object_unref (msg);
+}
+
+/**
+ * soup_session_connect_finish:
+ * @session: a #SoupSession
+ * @result: the #GAsyncResult passed to your callback
+ * @error: return location for a #GError, or %NULL
+ *
+ * Gets the #GIOStream created for the connection to communicate with the server.
+ *
+ * Return value: (transfer full): a new #GIOStream, or %NULL on error.
+ *
+ * Since: 2.62
+ */
+GIOStream *
+soup_session_connect_finish (SoupSession  *session,
+                             GAsyncResult *result,
+                             GError      **error)
+{
+        g_return_val_if_fail (SOUP_IS_SESSION (session), NULL);
+        g_return_val_if_fail (g_task_is_valid (result, session), NULL);
+
+        return g_task_propagate_pointer (G_TASK (result), error);
+}
diff --git a/libsoup/soup-session.h b/libsoup/soup-session.h
index 25dd02b8..24e90b29 100644
--- a/libsoup/soup-session.h
+++ b/libsoup/soup-session.h
@@ -234,6 +234,24 @@ SoupWebsocketConnection *soup_session_websocket_connect_finish (SoupSession
 								GAsyncResult         *result,
 								GError              **error);
 
+typedef void (*SoupSessionConnectProgressCallback) (SoupSession        *session,
+						    GSocketClientEvent  event,
+						    GIOStream          *connection,
+						    gpointer            user_data);
+
+SOUP_AVAILABLE_IN_2_62
+void       soup_session_connect_async  (SoupSession                       *session,
+					SoupURI                           *uri,
+					GCancellable                      *cancellable,
+					SoupSessionConnectProgressCallback progress_callback,
+					GAsyncReadyCallback                callback,
+					gpointer                           user_data);
+
+SOUP_AVAILABLE_IN_2_62
+GIOStream *soup_session_connect_finish (SoupSession                       *session,
+					GAsyncResult                      *result,
+					GError                           **error);
+
 G_END_DECLS
 
 #endif /* SOUP_SESSION_H */
diff --git a/libsoup/soup-version.h.in b/libsoup/soup-version.h.in
index 27ae7512..0af245be 100644
--- a/libsoup/soup-version.h.in
+++ b/libsoup/soup-version.h.in
@@ -65,6 +65,7 @@ G_BEGIN_DECLS
 #define SOUP_VERSION_2_54 (G_ENCODE_VERSION (2, 54))
 #define SOUP_VERSION_2_56 (G_ENCODE_VERSION (2, 56))
 #define SOUP_VERSION_2_58 (G_ENCODE_VERSION (2, 58))
+#define SOUP_VERSION_2_62 (G_ENCODE_VERSION (2, 62))
 
 /* evaluates to the current stable version; for development cycles,
  * this means the next stable target
@@ -359,6 +360,20 @@ G_BEGIN_DECLS
 # define SOUP_AVAILABLE_IN_2_58                 _SOUP_EXTERN
 #endif
 
+#if SOUP_VERSION_MIN_REQUIRED >= SOUP_VERSION_2_62
+# define SOUP_DEPRECATED_IN_2_62                G_DEPRECATED
+# define SOUP_DEPRECATED_IN_2_62_FOR(f)         G_DEPRECATED_FOR(f)
+#else
+# define SOUP_DEPRECATED_IN_2_62
+# define SOUP_DEPRECATED_IN_2_62_FOR(f)
+#endif
+
+#if SOUP_VERSION_MAX_ALLOWED < SOUP_VERSION_2_62
+# define SOUP_AVAILABLE_IN_2_62                 G_UNAVAILABLE(2, 62) _SOUP_EXTERN
+#else
+# define SOUP_AVAILABLE_IN_2_62                 _SOUP_EXTERN
+#endif
+
 SOUP_AVAILABLE_IN_2_42
 guint    soup_get_major_version (void);
 
diff --git a/tests/connection-test.c b/tests/connection-test.c
index f460b557..ec54daea 100644
--- a/tests/connection-test.c
+++ b/tests/connection-test.c
@@ -914,6 +914,186 @@ do_connection_event_test (void)
 	soup_test_session_abort_unref (session);
 }
 
+typedef struct {
+        GMainLoop *loop;
+        GIOStream *stream;
+        GError *error;
+        const char *events;
+} ConnectTestData;
+
+static void
+connect_progress (SoupSession *session, GSocketClientEvent event, GIOStream *connection, ConnectTestData *data)
+{
+        soup_test_assert (*data->events == event_abbrevs[event],
+                          "Unexpected event: %s (expected %s)",
+                          event_names[event],
+                          event_name_from_abbrev (*data->events));
+        data->events = data->events + 1;
+}
+
+static void
+connect_finished (SoupSession *session, GAsyncResult *result, ConnectTestData *data)
+{
+        data->stream = soup_session_connect_finish (session, result, &data->error);
+        g_main_loop_quit (data->loop);
+}
+
+static void
+do_one_connection_connect_test (SoupSession *session, SoupURI *uri, const char *response, const char *events)
+{
+        ConnectTestData data = { NULL, NULL, NULL, events };
+        static const char *request = "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n";
+        gsize bytes = 0;
+        char buffer[128];
+
+        data.loop = g_main_loop_new (NULL, FALSE);
+        soup_session_connect_async (session, uri, NULL,
+                                    (SoupSessionConnectProgressCallback)connect_progress,
+                                    (GAsyncReadyCallback)connect_finished,
+                                    &data);
+        g_main_loop_run (data.loop);
+
+        g_assert (G_IS_IO_STREAM (data.stream));
+        g_assert_no_error (data.error);
+        g_assert (g_output_stream_write_all (g_io_stream_get_output_stream (data.stream),
+                                             request, strlen (request), &bytes, NULL, NULL));
+        g_assert (g_input_stream_read_all (g_io_stream_get_input_stream (data.stream),
+                                           buffer, sizeof (buffer), &bytes, NULL, NULL));
+        buffer[strlen (response)] = '\0';
+        g_assert_cmpstr (buffer, ==, response);
+
+        while (*data.events) {
+                soup_test_assert (!*data.events,
+                                  "Expected %s",
+                                  event_name_from_abbrev (*data.events));
+                data.events++;
+        }
+
+        g_object_unref (data.stream);
+        g_main_loop_unref (data.loop);
+}
+
+static void
+do_one_connection_connect_fail_test (SoupSession *session, SoupURI *uri, GQuark domain, gint code, const char *events)
+{
+        ConnectTestData data = { NULL, NULL, NULL, events };
+
+        data.loop = g_main_loop_new (NULL, FALSE);
+        soup_session_connect_async (session, uri, NULL,
+                                    (SoupSessionConnectProgressCallback)connect_progress,
+                                    (GAsyncReadyCallback)connect_finished,
+                                    &data);
+        g_main_loop_run (data.loop);
+
+        g_assert (!data.stream);
+        g_assert_error (data.error, domain, code);
+
+        while (*data.events) {
+                soup_test_assert (!*data.events,
+                                  "Expected %s",
+                                  event_name_from_abbrev (*data.events));
+                data.events++;
+        }
+}
+
+static void
+do_connection_connect_test (void)
+{
+        SoupSession *session;
+        SoupURI *http_uri;
+        SoupURI *https_uri = NULL;
+        SoupURI *ws_uri;
+        SoupURI *wss_uri = NULL;
+        SoupURI *file_uri;
+        SoupURI *wrong_http_uri;
+        SoupURI *proxy_uri;
+
+        SOUP_TEST_SKIP_IF_NO_APACHE;
+
+        session = soup_test_session_new (SOUP_TYPE_SESSION_ASYNC,
+                                         SOUP_SESSION_USE_THREAD_CONTEXT, TRUE,
+                                         NULL);
+
+        debug_printf (1, "    http\n");
+        http_uri = soup_uri_new (HTTP_SERVER);
+        do_one_connection_connect_test (session, http_uri,
+                                        "HTTP/1.1 200 OK", "rRcCx");
+
+        if (tls_available) {
+                debug_printf (1, "    https\n");
+                https_uri = soup_uri_new (HTTPS_SERVER);
+                do_one_connection_connect_test (session, https_uri,
+                                                "HTTP/1.1 200 OK", "rRcCtTx");
+        } else
+                debug_printf (1, "    https -- SKIPPING\n");
+
+        debug_printf (1, "    ws\n");
+        ws_uri = soup_uri_new (HTTP_SERVER);
+        ws_uri->scheme = SOUP_URI_SCHEME_WS;
+        do_one_connection_connect_test (session, ws_uri,
+                                        "HTTP/1.1 200 OK", "rRcCx");
+
+        if (tls_available) {
+                debug_printf (1, "    wss\n");
+                wss_uri = soup_uri_new (HTTPS_SERVER);
+                do_one_connection_connect_test (session, wss_uri,
+                                                "HTTP/1.1 200 OK", "rRcCtTx");
+        } else
+                debug_printf (1, "    wss -- SKIPPING\n");
+
+        debug_printf (1, "    file\n");
+        file_uri = soup_uri_new ("file:///foo/bar");
+        do_one_connection_connect_fail_test (session, file_uri,
+                                             G_RESOLVER_ERROR, G_RESOLVER_ERROR_NOT_FOUND,
+                                             "r");
+
+        debug_printf (1, "    wrong http (invalid port)\n");
+        wrong_http_uri = soup_uri_new (HTTP_SERVER);
+        wrong_http_uri->port = 1234;
+        do_one_connection_connect_fail_test (session, wrong_http_uri,
+                                             G_IO_ERROR, G_IO_ERROR_CONNECTION_REFUSED,
+                                             "rRcr"); /* FIXME: why r again? GLib bug? */
+
+        proxy_uri = soup_uri_new (HTTP_PROXY);
+        g_object_set (G_OBJECT (session),
+                      SOUP_SESSION_PROXY_URI, proxy_uri,
+                      NULL);
+
+        debug_printf (1, "    http with proxy\n");
+        do_one_connection_connect_test (session, http_uri,
+                                        "HTTP/1.1 403 Forbidden", "rRcCx");
+
+        if (tls_available) {
+                debug_printf (1, "    https with proxy\n");
+                do_one_connection_connect_test (session, https_uri,
+                                                "HTTP/1.1 200 OK", "rRcCpPtTx");
+        } else
+                debug_printf (1, "    https with proxy -- SKIPPING\n");
+
+        debug_printf (1, "    ws with proxy\n");
+        do_one_connection_connect_test (session, ws_uri,
+                                        "HTTP/1.1 403 Forbidden", "rRcCx");
+
+        if (tls_available) {
+                debug_printf (1, "    wss with proxy\n");
+                do_one_connection_connect_test (session, wss_uri,
+                                                "HTTP/1.1 200 OK", "rRcCpPtTx");
+        } else
+                debug_printf (1, "    wss with proxy -- SKIPPING\n");
+
+        soup_uri_free (http_uri);
+        if (https_uri)
+                soup_uri_free (https_uri);
+        soup_uri_free (ws_uri);
+        if (wss_uri)
+                soup_uri_free (wss_uri);
+        soup_uri_free (file_uri);
+        soup_uri_free (wrong_http_uri);
+        soup_uri_free (proxy_uri);
+
+        soup_test_session_abort_unref (session);
+}
+
 int
 main (int argc, char **argv)
 {
@@ -933,6 +1113,7 @@ main (int argc, char **argv)
 	g_test_add_func ("/connection/non-idempotent", do_non_idempotent_connection_test);
 	g_test_add_func ("/connection/state", do_connection_state_test);
 	g_test_add_func ("/connection/event", do_connection_event_test);
+	g_test_add_func ("/connection/connect", do_connection_connect_test);
 
 	ret = g_test_run ();