libsoup/soup-session-async.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-session-async.c
   4  *
   5  * Copyright (C) 2000-2003, Ximian, Inc.
   6  */
   7
   8 #ifdef HAVE_CONFIG_H
   9 #include <config.h>
  10 #endif
  11
  12 #include "soup-session-async.h"
  13 #include "soup-session-private.h"
  14 #include "soup-address.h"
  15 #include "soup-message-private.h"
  16 #include "soup-misc.h"
  17
  18 /**
  19  * SECTION:soup-session-async
  20  * @short_description: Soup session for asynchronous (main-loop-based) I/O.
  21  *
  22  * #SoupSessionAsync is an implementation of #SoupSession that uses
  23  * non-blocking I/O via the glib main loop. It is intended for use in
  24  * single-threaded programs.
  25  **/
  26
  27 static gboolean run_queue (SoupSessionAsync *sa);
  28 static void do_idle_run_queue (SoupSession *session);
  29
  30 static void  queue_message   (SoupSession *session, SoupMessage *req,
  31                               SoupSessionCallback callback, gpointer user_data);
  32 static guint send_message    (SoupSession *session, SoupMessage *req);
  33
  34 G_DEFINE_TYPE (SoupSessionAsync, soup_session_async, SOUP_TYPE_SESSION)
  35
  36 typedef struct {
  37         GSource *idle_run_queue_source;
  38 } SoupSessionAsyncPrivate;
  39 #define SOUP_SESSION_ASYNC_GET_PRIVATE(o) (G_TYPE_INSTANCE_GET_PRIVATE ((o), SOUP_TYPE_SESSION_ASYNC, SoupSessionAsyncPrivate))
  40
  41 static void
  42 soup_session_async_init (SoupSessionAsync *sa)
  43 {
  44 }
  45
  46 static void
  47 finalize (GObject *object)
  48 {
  49         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (object);
  50
  51         if (priv->idle_run_queue_source)
  52                 g_source_destroy (priv->idle_run_queue_source);
  53
  54         G_OBJECT_CLASS (soup_session_async_parent_class)->finalize (object);
  55 }
  56
  57 static void
  58 soup_session_async_class_init (SoupSessionAsyncClass *soup_session_async_class)
  59 {
  60         SoupSessionClass *session_class = SOUP_SESSION_CLASS (soup_session_async_class);
  61         GObjectClass *object_class = G_OBJECT_CLASS (session_class);
  62
  63         g_type_class_add_private (soup_session_async_class,
  64                                   sizeof (SoupSessionAsyncPrivate));
  65
  66         /* virtual method override */
  67         session_class->queue_message = queue_message;
  68         session_class->send_message = send_message;
  69
  70         object_class->finalize = finalize;
  71 }
  72
  73
  74 /**
  75  * soup_session_async_new:
  76  *
  77  * Creates an asynchronous #SoupSession with the default options.
  78  *
  79  * Return value: the new session.
  80  **/
  81 SoupSession *
  82 soup_session_async_new (void)
  83 {
  84         return g_object_new (SOUP_TYPE_SESSION_ASYNC, NULL);
  85 }
  86
  87 /**
  88  * soup_session_async_new_with_options:
  89  * @optname1: name of first property to set
  90  * @...: value of @optname1, followed by additional property/value pairs
  91  *
  92  * Creates an asynchronous #SoupSession with the specified options.
  93  *
  94  * Return value: the new session.
  95  **/
  96 SoupSession *
  97 soup_session_async_new_with_options (const char *optname1, ...)
  98 {
  99         SoupSession *session;
 100         va_list ap;
 101
 102         va_start (ap, optname1);
 103         session = (SoupSession *)g_object_new_valist (SOUP_TYPE_SESSION_ASYNC,
 104                                                       optname1, ap);
 105         va_end (ap);
 106
 107         return session;
 108 }
 109
 110
 111 static void
 112 resolved_msg_addr (SoupAddress *addr, guint status, gpointer user_data)
 113 {
 114         SoupMessageQueueItem *item = user_data;
 115         SoupSession *session = item->session;
 116
 117         if (item->removed) {
 118                 /* Message was cancelled before its address resolved */
 119                 soup_message_queue_item_unref (item);
 120                 return;
 121         }
 122
 123         if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
 124                 soup_session_cancel_message (session, item->msg, status);
 125                 soup_message_queue_item_unref (item);
 126                 return;
 127         }
 128
 129         item->msg_addr = g_object_ref (addr);
 130         item->resolving_msg_addr = FALSE;
 131
 132         soup_message_queue_item_unref (item);
 133
 134         /* If we got here we know session still exists */
 135         run_queue ((SoupSessionAsync *)session);
 136 }
 137
 138 static void
 139 resolve_msg_addr (SoupMessageQueueItem *item)
 140 {
 141         if (item->resolving_msg_addr)
 142                 return;
 143         item->resolving_msg_addr = TRUE;
 144
 145         soup_message_queue_item_ref (item);
 146         soup_address_resolve_async (soup_message_get_address (item->msg),
 147                                     soup_session_get_async_context (item->session),
 148                                     item->cancellable,
 149                                     resolved_msg_addr, item);
 150 }
 151
 152 static void
 153 connection_closed (SoupConnection *conn, gpointer session)
 154 {
 155         /* Run the queue in case anyone was waiting for a connection
 156          * to be closed.
 157          */
 158         do_idle_run_queue (session);
 159 }
 160
 161 static void
 162 got_connection (SoupConnection *conn, guint status, gpointer user_data)
 163 {
 164         SoupSession *session = user_data;
 165
 166         if (status == SOUP_STATUS_OK) {
 167                 g_signal_connect (conn, "disconnected",
 168                                   G_CALLBACK (connection_closed), session);
 169
 170                 /* @conn has been marked reserved by SoupSession, but
 171                  * we don't actually have any specific message in mind
 172                  * for it. (In particular, the message we were
 173                  * originally planning to queue on it may have already
 174                  * been queued on some other connection that became
 175                  * available while we were waiting for this one to
 176                  * connect.) So we release the connection into the
 177                  * idle pool and then just run the queue and see what
 178                  * happens.
 179                  */
 180                 soup_connection_release (conn);
 181         }
 182
 183         /* Even if the connection failed, we run the queue, since
 184          * there may have been messages waiting for the connection
 185          * count to go down.
 186          */
 187         do_idle_run_queue (session);
 188         g_object_unref (session);
 189 }
 190
 191 static gboolean
 192 run_queue (SoupSessionAsync *sa)
 193 {
 194         SoupSession *session = SOUP_SESSION (sa);
 195         SoupMessageQueue *queue = soup_session_get_queue (session);
 196         SoupMessageQueueItem *item;
 197         SoupMessage *msg;
 198         SoupConnection *conn;
 199         gboolean try_pruning = TRUE, should_prune = FALSE;
 200         gboolean started_any = FALSE, is_new;
 201
 202         /* FIXME: prefer CONNECTING messages */
 203
 204  try_again:
 205         for (item = soup_message_queue_first (queue);
 206              item && !should_prune;
 207              item = soup_message_queue_next (queue, item)) {
 208                 msg = item->msg;
 209
 210                 if (!SOUP_MESSAGE_IS_STARTING (msg) ||
 211                     soup_message_io_in_progress (msg))
 212                         continue;
 213
 214                 if (!item->msg_addr) {
 215                         resolve_msg_addr (item);
 216                         continue;
 217                 }
 218
 219                 conn = soup_session_get_connection (session, msg,
 220                                                     &should_prune, &is_new);
 221                 if (!conn)
 222                         continue;
 223
 224                 if (is_new) {
 225                         soup_connection_connect_async (conn, got_connection,
 226                                                        g_object_ref (session));
 227                 } else
 228                         soup_connection_send_request (conn, msg);
 229         }
 230         if (item)
 231                 soup_message_queue_item_unref (item);
 232
 233         if (try_pruning && should_prune) {
 234                 /* There is at least one message in the queue that
 235                  * could be sent if we pruned an idle connection from
 236                  * some other server.
 237                  */
 238                 if (soup_session_try_prune_connection (session)) {
 239                         try_pruning = should_prune = FALSE;
 240                         goto try_again;
 241                 }
 242         }
 243
 244         return started_any;
 245 }
 246
 247 static void
 248 request_restarted (SoupMessage *req, gpointer user_data)
 249 {
 250         SoupMessageQueueItem *item = user_data;
 251
 252         if (item->msg_addr &&
 253             item->msg_addr != soup_message_get_address (item->msg)) {
 254                 g_object_unref (item->msg_addr);
 255                 item->msg_addr = NULL;
 256         }
 257
 258         run_queue ((SoupSessionAsync *)item->session);
 259 }
 260
 261 static void
 262 final_finished (SoupMessage *req, gpointer user_data)
 263 {
 264         SoupMessageQueueItem *item = user_data;
 265         SoupSession *session = item->session;
 266
 267         g_object_ref (session);
 268
 269         if (!SOUP_MESSAGE_IS_STARTING (req)) {
 270                 g_signal_handlers_disconnect_by_func (req, final_finished, item);
 271                 if (item->callback)
 272                         item->callback (session, req, item->callback_data);
 273
 274                 g_object_unref (req);
 275                 soup_message_queue_item_unref (item);
 276         }
 277
 278         do_idle_run_queue (session);
 279         g_object_unref (session);
 280 }
 281
 282 static gboolean
 283 idle_run_queue (gpointer sa)
 284 {
 285         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (sa);
 286
 287         priv->idle_run_queue_source = NULL;
 288         run_queue (sa);
 289         return FALSE;
 290 }
 291
 292 static void
 293 do_idle_run_queue (SoupSession *session)
 294 {
 295         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (session);
 296
 297         if (!priv->idle_run_queue_source) {
 298                 priv->idle_run_queue_source = soup_add_completion (
 299                         soup_session_get_async_context (session),
 300                         idle_run_queue, session);
 301         }
 302 }
 303
 304 static void
 305 queue_message (SoupSession *session, SoupMessage *req,
 306                SoupSessionCallback callback, gpointer user_data)
 307 {
 308         SoupMessageQueueItem *item;
 309
 310         SOUP_SESSION_CLASS (soup_session_async_parent_class)->queue_message (session, req, callback, user_data);
 311
 312         item = soup_message_queue_lookup (soup_session_get_queue (session), req);
 313         g_return_if_fail (item != NULL);
 314
 315         g_signal_connect (req, "restarted",
 316                           G_CALLBACK (request_restarted), item);
 317         g_signal_connect_after (req, "finished",
 318                                 G_CALLBACK (final_finished), item);
 319
 320         do_idle_run_queue (session);
 321 }
 322
 323 static guint
 324 send_message (SoupSession *session, SoupMessage *req)
 325 {
 326         GMainContext *async_context =
 327                 soup_session_get_async_context (session);
 328
 329         /* Balance out the unref that final_finished will do */
 330         g_object_ref (req);
 331
 332         queue_message (session, req, NULL, NULL);
 333
 334         while (soup_message_get_io_status (req) != SOUP_MESSAGE_IO_STATUS_FINISHED &&
 335                !SOUP_STATUS_IS_TRANSPORT_ERROR (req->status_code))
 336                 g_main_context_iteration (async_context, TRUE);
 337
 338         return req->status_code;
 339 }