libsoup/soup-message-io.c

   1 /* -*- Mode: C; tab-width: 8; indent-tabs-mode: t; c-basic-offset: 8 -*- */
   2 /*
   3  * soup-message-io.c: HTTP message I/O
   4  *
   5  * Copyright (C) 2000-2003, Ximian, Inc.
   6  */
   7
   8 #ifdef HAVE_CONFIG_H
   9 #include "config.h"
  10 #endif
  11
  12 #include <stdlib.h>
  13 #include <string.h>
  14
  15 #include "soup-connection.h"
  16 #include "soup-message.h"
  17 #include "soup-message-private.h"
  18 #include "soup-message-queue.h"
  19 #include "soup-misc.h"
  20 #include "soup-socket.h"
  21 #include "soup-ssl.h"
  22
  23 typedef enum {
  24         SOUP_MESSAGE_IO_CLIENT,
  25         SOUP_MESSAGE_IO_SERVER
  26 } SoupMessageIOMode;
  27
  28 typedef enum {
  29         SOUP_MESSAGE_IO_STATE_NOT_STARTED,
  30         SOUP_MESSAGE_IO_STATE_HEADERS,
  31         SOUP_MESSAGE_IO_STATE_BLOCKING,
  32         SOUP_MESSAGE_IO_STATE_BODY,
  33         SOUP_MESSAGE_IO_STATE_CHUNK_SIZE,
  34         SOUP_MESSAGE_IO_STATE_CHUNK,
  35         SOUP_MESSAGE_IO_STATE_CHUNK_END,
  36         SOUP_MESSAGE_IO_STATE_TRAILERS,
  37         SOUP_MESSAGE_IO_STATE_FINISHING,
  38         SOUP_MESSAGE_IO_STATE_DONE
  39 } SoupMessageIOState;
  40
  41 #define SOUP_MESSAGE_IO_STATE_ACTIVE(state) \
  42         (state != SOUP_MESSAGE_IO_STATE_NOT_STARTED && \
  43          state != SOUP_MESSAGE_IO_STATE_BLOCKING && \
  44          state != SOUP_MESSAGE_IO_STATE_DONE)
  45
  46 typedef struct {
  47         SoupSocket           *sock;
  48         SoupMessageQueueItem *item;
  49         SoupMessageIOMode     mode;
  50         GCancellable         *cancellable;
  51
  52         SoupMessageIOState    read_state;
  53         SoupEncoding          read_encoding;
  54         GByteArray           *read_meta_buf;
  55         SoupMessageBody      *read_body;
  56         goffset               read_length;
  57         gboolean              read_eof_ok;
  58
  59         gboolean              need_content_sniffed, need_got_chunk;
  60         SoupMessageBody      *sniff_data;
  61
  62         SoupMessageIOState    write_state;
  63         SoupEncoding          write_encoding;
  64         GString              *write_buf;
  65         SoupMessageBody      *write_body;
  66         SoupBuffer           *write_chunk;
  67         goffset               write_body_offset;
  68         goffset               write_length;
  69         goffset               written;
  70
  71         guint read_tag, write_tag, tls_signal_id;
  72         GSource *unpause_source;
  73
  74         SoupMessageGetHeadersFn   get_headers_cb;
  75         SoupMessageParseHeadersFn parse_headers_cb;
  76         gpointer                  header_data;
  77         SoupMessageCompletionFn   completion_cb;
  78         gpointer                  completion_data;
  79 } SoupMessageIOData;
  80
  81
  82 /* Put these around callback invocation if there is code afterward
  83  * that depends on the IO having not been cancelled.
  84  */
  85 #define dummy_to_make_emacs_happy {
  86 #define SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK { gboolean cancelled; g_object_ref (msg);
  87 #define SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED cancelled = (priv->io_data != io); g_object_unref (msg); if (cancelled || (!io->read_tag && !io->write_tag)) return; }
  88 #define SOUP_MESSAGE_IO_RETURN_VAL_IF_CANCELLED_OR_PAUSED(val) cancelled = (priv->io_data != io); g_object_unref (msg); if (cancelled || (!io->read_tag && !io->write_tag)) return val; }
  89
  90 #define RESPONSE_BLOCK_SIZE 8192
  91
  92 void
  93 soup_message_io_cleanup (SoupMessage *msg)
  94 {
  95         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
  96         SoupMessageIOData *io;
  97
  98         soup_message_io_stop (msg);
  99
 100         io = priv->io_data;
 101         if (!io)
 102                 return;
 103         priv->io_data = NULL;
 104
 105         if (io->tls_signal_id)
 106                 g_signal_handler_disconnect (io->sock, io->tls_signal_id);
 107         if (io->sock)
 108                 g_object_unref (io->sock);
 109         if (io->item)
 110                 soup_message_queue_item_unref (io->item);
 111
 112         g_byte_array_free (io->read_meta_buf, TRUE);
 113
 114         g_string_free (io->write_buf, TRUE);
 115         if (io->write_chunk)
 116                 soup_buffer_free (io->write_chunk);
 117
 118         if (io->sniff_data)
 119                 soup_message_body_free (io->sniff_data);
 120
 121         g_slice_free (SoupMessageIOData, io);
 122 }
 123
 124 void
 125 soup_message_io_stop (SoupMessage *msg)
 126 {
 127         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 128         SoupMessageIOData *io = priv->io_data;
 129
 130         if (!io)
 131                 return;
 132
 133         if (io->read_tag) {
 134                 g_signal_handler_disconnect (io->sock, io->read_tag);
 135                 io->read_tag = 0;
 136         }
 137         if (io->write_tag) {
 138                 g_signal_handler_disconnect (io->sock, io->write_tag);
 139                 io->write_tag = 0;
 140         }
 141
 142         if (io->unpause_source) {
 143                 g_source_destroy (io->unpause_source);
 144                 io->unpause_source = NULL;
 145         }
 146
 147         if (io->read_state < SOUP_MESSAGE_IO_STATE_FINISHING)
 148                 soup_socket_disconnect (io->sock);
 149         else if (io->item && io->item->conn) {
 150                 soup_connection_set_state (io->item->conn, SOUP_CONNECTION_IDLE);
 151                 g_object_unref (io->item->conn);
 152                 io->item->conn = NULL;
 153         }
 154 }
 155
 156 #define SOUP_MESSAGE_IO_EOL            "\r\n"
 157 #define SOUP_MESSAGE_IO_EOL_LEN        2
 158
 159 void
 160 soup_message_io_finished (SoupMessage *msg)
 161 {
 162         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 163         SoupMessageIOData *io = priv->io_data;
 164         SoupMessageCompletionFn completion_cb = io->completion_cb;
 165         gpointer completion_data = io->completion_data;
 166
 167         g_object_ref (msg);
 168         soup_message_io_cleanup (msg);
 169         if (completion_cb)
 170                 completion_cb (msg, completion_data);
 171         g_object_unref (msg);
 172 }
 173
 174 static void io_read (SoupSocket *sock, SoupMessage *msg);
 175
 176 static gboolean
 177 request_is_idempotent (SoupMessage *msg)
 178 {
 179         /* FIXME */
 180         return (msg->method == SOUP_METHOD_GET);
 181 }
 182
 183 static void
 184 io_error (SoupSocket *sock, SoupMessage *msg, GError *error)
 185 {
 186         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 187         SoupMessageIOData *io = priv->io_data;
 188
 189         if (error && error->domain == G_TLS_ERROR) {
 190                 soup_message_set_status_full (msg,
 191                                               SOUP_STATUS_SSL_FAILED,
 192                                               error->message);
 193         } else if (io->mode == SOUP_MESSAGE_IO_CLIENT &&
 194                    io->read_state <= SOUP_MESSAGE_IO_STATE_HEADERS &&
 195                    io->read_meta_buf->len == 0 &&
 196                    soup_connection_get_ever_used (io->item->conn) &&
 197                    !g_error_matches (error, G_IO_ERROR, G_IO_ERROR_TIMED_OUT) &&
 198                    request_is_idempotent (msg)) {
 199                 /* Connection got closed, but we can safely try again */
 200                 io->item->state = SOUP_MESSAGE_RESTARTING;
 201         } else if (!SOUP_STATUS_IS_TRANSPORT_ERROR (msg->status_code))
 202                 soup_message_set_status (msg, SOUP_STATUS_IO_ERROR);
 203
 204         if (error)
 205                 g_error_free (error);
 206
 207         soup_message_io_finished (msg);
 208 }
 209
 210 static gboolean
 211 io_handle_sniffing (SoupMessage *msg, gboolean done_reading)
 212 {
 213         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 214         SoupMessageIOData *io = priv->io_data;
 215         SoupBuffer *sniffed_buffer;
 216         char *sniffed_mime_type;
 217         GHashTable *params = NULL;
 218
 219         if (!priv->sniffer)
 220                 return TRUE;
 221
 222         if (!io->sniff_data) {
 223                 io->sniff_data = soup_message_body_new ();
 224                 io->need_content_sniffed = TRUE;
 225         }
 226
 227         if (io->need_content_sniffed) {
 228                 if (io->sniff_data->length < priv->bytes_for_sniffing &&
 229                     !done_reading)
 230                         return TRUE;
 231
 232                 io->need_content_sniffed = FALSE;
 233                 sniffed_buffer = soup_message_body_flatten (io->sniff_data);
 234                 sniffed_mime_type = soup_content_sniffer_sniff (priv->sniffer, msg, sniffed_buffer, &params);
 235
 236                 SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 237                 soup_message_content_sniffed (msg, sniffed_mime_type, params);
 238                 g_free (sniffed_mime_type);
 239                 if (params)
 240                         g_hash_table_destroy (params);
 241                 if (sniffed_buffer)
 242                         soup_buffer_free (sniffed_buffer);
 243                 SOUP_MESSAGE_IO_RETURN_VAL_IF_CANCELLED_OR_PAUSED (FALSE);
 244         }
 245
 246         if (io->need_got_chunk) {
 247                 io->need_got_chunk = FALSE;
 248                 sniffed_buffer = soup_message_body_flatten (io->sniff_data);
 249
 250                 SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 251                 soup_message_got_chunk (msg, sniffed_buffer);
 252                 soup_buffer_free (sniffed_buffer);
 253                 SOUP_MESSAGE_IO_RETURN_VAL_IF_CANCELLED_OR_PAUSED (FALSE);
 254         }
 255
 256         return TRUE;
 257 }
 258
 259 /* Reads data from io->sock into io->read_meta_buf. If @to_blank is
 260  * %TRUE, it reads up until a blank line ("CRLF CRLF" or "LF LF").
 261  * Otherwise, it reads up until a single CRLF or LF.
 262  *
 263  * This function is used to read metadata, and read_body_chunk() is
 264  * used to read the message body contents.
 265  *
 266  * read_metadata, read_body_chunk, and write_data all use the same
 267  * convention for return values: if they return %TRUE, it means
 268  * they've completely finished the requested read/write, and the
 269  * caller should move on to the next step. If they return %FALSE, it
 270  * means that either (a) the socket returned SOUP_SOCKET_WOULD_BLOCK,
 271  * so the caller should give up for now and wait for the socket to
 272  * emit a signal, or (b) the socket returned an error, and io_error()
 273  * was called to process it and cancel the I/O. So either way, if the
 274  * function returns %FALSE, the caller should return immediately.
 275  */
 276 static gboolean
 277 read_metadata (SoupMessage *msg, gboolean to_blank)
 278 {
 279         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 280         SoupMessageIOData *io = priv->io_data;
 281         SoupSocketIOStatus status;
 282         guchar read_buf[RESPONSE_BLOCK_SIZE];
 283         gsize nread;
 284         gboolean got_lf;
 285         GError *error = NULL;
 286
 287         while (1) {
 288                 status = soup_socket_read_until (io->sock, read_buf,
 289                                                  sizeof (read_buf),
 290                                                  "\n", 1, &nread, &got_lf,
 291                                                  io->cancellable, &error);
 292                 switch (status) {
 293                 case SOUP_SOCKET_OK:
 294                         g_byte_array_append (io->read_meta_buf, read_buf, nread);
 295                         break;
 296
 297                 case SOUP_SOCKET_EOF:
 298                         /* More lame server handling... deal with
 299                          * servers that don't send the final chunk.
 300                          */
 301                         if (io->read_state == SOUP_MESSAGE_IO_STATE_CHUNK_SIZE &&
 302                             io->read_meta_buf->len == 0) {
 303                                 g_byte_array_append (io->read_meta_buf,
 304                                                      (guchar *)"0\r\n", 3);
 305                                 got_lf = TRUE;
 306                                 break;
 307                         } else if (io->read_state == SOUP_MESSAGE_IO_STATE_TRAILERS &&
 308                                    io->read_meta_buf->len == 0) {
 309                                 g_byte_array_append (io->read_meta_buf,
 310                                                      (guchar *)"\r\n", 2);
 311                                 got_lf = TRUE;
 312                                 break;
 313                         }
 314                         /* else fall through */
 315
 316                 case SOUP_SOCKET_ERROR:
 317                         io_error (io->sock, msg, error);
 318                         return FALSE;
 319
 320                 case SOUP_SOCKET_WOULD_BLOCK:
 321                         return FALSE;
 322                 }
 323
 324                 if (got_lf) {
 325                         if (!to_blank)
 326                                 break;
 327                         if (nread == 1 &&
 328                             !strncmp ((char *)io->read_meta_buf->data +
 329                                       io->read_meta_buf->len - 2,
 330                                       "\n\n", 2))
 331                                 break;
 332                         else if (nread == 2 &&
 333                                  !strncmp ((char *)io->read_meta_buf->data +
 334                                            io->read_meta_buf->len - 3,
 335                                            "\n\r\n", 3))
 336                                 break;
 337                 }
 338         }
 339
 340         return TRUE;
 341 }
 342
 343 static SoupBuffer *
 344 content_decode_one (SoupBuffer *buf, GConverter *converter, GError **error)
 345 {
 346         gsize outbuf_length, outbuf_used, outbuf_cur, input_used, input_cur;
 347         char *outbuf;
 348         GConverterResult result;
 349
 350         outbuf_length = MAX (buf->length * 2, 1024);
 351         outbuf = g_malloc (outbuf_length);
 352         outbuf_cur = input_cur = 0;
 353
 354         do {
 355                 result = g_converter_convert (
 356                         converter,
 357                         buf->data + input_cur, buf->length - input_cur,
 358                         outbuf + outbuf_cur, outbuf_length - outbuf_cur,
 359                         0, &input_used, &outbuf_used, error);
 360                 input_cur += input_used;
 361                 outbuf_cur += outbuf_used;
 362
 363                 if (g_error_matches (*error, G_IO_ERROR, G_IO_ERROR_NO_SPACE) ||
 364                     (!*error && outbuf_cur == outbuf_length)) {
 365                         g_clear_error (error);
 366                         outbuf_length *= 2;
 367                         outbuf = g_realloc (outbuf, outbuf_length);
 368                 } else if (*error) {
 369                         /* GZlibDecompressor can't ever return
 370                          * G_IO_ERROR_PARTIAL_INPUT unless we pass it
 371                          * input_length = 0, which we don't. Other
 372                          * converters might of course, so eventually
 373                          * this code needs to be rewritten to deal
 374                          * with that.
 375                          */
 376                         g_free (outbuf);
 377                         return NULL;
 378                 }
 379         } while (input_cur < buf->length && result != G_CONVERTER_FINISHED);
 380
 381         if (outbuf_cur)
 382                 return soup_buffer_new (SOUP_MEMORY_TAKE, outbuf, outbuf_cur);
 383         else {
 384                 g_free (outbuf);
 385                 return NULL;
 386         }
 387 }
 388
 389 static SoupBuffer *
 390 content_decode (SoupMessage *msg, SoupBuffer *buf)
 391 {
 392         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 393         GConverter *decoder;
 394         SoupBuffer *decoded;
 395         GError *error = NULL;
 396         GSList *d;
 397
 398         for (d = priv->decoders; d; d = d->next) {
 399                 decoder = d->data;
 400
 401                 decoded = content_decode_one (buf, decoder, &error);
 402                 if (error) {
 403                         if (g_error_matches (error, G_IO_ERROR, G_IO_ERROR_FAILED))
 404                                 g_warning ("Content-Decoding error: %s\n", error->message);
 405                         g_error_free (error);
 406
 407                         soup_message_set_flags (msg, priv->msg_flags & ~SOUP_MESSAGE_CONTENT_DECODED);
 408                         break;
 409                 }
 410                 if (buf)
 411                         soup_buffer_free (buf);
 412
 413                 if (decoded)
 414                         buf = decoded;
 415                 else
 416                         return NULL;
 417         }
 418
 419         return buf;
 420 }
 421
 422 /* Reads as much message body data as is available on io->sock (but no
 423  * further than the end of the current message body or chunk). On a
 424  * successful read, emits "got_chunk" (possibly multiple times), and
 425  * (unless told not to) appends the chunk to io->read_body.
 426  *
 427  * See the note at read_metadata() for an explanation of the return
 428  * value.
 429  */
 430 static gboolean
 431 read_body_chunk (SoupMessage *msg)
 432 {
 433         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 434         SoupMessageIOData *io = priv->io_data;
 435         SoupSocketIOStatus status;
 436         guchar *stack_buf = NULL;
 437         gsize len;
 438         gboolean read_to_eof = (io->read_encoding == SOUP_ENCODING_EOF);
 439         gsize nread;
 440         GError *error = NULL;
 441         SoupBuffer *buffer;
 442
 443         if (!io_handle_sniffing (msg, FALSE))
 444                 return FALSE;
 445
 446         while (read_to_eof || io->read_length > 0) {
 447                 if (priv->chunk_allocator) {
 448                         buffer = priv->chunk_allocator (msg, io->read_length, priv->chunk_allocator_data);
 449                         if (!buffer) {
 450                                 soup_message_io_pause (msg);
 451                                 return FALSE;
 452                         }
 453                 } else {
 454                         if (!stack_buf)
 455                                 stack_buf = alloca (RESPONSE_BLOCK_SIZE);
 456                         buffer = soup_buffer_new (SOUP_MEMORY_TEMPORARY,
 457                                                   stack_buf,
 458                                                   RESPONSE_BLOCK_SIZE);
 459                 }
 460
 461                 if (read_to_eof)
 462                         len = buffer->length;
 463                 else
 464                         len = MIN (buffer->length, io->read_length);
 465
 466                 status = soup_socket_read (io->sock,
 467                                            (guchar *)buffer->data, len,
 468                                            &nread, io->cancellable, &error);
 469
 470                 if (status == SOUP_SOCKET_OK && nread) {
 471                         buffer->length = nread;
 472                         io->read_length -= nread;
 473
 474                         buffer = content_decode (msg, buffer);
 475                         if (!buffer)
 476                                 continue;
 477
 478                         soup_message_body_got_chunk (io->read_body, buffer);
 479
 480                         if (io->need_content_sniffed) {
 481                                 soup_message_body_append_buffer (io->sniff_data, buffer);
 482                                 soup_buffer_free (buffer);
 483                                 io->need_got_chunk = TRUE;
 484                                 if (!io_handle_sniffing (msg, FALSE))
 485                                         return FALSE;
 486                                 continue;
 487                         }
 488
 489                         SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 490                         soup_message_got_chunk (msg, buffer);
 491                         soup_buffer_free (buffer);
 492                         SOUP_MESSAGE_IO_RETURN_VAL_IF_CANCELLED_OR_PAUSED (FALSE);
 493                         continue;
 494                 }
 495
 496                 soup_buffer_free (buffer);
 497                 switch (status) {
 498                 case SOUP_SOCKET_OK:
 499                         break;
 500
 501                 case SOUP_SOCKET_EOF:
 502                         if (io->read_eof_ok) {
 503                                 io->read_length = 0;
 504                                 return TRUE;
 505                         }
 506                         /* else fall through */
 507
 508                 case SOUP_SOCKET_ERROR:
 509                         io_error (io->sock, msg, error);
 510                         return FALSE;
 511
 512                 case SOUP_SOCKET_WOULD_BLOCK:
 513                         return FALSE;
 514                 }
 515         }
 516
 517         return TRUE;
 518 }
 519
 520 /* Attempts to write @len bytes from @data. See the note at
 521  * read_metadata() for an explanation of the return value.
 522  */
 523 static gboolean
 524 write_data (SoupMessage *msg, const char *data, guint len, gboolean body)
 525 {
 526         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 527         SoupMessageIOData *io = priv->io_data;
 528         SoupSocketIOStatus status;
 529         gsize nwrote;
 530         GError *error = NULL;
 531         SoupBuffer *chunk;
 532         const char *start;
 533
 534         while (len > io->written) {
 535                 status = soup_socket_write (io->sock,
 536                                             data + io->written,
 537                                             len - io->written,
 538                                             &nwrote,
 539                                             io->cancellable, &error);
 540                 switch (status) {
 541                 case SOUP_SOCKET_EOF:
 542                 case SOUP_SOCKET_ERROR:
 543                         io_error (io->sock, msg, error);
 544                         return FALSE;
 545
 546                 case SOUP_SOCKET_WOULD_BLOCK:
 547                         return FALSE;
 548
 549                 case SOUP_SOCKET_OK:
 550                         start = data + io->written;
 551                         io->written += nwrote;
 552
 553                         if (body) {
 554                                 if (io->write_length)
 555                                         io->write_length -= nwrote;
 556
 557                                 chunk = soup_buffer_new (SOUP_MEMORY_TEMPORARY,
 558                                                          start, nwrote);
 559                                 SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 560                                 soup_message_wrote_body_data (msg, chunk);
 561                                 soup_buffer_free (chunk);
 562                                 SOUP_MESSAGE_IO_RETURN_VAL_IF_CANCELLED_OR_PAUSED (FALSE);
 563                         }
 564                         break;
 565                 }
 566         }
 567
 568         io->written = 0;
 569         return TRUE;
 570 }
 571
 572 static inline SoupMessageIOState
 573 io_body_state (SoupEncoding encoding)
 574 {
 575         if (encoding == SOUP_ENCODING_CHUNKED)
 576                 return SOUP_MESSAGE_IO_STATE_CHUNK_SIZE;
 577         else
 578                 return SOUP_MESSAGE_IO_STATE_BODY;
 579 }
 580
 581 /*
 582  * There are two request/response formats: the basic request/response,
 583  * possibly with one or more unsolicited informational responses (such
 584  * as the WebDAV "102 Processing" response):
 585  *
 586  *     Client                            Server
 587  *      W:HEADERS  / R:NOT_STARTED    ->  R:HEADERS  / W:NOT_STARTED
 588  *      W:BODY     / R:NOT_STARTED    ->  R:BODY     / W:NOT_STARTED
 589  *     [W:DONE     / R:HEADERS (1xx)  <-  R:DONE     / W:HEADERS (1xx) ...]
 590  *      W:DONE     / R:HEADERS        <-  R:DONE     / W:HEADERS
 591  *      W:DONE     / R:BODY           <-  R:DONE     / W:BODY
 592  *      W:DONE     / R:DONE               R:DONE     / W:DONE
 593  *
 594  * and the "Expect: 100-continue" request/response, with the client
 595  * blocking halfway through its request, and then either continuing or
 596  * aborting, depending on the server response:
 597  *
 598  *     Client                            Server
 599  *      W:HEADERS  / R:NOT_STARTED    ->  R:HEADERS  / W:NOT_STARTED
 600  *      W:BLOCKING / R:HEADERS        <-  R:BLOCKING / W:HEADERS
 601  *     [W:BODY     / R:BLOCKING       ->  R:BODY     / W:BLOCKING]
 602  *     [W:DONE     / R:HEADERS        <-  R:DONE     / W:HEADERS]
 603  *      W:DONE     / R:BODY           <-  R:DONE     / W:BODY
 604  *      W:DONE     / R:DONE               R:DONE     / W:DONE
 605  */
 606
 607 static void
 608 io_write (SoupSocket *sock, SoupMessage *msg)
 609 {
 610         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 611         SoupMessageIOData *io = priv->io_data;
 612
 613  write_more:
 614         switch (io->write_state) {
 615         case SOUP_MESSAGE_IO_STATE_NOT_STARTED:
 616                 return;
 617
 618
 619         case SOUP_MESSAGE_IO_STATE_HEADERS:
 620                 if (!io->write_buf->len) {
 621                         io->get_headers_cb (msg, io->write_buf,
 622                                             &io->write_encoding,
 623                                             io->header_data);
 624                         if (!io->write_buf->len) {
 625                                 soup_message_io_pause (msg);
 626                                 return;
 627                         }
 628                 }
 629
 630                 if (!write_data (msg, io->write_buf->str,
 631                                  io->write_buf->len, FALSE))
 632                         return;
 633
 634                 g_string_truncate (io->write_buf, 0);
 635
 636                 if (io->write_encoding == SOUP_ENCODING_CONTENT_LENGTH) {
 637                         SoupMessageHeaders *hdrs =
 638                                 (io->mode == SOUP_MESSAGE_IO_CLIENT) ?
 639                                 msg->request_headers : msg->response_headers;
 640                         io->write_length = soup_message_headers_get_content_length (hdrs);
 641                 }
 642
 643                 if (io->mode == SOUP_MESSAGE_IO_SERVER &&
 644                     SOUP_STATUS_IS_INFORMATIONAL (msg->status_code)) {
 645                         if (msg->status_code == SOUP_STATUS_CONTINUE) {
 646                                 /* Stop and wait for the body now */
 647                                 io->write_state =
 648                                         SOUP_MESSAGE_IO_STATE_BLOCKING;
 649                                 io->read_state = io_body_state (io->read_encoding);
 650                         } else {
 651                                 /* We just wrote a 1xx response
 652                                  * header, so stay in STATE_HEADERS.
 653                                  * (The caller will pause us from the
 654                                  * wrote_informational callback if he
 655                                  * is not ready to send the final
 656                                  * response.)
 657                                  */
 658                         }
 659                 } else if (io->mode == SOUP_MESSAGE_IO_CLIENT &&
 660                            soup_message_headers_get_expectations (msg->request_headers) & SOUP_EXPECTATION_CONTINUE) {
 661                         /* Need to wait for the Continue response */
 662                         io->write_state = SOUP_MESSAGE_IO_STATE_BLOCKING;
 663                         io->read_state = SOUP_MESSAGE_IO_STATE_HEADERS;
 664                 } else {
 665                         io->write_state = io_body_state (io->write_encoding);
 666
 667                         /* If the client was waiting for a Continue
 668                          * but we sent something else, then they're
 669                          * now done writing.
 670                          */
 671                         if (io->mode == SOUP_MESSAGE_IO_SERVER &&
 672                             io->read_state == SOUP_MESSAGE_IO_STATE_BLOCKING)
 673                                 io->read_state = SOUP_MESSAGE_IO_STATE_FINISHING;
 674                 }
 675
 676                 SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 677                 if (SOUP_STATUS_IS_INFORMATIONAL (msg->status_code)) {
 678                         soup_message_wrote_informational (msg);
 679                         soup_message_cleanup_response (msg);
 680                 } else
 681                         soup_message_wrote_headers (msg);
 682                 SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED;
 683                 break;
 684
 685
 686         case SOUP_MESSAGE_IO_STATE_BLOCKING:
 687                 io_read (sock, msg);
 688
 689                 /* If io_read reached a point where we could write
 690                  * again, it would have recursively called io_write.
 691                  * So (a) we don't need to try to keep writing, and
 692                  * (b) we can't anyway, because msg may have been
 693                  * destroyed.
 694                  */
 695                 return;
 696
 697
 698         case SOUP_MESSAGE_IO_STATE_BODY:
 699                 if (!io->write_length && io->write_encoding != SOUP_ENCODING_EOF) {
 700                 wrote_body:
 701                         io->write_state = SOUP_MESSAGE_IO_STATE_FINISHING;
 702
 703                         SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 704                         soup_message_wrote_body (msg);
 705                         SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED;
 706                         break;
 707                 }
 708
 709                 if (!io->write_chunk) {
 710                         io->write_chunk = soup_message_body_get_chunk (io->write_body, io->write_body_offset);
 711                         if (!io->write_chunk) {
 712                                 soup_message_io_pause (msg);
 713                                 return;
 714                         }
 715                         if (io->write_chunk->length > io->write_length &&
 716                             io->write_encoding != SOUP_ENCODING_EOF) {
 717                                 /* App is trying to write more than it
 718                                  * claimed it would; we have to truncate.
 719                                  */
 720                                 SoupBuffer *truncated =
 721                                         soup_buffer_new_subbuffer (io->write_chunk,
 722                                                                    0, io->write_length);
 723                                 soup_buffer_free (io->write_chunk);
 724                                 io->write_chunk = truncated;
 725                         } else if (io->write_encoding == SOUP_ENCODING_EOF &&
 726                                    !io->write_chunk->length)
 727                                 goto wrote_body;
 728                 }
 729
 730                 if (!write_data (msg, io->write_chunk->data,
 731                                  io->write_chunk->length, TRUE))
 732                         return;
 733
 734                 if (io->mode == SOUP_MESSAGE_IO_SERVER)
 735                         soup_message_body_wrote_chunk (io->write_body, io->write_chunk);
 736                 io->write_body_offset += io->write_chunk->length;
 737                 soup_buffer_free (io->write_chunk);
 738                 io->write_chunk = NULL;
 739
 740                 SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 741                 soup_message_wrote_chunk (msg);
 742                 SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED;
 743                 break;
 744
 745         case SOUP_MESSAGE_IO_STATE_CHUNK_SIZE:
 746                 if (!io->write_chunk) {
 747                         io->write_chunk = soup_message_body_get_chunk (io->write_body, io->write_body_offset);
 748                         if (!io->write_chunk) {
 749                                 soup_message_io_pause (msg);
 750                                 return;
 751                         }
 752                         g_string_append_printf (io->write_buf, "%lx\r\n",
 753                                                 (unsigned long) io->write_chunk->length);
 754                         io->write_body_offset += io->write_chunk->length;
 755                 }
 756
 757                 if (!write_data (msg, io->write_buf->str,
 758                                  io->write_buf->len, FALSE))
 759                         return;
 760
 761                 g_string_truncate (io->write_buf, 0);
 762
 763                 if (io->write_chunk->length == 0) {
 764                         /* The last chunk has no CHUNK_END... */
 765                         io->write_state = SOUP_MESSAGE_IO_STATE_TRAILERS;
 766                         break;
 767                 }
 768
 769                 io->write_state = SOUP_MESSAGE_IO_STATE_CHUNK;
 770                 /* fall through */
 771
 772
 773         case SOUP_MESSAGE_IO_STATE_CHUNK:
 774                 if (!write_data (msg, io->write_chunk->data,
 775                                  io->write_chunk->length, TRUE))
 776                         return;
 777
 778                 if (io->mode == SOUP_MESSAGE_IO_SERVER)
 779                         soup_message_body_wrote_chunk (io->write_body, io->write_chunk);
 780                 soup_buffer_free (io->write_chunk);
 781                 io->write_chunk = NULL;
 782
 783                 io->write_state = SOUP_MESSAGE_IO_STATE_CHUNK_END;
 784
 785                 SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 786                 soup_message_wrote_chunk (msg);
 787                 SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED;
 788
 789                 /* fall through */
 790
 791
 792         case SOUP_MESSAGE_IO_STATE_CHUNK_END:
 793                 if (!write_data (msg, SOUP_MESSAGE_IO_EOL,
 794                                  SOUP_MESSAGE_IO_EOL_LEN, FALSE))
 795                         return;
 796
 797                 io->write_state = SOUP_MESSAGE_IO_STATE_CHUNK_SIZE;
 798                 break;
 799
 800
 801         case SOUP_MESSAGE_IO_STATE_TRAILERS:
 802                 if (!write_data (msg, SOUP_MESSAGE_IO_EOL,
 803                                  SOUP_MESSAGE_IO_EOL_LEN, FALSE))
 804                         return;
 805
 806                 io->write_state = SOUP_MESSAGE_IO_STATE_FINISHING;
 807
 808                 SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 809                 soup_message_wrote_body (msg);
 810                 SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED;
 811                 /* fall through */
 812
 813
 814         case SOUP_MESSAGE_IO_STATE_FINISHING:
 815                 if (io->write_tag) {
 816                         g_signal_handler_disconnect (io->sock, io->write_tag);
 817                         io->write_tag = 0;
 818                 }
 819                 io->write_state = SOUP_MESSAGE_IO_STATE_DONE;
 820
 821                 if (io->mode == SOUP_MESSAGE_IO_CLIENT) {
 822                         io->read_state = SOUP_MESSAGE_IO_STATE_HEADERS;
 823                         io_read (sock, msg);
 824                 } else
 825                         soup_message_io_finished (msg);
 826                 return;
 827
 828
 829         case SOUP_MESSAGE_IO_STATE_DONE:
 830         default:
 831                 g_return_if_reached ();
 832         }
 833
 834         goto write_more;
 835 }
 836
 837 static void
 838 io_read (SoupSocket *sock, SoupMessage *msg)
 839 {
 840         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
 841         SoupMessageIOData *io = priv->io_data;
 842         guint status;
 843
 844  read_more:
 845         switch (io->read_state) {
 846         case SOUP_MESSAGE_IO_STATE_NOT_STARTED:
 847                 return;
 848
 849
 850         case SOUP_MESSAGE_IO_STATE_HEADERS:
 851                 if (!read_metadata (msg, TRUE))
 852                         return;
 853
 854                 /* We need to "rewind" io->read_meta_buf back one line.
 855                  * That SHOULD be two characters (CR LF), but if the
 856                  * web server was stupid, it might only be one.
 857                  */
 858                 if (io->read_meta_buf->len < 3 ||
 859                     io->read_meta_buf->data[io->read_meta_buf->len - 2] == '\n')
 860                         io->read_meta_buf->len--;
 861                 else
 862                         io->read_meta_buf->len -= 2;
 863                 io->read_meta_buf->data[io->read_meta_buf->len] = '\0';
 864                 status = io->parse_headers_cb (msg, (char *)io->read_meta_buf->data,
 865                                                io->read_meta_buf->len,
 866                                                &io->read_encoding,
 867                                                io->header_data);
 868                 g_byte_array_set_size (io->read_meta_buf, 0);
 869
 870                 if (status != SOUP_STATUS_OK) {
 871                         /* Either we couldn't parse the headers, or they
 872                          * indicated something that would mean we wouldn't
 873                          * be able to parse the body. (Eg, unknown
 874                          * Transfer-Encoding.). Skip the rest of the
 875                          * reading, and make sure the connection gets
 876                          * closed when we're done.
 877                          */
 878                         soup_message_set_status (msg, status);
 879                         soup_message_headers_append (msg->request_headers,
 880                                                      "Connection", "close");
 881                         io->read_state = SOUP_MESSAGE_IO_STATE_FINISHING;
 882                         break;
 883                 }
 884
 885                 if (io->read_encoding == SOUP_ENCODING_EOF)
 886                         io->read_eof_ok = TRUE;
 887
 888                 if (io->read_encoding == SOUP_ENCODING_CONTENT_LENGTH) {
 889                         SoupMessageHeaders *hdrs =
 890                                 (io->mode == SOUP_MESSAGE_IO_CLIENT) ?
 891                                 msg->response_headers : msg->request_headers;
 892                         io->read_length = soup_message_headers_get_content_length (hdrs);
 893
 894                         if (io->mode == SOUP_MESSAGE_IO_CLIENT &&
 895                             !soup_message_is_keepalive (msg)) {
 896                                 /* Some servers suck and send
 897                                  * incorrect Content-Length values, so
 898                                  * allow EOF termination in this case
 899                                  * (iff the message is too short) too.
 900                                  */
 901                                 io->read_eof_ok = TRUE;
 902                         }
 903                 }
 904
 905                 if (io->mode == SOUP_MESSAGE_IO_CLIENT &&
 906                     SOUP_STATUS_IS_INFORMATIONAL (msg->status_code)) {
 907                         if (msg->status_code == SOUP_STATUS_CONTINUE &&
 908                             io->write_state == SOUP_MESSAGE_IO_STATE_BLOCKING) {
 909                                 /* Pause the reader, unpause the writer */
 910                                 io->read_state =
 911                                         SOUP_MESSAGE_IO_STATE_BLOCKING;
 912                                 io->write_state =
 913                                         io_body_state (io->write_encoding);
 914                         } else {
 915                                 /* Just stay in HEADERS */
 916                                 io->read_state = SOUP_MESSAGE_IO_STATE_HEADERS;
 917                         }
 918                 } else if (io->mode == SOUP_MESSAGE_IO_SERVER &&
 919                            soup_message_headers_get_expectations (msg->request_headers) & SOUP_EXPECTATION_CONTINUE) {
 920                         /* The client requested a Continue response. The
 921                          * got_headers handler may change this to something
 922                          * else though.
 923                          */
 924                         soup_message_set_status (msg, SOUP_STATUS_CONTINUE);
 925                         io->write_state = SOUP_MESSAGE_IO_STATE_HEADERS;
 926                         io->read_state = SOUP_MESSAGE_IO_STATE_BLOCKING;
 927                 } else {
 928                         io->read_state = io_body_state (io->read_encoding);
 929
 930                         /* If the client was waiting for a Continue
 931                          * but got something else, then it's done
 932                          * writing.
 933                          */
 934                         if (io->mode == SOUP_MESSAGE_IO_CLIENT &&
 935                             io->write_state == SOUP_MESSAGE_IO_STATE_BLOCKING)
 936                                 io->write_state = SOUP_MESSAGE_IO_STATE_FINISHING;
 937                 }
 938
 939                 if (io->mode == SOUP_MESSAGE_IO_CLIENT &&
 940                     SOUP_STATUS_IS_INFORMATIONAL (msg->status_code)) {
 941                         SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 942                         soup_message_got_informational (msg);
 943                         soup_message_cleanup_response (msg);
 944                         SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED;
 945                 } else {
 946                         SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 947                         soup_message_got_headers (msg);
 948                         SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED;
 949                 }
 950                 break;
 951
 952
 953         case SOUP_MESSAGE_IO_STATE_BLOCKING:
 954                 io_write (sock, msg);
 955
 956                 /* As in the io_write case, we *must* return here. */
 957                 return;
 958
 959
 960         case SOUP_MESSAGE_IO_STATE_BODY:
 961                 if (!read_body_chunk (msg))
 962                         return;
 963
 964         got_body:
 965                 if (!io_handle_sniffing (msg, TRUE)) {
 966                         /* If the message was paused (as opposed to
 967                          * cancelled), we need to make sure we wind up
 968                          * back here when it's unpaused, even if it
 969                          * was doing a chunked or EOF-terminated read
 970                          * before.
 971                          */
 972                         if (io == priv->io_data) {
 973                                 io->read_state = SOUP_MESSAGE_IO_STATE_BODY;
 974                                 io->read_encoding = SOUP_ENCODING_CONTENT_LENGTH;
 975                                 io->read_length = 0;
 976                         }
 977                         return;
 978                 }
 979
 980                 io->read_state = SOUP_MESSAGE_IO_STATE_FINISHING;
 981
 982                 SOUP_MESSAGE_IO_PREPARE_FOR_CALLBACK;
 983                 soup_message_got_body (msg);
 984                 SOUP_MESSAGE_IO_RETURN_IF_CANCELLED_OR_PAUSED;
 985                 break;
 986
 987
 988         case SOUP_MESSAGE_IO_STATE_CHUNK_SIZE:
 989                 if (!read_metadata (msg, FALSE))
 990                         return;
 991
 992                 io->read_length = strtoul ((char *)io->read_meta_buf->data, NULL, 16);
 993                 g_byte_array_set_size (io->read_meta_buf, 0);
 994
 995                 if (io->read_length > 0)
 996                         io->read_state = SOUP_MESSAGE_IO_STATE_CHUNK;
 997                 else
 998                         io->read_state = SOUP_MESSAGE_IO_STATE_TRAILERS;
 999                 break;
1000
1001
1002         case SOUP_MESSAGE_IO_STATE_CHUNK:
1003                 if (!read_body_chunk (msg))
1004                         return;
1005
1006                 io->read_state = SOUP_MESSAGE_IO_STATE_CHUNK_END;
1007                 break;
1008
1009
1010         case SOUP_MESSAGE_IO_STATE_CHUNK_END:
1011                 if (!read_metadata (msg, FALSE))
1012                         return;
1013
1014                 g_byte_array_set_size (io->read_meta_buf, 0);
1015                 io->read_state = SOUP_MESSAGE_IO_STATE_CHUNK_SIZE;
1016                 break;
1017
1018
1019         case SOUP_MESSAGE_IO_STATE_TRAILERS:
1020                 if (!read_metadata (msg, FALSE))
1021                         return;
1022
1023                 if (io->read_meta_buf->len <= SOUP_MESSAGE_IO_EOL_LEN)
1024                         goto got_body;
1025
1026                 /* FIXME: process trailers */
1027                 g_byte_array_set_size (io->read_meta_buf, 0);
1028                 break;
1029
1030
1031         case SOUP_MESSAGE_IO_STATE_FINISHING:
1032                 if (io->read_tag) {
1033                         g_signal_handler_disconnect (io->sock, io->read_tag);
1034                         io->read_tag = 0;
1035                 }
1036                 io->read_state = SOUP_MESSAGE_IO_STATE_DONE;
1037
1038                 if (io->mode == SOUP_MESSAGE_IO_SERVER) {
1039                         io->write_state = SOUP_MESSAGE_IO_STATE_HEADERS;
1040                         io_write (sock, msg);
1041                 } else
1042                         soup_message_io_finished (msg);
1043                 return;
1044
1045
1046         case SOUP_MESSAGE_IO_STATE_DONE:
1047         default:
1048                 g_return_if_reached ();
1049         }
1050
1051         goto read_more;
1052 }
1053
1054 static void
1055 socket_tls_certificate_changed (GObject *sock, GParamSpec *pspec,
1056                                 gpointer msg)
1057 {
1058         GTlsCertificate *certificate;
1059         GTlsCertificateFlags errors;
1060
1061         g_object_get (sock,
1062                       SOUP_SOCKET_TLS_CERTIFICATE, &certificate,
1063                       SOUP_SOCKET_TLS_ERRORS, &errors,
1064                       NULL);
1065         g_object_set (msg,
1066                       SOUP_MESSAGE_TLS_CERTIFICATE, certificate,
1067                       SOUP_MESSAGE_TLS_ERRORS, errors,
1068                       NULL);
1069         if (certificate)
1070                 g_object_unref (certificate);
1071 }
1072
1073 static SoupMessageIOData *
1074 new_iostate (SoupMessage *msg, SoupSocket *sock, SoupMessageIOMode mode,
1075              SoupMessageGetHeadersFn get_headers_cb,
1076              SoupMessageParseHeadersFn parse_headers_cb,
1077              gpointer header_data,
1078              SoupMessageCompletionFn completion_cb,
1079              gpointer completion_data)
1080 {
1081         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
1082         SoupMessageIOData *io;
1083
1084         io = g_slice_new0 (SoupMessageIOData);
1085         io->sock = g_object_ref (sock);
1086         io->mode = mode;
1087         io->get_headers_cb   = get_headers_cb;
1088         io->parse_headers_cb = parse_headers_cb;
1089         io->header_data      = header_data;
1090         io->completion_cb    = completion_cb;
1091         io->completion_data  = completion_data;
1092
1093         io->read_meta_buf    = g_byte_array_new ();
1094         io->write_buf        = g_string_new (NULL);
1095
1096         io->read_tag  = g_signal_connect (io->sock, "readable",
1097                                           G_CALLBACK (io_read), msg);
1098         io->write_tag = g_signal_connect (io->sock, "writable",
1099                                           G_CALLBACK (io_write), msg);
1100
1101         io->read_state  = SOUP_MESSAGE_IO_STATE_NOT_STARTED;
1102         io->write_state = SOUP_MESSAGE_IO_STATE_NOT_STARTED;
1103
1104         if (soup_socket_is_ssl (io->sock)) {
1105                 io->tls_signal_id = g_signal_connect (io->sock, "notify::tls-certificate",
1106                                                       G_CALLBACK (socket_tls_certificate_changed), msg);
1107         }
1108
1109         if (priv->io_data)
1110                 soup_message_io_cleanup (msg);
1111         priv->io_data = io;
1112         return io;
1113 }
1114
1115 void
1116 soup_message_io_client (SoupMessageQueueItem *item,
1117                         SoupMessageGetHeadersFn get_headers_cb,
1118                         SoupMessageParseHeadersFn parse_headers_cb,
1119                         gpointer header_data,
1120                         SoupMessageCompletionFn completion_cb,
1121                         gpointer completion_data)
1122 {
1123         SoupMessageIOData *io;
1124         SoupSocket *sock = soup_connection_get_socket (item->conn);
1125
1126         io = new_iostate (item->msg, sock, SOUP_MESSAGE_IO_CLIENT,
1127                           get_headers_cb, parse_headers_cb, header_data,
1128                           completion_cb, completion_data);
1129
1130         io->item = item;
1131         soup_message_queue_item_ref (item);
1132         io->cancellable = item->cancellable;
1133
1134         io->read_body       = item->msg->response_body;
1135         io->write_body      = item->msg->request_body;
1136
1137         io->write_state     = SOUP_MESSAGE_IO_STATE_HEADERS;
1138         io_write (sock, item->msg);
1139 }
1140
1141 void
1142 soup_message_io_server (SoupMessage *msg, SoupSocket *sock,
1143                         SoupMessageGetHeadersFn get_headers_cb,
1144                         SoupMessageParseHeadersFn parse_headers_cb,
1145                         gpointer header_data,
1146                         SoupMessageCompletionFn completion_cb,
1147                         gpointer completion_data)
1148 {
1149         SoupMessageIOData *io;
1150
1151         io = new_iostate (msg, sock, SOUP_MESSAGE_IO_SERVER,
1152                           get_headers_cb, parse_headers_cb, header_data,
1153                           completion_cb, completion_data);
1154
1155         io->read_body       = msg->request_body;
1156         io->write_body      = msg->response_body;
1157
1158         io->read_state      = SOUP_MESSAGE_IO_STATE_HEADERS;
1159         io_read (sock, msg);
1160 }
1161
1162 void
1163 soup_message_io_pause (SoupMessage *msg)
1164 {
1165         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
1166         SoupMessageIOData *io = priv->io_data;
1167
1168         g_return_if_fail (io != NULL);
1169
1170         if (io->write_tag) {
1171                 g_signal_handler_disconnect (io->sock, io->write_tag);
1172                 io->write_tag = 0;
1173         }
1174         if (io->read_tag) {
1175                 g_signal_handler_disconnect (io->sock, io->read_tag);
1176                 io->read_tag = 0;
1177         }
1178
1179         if (io->unpause_source) {
1180                 g_source_destroy (io->unpause_source);
1181                 io->unpause_source = NULL;
1182         }
1183 }
1184
1185 static gboolean
1186 io_unpause_internal (gpointer msg)
1187 {
1188         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
1189         SoupMessageIOData *io = priv->io_data;
1190
1191         g_return_val_if_fail (io != NULL, FALSE);
1192         io->unpause_source = NULL;
1193
1194         if (io->write_tag || io->read_tag)
1195                 return FALSE;
1196
1197         if (io->write_state != SOUP_MESSAGE_IO_STATE_DONE) {
1198                 io->write_tag = g_signal_connect (io->sock, "writable",
1199                                                   G_CALLBACK (io_write), msg);
1200         }
1201
1202         if (io->read_state != SOUP_MESSAGE_IO_STATE_DONE) {
1203                 io->read_tag = g_signal_connect (io->sock, "readable",
1204                                                  G_CALLBACK (io_read), msg);
1205         }
1206
1207         if (SOUP_MESSAGE_IO_STATE_ACTIVE (io->write_state))
1208                 io_write (io->sock, msg);
1209         else if (SOUP_MESSAGE_IO_STATE_ACTIVE (io->read_state))
1210                 io_read (io->sock, msg);
1211
1212         return FALSE;
1213 }
1214
1215 void
1216 soup_message_io_unpause (SoupMessage *msg)
1217 {
1218         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
1219         SoupMessageIOData *io = priv->io_data;
1220         gboolean non_blocking;
1221         GMainContext *async_context;
1222
1223         g_return_if_fail (io != NULL);
1224
1225         g_object_get (io->sock,
1226                       SOUP_SOCKET_FLAG_NONBLOCKING, &non_blocking,
1227                       SOUP_SOCKET_ASYNC_CONTEXT, &async_context,
1228                       NULL);
1229         if (non_blocking) {
1230                 if (!io->unpause_source) {
1231                         io->unpause_source = soup_add_completion (
1232                                 async_context, io_unpause_internal, msg);
1233                 }
1234         } else
1235                 io_unpause_internal (msg);
1236         if (async_context)
1237                 g_main_context_unref (async_context);
1238 }
1239
1240 /**
1241  * soup_message_io_in_progress:
1242  * @msg: a #SoupMessage
1243  *
1244  * Tests whether or not I/O is currently in progress on @msg.
1245  *
1246  * Return value: whether or not I/O is currently in progress.
1247  **/
1248 gboolean
1249 soup_message_io_in_progress (SoupMessage *msg)
1250 {
1251         SoupMessagePrivate *priv = SOUP_MESSAGE_GET_PRIVATE (msg);
1252
1253         return priv->io_data != NULL;
1254 }