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