soup-server: add soup_client_context_steal_connection()

Add a method to allow a SoupServer handler to steal the connection
from the server, and use this in simple-proxy to implement CONNECT.

Incorporates a patch from Lionel Landwerlin.

6 files changed, 518 insertions, 1 deletions

diff --git a/docs/reference/libsoup-2.4-sections.txt b/docs/reference/libsoup-2.4-sections.txt
index 74391057..678c2ab1 100644
--- a/docs/reference/libsoup-2.4-sections.txt
+++ b/docs/reference/libsoup-2.4-sections.txt
@@ -242,6 +242,7 @@ soup_client_context_get_host
 soup_client_context_get_auth_domain
 soup_client_context_get_auth_user
 soup_client_context_get_gsocket
+soup_client_context_steal_connection
 <SUBSECTION>
 soup_server_add_auth_domain
 soup_server_remove_auth_domain
diff --git a/examples/simple-proxy.c b/examples/simple-proxy.c
index 08bd847c..7657355e 100644
--- a/examples/simple-proxy.c
+++ b/examples/simple-proxy.c
@@ -15,6 +15,217 @@
 static SoupSession *session;
 static SoupServer *server;
 
+typedef struct {
+	GIOStream *iostream;
+	GInputStream *istream;
+	GOutputStream *ostream;
+
+	gssize nread, nwrote;
+	guchar *buffer;
+} TunnelEnd;
+
+typedef struct {
+	SoupServer *self;
+	SoupMessage *msg;
+	SoupClientContext *context;
+	GCancellable *cancellable;
+
+	TunnelEnd client, server;
+} Tunnel;
+
+#define BUFSIZE 8192
+
+static void tunnel_read_cb (GObject      *object,
+			    GAsyncResult *result,
+			    gpointer      user_data);
+
+static void
+tunnel_close (Tunnel *tunnel)
+{
+	if (tunnel->cancellable) {
+		g_cancellable_cancel (tunnel->cancellable);
+		g_object_unref (tunnel->cancellable);
+	}
+
+	if (tunnel->client.iostream) {
+		g_io_stream_close (tunnel->client.iostream, NULL, NULL);
+		g_object_unref (tunnel->client.iostream);
+	}
+	if (tunnel->server.iostream) {
+		g_io_stream_close (tunnel->server.iostream, NULL, NULL);
+		g_object_unref (tunnel->server.iostream);
+	}
+
+	g_free (tunnel->client.buffer);
+	g_free (tunnel->server.buffer);
+
+	g_object_unref (tunnel->self);
+	g_object_unref (tunnel->msg);
+
+	g_free (tunnel);
+}
+
+static void
+tunnel_wrote_cb (GObject      *object,
+		 GAsyncResult *result,
+		 gpointer      user_data)
+{
+	Tunnel *tunnel = user_data;
+	TunnelEnd *write_end, *read_end;
+	GError *error = NULL;
+	gssize nwrote;
+
+	nwrote = g_output_stream_write_finish (G_OUTPUT_STREAM (object), result, &error);
+	if (nwrote <= 0) {
+		if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
+			g_error_free (error);
+			return;
+		} else if (error) {
+			g_print ("Tunnel write failed: %s\n", error->message);
+			g_error_free (error);
+		}
+		tunnel_close (tunnel);
+		return;
+	}
+
+	if (object == (GObject *)tunnel->client.ostream) {
+		write_end = &tunnel->client;
+		read_end = &tunnel->server;
+	} else {
+		write_end = &tunnel->server;
+		read_end = &tunnel->client;
+	}
+
+	write_end->nwrote += nwrote;
+	if (write_end->nwrote < read_end->nread) {
+		g_output_stream_write_async (write_end->ostream,
+					     read_end->buffer + write_end->nwrote,
+					     read_end->nread - write_end->nwrote,
+					     G_PRIORITY_DEFAULT, tunnel->cancellable,
+					     tunnel_wrote_cb, tunnel);
+	} else {
+		g_input_stream_read_async (read_end->istream,
+					   read_end->buffer, BUFSIZE,
+					   G_PRIORITY_DEFAULT, tunnel->cancellable,
+					   tunnel_read_cb, tunnel);
+	}
+}
+
+static void
+tunnel_read_cb (GObject      *object,
+		GAsyncResult *result,
+		gpointer      user_data)
+{
+	Tunnel *tunnel = user_data;
+	TunnelEnd *read_end, *write_end;
+	GError *error = NULL;
+	gssize nread;
+
+	nread = g_input_stream_read_finish (G_INPUT_STREAM (object), result, &error);
+	if (nread <= 0) {
+		if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
+			g_error_free (error);
+			return;
+		} else if (error) {
+			g_print ("Tunnel read failed: %s\n", error->message);
+			g_error_free (error);
+		}
+		tunnel_close (tunnel);
+		return;
+	}
+
+	if (object == (GObject *)tunnel->client.istream) {
+		read_end = &tunnel->client;
+		write_end = &tunnel->server;
+	} else {
+		read_end = &tunnel->server;
+		write_end = &tunnel->client;
+	}
+
+	read_end->nread = nread;
+	write_end->nwrote = 0;
+	g_output_stream_write_async (write_end->ostream,
+				     read_end->buffer, read_end->nread,
+				     G_PRIORITY_DEFAULT, tunnel->cancellable,
+				     tunnel_wrote_cb, tunnel);
+}
+
+static void
+start_tunnel (SoupMessage *msg, gpointer user_data)
+{
+	Tunnel *tunnel = user_data;
+
+	tunnel->client.iostream = soup_client_context_steal_connection (tunnel->context);
+	tunnel->client.istream = g_io_stream_get_input_stream (tunnel->client.iostream);
+	tunnel->client.ostream = g_io_stream_get_output_stream (tunnel->client.iostream);
+
+	tunnel->client.buffer = g_malloc (BUFSIZE);
+	tunnel->server.buffer = g_malloc (BUFSIZE);
+
+	tunnel->cancellable = g_cancellable_new ();
+
+	g_input_stream_read_async (tunnel->client.istream,
+				   tunnel->client.buffer, BUFSIZE,
+				   G_PRIORITY_DEFAULT, tunnel->cancellable,
+				   tunnel_read_cb, tunnel);
+	g_input_stream_read_async (tunnel->server.istream,
+				   tunnel->server.buffer, BUFSIZE,
+				   G_PRIORITY_DEFAULT, tunnel->cancellable,
+				   tunnel_read_cb, tunnel);
+}
+
+
+static void
+tunnel_connected_cb (GObject      *object,
+		     GAsyncResult *result,
+		     gpointer      user_data)
+{
+	Tunnel *tunnel = user_data;
+	GError *error = NULL;
+
+	tunnel->server.iostream = (GIOStream *)
+		g_socket_client_connect_to_host_finish (G_SOCKET_CLIENT (object), result, &error);
+	if (!tunnel->server.iostream) {
+		soup_message_set_status (tunnel->msg, SOUP_STATUS_BAD_GATEWAY);
+		soup_message_set_response (tunnel->msg, "text/plain",
+					   SOUP_MEMORY_COPY,
+					   error->message, strlen (error->message));
+		g_error_free (error);
+		soup_server_unpause_message (tunnel->self, tunnel->msg);
+		tunnel_close (tunnel);
+		return;
+	}
+
+	tunnel->server.istream = g_io_stream_get_input_stream (tunnel->server.iostream);
+	tunnel->server.ostream = g_io_stream_get_output_stream (tunnel->server.iostream);
+
+	soup_message_set_status (tunnel->msg, SOUP_STATUS_OK);
+	soup_server_unpause_message (tunnel->self, tunnel->msg);
+	g_signal_connect (tunnel->msg, "finished",
+			  G_CALLBACK (start_tunnel), tunnel);
+}
+
+static void
+try_tunnel (SoupServer *server, SoupMessage *msg, SoupClientContext *context)
+{
+	Tunnel *tunnel;
+	SoupURI *dest_uri;
+	GSocketClient *sclient;
+
+	soup_server_pause_message (server, msg);
+
+	tunnel = g_new0 (Tunnel, 1);
+	tunnel->self = g_object_ref (server);
+	tunnel->msg = g_object_ref (msg);
+	tunnel->context = context;
+
+	dest_uri = soup_message_get_uri (msg);
+	sclient = g_socket_client_new ();
+	g_socket_client_connect_to_host_async (sclient, dest_uri->host, dest_uri->port,
+					       NULL, tunnel_connected_cb, tunnel);
+	g_object_unref (sclient);
+}
+
 static void
 copy_header (const char *name, const char *value, gpointer dest_headers)
 {
@@ -78,7 +289,7 @@ server_callback (SoupServer *server, SoupMessage *msg,
 		 soup_message_get_http_version (msg));
 
 	if (msg->method == SOUP_METHOD_CONNECT) {
-		soup_message_set_status (msg, SOUP_STATUS_NOT_IMPLEMENTED);
+		try_tunnel (server, msg, context);
 		return;
 	}
 
diff --git a/libsoup/libsoup-2.4.sym b/libsoup/libsoup-2.4.sym
index e554702e..9d7eeecd 100644
--- a/libsoup/libsoup-2.4.sym
+++ b/libsoup/libsoup-2.4.sym
@@ -114,6 +114,7 @@ soup_client_context_get_local_address
 soup_client_context_get_remote_address
 soup_client_context_get_socket
 soup_client_context_get_type
+soup_client_context_steal_connection
 soup_connection_state_get_type
 soup_content_decoder_get_type
 soup_content_sniffer_get_buffer_size
diff --git a/libsoup/soup-server.c b/libsoup/soup-server.c
index 7c855324..2807439d 100644
--- a/libsoup/soup-server.c
+++ b/libsoup/soup-server.c
@@ -1166,6 +1166,13 @@ request_finished (SoupMessage *msg, SoupMessageIOCompletion completion, gpointer
 	SoupSocket *sock = client->sock;
 	gboolean failed;
 
+	if (completion == SOUP_MESSAGE_IO_STOLEN) {
+		soup_client_context_unref (client);
+		g_object_unref (msg);
+		g_object_unref (sock);
+		return;
+	}
+
 	soup_message_finished (msg);
 
 	failed = (completion == SOUP_MESSAGE_IO_INTERRUPTED ||
@@ -2307,6 +2314,52 @@ soup_client_context_get_auth_user (SoupClientContext *client)
 }
 
 /**
+ * soup_client_context_steal_connection:
+ * @client: a #SoupClientContext
+ *
+ * "Steals" the HTTP connection associated with @client from its
+ * #SoupServer. This happens immediately, regardless of the current
+ * state of the connection; if the response to the current
+ * #SoupMessage has not yet finished being sent, then it will be
+ * discarded; you can steal the connection from a
+ * #SoupMessage:wrote-informational or #SoupMessage:wrote-body signal
+ * handler if you need to wait for part or all of the response to be
+ * sent.
+ *
+ * Note that when calling this function from C, @client will most
+ * likely be freed as a side effect.
+ *
+ * Return value: (transfer full): the #GIOStream formerly associated
+ *   with @client (or %NULL if @client was no longer associated with a
+ *   connection). No guarantees are made about what kind of #GIOStream
+ *   is returned.
+ *
+ * Since: 2.50
+ **/
+GIOStream *
+soup_client_context_steal_connection (SoupClientContext *client)
+{
+	GIOStream *stream;
+
+	g_return_val_if_fail (client != NULL, NULL);
+
+	soup_client_context_ref (client);
+
+	stream = soup_message_io_steal (client->msg);
+	if (stream) {
+		g_object_set (G_OBJECT (client->sock),
+			      SOUP_SOCKET_CLOSE_ON_DISPOSE, FALSE,
+			      NULL);
+	}
+
+	socket_disconnected (client->sock, client);
+	soup_client_context_unref (client);
+
+	return stream;
+}
+
+
+/**
  * SoupServerCallback:
  * @server: the #SoupServer
  * @msg: the message being processed
diff --git a/libsoup/soup-server.h b/libsoup/soup-server.h
index 901f33b3..3cabc037 100644
--- a/libsoup/soup-server.h
+++ b/libsoup/soup-server.h
@@ -158,6 +158,8 @@ const char     *soup_client_context_get_host           (SoupClientContext *clien
 SoupAuthDomain *soup_client_context_get_auth_domain    (SoupClientContext *client);
 const char     *soup_client_context_get_auth_user      (SoupClientContext *client);
 
+SOUP_AVAILABLE_IN_2_50
+GIOStream      *soup_client_context_steal_connection   (SoupClientContext *client);
 
 /* Legacy API */
 
diff --git a/tests/server-test.c b/tests/server-test.c
index e00af693..80ad03ef 100644
--- a/tests/server-test.c
+++ b/tests/server-test.c
@@ -1053,6 +1053,253 @@ do_early_multi_test (ServerData *sd, gconstpointer test_data)
 	soup_test_session_abort_unref (session);
 }
 
+typedef struct {
+	GIOStream *iostream;
+	GInputStream *istream;
+	GOutputStream *ostream;
+
+	gssize nread, nwrote;
+	guchar *buffer;
+} TunnelEnd;
+
+typedef struct {
+	SoupServer *self;
+	SoupMessage *msg;
+	SoupClientContext *context;
+	GCancellable *cancellable;
+
+	TunnelEnd client, server;
+} Tunnel;
+
+#define BUFSIZE 8192
+
+static void tunnel_read_cb (GObject      *object,
+			    GAsyncResult *result,
+			    gpointer      user_data);
+
+static void
+tunnel_close (Tunnel *tunnel)
+{
+	if (tunnel->cancellable) {
+		g_cancellable_cancel (tunnel->cancellable);
+		g_object_unref (tunnel->cancellable);
+	}
+
+	if (tunnel->client.iostream) {
+		g_io_stream_close (tunnel->client.iostream, NULL, NULL);
+		g_object_unref (tunnel->client.iostream);
+	}
+	if (tunnel->server.iostream) {
+		g_io_stream_close (tunnel->server.iostream, NULL, NULL);
+		g_object_unref (tunnel->server.iostream);
+	}
+
+	g_free (tunnel->client.buffer);
+	g_free (tunnel->server.buffer);
+
+	g_object_unref (tunnel->self);
+	g_object_unref (tunnel->msg);
+
+	g_free (tunnel);
+}
+
+static void
+tunnel_wrote_cb (GObject      *object,
+		 GAsyncResult *result,
+		 gpointer      user_data)
+{
+	Tunnel *tunnel = user_data;
+	TunnelEnd *write_end, *read_end;
+	GError *error = NULL;
+	gssize nwrote;
+
+	nwrote = g_output_stream_write_finish (G_OUTPUT_STREAM (object), result, &error);
+	if (nwrote <= 0) {
+		if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
+			g_error_free (error);
+			return;
+		} else if (error) {
+			g_print ("Tunnel write failed: %s\n", error->message);
+			g_error_free (error);
+		}
+		tunnel_close (tunnel);
+		return;
+	}
+
+	if (object == (GObject *)tunnel->client.ostream) {
+		write_end = &tunnel->client;
+		read_end = &tunnel->server;
+	} else {
+		write_end = &tunnel->server;
+		read_end = &tunnel->client;
+	}
+
+	write_end->nwrote += nwrote;
+	if (write_end->nwrote < read_end->nread) {
+		g_output_stream_write_async (write_end->ostream,
+					     read_end->buffer + write_end->nwrote,
+					     read_end->nread - write_end->nwrote,
+					     G_PRIORITY_DEFAULT, tunnel->cancellable,
+					     tunnel_wrote_cb, tunnel);
+	} else {
+		g_input_stream_read_async (read_end->istream,
+					   read_end->buffer, BUFSIZE,
+					   G_PRIORITY_DEFAULT, tunnel->cancellable,
+					   tunnel_read_cb, tunnel);
+	}
+}
+
+static void
+tunnel_read_cb (GObject      *object,
+		GAsyncResult *result,
+		gpointer      user_data)
+{
+	Tunnel *tunnel = user_data;
+	TunnelEnd *read_end, *write_end;
+	GError *error = NULL;
+	gssize nread;
+
+	nread = g_input_stream_read_finish (G_INPUT_STREAM (object), result, &error);
+	if (nread <= 0) {
+		if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_CANCELLED)) {
+			g_error_free (error);
+			return;
+		} else if (error) {
+			g_print ("Tunnel read failed: %s\n", error->message);
+			g_error_free (error);
+		}
+		tunnel_close (tunnel);
+		return;
+	}
+
+	if (object == (GObject *)tunnel->client.istream) {
+		read_end = &tunnel->client;
+		write_end = &tunnel->server;
+	} else {
+		read_end = &tunnel->server;
+		write_end = &tunnel->client;
+	}
+
+	read_end->nread = nread;
+	write_end->nwrote = 0;
+	g_output_stream_write_async (write_end->ostream,
+				     read_end->buffer, read_end->nread,
+				     G_PRIORITY_DEFAULT, tunnel->cancellable,
+				     tunnel_wrote_cb, tunnel);
+}
+
+static void
+start_tunnel (SoupMessage *msg, gpointer user_data)
+{
+	Tunnel *tunnel = user_data;
+
+	tunnel->client.iostream = soup_client_context_steal_connection (tunnel->context);
+	tunnel->client.istream = g_io_stream_get_input_stream (tunnel->client.iostream);
+	tunnel->client.ostream = g_io_stream_get_output_stream (tunnel->client.iostream);
+
+	tunnel->client.buffer = g_malloc (BUFSIZE);
+	tunnel->server.buffer = g_malloc (BUFSIZE);
+
+	tunnel->cancellable = g_cancellable_new ();
+
+	g_input_stream_read_async (tunnel->client.istream,
+				   tunnel->client.buffer, BUFSIZE,
+				   G_PRIORITY_DEFAULT, tunnel->cancellable,
+				   tunnel_read_cb, tunnel);
+	g_input_stream_read_async (tunnel->server.istream,
+				   tunnel->server.buffer, BUFSIZE,
+				   G_PRIORITY_DEFAULT, tunnel->cancellable,
+				   tunnel_read_cb, tunnel);
+}
+
+
+static void
+tunnel_connected_cb (GObject      *object,
+		     GAsyncResult *result,
+		     gpointer      user_data)
+{
+	Tunnel *tunnel = user_data;
+	GError *error = NULL;
+
+	tunnel->server.iostream = (GIOStream *)
+		g_socket_client_connect_to_host_finish (G_SOCKET_CLIENT (object), result, &error);
+	if (!tunnel->server.iostream) {
+		soup_message_set_status (tunnel->msg, SOUP_STATUS_BAD_GATEWAY);
+		soup_message_set_response (tunnel->msg, "text/plain",
+					   SOUP_MEMORY_COPY,
+					   error->message, strlen (error->message));
+		g_error_free (error);
+		soup_server_unpause_message (tunnel->self, tunnel->msg);
+		tunnel_close (tunnel);
+		return;
+	}
+
+	tunnel->server.istream = g_io_stream_get_input_stream (tunnel->server.iostream);
+	tunnel->server.ostream = g_io_stream_get_output_stream (tunnel->server.iostream);
+
+	soup_message_set_status (tunnel->msg, SOUP_STATUS_OK);
+	soup_server_unpause_message (tunnel->self, tunnel->msg);
+	g_signal_connect (tunnel->msg, "wrote-body",
+			  G_CALLBACK (start_tunnel), tunnel);
+}
+
+static void
+proxy_server_callback (SoupServer *server, SoupMessage *msg,
+		       const char *path, GHashTable *query,
+		       SoupClientContext *context, gpointer data)
+{
+	GSocketClient *sclient;
+	SoupURI *dest_uri;
+	Tunnel *tunnel;
+
+	if (msg->method != SOUP_METHOD_CONNECT) {
+		soup_message_set_status (msg, SOUP_STATUS_NOT_IMPLEMENTED);
+		return;
+	}
+
+	soup_server_pause_message (server, msg);
+
+	tunnel = g_new0 (Tunnel, 1);
+	tunnel->self = g_object_ref (server);
+	tunnel->msg = g_object_ref (msg);
+	tunnel->context = context;
+
+	dest_uri = soup_message_get_uri (msg);
+	sclient = g_socket_client_new ();
+	g_socket_client_connect_to_host_async (sclient, dest_uri->host, dest_uri->port,
+					       NULL, tunnel_connected_cb, tunnel);
+	g_object_unref (sclient);
+}
+
+static void
+do_steal_connect_test (ServerData *sd, gconstpointer test_data)
+{
+	SoupServer *proxy;
+	SoupURI *proxy_uri;
+	SoupSession *session;
+	SoupMessage *msg;
+	const char *handled_by;
+
+	proxy = soup_test_server_new (SOUP_TEST_SERVER_IN_THREAD);
+	proxy_uri = soup_test_server_get_uri (proxy, SOUP_URI_SCHEME_HTTP, "127.0.0.1");
+	soup_server_add_handler (proxy, NULL, proxy_server_callback, NULL, NULL);
+
+	session = soup_test_session_new (SOUP_TYPE_SESSION,
+					 SOUP_SESSION_PROXY_URI, proxy_uri,
+					 NULL);
+	msg = soup_message_new_from_uri ("GET", sd->ssl_base_uri);
+	soup_session_send_message (session, msg);
+
+	soup_test_assert_message_status (msg, SOUP_STATUS_OK);
+	handled_by = soup_message_headers_get_one (msg->response_headers, "X-Handled-By");
+	g_assert_cmpstr (handled_by, ==, "server_callback");
+
+	g_object_unref (msg);
+	soup_test_session_abort_unref (session);
+	soup_test_server_quit_unref (proxy);
+	soup_uri_free (proxy_uri);
+}
+
 int
 main (int argc, char **argv)
 {
@@ -1089,6 +1336,8 @@ main (int argc, char **argv)
 		    server_setup, do_early_respond_test, server_teardown);
 	g_test_add ("/server/early/multi", ServerData, NULL,
 		    server_setup_nohandler, do_early_multi_test, server_teardown);
+	g_test_add ("/server/steal/CONNECT", ServerData, NULL,
+		    server_setup, do_steal_connect_test, server_teardown);
 
 	ret = g_test_run ();