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-uri.h"
  23
  24 /**
  25  * SECTION:soup-session-async
  26  * @short_description: Soup session for asynchronous (main-loop-based) I/O.
  27  *
  28  * #SoupSessionAsync is an implementation of #SoupSession that uses
  29  * non-blocking I/O via the glib main loop. It is intended for use in
  30  * single-threaded programs.
  31  **/
  32
  33 static void run_queue (SoupSessionAsync *sa);
  34 static void do_idle_run_queue (SoupSession *session);
  35
  36 static void send_request_running   (SoupSession *session, SoupMessageQueueItem *item);
  37 static void send_request_restarted (SoupSession *session, SoupMessageQueueItem *item);
  38 static void send_request_finished  (SoupSession *session, SoupMessageQueueItem *item);
  39
  40 static void  queue_message   (SoupSession *session, SoupMessage *req,
  41                               SoupSessionCallback callback, gpointer user_data);
  42 static guint send_message    (SoupSession *session, SoupMessage *req);
  43 static void  cancel_message  (SoupSession *session, SoupMessage *msg,
  44                               guint status_code);
  45 static void  kick            (SoupSession *session);
  46
  47 static void  auth_required   (SoupSession *session, SoupMessage *msg,
  48                               SoupAuth *auth, gboolean retrying);
  49
  50 G_DEFINE_TYPE (SoupSessionAsync, soup_session_async, SOUP_TYPE_SESSION)
  51
  52 typedef struct {
  53         SoupSessionAsync *sa;
  54         gboolean disposed;
  55
  56 } SoupSessionAsyncPrivate;
  57 #define SOUP_SESSION_ASYNC_GET_PRIVATE(o) (G_TYPE_INSTANCE_GET_PRIVATE ((o), SOUP_TYPE_SESSION_ASYNC, SoupSessionAsyncPrivate))
  58
  59 static void
  60 soup_session_async_init (SoupSessionAsync *sa)
  61 {
  62         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (sa);
  63
  64         priv->sa = sa;
  65 }
  66
  67 static void
  68 dispose (GObject *object)
  69 {
  70         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (object);
  71
  72         priv->disposed = TRUE;
  73
  74         G_OBJECT_CLASS (soup_session_async_parent_class)->dispose (object);
  75 }
  76
  77 static void
  78 soup_session_async_class_init (SoupSessionAsyncClass *soup_session_async_class)
  79 {
  80         SoupSessionClass *session_class = SOUP_SESSION_CLASS (soup_session_async_class);
  81         GObjectClass *object_class = G_OBJECT_CLASS (session_class);
  82
  83         g_type_class_add_private (soup_session_async_class,
  84                                   sizeof (SoupSessionAsyncPrivate));
  85
  86         /* virtual method override */
  87         session_class->queue_message = queue_message;
  88         session_class->send_message = send_message;
  89         session_class->cancel_message = cancel_message;
  90         session_class->auth_required = auth_required;
  91         session_class->kick = kick;
  92
  93         object_class->dispose = dispose;
  94 }
  95
  96
  97 /**
  98  * soup_session_async_new:
  99  *
 100  * Creates an asynchronous #SoupSession with the default options.
 101  *
 102  * Return value: the new session.
 103  **/
 104 SoupSession *
 105 soup_session_async_new (void)
 106 {
 107         return g_object_new (SOUP_TYPE_SESSION_ASYNC, NULL);
 108 }
 109
 110 /**
 111  * soup_session_async_new_with_options:
 112  * @optname1: name of first property to set
 113  * @...: value of @optname1, followed by additional property/value pairs
 114  *
 115  * Creates an asynchronous #SoupSession with the specified options.
 116  *
 117  * Return value: the new session.
 118  **/
 119 SoupSession *
 120 soup_session_async_new_with_options (const char *optname1, ...)
 121 {
 122         SoupSession *session;
 123         va_list ap;
 124
 125         va_start (ap, optname1);
 126         session = (SoupSession *)g_object_new_valist (SOUP_TYPE_SESSION_ASYNC,
 127                                                       optname1, ap);
 128         va_end (ap);
 129
 130         return session;
 131 }
 132
 133 static void
 134 connection_closed (SoupConnection *conn, gpointer session)
 135 {
 136         /* Run the queue in case anyone was waiting for a connection
 137          * to be closed.
 138          */
 139         do_idle_run_queue (session);
 140 }
 141
 142 static void
 143 message_completed (SoupMessage *msg, gpointer user_data)
 144 {
 145         SoupMessageQueueItem *item = user_data;
 146
 147         do_idle_run_queue (item->session);
 148
 149         if (item->state != SOUP_MESSAGE_RESTARTING)
 150                 item->state = SOUP_MESSAGE_FINISHING;
 151 }
 152
 153 static void
 154 tunnel_complete (SoupMessageQueueItem *item)
 155 {
 156         SoupSession *session = item->session;
 157
 158         soup_message_finished (item->msg);
 159         if (item->related->msg->status_code)
 160                 item->related->state = SOUP_MESSAGE_FINISHING;
 161         else
 162                 soup_message_set_https_status (item->related->msg, item->conn);
 163
 164         do_idle_run_queue (session);
 165         soup_message_queue_item_unref (item->related);
 166         soup_session_unqueue_item (session, item);
 167         soup_message_queue_item_unref (item);
 168         g_object_unref (session);
 169 }
 170
 171 static void
 172 ssl_tunnel_completed (SoupConnection *conn, guint status, gpointer user_data)
 173 {
 174         SoupMessageQueueItem *item = user_data;
 175
 176         if (SOUP_STATUS_IS_SUCCESSFUL (status)) {
 177                 g_signal_connect (item->conn, "disconnected",
 178                                   G_CALLBACK (connection_closed), item->session);
 179                 soup_connection_set_state (item->conn, SOUP_CONNECTION_IDLE);
 180                 soup_connection_set_state (item->conn, SOUP_CONNECTION_IN_USE);
 181
 182                 item->related->state = SOUP_MESSAGE_READY;
 183         } else {
 184                 if (item->conn)
 185                         soup_connection_disconnect (item->conn);
 186                 soup_message_set_status (item->related->msg, SOUP_STATUS_SSL_FAILED);
 187         }
 188
 189         tunnel_complete (item);
 190 }
 191
 192 static void
 193 tunnel_message_completed (SoupMessage *msg, gpointer user_data)
 194 {
 195         SoupMessageQueueItem *item = user_data;
 196         SoupSession *session = item->session;
 197
 198         if (item->state == SOUP_MESSAGE_RESTARTING) {
 199                 soup_message_restarted (msg);
 200                 if (item->conn) {
 201                         item->state = SOUP_MESSAGE_RUNNING;
 202                         soup_session_send_queue_item (session, item, tunnel_message_completed);
 203                         return;
 204                 }
 205
 206                 soup_message_set_status (msg, SOUP_STATUS_TRY_AGAIN);
 207         }
 208
 209         item->state = SOUP_MESSAGE_FINISHED;
 210
 211         if (!SOUP_STATUS_IS_SUCCESSFUL (msg->status_code)) {
 212                 if (item->conn)
 213                         soup_connection_disconnect (item->conn);
 214                 if (msg->status_code == SOUP_STATUS_TRY_AGAIN) {
 215                         item->related->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 216                         soup_message_queue_item_set_connection (item->related, NULL);
 217                 } else
 218                         soup_message_set_status (item->related->msg, msg->status_code);
 219
 220                 tunnel_complete (item);
 221                 return;
 222         }
 223
 224         soup_connection_start_ssl_async (item->conn, item->cancellable,
 225                                          ssl_tunnel_completed, item);
 226 }
 227
 228 static void
 229 got_connection (SoupConnection *conn, guint status, gpointer user_data)
 230 {
 231         SoupMessageQueueItem *item = user_data;
 232         SoupSession *session = item->session;
 233
 234         if (item->state != SOUP_MESSAGE_CONNECTING) {
 235                 soup_connection_disconnect (conn);
 236                 do_idle_run_queue (session);
 237                 soup_message_queue_item_unref (item);
 238                 g_object_unref (session);
 239                 return;
 240         }
 241
 242         soup_message_set_https_status (item->msg, conn);
 243
 244         if (status != SOUP_STATUS_OK) {
 245                 soup_connection_disconnect (conn);
 246
 247                 if (status == SOUP_STATUS_TRY_AGAIN) {
 248                         soup_message_queue_item_set_connection (item, NULL);
 249                         item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 250                 } else {
 251                         soup_session_set_item_status (session, item, status);
 252                         item->state = SOUP_MESSAGE_FINISHING;
 253                 }
 254
 255                 do_idle_run_queue (session);
 256                 soup_message_queue_item_unref (item);
 257                 g_object_unref (session);
 258                 return;
 259         }
 260
 261         if (soup_connection_is_tunnelled (conn)) {
 262                 SoupMessageQueueItem *tunnel_item;
 263
 264                 item->state = SOUP_MESSAGE_TUNNELING;
 265
 266                 tunnel_item = soup_session_make_connect_message (session, conn);
 267                 tunnel_item->related = item;
 268                 soup_session_send_queue_item (session, tunnel_item, tunnel_message_completed);
 269                 return;
 270         }
 271
 272         item->state = SOUP_MESSAGE_READY;
 273         g_signal_connect (conn, "disconnected",
 274                           G_CALLBACK (connection_closed), session);
 275         run_queue ((SoupSessionAsync *)session);
 276         soup_message_queue_item_unref (item);
 277         g_object_unref (session);
 278 }
 279
 280 static void
 281 process_queue_item (SoupMessageQueueItem *item,
 282                     gboolean             *should_prune,
 283                     gboolean              loop)
 284 {
 285         SoupSession *session = item->session;
 286
 287         if (item->async_context != soup_session_get_async_context (session))
 288                 return;
 289
 290         do {
 291                 if (item->paused)
 292                         return;
 293
 294                 switch (item->state) {
 295                 case SOUP_MESSAGE_STARTING:
 296                         item->state = SOUP_MESSAGE_AWAITING_CONNECTION;
 297                         break;
 298
 299                 case SOUP_MESSAGE_AWAITING_CONNECTION:
 300                         if (!soup_session_get_connection (session, item, should_prune))
 301                                 return;
 302
 303                         if (soup_connection_get_state (item->conn) != SOUP_CONNECTION_NEW) {
 304                                 item->state = SOUP_MESSAGE_READY;
 305                                 break;
 306                         }
 307
 308                         item->state = SOUP_MESSAGE_CONNECTING;
 309                         soup_message_queue_item_ref (item);
 310                         g_object_ref (session);
 311                         soup_connection_connect_async (item->conn, item->cancellable,
 312                                                        got_connection, item);
 313                         return;
 314
 315                 case SOUP_MESSAGE_READY:
 316                         item->state = SOUP_MESSAGE_RUNNING;
 317                         soup_session_send_queue_item (session, item, message_completed);
 318                         if (item->new_api)
 319                                 send_request_running (session, item);
 320                         break;
 321
 322                 case SOUP_MESSAGE_RESTARTING:
 323                         item->state = SOUP_MESSAGE_STARTING;
 324                         soup_message_restarted (item->msg);
 325                         if (item->new_api)
 326                                 send_request_restarted (session, item);
 327                         break;
 328
 329                 case SOUP_MESSAGE_FINISHING:
 330                         item->state = SOUP_MESSAGE_FINISHED;
 331                         soup_message_finished (item->msg);
 332                         if (item->state != SOUP_MESSAGE_FINISHED)
 333                                 break;
 334
 335                         g_object_ref (session);
 336                         soup_session_unqueue_item (session, item);
 337                         if (item->callback)
 338                                 item->callback (session, item->msg, item->callback_data);
 339                         else if (item->new_api)
 340                                 send_request_finished (session, item);
 341                         g_object_unref (item->msg);
 342                         do_idle_run_queue (session);
 343                         g_object_unref (session);
 344                         return;
 345
 346                 default:
 347                         /* Nothing to do with this message in any
 348                          * other state.
 349                          */
 350                         return;
 351                 }
 352         } while (loop && item->state != SOUP_MESSAGE_FINISHED);
 353 }
 354
 355 static void
 356 run_queue (SoupSessionAsync *sa)
 357 {
 358         SoupSession *session = SOUP_SESSION (sa);
 359         SoupMessageQueue *queue = soup_session_get_queue (session);
 360         SoupMessageQueueItem *item;
 361         SoupMessage *msg;
 362         gboolean try_pruning = TRUE, should_prune = FALSE;
 363
 364         g_object_ref (session);
 365         soup_session_cleanup_connections (session, FALSE);
 366
 367  try_again:
 368         for (item = soup_message_queue_first (queue);
 369              item;
 370              item = soup_message_queue_next (queue, item)) {
 371                 msg = item->msg;
 372
 373                 /* CONNECT messages are handled specially */
 374                 if (msg->method != SOUP_METHOD_CONNECT)
 375                         process_queue_item (item, &should_prune, TRUE);
 376         }
 377
 378         if (try_pruning && should_prune) {
 379                 /* There is at least one message in the queue that
 380                  * could be sent if we pruned an idle connection from
 381                  * some other server.
 382                  */
 383                 if (soup_session_cleanup_connections (session, TRUE)) {
 384                         try_pruning = should_prune = FALSE;
 385                         goto try_again;
 386                 }
 387         }
 388
 389         g_object_unref (session);
 390 }
 391
 392 static gboolean
 393 idle_run_queue (gpointer user_data)
 394 {
 395         SoupSessionAsyncPrivate *priv = user_data;
 396
 397         if (priv->disposed)
 398                 return FALSE;
 399
 400         /* Ensure that the source is destroyed before running the queue */
 401         g_source_destroy (g_main_current_source ());
 402
 403         run_queue (priv->sa);
 404         return FALSE;
 405 }
 406
 407 static void
 408 do_idle_run_queue (SoupSession *session)
 409 {
 410         SoupSessionAsyncPrivate *priv = SOUP_SESSION_ASYNC_GET_PRIVATE (session);
 411         GMainContext *async_context = soup_session_get_async_context (session);
 412         GSource *source;
 413
 414         if (priv->disposed)
 415                 return;
 416
 417         /* We use priv rather than session as the source data, because
 418          * other parts of libsoup (or the calling app) may have sources
 419          * using the session as the source data.
 420          */
 421
 422         source = g_main_context_find_source_by_user_data (async_context, priv);
 423         if (source)
 424                 return;
 425
 426         source = soup_add_completion (async_context, idle_run_queue, priv);
 427 }
 428
 429 static void
 430 queue_message (SoupSession *session, SoupMessage *req,
 431                SoupSessionCallback callback, gpointer user_data)
 432 {
 433         SOUP_SESSION_CLASS (soup_session_async_parent_class)->queue_message (session, req, callback, user_data);
 434
 435         do_idle_run_queue (session);
 436 }
 437
 438 static guint
 439 send_message (SoupSession *session, SoupMessage *req)
 440 {
 441         SoupMessageQueueItem *item;
 442         GMainContext *async_context =
 443                 soup_session_get_async_context (session);
 444
 445         /* Balance out the unref that queuing will eventually do */
 446         g_object_ref (req);
 447
 448         queue_message (session, req, NULL, NULL);
 449
 450         item = soup_message_queue_lookup (soup_session_get_queue (session), req);
 451         g_return_val_if_fail (item != NULL, SOUP_STATUS_MALFORMED);
 452
 453         while (item->state != SOUP_MESSAGE_FINISHED)
 454                 g_main_context_iteration (async_context, TRUE);
 455
 456         soup_message_queue_item_unref (item);
 457
 458         return req->status_code;
 459 }
 460
 461 static void
 462 cancel_message (SoupSession *session, SoupMessage *msg,
 463                 guint status_code)
 464 {
 465         SoupMessageQueue *queue;
 466         SoupMessageQueueItem *item;
 467         gboolean dummy;
 468
 469         SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 470                 cancel_message (session, msg, status_code);
 471
 472         queue = soup_session_get_queue (session);
 473         item = soup_message_queue_lookup (queue, msg);
 474         if (!item)
 475                 return;
 476
 477         /* Force it to finish immediately, so that
 478          * soup_session_abort (session); g_object_unref (session);
 479          * will work. (The soup_session_cancel_message() docs
 480          * point out that the callback will be invoked from
 481          * within the cancel call.)
 482          */
 483         if (soup_message_io_in_progress (msg))
 484                 soup_message_io_finished (msg);
 485         else if (item->state != SOUP_MESSAGE_FINISHED)
 486                 item->state = SOUP_MESSAGE_FINISHING;
 487
 488         if (item->state != SOUP_MESSAGE_FINISHED)
 489                 process_queue_item (item, &dummy, FALSE);
 490
 491         soup_message_queue_item_unref (item);
 492 }
 493
 494 static void
 495 got_passwords (SoupPasswordManager *password_manager, SoupMessage *msg,
 496                SoupAuth *auth, gboolean retrying, gpointer session)
 497 {
 498         soup_session_unpause_message (session, msg);
 499         SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 500                 auth_required (session, msg, auth, retrying);
 501         g_object_unref (auth);
 502 }
 503
 504 static void
 505 auth_required (SoupSession *session, SoupMessage *msg,
 506                SoupAuth *auth, gboolean retrying)
 507 {
 508         SoupSessionFeature *password_manager;
 509
 510         password_manager = soup_session_get_feature_for_message (
 511                 session, SOUP_TYPE_PASSWORD_MANAGER, msg);
 512         if (password_manager) {
 513                 soup_session_pause_message (session, msg);
 514                 g_object_ref (auth);
 515                 soup_password_manager_get_passwords_async (
 516                         SOUP_PASSWORD_MANAGER (password_manager),
 517                         msg, auth, retrying,
 518                         soup_session_get_async_context (session),
 519                         NULL, /* FIXME cancellable */
 520                         got_passwords, session);
 521         } else {
 522                 SOUP_SESSION_CLASS (soup_session_async_parent_class)->
 523                         auth_required (session, msg, auth, retrying);
 524         }
 525 }
 526
 527 static void
 528 kick (SoupSession *session)
 529 {
 530         do_idle_run_queue (session);
 531 }
 532
 533
 534 static void
 535 send_request_return_result (SoupMessageQueueItem *item,
 536                             gpointer stream, GError *error)
 537 {
 538         GSimpleAsyncResult *simple;
 539
 540         simple = item->result;
 541         item->result = NULL;
 542
 543         if (error)
 544                 g_simple_async_result_take_error (simple, error);
 545         else if (SOUP_STATUS_IS_TRANSPORT_ERROR (item->msg->status_code)) {
 546                 if (stream)
 547                         g_object_unref (stream);
 548                 g_simple_async_result_set_error (simple,
 549                                                  SOUP_HTTP_ERROR,
 550                                                  item->msg->status_code,
 551                                                  "%s",
 552                                                  item->msg->reason_phrase);
 553         } else
 554                 g_simple_async_result_set_op_res_gpointer (simple, stream, g_object_unref);
 555
 556         g_simple_async_result_complete (simple);
 557         g_object_unref (simple);
 558 }
 559
 560 static void
 561 send_request_restarted (SoupSession *session, SoupMessageQueueItem *item)
 562 {
 563         /* We won't be needing this, then. */
 564         g_object_set_data (G_OBJECT (item->msg), "SoupSessionAsync:ostream", NULL);
 565         item->io_started = FALSE;
 566 }
 567
 568 static void
 569 send_request_finished (SoupSession *session, SoupMessageQueueItem *item)
 570 {
 571         GMemoryOutputStream *mostream;
 572         GInputStream *istream = NULL;
 573         GError *error = NULL;
 574
 575         if (!item->result) {
 576                 /* Something else already took care of it. */
 577                 return;
 578         }
 579
 580         mostream = g_object_get_data (G_OBJECT (item->msg), "SoupSessionAsync:ostream");
 581         if (mostream) {
 582                 gpointer data;
 583                 gssize size;
 584
 585                 /* We thought it would be requeued, but it wasn't, so
 586                  * return the original body.
 587                  */
 588                 size = g_memory_output_stream_get_data_size (mostream);
 589                 data = size ? g_memory_output_stream_steal_data (mostream) : g_strdup ("");
 590                 istream = g_memory_input_stream_new_from_data (data, size, g_free);
 591         } else if (item->io_started) {
 592                 /* The message finished before becoming readable. This
 593                  * will happen, eg, if it's cancelled from got-headers.
 594                  * Do nothing; the op will complete via read_ready_cb()
 595                  * after we return;
 596                  */
 597                 return;
 598         } else {
 599                 /* The message finished before even being started;
 600                  * probably a tunnel connect failure.
 601                  */
 602                 istream = g_memory_input_stream_new ();
 603         }
 604
 605         send_request_return_result (item, istream, error);
 606 }
 607
 608 static void
 609 send_async_spliced (GObject *source, GAsyncResult *result, gpointer user_data)
 610 {
 611         SoupMessageQueueItem *item = user_data;
 612         GInputStream *istream = g_object_get_data (source, "istream");
 613
 614         GError *error = NULL;
 615
 616         /* If the message was cancelled, it will be completed via other means */
 617         if (g_cancellable_is_cancelled (item->cancellable) ||
 618             !item->result) {
 619                 soup_message_queue_item_unref (item);
 620                 return;
 621         }
 622
 623         if (g_output_stream_splice_finish (G_OUTPUT_STREAM (source),
 624                                            result, &error) == -1) {
 625                 send_request_return_result (item, NULL, error);
 626                 return;
 627         }
 628
 629         /* Otherwise either restarted or finished will eventually be called.
 630          * It should be safe to call the sync close() method here since
 631          * the message body has already been written.
 632          */
 633         g_input_stream_close (istream, NULL, NULL);
 634         do_idle_run_queue (item->session);
 635         soup_message_queue_item_unref (item);
 636 }
 637
 638 static void
 639 send_async_maybe_complete (SoupMessageQueueItem *item,
 640                            GInputStream         *stream)
 641 {
 642         if (item->msg->status_code == SOUP_STATUS_UNAUTHORIZED ||
 643             item->msg->status_code == SOUP_STATUS_PROXY_UNAUTHORIZED ||
 644             soup_session_would_redirect (item->session, item->msg)) {
 645                 GOutputStream *ostream;
 646
 647                 /* Message may be requeued, so gather the current message body... */
 648                 ostream = g_memory_output_stream_new (NULL, 0, g_realloc, g_free);
 649                 g_object_set_data_full (G_OBJECT (item->msg), "SoupSessionAsync:ostream",
 650                                         ostream, g_object_unref);
 651
 652                 g_object_set_data_full (G_OBJECT (ostream), "istream",
 653                                         stream, g_object_unref);
 654
 655                 /* Give the splice op its own ref on item */
 656                 soup_message_queue_item_ref (item);
 657                 g_output_stream_splice_async (ostream, stream,
 658                                               /* We can't use CLOSE_SOURCE because it
 659                                                * might get closed in the wrong thread.
 660                                                */
 661                                               G_OUTPUT_STREAM_SPLICE_CLOSE_TARGET,
 662                                               G_PRIORITY_DEFAULT,
 663                                               item->cancellable,
 664                                               send_async_spliced, item);
 665                 return;
 666         }
 667
 668         send_request_return_result (item, stream, NULL);
 669 }
 670
 671 static void try_run_until_read (SoupMessageQueueItem *item);
 672
 673 static gboolean
 674 read_ready_cb (SoupMessage *msg, gpointer user_data)
 675 {
 676         SoupMessageQueueItem *item = user_data;
 677
 678         try_run_until_read (item);
 679         return FALSE;
 680 }
 681
 682 static void
 683 try_run_until_read (SoupMessageQueueItem *item)
 684 {
 685         GError *error = NULL;
 686         GInputStream *stream = NULL;
 687         GSource *source;
 688
 689         if (soup_message_io_run_until_read (item->msg, item->cancellable, &error))
 690                 stream = soup_message_io_get_response_istream (item->msg, &error);
 691         if (stream) {
 692                 send_async_maybe_complete (item, stream);
 693                 return;
 694         }
 695
 696         if (!g_error_matches (error, G_IO_ERROR, G_IO_ERROR_WOULD_BLOCK)) {
 697                 send_request_return_result (item, NULL, error);
 698                 return;
 699         }
 700
 701         g_clear_error (&error);
 702         source = soup_message_io_get_source (item->msg, item->cancellable,
 703                                              read_ready_cb, item);
 704         g_source_attach (source, soup_session_get_async_context (item->session));
 705         g_source_unref (source);
 706 }
 707
 708 static void
 709 send_request_running (SoupSession *session, SoupMessageQueueItem *item)
 710 {
 711         item->io_started = TRUE;
 712         try_run_until_read (item);
 713 }
 714
 715 void
 716 soup_session_send_request_async (SoupSession         *session,
 717                                  SoupMessage         *msg,
 718                                  GCancellable        *cancellable,
 719                                  GAsyncReadyCallback  callback,
 720                                  gpointer             user_data)
 721 {
 722         SoupMessageQueueItem *item;
 723         gboolean use_thread_context;
 724
 725         g_return_if_fail (SOUP_IS_SESSION_ASYNC (session));
 726
 727         g_object_get (G_OBJECT (session),
 728                       SOUP_SESSION_USE_THREAD_CONTEXT, &use_thread_context,
 729                       NULL);
 730         g_return_if_fail (use_thread_context);
 731
 732         /* Balance out the unref that queuing will eventually do */
 733         g_object_ref (msg);
 734
 735         queue_message (session, msg, NULL, NULL);
 736
 737         item = soup_message_queue_lookup (soup_session_get_queue (session), msg);
 738         g_return_if_fail (item != NULL);
 739
 740         item->new_api = TRUE;
 741         item->result = g_simple_async_result_new (G_OBJECT (session),
 742                                                   callback, user_data,
 743                                                   soup_session_send_request_async);
 744         g_simple_async_result_set_op_res_gpointer (item->result, item, (GDestroyNotify) soup_message_queue_item_unref);
 745
 746         if (cancellable) {
 747                 g_object_unref (item->cancellable);
 748                 item->cancellable = g_object_ref (cancellable);
 749         }
 750 }
 751
 752 GInputStream *
 753 soup_session_send_request_finish (SoupSession   *session,
 754                                   GAsyncResult  *result,
 755                                   GError       **error)
 756 {
 757         GSimpleAsyncResult *simple;
 758
 759         g_return_val_if_fail (SOUP_IS_SESSION_ASYNC (session), NULL);
 760         g_return_val_if_fail (g_simple_async_result_is_valid (result, G_OBJECT (session), soup_session_send_request_async), NULL);
 761
 762         simple = G_SIMPLE_ASYNC_RESULT (result);
 763         if (g_simple_async_result_propagate_error (simple, error))
 764                 return NULL;
 765         return g_object_ref (g_simple_async_result_get_op_res_gpointer (simple));
 766 }