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.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         g_cancellable_reset (item->cancellable);
  71 }
  72
  73 /**
  74  * soup_message_queue_append:
  75  * @queue: a #SoupMessageQueue
  76  * @msg: a #SoupMessage
  77  * @callback: the callback for @msg
  78  * @user_data: the data to pass to @callback
  79  *
  80  * Creates a new #SoupMessageQueueItem and appends it to @queue.
  81  *
  82  * Return value: the new item, which you must unref with
  83  * soup_message_queue_unref_item() when you are done with.
  84  **/
  85 SoupMessageQueueItem *
  86 soup_message_queue_append (SoupMessageQueue *queue, SoupMessage *msg,
  87                            SoupSessionCallback callback, gpointer user_data)
  88 {
  89         SoupMessageQueueItem *item;
  90
  91         item = g_slice_new0 (SoupMessageQueueItem);
  92         item->session = g_object_ref (queue->session);
  93         item->async_context = soup_session_get_async_context (item->session);
  94         item->queue = queue;
  95         item->msg = g_object_ref (msg);
  96         item->callback = callback;
  97         item->callback_data = user_data;
  98         item->cancellable = g_cancellable_new ();
  99
 100         g_signal_connect (msg, "restarted",
 101                           G_CALLBACK (queue_message_restarted), item);
 102
 103         /* Note: the initial ref_count of 1 represents the caller's
 104          * ref; the queue's own ref is indicated by the absence of the
 105          * "removed" flag.
 106          */
 107         item->ref_count = 1;
 108
 109         g_mutex_lock (&queue->mutex);
 110         if (queue->head) {
 111                 queue->tail->next = item;
 112                 item->prev = queue->tail;
 113                 queue->tail = item;
 114         } else
 115                 queue->head = queue->tail = item;
 116
 117         g_mutex_unlock (&queue->mutex);
 118         return item;
 119 }
 120
 121 /**
 122  * soup_message_queue_item_ref:
 123  * @item: a #SoupMessageQueueItem
 124  *
 125  * Refs @item.
 126  **/
 127 void
 128 soup_message_queue_item_ref (SoupMessageQueueItem *item)
 129 {
 130         item->ref_count++;
 131 }
 132
 133 /**
 134  * soup_message_queue_item_unref:
 135  * @item: a #SoupMessageQueueItem
 136  *
 137  * Unrefs @item; use this on a #SoupMessageQueueItem that you are done
 138  * with (but that you aren't passing to
 139  * soup_message_queue_item_next()).
 140  **/
 141 void
 142 soup_message_queue_item_unref (SoupMessageQueueItem *item)
 143 {
 144         g_mutex_lock (&item->queue->mutex);
 145
 146         /* Decrement the ref_count; if it's still non-zero OR if the
 147          * item is still in the queue, then return.
 148          */
 149         if (--item->ref_count || !item->removed) {
 150                 g_mutex_unlock (&item->queue->mutex);
 151                 return;
 152         }
 153
 154         g_warn_if_fail (item->conn == NULL);
 155
 156         /* OK, @item is dead. Rewrite @queue around it */
 157         if (item->prev)
 158                 item->prev->next = item->next;
 159         else
 160                 item->queue->head = item->next;
 161         if (item->next)
 162                 item->next->prev = item->prev;
 163         else
 164                 item->queue->tail = item->prev;
 165
 166         g_mutex_unlock (&item->queue->mutex);
 167
 168         /* And free it */
 169         g_signal_handlers_disconnect_by_func (item->msg,
 170                                               queue_message_restarted, item);
 171         g_object_unref (item->session);
 172         g_object_unref (item->msg);
 173         g_object_unref (item->cancellable);
 174         g_clear_object (&item->proxy_addr);
 175         g_clear_pointer (&item->proxy_uri, soup_uri_free);
 176         g_clear_object (&item->task);
 177         if (item->io_source) {
 178                 g_source_destroy (item->io_source);
 179                 g_source_unref (item->io_source);
 180         }
 181         g_slice_free (SoupMessageQueueItem, item);
 182 }
 183
 184 /**
 185  * soup_message_queue_lookup:
 186  * @queue: a #SoupMessageQueue
 187  * @msg: a #SoupMessage
 188  *
 189  * Finds the #SoupMessageQueueItem for @msg in @queue. You must unref
 190  * the item with soup_message_queue_unref_item() when you are done
 191  * with it.
 192  *
 193  * Return value: the queue item for @msg, or %NULL
 194  **/
 195 SoupMessageQueueItem *
 196 soup_message_queue_lookup (SoupMessageQueue *queue, SoupMessage *msg)
 197 {
 198         SoupMessageQueueItem *item;
 199
 200         g_mutex_lock (&queue->mutex);
 201
 202         item = queue->tail;
 203         while (item && (item->removed || item->msg != msg))
 204                 item = item->prev;
 205
 206         if (item)
 207                 item->ref_count++;
 208
 209         g_mutex_unlock (&queue->mutex);
 210         return item;
 211 }
 212
 213 /**
 214  * soup_message_queue_first:
 215  * @queue: a #SoupMessageQueue
 216  *
 217  * Gets the first item in @queue. You must unref the item by calling
 218  * soup_message_queue_unref_item() on it when you are done.
 219  * (soup_message_queue_next() does this for you automatically, so you
 220  * only need to unref the item yourself if you are not going to
 221  * finishing walking the queue.)
 222  *
 223  * Return value: the first item in @queue.
 224  **/
 225 SoupMessageQueueItem *
 226 soup_message_queue_first (SoupMessageQueue *queue)
 227 {
 228         SoupMessageQueueItem *item;
 229
 230         g_mutex_lock (&queue->mutex);
 231
 232         item = queue->head;
 233         while (item && item->removed)
 234                 item = item->next;
 235
 236         if (item)
 237                 item->ref_count++;
 238
 239         g_mutex_unlock (&queue->mutex);
 240         return item;
 241 }
 242
 243 /**
 244  * soup_message_queue_next:
 245  * @queue: a #SoupMessageQueue
 246  * @item: a #SoupMessageQueueItem
 247  *
 248  * Unrefs @item and gets the next item after it in @queue. As with
 249  * soup_message_queue_first(), you must unref the returned item
 250  * yourself with soup_message_queue_unref_item() if you do not finish
 251  * walking the queue.
 252  *
 253  * Return value: the next item in @queue.
 254  **/
 255 SoupMessageQueueItem *
 256 soup_message_queue_next (SoupMessageQueue *queue, SoupMessageQueueItem *item)
 257 {
 258         SoupMessageQueueItem *next;
 259
 260         g_mutex_lock (&queue->mutex);
 261
 262         next = item->next;
 263         while (next && next->removed)
 264                 next = next->next;
 265         if (next)
 266                 next->ref_count++;
 267
 268         g_mutex_unlock (&queue->mutex);
 269         soup_message_queue_item_unref (item);
 270         return next;
 271 }
 272
 273 /**
 274  * soup_message_queue_remove:
 275  * @queue: a #SoupMessageQueue
 276  * @item: a #SoupMessageQueueItem
 277  *
 278  * Removes @item from @queue. Note that you probably also need to call
 279  * soup_message_queue_unref_item() after this.
 280  **/
 281 void
 282 soup_message_queue_remove (SoupMessageQueue *queue, SoupMessageQueueItem *item)
 283 {
 284         g_return_if_fail (!item->removed);
 285
 286         g_mutex_lock (&queue->mutex);
 287         item->removed = TRUE;
 288         g_mutex_unlock (&queue->mutex);
 289 }