[libsoup] cache: Do not unqueue messages of cached resources until finished

[Carlos Garcia Campos <carlosgc src gnome org>]

commit af2b1e75131cd7043fb243f7e84b5199372103fb
Author: Carlos Garcia Campos <cgarcia igalia com>
Date:   Thu Feb 19 14:59:12 2015 +0100

    cache: Do not unqueue messages of cached resources until finished
    
    We were marking the message is FINISHING right after returning the
    response, so that the message was unqueued while the client is still
    reading the input stream. This is inconsistent with normal messages and
    can confuse clients that connect to request-queued and request-unqueued
    signals to track messages.
    
    https://bugzilla.gnome.org/show_bug.cgi?id=744788

 libsoup/Makefile.am                      |    2 +
 libsoup/soup-cache-client-input-stream.c |   98 ++++++++++++++++++++++++++++++
 libsoup/soup-cache-client-input-stream.h |   36 +++++++++++
 libsoup/soup-cache.c                     |    8 ++-
 libsoup/soup-session.c                   |   17 +++++-
 tests/cache-test.c                       |   16 +++++-
 6 files changed, 172 insertions(+), 5 deletions(-)
---
diff --git a/libsoup/Makefile.am b/libsoup/Makefile.am
index 584f24a..1c71dd2 100644
--- a/libsoup/Makefile.am
+++ b/libsoup/Makefile.am
@@ -113,6 +113,8 @@ libsoup_2_4_la_SOURCES =            \
        soup-body-output-stream.h       \
        soup-body-output-stream.c       \
        soup-cache.c                    \
+       soup-cache-client-input-stream.h\
+       soup-cache-client-input-stream.c\
        soup-cache-input-stream.h       \
        soup-cache-input-stream.c       \
        soup-cache-private.h            \
diff --git a/libsoup/soup-cache-client-input-stream.c b/libsoup/soup-cache-client-input-stream.c
new file mode 100644
index 0000000..cea39a3
--- /dev/null
+++ b/libsoup/soup-cache-client-input-stream.c
@@ -0,0 +1,98 @@
+/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
+/*
+ * soup-cache-client-input-stream.c
+ *
+ * Copyright 2015 Igalia S.L.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include "soup-cache-client-input-stream.h"
+#include "soup.h"
+#include "soup-message-private.h"
+
+enum {
+       EOF,
+       CLOSED,
+       LAST_SIGNAL
+};
+
+static guint signals[LAST_SIGNAL] = { 0 };
+
+G_DEFINE_TYPE (SoupCacheClientInputStream, soup_cache_client_input_stream, G_TYPE_FILTER_INPUT_STREAM)
+
+static void
+soup_cache_client_input_stream_init (SoupCacheClientInputStream *stream)
+{
+}
+
+static gssize
+soup_cache_client_input_stream_read_fn (GInputStream  *stream,
+                                       void          *buffer,
+                                       gsize          count,
+                                       GCancellable  *cancellable,
+                                       GError       **error)
+{
+       gssize nread;
+
+       nread = G_INPUT_STREAM_CLASS (soup_cache_client_input_stream_parent_class)->
+               read_fn (stream, buffer, count, cancellable, error);
+
+       if (nread == 0)
+               g_signal_emit (stream, signals[EOF], 0);
+
+       return nread;
+}
+
+
+static gboolean
+soup_cache_client_input_stream_close_fn (GInputStream  *stream,
+                                        GCancellable  *cancellable,
+                                        GError       **error)
+{
+       gboolean success;
+
+       success = G_INPUT_STREAM_CLASS (soup_cache_client_input_stream_parent_class)->
+               close_fn (stream, cancellable, error);
+
+       g_signal_emit (stream, signals[CLOSED], 0);
+
+       return success;
+}
+
+static void
+soup_cache_client_input_stream_class_init (SoupCacheClientInputStreamClass *stream_class)
+{
+       GObjectClass *object_class = G_OBJECT_CLASS (stream_class);
+       GInputStreamClass *input_stream_class = G_INPUT_STREAM_CLASS (stream_class);
+
+       input_stream_class->read_fn = soup_cache_client_input_stream_read_fn;
+       input_stream_class->close_fn = soup_cache_client_input_stream_close_fn;
+
+       signals[EOF] =
+               g_signal_new ("eof",
+                             G_OBJECT_CLASS_TYPE (object_class),
+                             G_SIGNAL_RUN_LAST,
+                             0,
+                             NULL, NULL,
+                             NULL,
+                             G_TYPE_NONE, 0);
+       signals[CLOSED] =
+               g_signal_new ("closed",
+                             G_OBJECT_CLASS_TYPE (object_class),
+                             G_SIGNAL_RUN_LAST,
+                             0,
+                             NULL, NULL,
+                             NULL,
+                             G_TYPE_NONE, 0);
+}
+
+GInputStream *
+soup_cache_client_input_stream_new (GInputStream *base_stream)
+{
+       return g_object_new (SOUP_TYPE_CACHE_CLIENT_INPUT_STREAM,
+                            "base-stream", base_stream,
+                            NULL);
+}
diff --git a/libsoup/soup-cache-client-input-stream.h b/libsoup/soup-cache-client-input-stream.h
new file mode 100644
index 0000000..17f8359
--- /dev/null
+++ b/libsoup/soup-cache-client-input-stream.h
@@ -0,0 +1,36 @@
+/* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
+/*
+ * Copyright 2015 Igalia S.L.
+ */
+
+#ifndef SOUP_CACHE_CLIENT_INPUT_STREAM_H
+#define SOUP_CACHE_CLIENT_INPUT_STREAM_H 1
+
+#include "soup-types.h"
+
+G_BEGIN_DECLS
+
+#define SOUP_TYPE_CACHE_CLIENT_INPUT_STREAM            (soup_cache_client_input_stream_get_type ())
+#define SOUP_CACHE_CLIENT_INPUT_STREAM(obj)            (G_TYPE_CHECK_INSTANCE_CAST ((obj), 
SOUP_TYPE_CACHE_CLIENT_INPUT_STREAM, SoupCacheClientInputStream))
+#define SOUP_CACHE_CLIENT_INPUT_STREAM_CLASS(klass)    (G_TYPE_CHECK_CLASS_CAST ((klass), 
SOUP_TYPE_CACHE_CLIENT_INPUT_STREAM, SoupCacheClientInputStreamClass))
+#define SOUP_IS_CACHE_CLIENT_INPUT_STREAM(obj)         (G_TYPE_CHECK_INSTANCE_TYPE ((obj), 
SOUP_TYPE_CACHE_CLIENT_INPUT_STREAM))
+#define SOUP_IS_CACHE_CLIENT_INPUT_STREAM_CLASS(klass) (G_TYPE_CHECK_CLASS_TYPE ((obj), 
SOUP_TYPE_CACHE_CLIENT_INPUT_STREAM))
+#define SOUP_CACHE_CLIENT_INPUT_STREAM_GET_CLASS(obj)  (G_TYPE_INSTANCE_GET_CLASS ((obj), 
SOUP_TYPE_CACHE_CLIENT_INPUT_STREAM, SoupCacheClientInputStreamClass))
+
+typedef struct _SoupCacheClientInputStreamPrivate SoupCacheClientInputStreamPrivate;
+
+typedef struct {
+       GFilterInputStream parent;
+} SoupCacheClientInputStream;
+
+typedef struct {
+       GFilterInputStreamClass parent_class;
+} SoupCacheClientInputStreamClass;
+
+GType soup_cache_client_input_stream_get_type (void);
+
+GInputStream *soup_cache_client_input_stream_new (GInputStream *base_stream);
+
+G_END_DECLS
+
+#endif /* SOUP_CACHE_CLIENT_INPUT_STREAM_H */
diff --git a/libsoup/soup-cache.c b/libsoup/soup-cache.c
index 90fce5a..d5f5586 100644
--- a/libsoup/soup-cache.c
+++ b/libsoup/soup-cache.c
@@ -34,6 +34,7 @@
 
 #include "soup-cache.h"
 #include "soup-body-input-stream.h"
+#include "soup-cache-client-input-stream.h"
 #include "soup-cache-input-stream.h"
 #include "soup-cache-private.h"
 #include "soup-content-processor.h"
@@ -671,7 +672,7 @@ GInputStream *
 soup_cache_send_response (SoupCache *cache, SoupMessage *msg)
 {
        SoupCacheEntry *entry;
-       GInputStream *file_stream, *body_stream, *cache_stream;
+       GInputStream *file_stream, *body_stream, *cache_stream, *client_stream;
        GFile *file;
 
        g_return_val_if_fail (SOUP_IS_CACHE (cache), NULL);
@@ -714,7 +715,10 @@ soup_cache_send_response (SoupCache *cache, SoupMessage *msg)
                                                        SOUP_STAGE_ENTITY_BODY);
        g_object_unref (body_stream);
 
-       return cache_stream;
+       client_stream = soup_cache_client_input_stream_new (cache_stream);
+       g_object_unref (cache_stream);
+
+       return client_stream;
 }
 
 static void
diff --git a/libsoup/soup-session.c b/libsoup/soup-session.c
index 1cd67b8..a01aa95 100644
--- a/libsoup/soup-session.c
+++ b/libsoup/soup-session.c
@@ -4086,6 +4086,17 @@ async_send_request_running (SoupSession *session, SoupMessageQueueItem *item)
 }
 
 static void
+cache_stream_finished (GInputStream         *stream,
+                      SoupMessageQueueItem *item)
+{
+       g_signal_handlers_disconnect_matched (stream, G_SIGNAL_MATCH_DATA,
+                                             0, 0, NULL, NULL, item);
+       item->state = SOUP_MESSAGE_FINISHING;
+       soup_session_kick_queue (item->session);
+       soup_message_queue_item_unref (item);
+}
+
+static void
 async_return_from_cache (SoupMessageQueueItem *item,
                         GInputStream         *stream)
 {
@@ -4100,9 +4111,11 @@ async_return_from_cache (SoupMessageQueueItem *item,
                g_hash_table_unref (params);
        }
 
+       soup_message_queue_item_ref (item);
+       g_signal_connect (stream, "eof", G_CALLBACK (cache_stream_finished), item);
+       g_signal_connect (stream, "closed", G_CALLBACK (cache_stream_finished), item);
+
        async_send_request_return_result (item, g_object_ref (stream), NULL);
-       item->state = SOUP_MESSAGE_FINISHING;
-       soup_session_kick_queue (item->session);
 }
 
 typedef struct {
diff --git a/tests/cache-test.c b/tests/cache-test.c
index b8d9a25..ec99a2d 100644
--- a/tests/cache-test.c
+++ b/tests/cache-test.c
@@ -183,6 +183,12 @@ do_request (SoupSession        *session,
 
        g_object_unref (msg);
 
+       if (last_request_validated)
+               last_request_unqueued = FALSE;
+       else
+               soup_test_assert (!last_request_unqueued,
+                                 "Request unqueued before finishing");
+
        last_request_hit_network = is_network_stream (stream);
 
        g_input_stream_read_all (stream, buf, sizeof (buf), &nread,
@@ -221,7 +227,7 @@ do_request_with_cancel (SoupSession          *session,
        GError *error = NULL;
        GCancellable *cancellable;
 
-       last_request_validated = last_request_hit_network = FALSE;
+       last_request_validated = last_request_hit_network = last_request_unqueued = FALSE;
        cancelled_requests = 0;
 
        uri = soup_uri_new_with_base (base_uri, path);
@@ -503,11 +509,15 @@ do_cancel_test (gconstpointer data)
        flags = SOUP_TEST_REQUEST_CANCEL_MESSAGE | SOUP_TEST_REQUEST_CANCEL_IMMEDIATE;
        do_request_with_cancel (session, base_uri, "GET", "/1", flags);
        g_assert_cmpint (cancelled_requests, ==, 1);
+       soup_test_assert (last_request_unqueued,
+                         "Cancelled request /1 not unqueued");
 
        debug_printf (1, "  Cancel fresh resource with g_cancellable_cancel()\n");
        flags = SOUP_TEST_REQUEST_CANCEL_CANCELLABLE | SOUP_TEST_REQUEST_CANCEL_IMMEDIATE;
        do_request_with_cancel (session, base_uri, "GET", "/1", flags);
        g_assert_cmpint (cancelled_requests, ==, 1);
+       soup_test_assert (last_request_unqueued,
+                         "Cancelled request /1 not unqueued");
 
        soup_test_session_abort_unref (session);
 
@@ -523,11 +533,15 @@ do_cancel_test (gconstpointer data)
        flags = SOUP_TEST_REQUEST_CANCEL_MESSAGE | SOUP_TEST_REQUEST_CANCEL_IMMEDIATE;
        do_request_with_cancel (session, base_uri, "GET", "/2", flags);
        g_assert_cmpint (cancelled_requests, ==, 2);
+       soup_test_assert (last_request_unqueued,
+                         "Cancelled request /2 not unqueued");
 
        debug_printf (1, "  Cancel a revalidating resource with g_cancellable_cancel()\n");
        flags = SOUP_TEST_REQUEST_CANCEL_CANCELLABLE | SOUP_TEST_REQUEST_CANCEL_IMMEDIATE;
        do_request_with_cancel (session, base_uri, "GET", "/2", flags);
        g_assert_cmpint (cancelled_requests, ==, 2);
+       soup_test_assert (last_request_unqueued,
+                         "Cancelled request /2 not unqueued");
 
        soup_test_session_abort_unref (session);