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_complete (SoupMessageQueueItem *item)
 229 {
 230         SoupSession *session = item->session;
 231
 232         soup_message_finished (item->msg);
 233         if (item->related->msg->status_code)
 234                 item->related->state = SOUP_MESSAGE_FINISHING;
 235
 236         do_idle_run_queue (session);
 237         soup_message_queue_item_unref (item->related);
 238         soup_session_unqueue_item (session, item);
 239         soup_message_queue_item_unref (item);
 240         g_object_unref (session);
 241 }
 242
 243 static void
 244 ssl_tunnel_completed (SoupConnection *conn, guint status, gpointer user_data)
 245 {
 246         SoupMessageQueueItem *item = user_data;
 247
 248         if (SOUP_STATUS_IS_SUCCESSFUL (status)) {
 249                 g_signal_connect (item->conn, "disconnected",
 250                                   G_CALLBACK (connection_closed), item->session);
 251                 soup_connection_set_state (item->conn, SOUP_CONNECTION_IDLE);
 252                 soup_connection_set_state (item->conn, SOUP_CONNECTION_IN_USE);
 253
 254                 item->related->state = SOUP_MESSAGE_READY;
 255         } else {
 256                 if (item->conn)
 257                         soup_connection_disconnect (item->conn);
 258                 soup_message_set_status (item->related->msg, SOUP_STATUS_SSL_FAILED);
 259         }
 260
 261         tunnel_complete (item);
 262 }
 263
 264 static void
 265 tunnel_message_completed (SoupMessage *msg, gpointer user_data)
 266 {
 267         SoupMessageQueueItem *item = user_data;
 268         SoupSession *session = item->session;
 269
 270         if (item->state == SOUP_MESSAGE_RESTARTING) {
 271                 soup_message_restarted (msg);
 272                 if (item->conn) {
 273                         soup_session_send_queue_item (session, item, tunnel_message_completed);
 274                         return;
 275                 }
 276
 277                 soup_message_set_status (msg, SOUP_STATUS_TRY_AGAIN);
 278         }
 279
 280         item->state = SOUP_MESSAGE_FINISHED;
 281
 282         if (!SOUP_STATUS_IS_SUCCESSFUL (msg->status_code)) {
 283                 if (item->conn)
 284                         soup_connection_disconnect (item->conn);
 285                 if (msg->status_code == SOUP_STATUS_TRY_AGAIN) {
 286                         item->related->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 287                         g_object_unref (item->related->conn);
 288                         item->related->conn = NULL;
 289                 } else
 290                         soup_message_set_status (item->related->msg, msg->status_code);
 291
 292                 tunnel_complete (item);
 293                 return;
 294         }
 295
 296         soup_connection_start_ssl_async (item->conn, item->cancellable,
 297                                          ssl_tunnel_completed, item);
 298 }
 299
 300 static void
 301 got_connection (SoupConnection *conn, guint status, gpointer user_data)
 302 {
 303         SoupMessageQueueItem *item = user_data;
 304         SoupSession *session = item->session;
 305         SoupAddress *tunnel_addr;
 306
 307         if (item->state != SOUP_MESSAGE_CONNECTING) {
 308                 soup_connection_disconnect (conn);
 309                 do_idle_run_queue (session);
 310                 soup_message_queue_item_unref (item);
 311                 g_object_unref (session);
 312                 return;
 313         }
 314
 315         if (status != SOUP_STATUS_OK) {
 316                 soup_connection_disconnect (conn);
 317
 318                 if (status == SOUP_STATUS_TRY_AGAIN) {
 319                         g_object_unref (item->conn);
 320                         item->conn = NULL;
 321                         item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 322                 } else {
 323                         soup_session_set_item_status (session, item, status);
 324                         item->state = SOUP_MESSAGE_FINISHING;
 325                 }
 326
 327                 do_idle_run_queue (session);
 328                 soup_message_queue_item_unref (item);
 329                 g_object_unref (session);
 330                 return;
 331         }
 332
 333         tunnel_addr = soup_connection_get_tunnel_addr (conn);
 334         if (tunnel_addr) {
 335                 SoupMessageQueueItem *tunnel_item;
 336
 337                 item->state = SOUP_MESSAGE_TUNNELING;
 338
 339                 tunnel_item = soup_session_make_connect_message (session, conn);
 340                 tunnel_item->related = item;
 341                 soup_session_send_queue_item (session, tunnel_item, tunnel_message_completed);
 342                 return;
 343         }
 344
 345         item->state = SOUP_MESSAGE_READY;
 346         g_signal_connect (conn, "disconnected",
 347                           G_CALLBACK (connection_closed), session);
 348         run_queue ((SoupSessionAsync *)session);
 349         soup_message_queue_item_unref (item);
 350         g_object_unref (session);
 351 }
 352
 353 static void
 354 process_queue_item (SoupMessageQueueItem *item,
 355                     gboolean             *should_prune,
 356                     gboolean              loop)
 357 {
 358         SoupSession *session = item->session;
 359         SoupProxyURIResolver *proxy_resolver;
 360
 361         do {
 362                 switch (item->state) {
 363                 case SOUP_MESSAGE_STARTING:
 364                         proxy_resolver = (SoupProxyURIResolver *)soup_session_get_feature_for_message (session, SOUP_TYPE_PROXY_URI_RESOLVER, item->msg);
 365                         if (!proxy_resolver) {
 366                                 item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 367                                 break;
 368                         }
 369                         resolve_proxy_addr (item, proxy_resolver);
 370                         return;
 371
 372                 case SOUP_MESSAGE_AWAITING_CONNECTION:
 373                         if (!soup_session_get_connection (session, item, should_prune))
 374                                 return;
 375
 376                         if (soup_connection_get_state (item->conn) != SOUP_CONNECTION_NEW) {
 377                                 item->state = SOUP_MESSAGE_READY;
 378                                 break;
 379                         }
 380
 381                         item->state = SOUP_MESSAGE_CONNECTING;
 382                         soup_message_queue_item_ref (item);
 383                         g_object_ref (session);
 384                         soup_connection_connect_async (item->conn, item->cancellable,
 385                                                        got_connection, item);
 386                         return;
 387
 388                 case SOUP_MESSAGE_READY:
 389                         item->state = SOUP_MESSAGE_RUNNING;
 390                         soup_session_send_queue_item (session, item, message_completed);
 391                         break;
 392
 393                 case SOUP_MESSAGE_RESTARTING:
 394                         item->state = SOUP_MESSAGE_STARTING;
 395                         soup_message_restarted (item->msg);
 396                         break;
 397
 398                 case SOUP_MESSAGE_FINISHING:
 399                         item->state = SOUP_MESSAGE_FINISHED;
 400                         soup_message_finished (item->msg);
 401                         if (item->state != SOUP_MESSAGE_FINISHED)
 402                                 break;
 403
 404                         g_object_ref (session);
 405                         soup_session_unqueue_item (session, item);
 406                         if (item->callback)
 407                                 item->callback (session, item->msg, item->callback_data);
 408                         g_object_unref (item->msg);
 409                         do_idle_run_queue (session);
 410                         g_object_unref (session);
 411                         return;
 412
 413                 default:
 414                         /* Nothing to do with this message in any
 415                          * other state.
 416                          */
 417                         return;
 418                 }
 419         } while (loop && item->state != SOUP_MESSAGE_FINISHED);
 420 }
 421
 422 static void
 423 run_queue (SoupSessionAsync *sa)
 424 {
 425         SoupSession *session = SOUP_SESSION (sa);
 426         SoupMessageQueue *queue = soup_session_get_queue (session);
 427         SoupMessageQueueItem *item;
 428         SoupMessage *msg;
 429         gboolean try_pruning = TRUE, should_prune = FALSE;
 430
 431         g_object_ref (session);
 432         soup_session_cleanup_connections (session, FALSE);
 433
 434  try_again:
 435         for (item = soup_message_queue_first (queue);
 436              item;
 437              item = soup_message_queue_next (queue, item)) {
 438                 msg = item->msg;
 439
 440                 /* CONNECT messages are handled specially */
 441                 if (msg->method != SOUP_METHOD_CONNECT)
 442                         process_queue_item (item, &should_prune, TRUE);
 443         }
 444
 445         if (try_pruning && should_prune) {
 446                 /* There is at least one message in the queue that
 447                  * could be sent if we pruned an idle connection from
 448                  * some other server.
 449                  */
 450                 if (soup_session_cleanup_connections (session, TRUE)) {
 451                         try_pruning = should_prune = FALSE;
 452                         goto try_again;
 453                 }
 454         }
 455
 456         g_object_unref (session);
 457 }
 458
 459 static gboolean
 460 idle_run_queue (gpointer sa)
 461 {
 462         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (sa);
 463
 464         priv->idle_run_queue_source = NULL;
 465         run_queue (sa);
 466         return FALSE;
 467 }
 468
 469 static void
 470 do_idle_run_queue (SoupSession *session)
 471 {
 472         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (session);
 473
 474         if (!priv->idle_run_queue_source) {
 475                 priv->idle_run_queue_source = soup_add_completion (
 476                         soup_session_get_async_context (session),
 477                         idle_run_queue, session);
 478         }
 479 }
 480
 481 static void
 482 queue_message (SoupSession *session, SoupMessage *req,
 483                SoupSessionCallback callback, gpointer user_data)
 484 {
 485         SOUP_SESSION_CLASS (soup_session_async_parent_class)->queue_message (session, req, callback, user_data);
 486
 487         do_idle_run_queue (session);
 488 }
 489
 490 static guint
 491 send_message (SoupSession *session, SoupMessage *req)
 492 {
 493         SoupMessageQueueItem *item;
 494         GMainContext *async_context =
 495                 soup_session_get_async_context (session);
 496
 497         /* Balance out the unref that queuing will eventually do */
 498         g_object_ref (req);
 499
 500         queue_message (session, req, NULL, NULL);
 501
 502         item = soup_message_queue_lookup (soup_session_get_queue (session), req);
 503         g_return_val_if_fail (item != NULL, SOUP_STATUS_MALFORMED);
 504
 505         while (item->state != SOUP_MESSAGE_FINISHED)
 506                 g_main_context_iteration (async_context, TRUE);
 507
 508         soup_message_queue_item_unref (item);
 509
 510         return req->status_code;
 511 }
 512
 513 static void
 514 cancel_message (SoupSession *session, SoupMessage *msg,
 515                 guint status_code)
 516 {
 517         SoupMessageQueue *queue;
 518         SoupMessageQueueItem *item;
 519         gboolean dummy;
 520
 521         SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 522                 cancel_message (session, msg, status_code);
 523
 524         queue = soup_session_get_queue (session);
 525         item = soup_message_queue_lookup (queue, msg);
 526         if (!item)
 527                 return;
 528
 529         /* Force it to finish immediately, so that
 530          * soup_session_abort (session); g_object_unref (session);
 531          * will work. (The soup_session_cancel_message() docs
 532          * point out that the callback will be invoked from
 533          * within the cancel call.)
 534          */
 535         if (soup_message_io_in_progress (msg))
 536                 soup_message_io_finished (msg);
 537         else if (item->state != SOUP_MESSAGE_FINISHED)
 538                 item->state = SOUP_MESSAGE_FINISHING;
 539
 540         if (item->state != SOUP_MESSAGE_FINISHED)
 541                 process_queue_item (item, &dummy, FALSE);
 542
 543         soup_message_queue_item_unref (item);
 544 }
 545
 546 static void
 547 got_passwords (SoupPasswordManager *password_manager, SoupMessage *msg,
 548                SoupAuth *auth, gboolean retrying, gpointer session)
 549 {
 550         soup_session_unpause_message (session, msg);
 551         SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 552                 auth_required (session, msg, auth, retrying);
 553         g_object_unref (auth);
 554 }
 555
 556 static void
 557 auth_required (SoupSession *session, SoupMessage *msg,
 558                SoupAuth *auth, gboolean retrying)
 559 {
 560         SoupSessionFeature *password_manager;
 561
 562         password_manager = soup_session_get_feature_for_message (
 563                 session, SOUP_TYPE_PASSWORD_MANAGER, msg);
 564         if (password_manager) {
 565                 soup_session_pause_message (session, msg);
 566                 g_object_ref (auth);
 567                 soup_password_manager_get_passwords_async (
 568                         SOUP_PASSWORD_MANAGER (password_manager),
 569                         msg, auth, retrying,
 570                         soup_session_get_async_context (session),
 571                         NULL, /* FIXME cancellable */
 572                         got_passwords, session);
 573         } else {
 574                 SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 575                         auth_required (session, msg, auth, retrying);
 576         }
 577 }