message: add SoupMessage::got-body-data signal

It can be used to monitor the progress of the read operation.

Fixes #319

6 files changed, 90 insertions, 16 deletions

diff --git a/libsoup/http1/soup-client-message-io-http1.c b/libsoup/http1/soup-client-message-io-http1.c
index 8c647bfa..d4152ec5 100644
--- a/libsoup/http1/soup-client-message-io-http1.c
+++ b/libsoup/http1/soup-client-message-io-http1.c
@@ -36,6 +36,7 @@ typedef struct {
 
         SoupMessageQueueItem *item;
 
+        gint64 response_header_bytes_received;
         SoupMessageMetrics *metrics;
 
         /* Request body logger */
@@ -495,10 +496,17 @@ response_network_stream_read_data_cb (SoupMessage *msg,
 {
         SoupClientMessageIOHTTP1 *client_io = (SoupClientMessageIOHTTP1 *)soup_message_get_io_data (msg);
 
-        if (client_io->msg_io->base.read_state < SOUP_MESSAGE_IO_STATE_BODY_START)
-                client_io->msg_io->metrics->response_header_bytes_received += count;
-        else
+        if (client_io->msg_io->base.read_state < SOUP_MESSAGE_IO_STATE_BODY_START) {
+                client_io->msg_io->response_header_bytes_received += count;
+                if (client_io->msg_io->metrics)
+                        client_io->msg_io->metrics->response_header_bytes_received += count;
+                return;
+        }
+
+        if (client_io->msg_io->metrics)
                 client_io->msg_io->metrics->response_body_bytes_received += count;
+
+        soup_message_got_body_data (msg, count);
 }
 
 /* Attempts to push forward the reading side of @msg's I/O. Returns
@@ -518,6 +526,7 @@ io_read (SoupClientMessageIOHTTP1 *client_io,
         gboolean succeeded;
         gboolean is_first_read;
         gushort extra_bytes;
+        gsize response_body_bytes_received = 0;
 
         switch (io->read_state) {
         case SOUP_MESSAGE_IO_STATE_HEADERS:
@@ -531,16 +540,17 @@ io_read (SoupClientMessageIOHTTP1 *client_io,
                 if (!succeeded)
                         return FALSE;
 
-                if (client_io->msg_io->metrics) {
-                        /* Adjust the header and body bytes received, since we might
-                         * have read part of the body already that is queued by the stream.
-                         */
-                        if (client_io->msg_io->metrics->response_header_bytes_received > io->read_header_buf->len + extra_bytes) {
-                                client_io->msg_io->metrics->response_body_bytes_received =
-                                        client_io->msg_io->metrics->response_header_bytes_received - io->read_header_buf->len - extra_bytes;
-                                client_io->msg_io->metrics->response_header_bytes_received -= client_io->msg_io->metrics->response_body_bytes_received;
+                /* Adjust the header and body bytes received, since we might
+                 * have read part of the body already that is queued by the stream.
+                 */
+                if (client_io->msg_io->response_header_bytes_received > io->read_header_buf->len + extra_bytes) {
+                        response_body_bytes_received = client_io->msg_io->response_header_bytes_received - io->read_header_buf->len - extra_bytes;
+                        if (client_io->msg_io->metrics) {
+                                client_io->msg_io->metrics->response_body_bytes_received = response_body_bytes_received;
+                                client_io->msg_io->metrics->response_header_bytes_received -= response_body_bytes_received;
                         }
                 }
+                client_io->msg_io->response_header_bytes_received = 0;
 
                 succeeded = parse_headers (msg,
                                            (char *)io->read_header_buf->data,
@@ -616,6 +626,9 @@ io_read (SoupClientMessageIOHTTP1 *client_io,
                         io->read_length = -1;
 
                 soup_message_got_headers (msg);
+
+                if (response_body_bytes_received > 0)
+                        soup_message_got_body_data (msg, response_body_bytes_received);
                 break;
 
         case SOUP_MESSAGE_IO_STATE_BODY_START:
@@ -1043,11 +1056,9 @@ soup_client_message_io_http1_send_item (SoupClientMessageIO       *iface,
         msg_io->base.read_state = SOUP_MESSAGE_IO_STATE_NOT_STARTED;
         msg_io->base.write_state = SOUP_MESSAGE_IO_STATE_HEADERS;
         msg_io->metrics = soup_message_get_metrics (msg_io->item->msg);
-        if (msg_io->metrics) {
-                g_signal_connect_object (io->istream, "read-data",
-                                         G_CALLBACK (response_network_stream_read_data_cb),
-                                         msg_io->item->msg, G_CONNECT_SWAPPED);
-        }
+        g_signal_connect_object (io->istream, "read-data",
+                                 G_CALLBACK (response_network_stream_read_data_cb),
+                                 msg_io->item->msg, G_CONNECT_SWAPPED);
 
 #ifdef HAVE_SYSPROF
         msg_io->begin_time_nsec = SYSPROF_CAPTURE_CURRENT_TIME;
diff --git a/libsoup/http2/soup-client-message-io-http2.c b/libsoup/http2/soup-client-message-io-http2.c
index 15278786..450b0a8d 100644
--- a/libsoup/http2/soup-client-message-io-http2.c
+++ b/libsoup/http2/soup-client-message-io-http2.c
@@ -740,6 +740,7 @@ on_frame_recv_callback (nghttp2_session     *session,
         case NGHTTP2_DATA:
                 if (data->metrics)
                         data->metrics->response_body_bytes_received += frame->data.hd.length + FRAME_HEADER_SIZE;
+                soup_message_got_body_data (data->msg, frame->data.hd.length + FRAME_HEADER_SIZE);
                 if (frame->hd.flags & NGHTTP2_FLAG_END_STREAM) {
                         if (data->body_istream) {
                                 soup_body_input_stream_http2_complete (SOUP_BODY_INPUT_STREAM_HTTP2 (data->body_istream));
diff --git a/libsoup/soup-message-private.h b/libsoup/soup-message-private.h
index c036630e..a4adabab 100644
--- a/libsoup/soup-message-private.h
+++ b/libsoup/soup-message-private.h
@@ -80,6 +80,8 @@ void soup_message_wrote_body_data   (SoupMessage *msg,
 void soup_message_wrote_body        (SoupMessage *msg);
 void soup_message_got_informational (SoupMessage *msg);
 void soup_message_got_headers       (SoupMessage *msg);
+void soup_message_got_body_data     (SoupMessage *msg,
+                                     gsize        chunk_size);
 void soup_message_got_body          (SoupMessage *msg);
 void soup_message_content_sniffed   (SoupMessage *msg,
 				     const char  *content_type,
diff --git a/libsoup/soup-message.c b/libsoup/soup-message.c
index ab807935..a7b7349c 100644
--- a/libsoup/soup-message.c
+++ b/libsoup/soup-message.c
@@ -111,6 +111,7 @@ enum {
 
 	GOT_INFORMATIONAL,
 	GOT_HEADERS,
+        GOT_BODY_DATA,
 	GOT_BODY,
 	CONTENT_SNIFFED,
 
@@ -436,6 +437,26 @@ soup_message_class_init (SoupMessageClass *message_class)
 			      NULL,
 			      G_TYPE_NONE, 0);
 
+        /**
+         * SoupMessage::got-body-data:
+         * @msg: the message
+         * @chunk_size: the number of bytes read
+         *
+         * Emitted after reading a portion of the message
+         * body from the network.
+         *
+         * Since: 3.4
+         */
+        signals[GOT_BODY_DATA] =
+                g_signal_new ("got-body-data",
+                              G_OBJECT_CLASS_TYPE (object_class),
+                              G_SIGNAL_RUN_FIRST,
+                              0,
+                              NULL, NULL,
+                              NULL,
+                              G_TYPE_NONE, 1,
+                              G_TYPE_UINT);
+
 	/**
 	 * SoupMessage::got-body:
 	 * @msg: the message
@@ -1213,6 +1234,13 @@ soup_message_got_headers (SoupMessage *msg)
 }
 
 void
+soup_message_got_body_data (SoupMessage *msg,
+                            gsize        chunk_size)
+{
+        g_signal_emit (msg, signals[GOT_BODY_DATA], 0, chunk_size);
+}
+
+void
 soup_message_got_body (SoupMessage *msg)
 {
 	g_signal_emit (msg, signals[GOT_BODY], 0);
diff --git a/tests/http2-test.c b/tests/http2-test.c
index d8a1e5f7..19cee8e1 100644
--- a/tests/http2-test.c
+++ b/tests/http2-test.c
@@ -718,16 +718,28 @@ do_logging_test (Test *test, gconstpointer data)
 }
 
 static void
+msg_got_body_data_cb (SoupMessage *msg,
+                      guint        chunk_size,
+                      guint64     *response_body_bytes_received)
+{
+        *response_body_bytes_received += chunk_size;
+}
+
+static void
 do_metrics_size_test (Test *test, gconstpointer data)
 {
         GUri *uri;
         SoupMessage *msg;
         GBytes *response;
         GError *error = NULL;
+        guint64 response_body_bytes_received = 0;
         GBytes *bytes = g_bytes_new_static ("Test", sizeof ("Test"));
 
         uri = g_uri_parse_relative (base_uri, "/echo_post", SOUP_HTTP_URI_FLAGS, NULL);
         msg = soup_message_new_from_uri (SOUP_METHOD_POST, uri);
+        g_signal_connect (msg, "got-body-data",
+                          G_CALLBACK (msg_got_body_data_cb),
+                          &response_body_bytes_received);
         soup_message_set_request_body_from_bytes (msg, "text/plain", bytes);
         soup_message_add_flags (msg, SOUP_MESSAGE_COLLECT_METRICS);
 
@@ -745,6 +757,7 @@ do_metrics_size_test (Test *test, gconstpointer data)
         g_assert_cmpuint (soup_message_metrics_get_response_header_bytes_received (metrics), >, 0);
         g_assert_cmpuint (soup_message_metrics_get_response_body_size (metrics), ==, g_bytes_get_size (response));
         g_assert_cmpuint (soup_message_metrics_get_response_body_bytes_received (metrics), >, soup_message_metrics_get_response_body_size (metrics));
+        g_assert_cmpuint (soup_message_metrics_get_response_body_bytes_received (metrics), ==, response_body_bytes_received);
 
         g_bytes_unref (response);
         g_bytes_unref (bytes);
diff --git a/tests/streaming-test.c b/tests/streaming-test.c
index 7b776d4c..22153325 100644
--- a/tests/streaming-test.c
+++ b/tests/streaming-test.c
@@ -98,6 +98,14 @@ msg_got_headers_cb (SoupMessage        *msg,
 }
 
 static void
+msg_got_body_data_cb (SoupMessage *msg,
+                      guint        chunk_size,
+                      guint64     *response_body_bytes_received)
+{
+        *response_body_bytes_received += chunk_size;
+}
+
+static void
 msg_got_body_cb (SoupMessage        *msg,
                  SoupMessageMetrics *metrics)
 {
@@ -113,6 +121,7 @@ do_request (SoupSession *session, GUri *base_uri, char *path)
 	GBytes *body;
 	char *md5;
         SoupMessageMetrics *metrics;
+        guint64 response_body_bytes_received = 0;
 
 	uri = g_uri_parse_relative (base_uri, path, SOUP_HTTP_URI_FLAGS, NULL);
 	msg = soup_message_new_from_uri ("GET", uri);
@@ -134,6 +143,9 @@ do_request (SoupSession *session, GUri *base_uri, char *path)
         g_signal_connect (msg, "got-headers",
                           G_CALLBACK (msg_got_headers_cb),
                           metrics);
+        g_signal_connect (msg, "got-body-data",
+                          G_CALLBACK (msg_got_body_data_cb),
+                          &response_body_bytes_received);
         g_signal_connect (msg, "got-body",
                           G_CALLBACK (msg_got_body_cb),
                           metrics);
@@ -143,6 +155,7 @@ do_request (SoupSession *session, GUri *base_uri, char *path)
 	soup_test_assert_message_status (msg, SOUP_STATUS_OK);
 	g_assert_cmpint (g_bytes_get_size (body), ==, g_bytes_get_size (full_response));
         g_assert_cmpint (soup_message_metrics_get_response_body_size (metrics), ==, g_bytes_get_size (body));
+        g_assert_cmpuint (soup_message_metrics_get_response_body_bytes_received (metrics), ==, response_body_bytes_received);
         g_assert_cmpuint (soup_message_metrics_get_request_header_bytes_sent (metrics), >, 0);
         g_assert_cmpuint (soup_message_metrics_get_request_body_bytes_sent (metrics), ==, 0);
         g_assert_cmpuint (soup_message_metrics_get_request_body_size (metrics), ==, 0);
@@ -152,6 +165,12 @@ do_request (SoupSession *session, GUri *base_uri, char *path)
         } else {
                 g_assert_cmpuint (soup_message_metrics_get_response_body_bytes_received (metrics), ==, soup_message_metrics_get_response_body_size (metrics));
         }
+        if (g_str_equal (path, "content-length")) {
+                goffset content_length;
+
+                content_length = soup_message_headers_get_content_length (soup_message_get_response_headers (msg));
+                g_assert_cmpuint (content_length, ==, response_body_bytes_received);
+        }
 
 	md5 = g_compute_checksum_for_data (G_CHECKSUM_MD5,
 					   (guchar *)g_bytes_get_data (body, NULL),