libsoup/soup-session-sync.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-session-sync.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-sync.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-sync
  27  * @short_description: Soup session for blocking I/O in multithreaded
  28  * programs.
  29  *
  30  * #SoupSessionSync is an implementation of #SoupSession that uses
  31  * synchronous I/O, intended for use in multi-threaded programs.
  32  *
  33  * You can use #SoupSessionSync from multiple threads concurrently.
  34  * Eg, you can send a #SoupMessage in one thread, and then while
  35  * waiting for the response, send another #SoupMessage from another
  36  * thread. You can also send a message from one thread and then call
  37  * soup_session_cancel_message() on it from any other thread (although
  38  * you need to be careful to avoid race conditions, where the message
  39  * finishes and is then unreffed by the sending thread just before you
  40  * cancel it).
  41  *
  42  * However, the majority of other types and methods in libsoup are not
  43  * MT-safe. In particular, you <emphasis>cannot</emphasis> modify or
  44  * examine a #SoupMessage while it is being transmitted by
  45  * #SoupSessionSync in another thread. Once a message has been handed
  46  * off to #SoupSessionSync, it can only be manipulated from its signal
  47  * handler callbacks, until I/O is complete.
  48  **/
  49
  50 typedef struct {
  51         GMutex *lock;
  52         GCond *cond;
  53 } SoupSessionSyncPrivate;
  54 #define SOUP_SESSION_SYNC_GET_PRIVATE(o) (G_TYPE_INSTANCE_GET_PRIVATE ((o), SOUP_TYPE_SESSION_SYNC, SoupSessionSyncPrivate))
  55
  56 static void  queue_message  (SoupSession *session, SoupMessage *msg,
  57                              SoupSessionCallback callback, gpointer user_data);
  58 static guint send_message   (SoupSession *session, SoupMessage *msg);
  59 static void  cancel_message (SoupSession *session, SoupMessage *msg,
  60                              guint status_code);
  61 static void  auth_required  (SoupSession *session, SoupMessage *msg,
  62                              SoupAuth *auth, gboolean retrying);
  63 static void  flush_queue    (SoupSession *session);
  64
  65 G_DEFINE_TYPE (SoupSessionSync, soup_session_sync, SOUP_TYPE_SESSION)
  66
  67 static void
  68 soup_session_sync_init (SoupSessionSync *ss)
  69 {
  70         SoupSessionSyncPrivate *priv = SOUP_SESSION_SYNC_GET_PRIVATE (ss);
  71
  72         priv->lock = g_mutex_new ();
  73         priv->cond = g_cond_new ();
  74 }
  75
  76 static void
  77 finalize (GObject *object)
  78 {
  79         SoupSessionSyncPrivate *priv = SOUP_SESSION_SYNC_GET_PRIVATE (object);
  80
  81         g_mutex_free (priv->lock);
  82         g_cond_free (priv->cond);
  83
  84         G_OBJECT_CLASS (soup_session_sync_parent_class)->finalize (object);
  85 }
  86
  87 static void
  88 soup_session_sync_class_init (SoupSessionSyncClass *session_sync_class)
  89 {
  90         GObjectClass *object_class = G_OBJECT_CLASS (session_sync_class);
  91         SoupSessionClass *session_class = SOUP_SESSION_CLASS (session_sync_class);
  92
  93         g_type_class_add_private (session_sync_class, sizeof (SoupSessionSyncPrivate));
  94
  95         /* virtual method override */
  96         session_class->queue_message = queue_message;
  97         session_class->send_message = send_message;
  98         session_class->cancel_message = cancel_message;
  99         session_class->auth_required = auth_required;
 100         session_class->flush_queue = flush_queue;
 101
 102         object_class->finalize = finalize;
 103 }
 104
 105
 106 /**
 107  * soup_session_sync_new:
 108  *
 109  * Creates an synchronous #SoupSession with the default options.
 110  *
 111  * Return value: the new session.
 112  **/
 113 SoupSession *
 114 soup_session_sync_new (void)
 115 {
 116         return g_object_new (SOUP_TYPE_SESSION_SYNC, NULL);
 117 }
 118
 119 /**
 120  * soup_session_sync_new_with_options:
 121  * @optname1: name of first property to set
 122  * @...: value of @optname1, followed by additional property/value pairs
 123  *
 124  * Creates an synchronous #SoupSession with the specified options.
 125  *
 126  * Return value: the new session.
 127  **/
 128 SoupSession *
 129 soup_session_sync_new_with_options (const char *optname1, ...)
 130 {
 131         SoupSession *session;
 132         va_list ap;
 133
 134         va_start (ap, optname1);
 135         session = (SoupSession *)g_object_new_valist (SOUP_TYPE_SESSION_SYNC,
 136                                                       optname1, ap);
 137         va_end (ap);
 138
 139         return session;
 140 }
 141
 142 static guint
 143 tunnel_connect (SoupSession *session, SoupMessageQueueItem *related)
 144 {
 145         SoupConnection *conn = related->conn;
 146         SoupMessageQueueItem *item;
 147         guint status;
 148
 149         g_object_ref (conn);
 150
 151         item = soup_session_make_connect_message (session, conn);
 152         do {
 153                 soup_session_send_queue_item (session, item, NULL);
 154                 status = item->msg->status_code;
 155                 if (item->state == SOUP_MESSAGE_RESTARTING &&
 156                     soup_connection_get_state (conn) != SOUP_CONNECTION_DISCONNECTED) {
 157                         item->state = SOUP_MESSAGE_STARTING;
 158                         soup_message_restarted (item->msg);
 159                 } else {
 160                         if (item->state == SOUP_MESSAGE_RESTARTING)
 161                                 status = SOUP_STATUS_TRY_AGAIN;
 162                         item->state = SOUP_MESSAGE_FINISHED;
 163                         soup_message_finished (item->msg);
 164                 }
 165         } while (item->state == SOUP_MESSAGE_STARTING);
 166         soup_session_unqueue_item (session, item);
 167         soup_message_queue_item_unref (item);
 168
 169         if (SOUP_STATUS_IS_SUCCESSFUL (status)) {
 170                 if (!soup_connection_start_ssl_sync (conn, related->cancellable))
 171                         status = SOUP_STATUS_SSL_FAILED;
 172         }
 173
 174         if (!SOUP_STATUS_IS_SUCCESSFUL (status))
 175                 soup_connection_disconnect (conn);
 176
 177         g_object_unref (conn);
 178         return status;
 179 }
 180
 181 static void
 182 get_connection (SoupMessageQueueItem *item)
 183 {
 184         SoupSession *session = item->session;
 185         SoupMessage *msg = item->msg;
 186         gboolean try_pruning = FALSE;
 187         guint status;
 188
 189 try_again:
 190         soup_session_cleanup_connections (session, FALSE);
 191
 192         if (!soup_session_get_connection (session, item, &try_pruning)) {
 193                 if (!try_pruning)
 194                         return;
 195                 soup_session_cleanup_connections (session, TRUE);
 196                 if (!soup_session_get_connection (session, item, &try_pruning))
 197                         return;
 198                 try_pruning = FALSE;
 199         }
 200
 201         if (soup_connection_get_state (item->conn) != SOUP_CONNECTION_NEW) {
 202                 item->state = SOUP_MESSAGE_READY;
 203                 return;
 204         }
 205
 206         status = soup_connection_connect_sync (item->conn, item->cancellable);
 207         if (status == SOUP_STATUS_TRY_AGAIN) {
 208                 soup_connection_disconnect (item->conn);
 209                 g_object_unref (item->conn);
 210                 item->conn = NULL;
 211                 goto try_again;
 212         }
 213
 214         if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
 215                 if (!msg->status_code)
 216                         soup_session_set_item_status (session, item, status);
 217                 item->state = SOUP_MESSAGE_FINISHING;
 218                 soup_connection_disconnect (item->conn);
 219                 g_object_unref (item->conn);
 220                 item->conn = NULL;
 221                 return;
 222         }
 223
 224         if (soup_connection_get_tunnel_addr (item->conn)) {
 225                 status = tunnel_connect (session, item);
 226                 if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
 227                         soup_connection_disconnect (item->conn);
 228                         g_object_unref (item->conn);
 229                         item->conn = NULL;
 230                         if (status == SOUP_STATUS_TRY_AGAIN)
 231                                 goto try_again;
 232                         soup_session_set_item_status (session, item, status);
 233                         item->state = SOUP_MESSAGE_FINISHING;
 234                         return;
 235                 }
 236         }
 237
 238         item->state = SOUP_MESSAGE_READY;
 239 }
 240
 241 static void
 242 process_queue_item (SoupMessageQueueItem *item)
 243 {
 244         SoupSession *session = item->session;
 245         SoupSessionSyncPrivate *priv = SOUP_SESSION_SYNC_GET_PRIVATE (session);
 246         SoupMessage *msg = item->msg;
 247         SoupProxyURIResolver *proxy_resolver;
 248         guint status;
 249
 250         item->state = SOUP_MESSAGE_STARTING;
 251         do {
 252                 switch (item->state) {
 253                 case SOUP_MESSAGE_STARTING:
 254                         proxy_resolver = (SoupProxyURIResolver *)soup_session_get_feature_for_message (session, SOUP_TYPE_PROXY_URI_RESOLVER, msg);
 255                         if (!proxy_resolver) {
 256                                 item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 257                                 break;
 258                         }
 259
 260                         status = soup_proxy_uri_resolver_get_proxy_uri_sync (
 261                                 proxy_resolver, soup_message_get_uri (msg),
 262                                 item->cancellable, &item->proxy_uri);
 263                         if (!SOUP_STATUS_IS_SUCCESSFUL (status)) {
 264                                 soup_session_set_item_status (session, item, status);
 265                                 item->state = SOUP_MESSAGE_FINISHING;
 266                                 break;
 267                         }
 268                         if (!item->proxy_uri) {
 269                                 item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 270                                 break;
 271                         }
 272
 273                         item->proxy_addr = soup_address_new (
 274                                 item->proxy_uri->host, item->proxy_uri->port);
 275                         status = soup_address_resolve_sync (item->proxy_addr,
 276                                                             item->cancellable);
 277                         if (SOUP_STATUS_IS_SUCCESSFUL (status))
 278                                 item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 279                         else {
 280                                 soup_session_set_item_status (session, item, soup_status_proxify (status));
 281                                 item->state = SOUP_MESSAGE_FINISHING;
 282                         }
 283                         break;
 284
 285                 case SOUP_MESSAGE_AWAITING_CONNECTION:
 286                         g_mutex_lock (priv->lock);
 287                         do {
 288                                 get_connection (item);
 289                                 if (item->state == SOUP_MESSAGE_AWAITING_CONNECTION)
 290                                         g_cond_wait (priv->cond, priv->lock);
 291                         } while (item->state == SOUP_MESSAGE_AWAITING_CONNECTION);
 292                         g_mutex_unlock (priv->lock);
 293                         break;
 294
 295                 case SOUP_MESSAGE_READY:
 296                         item->state = SOUP_MESSAGE_RUNNING;
 297                         soup_session_send_queue_item (item->session, item, NULL);
 298                         if (item->state != SOUP_MESSAGE_RESTARTING)
 299                                 item->state = SOUP_MESSAGE_FINISHING;
 300                         break;
 301
 302                 case SOUP_MESSAGE_RESTARTING:
 303                         item->state = SOUP_MESSAGE_STARTING;
 304                         soup_message_restarted (item->msg);
 305                         break;
 306
 307                 case SOUP_MESSAGE_FINISHING:
 308                         item->state = SOUP_MESSAGE_FINISHED;
 309                         soup_message_finished (item->msg);
 310                         soup_session_unqueue_item (session, item);
 311                         g_cond_broadcast (priv->cond);
 312                         break;
 313
 314                 default:
 315                         g_warn_if_reached ();
 316                         item->state = SOUP_MESSAGE_FINISHING;
 317                         break;
 318                 }
 319         } while (item->state != SOUP_MESSAGE_FINISHED);
 320 }
 321
 322 static gboolean
 323 queue_message_callback (gpointer data)
 324 {
 325         SoupMessageQueueItem *item = data;
 326
 327         item->callback (item->session, item->msg, item->callback_data);
 328         g_object_unref (item->session);
 329         g_object_unref (item->msg);
 330         soup_message_queue_item_unref (item);
 331         return FALSE;
 332 }
 333
 334 static gpointer
 335 queue_message_thread (gpointer data)
 336 {
 337         SoupMessageQueueItem *item = data;
 338
 339         process_queue_item (item);
 340         if (item->callback) {
 341                 soup_add_completion (soup_session_get_async_context (item->session),
 342                                      queue_message_callback, item);
 343         } else {
 344                 g_object_unref (item->session);
 345                 g_object_unref (item->msg);
 346                 soup_message_queue_item_unref (item);
 347         }
 348
 349         return NULL;
 350 }
 351
 352 static void
 353 queue_message (SoupSession *session, SoupMessage *msg,
 354                SoupSessionCallback callback, gpointer user_data)
 355 {
 356         SoupMessageQueueItem *item;
 357
 358         SOUP_SESSION_CLASS (soup_session_sync_parent_class)->
 359                 queue_message (g_object_ref (session), msg, callback, user_data);
 360
 361         item = soup_message_queue_lookup (soup_session_get_queue (session), msg);
 362         g_return_if_fail (item != NULL);
 363
 364         g_thread_create (queue_message_thread, item, FALSE, NULL);
 365 }
 366
 367 static guint
 368 send_message (SoupSession *session, SoupMessage *msg)
 369 {
 370         SoupMessageQueueItem *item;
 371         guint status;
 372
 373         SOUP_SESSION_CLASS (soup_session_sync_parent_class)->queue_message (session, msg, NULL, NULL);
 374
 375         item = soup_message_queue_lookup (soup_session_get_queue (session), msg);
 376         g_return_val_if_fail (item != NULL, SOUP_STATUS_MALFORMED);
 377
 378         process_queue_item (item);
 379         status = msg->status_code;
 380         soup_message_queue_item_unref (item);
 381         return status;
 382 }
 383
 384 static void
 385 cancel_message (SoupSession *session, SoupMessage *msg, guint status_code)
 386 {
 387         SoupSessionSyncPrivate *priv = SOUP_SESSION_SYNC_GET_PRIVATE (session);
 388
 389         g_mutex_lock (priv->lock);
 390         SOUP_SESSION_CLASS (soup_session_sync_parent_class)->cancel_message (session, msg, status_code);
 391         g_cond_broadcast (priv->cond);
 392         g_mutex_unlock (priv->lock);
 393 }
 394
 395 static void
 396 auth_required (SoupSession *session, SoupMessage *msg,
 397                SoupAuth *auth, gboolean retrying)
 398 {
 399         SoupSessionFeature *password_manager;
 400
 401         password_manager = soup_session_get_feature_for_message (
 402                 session, SOUP_TYPE_PASSWORD_MANAGER, msg);
 403         if (password_manager) {
 404                 soup_password_manager_get_passwords_sync (
 405                         SOUP_PASSWORD_MANAGER (password_manager),
 406                         msg, auth, NULL); /* FIXME cancellable */
 407         }
 408
 409         SOUP_SESSION_CLASS (soup_session_sync_parent_class)->
 410                 auth_required (session, msg, auth, retrying);
 411 }
 412
 413 static void
 414 flush_queue (SoupSession *session)
 415 {
 416         SoupSessionSyncPrivate *priv = SOUP_SESSION_SYNC_GET_PRIVATE (session);
 417         SoupMessageQueue *queue;
 418         SoupMessageQueueItem *item;
 419         GHashTable *current;
 420         gboolean done = FALSE;
 421
 422         /* Record the current contents of the queue */
 423         current = g_hash_table_new (NULL, NULL);
 424         queue = soup_session_get_queue (session);
 425         for (item = soup_message_queue_first (queue);
 426              item;
 427              item = soup_message_queue_next (queue, item))
 428                 g_hash_table_insert (current, item, item);
 429
 430         /* Cancel everything */
 431         SOUP_SESSION_CLASS (soup_session_sync_parent_class)->flush_queue (session);
 432
 433         /* Wait until all of the items in @current have been removed
 434          * from the queue. (This is not the same as "wait for the
 435          * queue to be empty", because the app may queue new requests
 436          * in response to the cancellation of the old ones. We don't
 437          * try to cancel those requests as well, since we'd likely
 438          * just end up looping forever.)
 439          */
 440         g_mutex_lock (priv->lock);
 441         do {
 442                 done = TRUE;
 443                 for (item = soup_message_queue_first (queue);
 444                      item;
 445                      item = soup_message_queue_next (queue, item)) {
 446                         if (g_hash_table_lookup (current, item))
 447                                 done = FALSE;
 448                 }
 449
 450                 if (!done)
 451                         g_cond_wait (priv->cond, priv->lock);
 452         } while (!done);
 453         g_mutex_unlock (priv->lock);
 454
 455         g_hash_table_destroy (current);
 456 }