[libsoup/http2] queue move wip

[Dan Winship <danw src gnome org>]

commit b27940638c6ce2ec22bd8975a6ffc49834a68f09
Author: Dan Winship <danw gnome org>
Date:   Tue Dec 10 17:03:11 2013 +0100

    queue move wip

 libsoup/soup-session-host.c |   74 +++++++++++++++++++++++++++++++++++++++---
 libsoup/soup-session-host.h |    6 ++-
 libsoup/soup-session.c      |   33 ++++++++++---------
 3 files changed, 89 insertions(+), 24 deletions(-)
---
diff --git a/libsoup/soup-session-host.c b/libsoup/soup-session-host.c
index 5645ca4..d362a5b 100644
--- a/libsoup/soup-session-host.c
+++ b/libsoup/soup-session-host.c
@@ -12,8 +12,8 @@
 #include "soup-session-host.h"
 #include "soup.h"
 #include "soup-connection.h"
+#include "soup-message-queue.h"
 #include "soup-misc-private.h"
-#include "soup-session-private.h"
 #include "soup-socket-properties.h"
 
 G_DEFINE_TYPE (SoupSessionHost, soup_session_host, G_TYPE_OBJECT)
@@ -24,6 +24,8 @@ typedef struct {
        SoupURI     *uri;
        SoupAddress *addr;
 
+       SoupMessageQueue *queue;
+
        GSList      *connections;      /* CONTAINS: SoupConnection */
        guint        num_conns;
        guint        max_conns;
@@ -52,6 +54,8 @@ soup_session_host_init (SoupSessionHost *host)
        SoupSessionHostPrivate *priv = SOUP_SESSION_HOST_GET_PRIVATE (host);
 
        g_mutex_init (&priv->mutex);
+
+       priv->queue = soup_message_queue_new ();
 }
 
 static void
@@ -69,6 +73,8 @@ soup_session_host_finalize (GObject *object)
        soup_uri_free (priv->uri);
        g_object_unref (priv->addr);
 
+       soup_message_queue_destroy (priv->queue);
+
        g_mutex_clear (&priv->mutex);
 
        G_OBJECT_CLASS (soup_session_host_parent_class)->finalize (object);
@@ -132,18 +138,74 @@ soup_session_host_get_address (SoupSessionHost *host)
        return SOUP_SESSION_HOST_GET_PRIVATE (host)->addr;
 }
 
-void
+static void
+async_run_queue (SoupSessionHost *host)
+{
+       SoupSessionHostPrivate *priv = SOUP_SESSION_HOST_GET_PRIVATE (host);
+       SoupMessageQueueItem *item;
+       SoupMessage *msg;
+       gboolean try_cleanup = TRUE, should_cleanup = FALSE;
+
+       g_object_ref (host);
+       soup_session_host_cleanup_connections (host, FALSE);
+
+ try_again:
+       for (item = soup_message_queue_first (priv->queue);
+            item;
+            item = soup_message_queue_next (priv->queue, item)) {
+               msg = item->msg;
+
+               /* CONNECT messages are handled specially */
+               if (msg->method == SOUP_METHOD_CONNECT)
+                       continue;
+
+               if (item->async_context != soup_session_get_async_context (session))
+                       continue;
+
+               soup_session_process_queue_item (session, item, &should_cleanup, TRUE);
+       }
+
+       if (try_cleanup && should_cleanup) {
+               /* There is at least one message in the queue that
+                * could be sent if we cleanupd an idle connection from
+                * some other server.
+                */
+               if (soup_session_host_cleanup_connections (host, TRUE)) {
+                       try_cleanup = should_cleanup = FALSE;
+                       goto try_again;
+               }
+       }
+
+       g_object_unref (host);
+}
+
+SoupMessageQueueItem *
 soup_session_host_add_message (SoupSessionHost *host,
                               SoupMessage     *msg)
 {
-       SOUP_SESSION_HOST_GET_PRIVATE (host)->num_messages++;
+       SoupSessionHostPrivate *priv = SOUP_SESSION_HOST_GET_PRIVATE (host);
+
+       priv->num_messages++;
+       return soup_message_queue_append (priv->queue, msg);
+}
+
+SoupMessageQueueItem *
+soup_session_host_lookup_queue_item (SoupSessionHost *host,
+                                    SoupMessage     *msg)
+{
+       SoupSessionHostPrivate *priv = SOUP_SESSION_HOST_GET_PRIVATE (host);
+
+       return soup_message_queue_lookup (priv->queue, msg);
 }
 
 void
-soup_session_host_remove_message (SoupSessionHost *host,
-                                 SoupMessage     *msg)
+soup_session_host_remove_item (SoupSessionHost      *host,
+                              SoupMessageQueueItem *item)
 {
-       SOUP_SESSION_HOST_GET_PRIVATE (host)->num_messages--;
+       SoupSessionHostPrivate *priv = SOUP_SESSION_HOST_GET_PRIVATE (host);
+
+       soup_message_queue_remove (priv->queue, item);
+       priv->num_messages--;
 }
 
 static gboolean
diff --git a/libsoup/soup-session-host.h b/libsoup/soup-session-host.h
index 4e5694d..b95f7d0 100644
--- a/libsoup/soup-session-host.h
+++ b/libsoup/soup-session-host.h
@@ -35,8 +35,10 @@ SoupSessionHost *soup_session_host_new                 (SoupSession          *se
 SoupURI         *soup_session_host_get_uri             (SoupSessionHost      *host);
 SoupAddress     *soup_session_host_get_address         (SoupSessionHost      *host);
 
-void             soup_session_host_add_message         (SoupSessionHost      *host,
-                                                       SoupMessage          *msg);
+SoupMessageQueueItem *soup_session_host_add_message         (SoupSessionHost      *host,
+                                                            SoupMessage          *msg);
+SoupMessageQueueItem *soup_session_host_lookup_queue_item   (SoupSessionHost      *host,
+                                                            SoupMessage          *msg);
 void             soup_session_host_remove_message      (SoupSessionHost      *host,
                                                        SoupMessage          *msg);
 
diff --git a/libsoup/soup-session.c b/libsoup/soup-session.c
index 77f63da..bdcee7d 100644
--- a/libsoup/soup-session.c
+++ b/libsoup/soup-session.c
@@ -77,8 +77,6 @@ typedef struct {
        gboolean ssl_strict;
        gboolean tlsdb_use_default;
 
-       SoupMessageQueue *queue;
-
        char *user_agent;
        char *accept_language;
        gboolean accept_language_auto;
@@ -194,8 +192,6 @@ soup_session_init (SoupSession *session)
 
        priv->session = session;
 
-       priv->queue = soup_message_queue_new ();
-
        g_mutex_init (&priv->conn_lock);
        g_cond_init (&priv->conn_cond);
        priv->http_hosts = g_hash_table_new_full (soup_host_uri_hash,
@@ -305,8 +301,6 @@ soup_session_finalize (GObject *object)
        SoupSession *session = SOUP_SESSION (object);
        SoupSessionPrivate *priv = SOUP_SESSION_GET_PRIVATE (session);
 
-       soup_message_queue_destroy (priv->queue);
-
        g_mutex_clear (&priv->conn_lock);
        g_cond_clear (&priv->conn_cond);
        g_hash_table_destroy (priv->http_hosts);
@@ -1177,7 +1171,11 @@ soup_session_append_queue_item (SoupSession *session, SoupMessage *msg,
 
        soup_message_cleanup_response (msg);
 
-       item = soup_message_queue_append (priv->queue, msg);
+       g_mutex_lock (&priv->conn_lock);
+       host = get_host_for_message (session, item->msg);
+       item = soup_session_host_add_message (host, msg);
+       g_mutex_unlock (&priv->conn_lock);
+
        item->session = g_object_ref (session);
        item->async_context = soup_session_get_async_context (session);
        if (item->async_context)
@@ -1187,11 +1185,6 @@ soup_session_append_queue_item (SoupSession *session, SoupMessage *msg,
        item->callback = callback;
        item->callback_data = user_data;
 
-       g_mutex_lock (&priv->conn_lock);
-       host = get_host_for_message (session, item->msg);
-       soup_session_host_add_message (host, msg);
-       g_mutex_unlock (&priv->conn_lock);
-
        if (!(soup_message_get_flags (msg) & SOUP_MESSAGE_NO_REDIRECT)) {
                soup_message_add_header_handler (
                        msg, "got_body", "Location",
@@ -1359,8 +1352,18 @@ soup_session_lookup_queue_item (SoupSession *session,
                                SoupMessage *msg)
 {
        SoupSessionPrivate *priv = SOUP_SESSION_GET_PRIVATE (session);
+       GSList *hosts, *h;
+       SoupMessageQueueItem *item = NULL;
 
-       return soup_message_queue_lookup (priv->queue, msg);
+       g_mutex_lock (&priv->conn_lock);
+       hosts = get_all_hosts (session);
+       g_mutex_unlock (&priv->conn_lock);
+
+       for (h = hosts; h && !item; h = h->next)
+               item = soup_session_host_lookup_queue_item (h->data, msg);
+
+       g_slist_free_full (hosts, g_object_unref);
+       return item;
 }
 
 static void
@@ -1382,11 +1385,9 @@ soup_session_unqueue_item (SoupSession          *session,
                return;
        }
 
-       soup_message_queue_remove (priv->queue, item);
-
        g_mutex_lock (&priv->conn_lock);
        host = get_host_for_message (session, item->msg);
-       soup_session_host_remove_message (host, item->msg);
+       soup_session_host_remove_item (host, item);
        g_cond_broadcast (&priv->conn_cond);
        g_mutex_unlock (&priv->conn_lock);