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 #define LIBSOUP_I_HAVE_READ_BUG_594377_AND_KNOW_SOUP_PASSWORD_MANAGER_MIGHT_GO_AWAY
  13
  14 #include "soup-address.h"
  15 #include "soup-session-async.h"
  16 #include "soup-session-private.h"
  17 #include "soup-address.h"
  18 #include "soup-message-private.h"
  19 #include "soup-message-queue.h"
  20 #include "soup-misc.h"
  21 #include "soup-password-manager.h"
  22 #include "soup-proxy-uri-resolver.h"
  23 #include "soup-uri.h"
  24
  25 /**
  26  * SECTION:soup-session-async
  27  * @short_description: Soup session for asynchronous (main-loop-based) I/O.
  28  *
  29  * #SoupSessionAsync is an implementation of #SoupSession that uses
  30  * non-blocking I/O via the glib main loop. It is intended for use in
  31  * single-threaded programs.
  32  **/
  33
  34 static void run_queue (SoupSessionAsync *sa);
  35 static void do_idle_run_queue (SoupSession *session);
  36
  37 static void  queue_message   (SoupSession *session, SoupMessage *req,
  38                               SoupSessionCallback callback, gpointer user_data);
  39 static guint send_message    (SoupSession *session, SoupMessage *req);
  40 static void  cancel_message  (SoupSession *session, SoupMessage *msg,
  41                               guint status_code);
  42
  43 static void  auth_required   (SoupSession *session, SoupMessage *msg,
  44                               SoupAuth *auth, gboolean retrying);
  45
  46 G_DEFINE_TYPE (SoupSessionAsync, soup_session_async, SOUP_TYPE_SESSION)
  47
  48 typedef struct {
  49         GSource *idle_run_queue_source;
  50 } SoupSessionAsyncPrivate;
  51 #define SOUP_SESSION_ASYNC_GET_PRIVATE(o) (G_TYPE_INSTANCE_GET_PRIVATE ((o), SOUP_TYPE_SESSION_ASYNC, SoupSessionAsyncPrivate))
  52
  53 static void
  54 soup_session_async_init (SoupSessionAsync *sa)
  55 {
  56 }
  57
  58 static void
  59 finalize (GObject *object)
  60 {
  61         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (object);
  62
  63         if (priv->idle_run_queue_source)
  64                 g_source_destroy (priv->idle_run_queue_source);
  65
  66         G_OBJECT_CLASS (soup_session_async_parent_class)->finalize (object);
  67 }
  68
  69 static void
  70 soup_session_async_class_init (SoupSessionAsyncClass *soup_session_async_class)
  71 {
  72         SoupSessionClass *session_class = SOUP_SESSION_CLASS (soup_session_async_class);
  73         GObjectClass *object_class = G_OBJECT_CLASS (session_class);
  74
  75         g_type_class_add_private (soup_session_async_class,
  76                                   sizeof (SoupSessionAsyncPrivate));
  77
  78         /* virtual method override */
  79         session_class->queue_message = queue_message;
  80         session_class->send_message = send_message;
  81         session_class->cancel_message = cancel_message;
  82         session_class->auth_required = auth_required;
  83
  84         object_class->finalize = finalize;
  85 }
  86
  87
  88 /**
  89  * soup_session_async_new:
  90  *
  91  * Creates an asynchronous #SoupSession with the default options.
  92  *
  93  * Return value: the new session.
  94  **/
  95 SoupSession *
  96 soup_session_async_new (void)
  97 {
  98         return g_object_new (SOUP_TYPE_SESSION_ASYNC, NULL);
  99 }
 100
 101 /**
 102  * soup_session_async_new_with_options:
 103  * @optname1: name of first property to set
 104  * @...: value of @optname1, followed by additional property/value pairs
 105  *
 106  * Creates an asynchronous #SoupSession with the specified options.
 107  *
 108  * Return value: the new session.
 109  **/
 110 SoupSession *
 111 soup_session_async_new_with_options (const char *optname1, ...)
 112 {
 113         SoupSession *session;
 114         va_list ap;
 115
 116         va_start (ap, optname1);
 117         session = (SoupSession *)g_object_new_valist (SOUP_TYPE_SESSION_ASYNC,
 118                                                       optname1, ap);
 119         va_end (ap);
 120
 121         return session;
 122 }
 123
 124 static gboolean
 125 item_failed (SoupMessageQueueItem *item, guint status)
 126 {
 127         if (item->removed) {
 128                 soup_message_queue_item_unref (item);
 129                 return TRUE;
 130         }
 131
 132         if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
 133                 item->state = SOUP_MESSAGE_FINISHING;
 134                 if (!item->msg->status_code)
 135                         soup_session_set_item_status (item->session, item, status);
 136                 do_idle_run_queue (item->session);
 137                 soup_message_queue_item_unref (item);
 138                 return TRUE;
 139         }
 140
 141         return FALSE;
 142 }
 143
 144 static void
 145 resolved_proxy_addr (SoupAddress *addr, guint status, gpointer user_data)
 146 {
 147         SoupMessageQueueItem *item = user_data;
 148         SoupSession *session = item->session;
 149
 150         if (item_failed (item, soup_status_proxify (status)))
 151                 return;
 152
 153         item->proxy_addr = g_object_ref (addr);
 154         item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 155
 156         soup_message_queue_item_unref (item);
 157
 158         /* If we got here we know session still exists */
 159         run_queue ((SoupSessionAsync *)session);
 160 }
 161
 162 static void
 163 resolved_proxy_uri (SoupProxyURIResolver *proxy_resolver,
 164                     guint status, SoupURI *proxy_uri, gpointer user_data)
 165 {
 166         SoupMessageQueueItem *item = user_data;
 167         SoupSession *session = item->session;
 168
 169         if (item_failed (item, status))
 170                 return;
 171
 172         if (proxy_uri) {
 173                 SoupAddress *proxy_addr;
 174
 175                 item->state = SOUP_MESSAGE_RESOLVING_PROXY_ADDRESS;
 176
 177                 item->proxy_uri = soup_uri_copy (proxy_uri);
 178                 proxy_addr = soup_address_new (proxy_uri->host,
 179                                                proxy_uri->port);
 180                 soup_address_resolve_async (proxy_addr,
 181                                             soup_session_get_async_context (session),
 182                                             item->cancellable,
 183                                             resolved_proxy_addr, item);
 184                 g_object_unref (proxy_addr);
 185                 return;
 186         }
 187
 188         item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 189         soup_message_queue_item_unref (item);
 190
 191         /* If we got here we know session still exists */
 192         run_queue ((SoupSessionAsync *)session);
 193 }
 194
 195 static void
 196 resolve_proxy_addr (SoupMessageQueueItem *item,
 197                     SoupProxyURIResolver *proxy_resolver)
 198 {
 199         item->state = SOUP_MESSAGE_RESOLVING_PROXY_URI;
 200
 201         soup_message_queue_item_ref (item);
 202         soup_proxy_uri_resolver_get_proxy_uri_async (
 203                 proxy_resolver, soup_message_get_uri (item->msg),
 204                 soup_session_get_async_context (item->session),
 205                 item->cancellable, resolved_proxy_uri, item);
 206 }
 207
 208 static void
 209 connection_closed (SoupConnection *conn, gpointer session)
 210 {
 211         /* Run the queue in case anyone was waiting for a connection
 212          * to be closed.
 213          */
 214         do_idle_run_queue (session);
 215 }
 216
 217 static void
 218 message_completed (SoupMessage *msg, gpointer user_data)
 219 {
 220         SoupMessageQueueItem *item = user_data;
 221
 222         if (item->state != SOUP_MESSAGE_RESTARTING)
 223                 item->state = SOUP_MESSAGE_FINISHING;
 224         do_idle_run_queue (item->session);
 225 }
 226
 227 static void
 228 tunnel_message_completed (SoupMessage *msg, gpointer user_data)
 229 {
 230         SoupMessageQueueItem *item = user_data;
 231         SoupSession *session = item->session;
 232
 233         if (item->state == SOUP_MESSAGE_RESTARTING) {
 234                 soup_message_restarted (msg);
 235                 if (item->conn) {
 236                         soup_session_send_queue_item (session, item, tunnel_message_completed);
 237                         return;
 238                 }
 239
 240                 soup_message_set_status (msg, SOUP_STATUS_TRY_AGAIN);
 241         }
 242
 243         item->state = SOUP_MESSAGE_FINISHED;
 244
 245         if (!SOUP_STATUS_IS_SUCCESSFUL (msg->status_code)) {
 246                 if (item->conn)
 247                         soup_connection_disconnect (item->conn);
 248                 if (msg->status_code == SOUP_STATUS_TRY_AGAIN) {
 249                         item->related->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 250                         g_object_unref (item->related->conn);
 251                         item->related->conn = NULL;
 252                 } else
 253                         soup_message_set_status (item->related->msg, msg->status_code);
 254                 goto done;
 255         }
 256
 257         if (!soup_connection_start_ssl (item->conn)) {
 258                 if (item->conn)
 259                         soup_connection_disconnect (item->conn);
 260                 soup_message_set_status (item->related->msg, SOUP_STATUS_SSL_FAILED);
 261                 goto done;
 262         }
 263
 264         g_signal_connect (item->conn, "disconnected",
 265                           G_CALLBACK (connection_closed), item->session);
 266         soup_connection_set_state (item->conn, SOUP_CONNECTION_IDLE);
 267         soup_connection_set_state (item->conn, SOUP_CONNECTION_IN_USE);
 268
 269         item->related->state = SOUP_MESSAGE_READY;
 270
 271 done:
 272         soup_message_finished (msg);
 273         if (item->related->msg->status_code)
 274                 item->related->state = SOUP_MESSAGE_FINISHING;
 275
 276         do_idle_run_queue (item->session);
 277         soup_message_queue_item_unref (item->related);
 278         soup_session_unqueue_item (session, item);
 279         soup_message_queue_item_unref (item);
 280         g_object_unref (session);
 281 }
 282
 283 static void
 284 got_connection (SoupConnection *conn, guint status, gpointer user_data)
 285 {
 286         SoupMessageQueueItem *item = user_data;
 287         SoupSession *session = item->session;
 288         SoupAddress *tunnel_addr;
 289
 290         if (item->state != SOUP_MESSAGE_CONNECTING) {
 291                 soup_connection_disconnect (conn);
 292                 do_idle_run_queue (session);
 293                 soup_message_queue_item_unref (item);
 294                 g_object_unref (session);
 295                 return;
 296         }
 297
 298         if (status != SOUP_STATUS_OK) {
 299                 soup_session_set_item_status (session, item, status);
 300                 item->state = SOUP_MESSAGE_FINISHING;
 301
 302                 soup_connection_disconnect (conn);
 303                 do_idle_run_queue (session);
 304                 soup_message_queue_item_unref (item);
 305                 g_object_unref (session);
 306                 return;
 307         }
 308
 309         tunnel_addr = soup_connection_get_tunnel_addr (conn);
 310         if (tunnel_addr) {
 311                 SoupMessageQueueItem *tunnel_item;
 312
 313                 item->state = SOUP_MESSAGE_TUNNELING;
 314
 315                 tunnel_item = soup_session_make_connect_message (session, conn);
 316                 tunnel_item->related = item;
 317                 soup_session_send_queue_item (session, tunnel_item, tunnel_message_completed);
 318                 return;
 319         }
 320
 321         item->state = SOUP_MESSAGE_READY;
 322         g_signal_connect (conn, "disconnected",
 323                           G_CALLBACK (connection_closed), session);
 324         run_queue ((SoupSessionAsync *)session);
 325         soup_message_queue_item_unref (item);
 326         g_object_unref (session);
 327 }
 328
 329 static void
 330 process_queue_item (SoupMessageQueueItem *item,
 331                     gboolean             *should_prune,
 332                     gboolean              loop)
 333 {
 334         SoupSession *session = item->session;
 335         SoupProxyURIResolver *proxy_resolver;
 336
 337         do {
 338                 switch (item->state) {
 339                 case SOUP_MESSAGE_STARTING:
 340                         proxy_resolver = (SoupProxyURIResolver *)soup_session_get_feature_for_message (session, SOUP_TYPE_PROXY_URI_RESOLVER, item->msg);
 341                         if (!proxy_resolver) {
 342                                 item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 343                                 break;
 344                         }
 345                         resolve_proxy_addr (item, proxy_resolver);
 346                         return;
 347
 348                 case SOUP_MESSAGE_AWAITING_CONNECTION:
 349                         if (!soup_session_get_connection (session, item, should_prune))
 350                                 return;
 351
 352                         if (soup_connection_get_state (item->conn) != SOUP_CONNECTION_NEW) {
 353                                 item->state = SOUP_MESSAGE_READY;
 354                                 break;
 355                         }
 356
 357                         item->state = SOUP_MESSAGE_CONNECTING;
 358                         soup_message_queue_item_ref (item);
 359                         g_object_ref (session);
 360                         soup_connection_connect_async (item->conn, item->cancellable,
 361                                                        got_connection, item);
 362                         return;
 363
 364                 case SOUP_MESSAGE_READY:
 365                         item->state = SOUP_MESSAGE_RUNNING;
 366                         soup_session_send_queue_item (session, item, message_completed);
 367                         break;
 368
 369                 case SOUP_MESSAGE_RESTARTING:
 370                         item->state = SOUP_MESSAGE_STARTING;
 371                         soup_message_restarted (item->msg);
 372                         break;
 373
 374                 case SOUP_MESSAGE_FINISHING:
 375                         item->state = SOUP_MESSAGE_FINISHED;
 376                         soup_message_finished (item->msg);
 377                         if (item->state != SOUP_MESSAGE_FINISHED)
 378                                 break;
 379
 380                         g_object_ref (session);
 381                         soup_session_unqueue_item (session, item);
 382                         if (item->callback)
 383                                 item->callback (session, item->msg, item->callback_data);
 384                         g_object_unref (item->msg);
 385                         do_idle_run_queue (session);
 386                         g_object_unref (session);
 387                         return;
 388
 389                 default:
 390                         /* Nothing to do with this message in any
 391                          * other state.
 392                          */
 393                         return;
 394                 }
 395         } while (loop && item->state != SOUP_MESSAGE_FINISHED);
 396 }
 397
 398 static void
 399 run_queue (SoupSessionAsync *sa)
 400 {
 401         SoupSession *session = SOUP_SESSION (sa);
 402         SoupMessageQueue *queue = soup_session_get_queue (session);
 403         SoupMessageQueueItem *item;
 404         SoupMessage *msg;
 405         gboolean try_pruning = TRUE, should_prune = FALSE;
 406
 407         g_object_ref (session);
 408         soup_session_cleanup_connections (session, FALSE);
 409
 410  try_again:
 411         for (item = soup_message_queue_first (queue);
 412              item && !should_prune;
 413              item = soup_message_queue_next (queue, item)) {
 414                 msg = item->msg;
 415
 416                 /* CONNECT messages are handled specially */
 417                 if (msg->method != SOUP_METHOD_CONNECT)
 418                         process_queue_item (item, &should_prune, TRUE);
 419         }
 420         if (item)
 421                 soup_message_queue_item_unref (item);
 422
 423         if (try_pruning && should_prune) {
 424                 /* There is at least one message in the queue that
 425                  * could be sent if we pruned an idle connection from
 426                  * some other server.
 427                  */
 428                 if (soup_session_cleanup_connections (session, TRUE)) {
 429                         try_pruning = should_prune = FALSE;
 430                         goto try_again;
 431                 }
 432         }
 433
 434         g_object_unref (session);
 435 }
 436
 437 static gboolean
 438 idle_run_queue (gpointer sa)
 439 {
 440         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (sa);
 441
 442         priv->idle_run_queue_source = NULL;
 443         run_queue (sa);
 444         return FALSE;
 445 }
 446
 447 static void
 448 do_idle_run_queue (SoupSession *session)
 449 {
 450         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (session);
 451
 452         if (!priv->idle_run_queue_source) {
 453                 priv->idle_run_queue_source = soup_add_completion (
 454                         soup_session_get_async_context (session),
 455                         idle_run_queue, session);
 456         }
 457 }
 458
 459 static void
 460 queue_message (SoupSession *session, SoupMessage *req,
 461                SoupSessionCallback callback, gpointer user_data)
 462 {
 463         SOUP_SESSION_CLASS (soup_session_async_parent_class)->queue_message (session, req, callback, user_data);
 464
 465         do_idle_run_queue (session);
 466 }
 467
 468 static guint
 469 send_message (SoupSession *session, SoupMessage *req)
 470 {
 471         SoupMessageQueueItem *item;
 472         GMainContext *async_context =
 473                 soup_session_get_async_context (session);
 474
 475         /* Balance out the unref that queuing will eventually do */
 476         g_object_ref (req);
 477
 478         queue_message (session, req, NULL, NULL);
 479
 480         item = soup_message_queue_lookup (soup_session_get_queue (session), req);
 481         g_return_val_if_fail (item != NULL, SOUP_STATUS_MALFORMED);
 482
 483         while (item->state != SOUP_MESSAGE_FINISHED)
 484                 g_main_context_iteration (async_context, TRUE);
 485
 486         soup_message_queue_item_unref (item);
 487
 488         return req->status_code;
 489 }
 490
 491 static void
 492 cancel_message (SoupSession *session, SoupMessage *msg,
 493                 guint status_code)
 494 {
 495         SoupMessageQueue *queue;
 496         SoupMessageQueueItem *item;
 497         gboolean dummy;
 498
 499         SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 500                 cancel_message (session, msg, status_code);
 501
 502         queue = soup_session_get_queue (session);
 503         item = soup_message_queue_lookup (queue, msg);
 504         if (!item)
 505                 return;
 506
 507         /* Force it to finish immediately, so that
 508          * soup_session_abort (session); g_object_unref (session);
 509          * will work. (The soup_session_cancel_message() docs
 510          * point out that the callback will be invoked from
 511          * within the cancel call.)
 512          */
 513         if (soup_message_io_in_progress (msg))
 514                 soup_message_io_finished (msg);
 515         else if (item->state != SOUP_MESSAGE_FINISHED)
 516                 item->state = SOUP_MESSAGE_FINISHING;
 517
 518         if (item->state != SOUP_MESSAGE_FINISHED)
 519                 process_queue_item (item, &dummy, FALSE);
 520
 521         soup_message_queue_item_unref (item);
 522 }
 523
 524 static void
 525 got_passwords (SoupPasswordManager *password_manager, SoupMessage *msg,
 526                SoupAuth *auth, gboolean retrying, gpointer session)
 527 {
 528         soup_session_unpause_message (session, msg);
 529         SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 530                 auth_required (session, msg, auth, retrying);
 531         g_object_unref (auth);
 532 }
 533
 534 static void
 535 auth_required (SoupSession *session, SoupMessage *msg,
 536                SoupAuth *auth, gboolean retrying)
 537 {
 538         SoupSessionFeature *password_manager;
 539
 540         password_manager = soup_session_get_feature_for_message (
 541                 session, SOUP_TYPE_PASSWORD_MANAGER, msg);
 542         if (password_manager) {
 543                 soup_session_pause_message (session, msg);
 544                 g_object_ref (auth);
 545                 soup_password_manager_get_passwords_async (
 546                         SOUP_PASSWORD_MANAGER (password_manager),
 547                         msg, auth, retrying,
 548                         soup_session_get_async_context (session),
 549                         NULL, /* FIXME cancellable */
 550                         got_passwords, session);
 551         } else {
 552                 SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 553                         auth_required (session, msg, auth, retrying);
 554         }
 555 }