libsoup/soup-message-queue.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-message-queue.c: Message queue
   4  *
   5  * Copyright (C) 2003 Novell, Inc.
   6  * Copyright (C) 2008 Red Hat, Inc.
   7  */
   8
   9 #ifdef HAVE_CONFIG_H
  10 #include <config.h>
  11 #endif
  12
  13 #include "soup-message-queue.h"
  14 #include "soup-uri.h"
  15
  16 /* This is an internal structure used by #SoupSession and its
  17  * subclasses to keep track of the status of messages currently being
  18  * processed.
  19  *
  20  * The #SoupMessageQueue itself is mostly just a linked list of
  21  * #SoupMessageQueueItem, with some added cleverness to allow the list
  22  * to be walked safely while other threads / re-entrant loops are
  23  * adding items to and removing items from it. In particular, this is
  24  * handled by refcounting items and then keeping "removed" items in
  25  * the list until their ref_count drops to 0, but skipping over the
  26  * "removed" ones when walking the queue.
  27  **/
  28
  29 struct _SoupMessageQueue {
  30         SoupSession *session;
  31
  32         GMutex mutex;
  33         SoupMessageQueueItem *head, *tail;
  34 };
  35
  36 SoupMessageQueue *
  37 soup_message_queue_new (SoupSession *session)
  38 {
  39         SoupMessageQueue *queue;
  40
  41         queue = g_slice_new0 (SoupMessageQueue);
  42         queue->session = session;
  43         g_mutex_init (&queue->mutex);
  44         return queue;
  45 }
  46
  47 void
  48 soup_message_queue_destroy (SoupMessageQueue *queue)
  49 {
  50         g_return_if_fail (queue->head == NULL);
  51
  52         g_mutex_clear (&queue->mutex);
  53         g_slice_free (SoupMessageQueue, queue);
  54 }
  55
  56 static void
  57 queue_message_restarted (SoupMessage *msg, gpointer user_data)
  58 {
  59         SoupMessageQueueItem *item = user_data;
  60
  61         if (item->proxy_addr) {
  62                 g_object_unref (item->proxy_addr);
  63                 item->proxy_addr = NULL;
  64         }
  65         if (item->proxy_uri) {
  66                 soup_uri_free (item->proxy_uri);
  67                 item->proxy_uri = NULL;
  68         }
  69
  70         if (item->conn &&
  71             (!soup_message_is_keepalive (msg) ||
  72              SOUP_STATUS_IS_REDIRECTION (msg->status_code))) {
  73                 if (soup_connection_get_state (item->conn) == SOUP_CONNECTION_IN_USE)
  74                         soup_connection_set_state (item->conn, SOUP_CONNECTION_IDLE);
  75                 soup_message_queue_item_set_connection (item, NULL);
  76         }
  77
  78         soup_message_cleanup_response (msg);
  79
  80         g_cancellable_reset (item->cancellable);
  81
  82         item->state = SOUP_MESSAGE_STARTING;
  83 }
  84
  85 /**
  86  * soup_message_queue_append:
  87  * @queue: a #SoupMessageQueue
  88  * @msg: a #SoupMessage
  89  * @callback: the callback for @msg
  90  * @user_data: the data to pass to @callback
  91  *
  92  * Creates a new #SoupMessageQueueItem and appends it to @queue.
  93  *
  94  * Return value: the new item, which you must unref with
  95  * soup_message_queue_unref_item() when you are done with.
  96  **/
  97 SoupMessageQueueItem *
  98 soup_message_queue_append (SoupMessageQueue *queue, SoupMessage *msg,
  99                            SoupSessionCallback callback, gpointer user_data)
 100 {
 101         SoupMessageQueueItem *item;
 102
 103         item = g_slice_new0 (SoupMessageQueueItem);
 104         item->session = queue->session;
 105         item->async_context = soup_session_get_async_context (item->session);
 106         item->queue = queue;
 107         item->msg = g_object_ref (msg);
 108         item->callback = callback;
 109         item->callback_data = user_data;
 110         item->cancellable = g_cancellable_new ();
 111
 112         g_signal_connect (msg, "restarted",
 113                           G_CALLBACK (queue_message_restarted), item);
 114
 115         /* Note: the initial ref_count of 1 represents the caller's
 116          * ref; the queue's own ref is indicated by the absence of the
 117          * "removed" flag.
 118          */
 119         item->ref_count = 1;
 120
 121         g_mutex_lock (&queue->mutex);
 122         if (queue->head) {
 123                 queue->tail->next = item;
 124                 item->prev = queue->tail;
 125                 queue->tail = item;
 126         } else
 127                 queue->head = queue->tail = item;
 128
 129         g_mutex_unlock (&queue->mutex);
 130         return item;
 131 }
 132
 133 /**
 134  * soup_message_queue_item_ref:
 135  * @item: a #SoupMessageQueueItem
 136  *
 137  * Refs @item.
 138  **/
 139 void
 140 soup_message_queue_item_ref (SoupMessageQueueItem *item)
 141 {
 142         item->ref_count++;
 143 }
 144
 145 /**
 146  * soup_message_queue_item_unref:
 147  * @item: a #SoupMessageQueueItem
 148  *
 149  * Unrefs @item; use this on a #SoupMessageQueueItem that you are done
 150  * with (but that you aren't passing to
 151  * soup_message_queue_item_next()).
 152  **/
 153 void
 154 soup_message_queue_item_unref (SoupMessageQueueItem *item)
 155 {
 156         g_mutex_lock (&item->queue->mutex);
 157
 158         /* Decrement the ref_count; if it's still non-zero OR if the
 159          * item is still in the queue, then return.
 160          */
 161         if (--item->ref_count || !item->removed) {
 162                 g_mutex_unlock (&item->queue->mutex);
 163                 return;
 164         }
 165
 166         /* OK, @item is dead. Rewrite @queue around it */
 167         if (item->prev)
 168                 item->prev->next = item->next;
 169         else
 170                 item->queue->head = item->next;
 171         if (item->next)
 172                 item->next->prev = item->prev;
 173         else
 174                 item->queue->tail = item->prev;
 175
 176         g_mutex_unlock (&item->queue->mutex);
 177
 178         /* And free it */
 179         g_signal_handlers_disconnect_by_func (item->msg,
 180                                               queue_message_restarted, item);
 181         g_object_unref (item->msg);
 182         g_object_unref (item->cancellable);
 183         if (item->proxy_addr)
 184                 g_object_unref (item->proxy_addr);
 185         if (item->proxy_uri)
 186                 soup_uri_free (item->proxy_uri);
 187         if (item->result)
 188                 g_object_unref (item->result);
 189         soup_message_queue_item_set_connection (item, NULL);
 190         g_slice_free (SoupMessageQueueItem, item);
 191 }
 192
 193 static void
 194 proxy_connection_event (SoupConnection      *conn,
 195                         GSocketClientEvent   event,
 196                         GIOStream           *connection,
 197                         gpointer             user_data)
 198 {
 199         SoupMessageQueueItem *item = user_data;
 200
 201         soup_message_network_event (item->msg, event, connection);
 202 }
 203
 204 void
 205 soup_message_queue_item_set_connection (SoupMessageQueueItem *item,
 206                                         SoupConnection       *conn)
 207 {
 208         if (item->conn) {
 209                 g_signal_handlers_disconnect_by_func (item->conn, proxy_connection_event, item);
 210                 g_object_unref (item->conn);
 211         }
 212
 213         item->conn = conn;
 214
 215         if (item->conn) {
 216                 g_object_ref (item->conn);
 217                 g_signal_connect (item->conn, "event",
 218                                   G_CALLBACK (proxy_connection_event), item);
 219         }
 220 }
 221
 222 /**
 223  * soup_message_queue_lookup:
 224  * @queue: a #SoupMessageQueue
 225  * @msg: a #SoupMessage
 226  *
 227  * Finds the #SoupMessageQueueItem for @msg in @queue. You must unref
 228  * the item with soup_message_queue_unref_item() when you are done
 229  * with it.
 230  *
 231  * Return value: the queue item for @msg, or %NULL
 232  **/
 233 SoupMessageQueueItem *
 234 soup_message_queue_lookup (SoupMessageQueue *queue, SoupMessage *msg)
 235 {
 236         SoupMessageQueueItem *item;
 237
 238         g_mutex_lock (&queue->mutex);
 239
 240         item = queue->tail;
 241         while (item && (item->removed || item->msg != msg))
 242                 item = item->prev;
 243
 244         if (item)
 245                 item->ref_count++;
 246
 247         g_mutex_unlock (&queue->mutex);
 248         return item;
 249 }
 250
 251 /**
 252  * soup_message_queue_first:
 253  * @queue: a #SoupMessageQueue
 254  *
 255  * Gets the first item in @queue. You must unref the item by calling
 256  * soup_message_queue_unref_item() on it when you are done.
 257  * (soup_message_queue_next() does this for you automatically, so you
 258  * only need to unref the item yourself if you are not going to
 259  * finishing walking the queue.)
 260  *
 261  * Return value: the first item in @queue.
 262  **/
 263 SoupMessageQueueItem *
 264 soup_message_queue_first (SoupMessageQueue *queue)
 265 {
 266         SoupMessageQueueItem *item;
 267
 268         g_mutex_lock (&queue->mutex);
 269
 270         item = queue->head;
 271         while (item && item->removed)
 272                 item = item->next;
 273
 274         if (item)
 275                 item->ref_count++;
 276
 277         g_mutex_unlock (&queue->mutex);
 278         return item;
 279 }
 280
 281 /**
 282  * soup_message_queue_next:
 283  * @queue: a #SoupMessageQueue
 284  * @item: a #SoupMessageQueueItem
 285  *
 286  * Unrefs @item and gets the next item after it in @queue. As with
 287  * soup_message_queue_first(), you must unref the returned item
 288  * yourself with soup_message_queue_unref_item() if you do not finish
 289  * walking the queue.
 290  *
 291  * Return value: the next item in @queue.
 292  **/
 293 SoupMessageQueueItem *
 294 soup_message_queue_next (SoupMessageQueue *queue, SoupMessageQueueItem *item)
 295 {
 296         SoupMessageQueueItem *next;
 297
 298         g_mutex_lock (&queue->mutex);
 299
 300         next = item->next;
 301         while (next && next->removed)
 302                 next = next->next;
 303         if (next)
 304                 next->ref_count++;
 305
 306         g_mutex_unlock (&queue->mutex);
 307         soup_message_queue_item_unref (item);
 308         return next;
 309 }
 310
 311 /**
 312  * soup_message_queue_remove:
 313  * @queue: a #SoupMessageQueue
 314  * @item: a #SoupMessageQueueItem
 315  *
 316  * Removes @item from @queue. Note that you probably also need to call
 317  * soup_message_queue_unref_item() after this.
 318  **/
 319 void
 320 soup_message_queue_remove (SoupMessageQueue *queue, SoupMessageQueueItem *item)
 321 {
 322         g_return_if_fail (!item->removed);
 323
 324         g_mutex_lock (&queue->mutex);
 325         item->removed = TRUE;
 326         g_mutex_unlock (&queue->mutex);
 327 }