contrib/serf/outgoing.c

   1 /* Copyright 2002-2004 Justin Erenkrantz and Greg Stein
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at
   6  *
   7  *     http://www.apache.org/licenses/LICENSE-2.0
   8  *
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #include <apr_pools.h>
  17 #include <apr_poll.h>
  18 #include <apr_version.h>
  19 #include <apr_portable.h>
  20
  21 #include "serf.h"
  22 #include "serf_bucket_util.h"
  23
  24 #include "serf_private.h"
  25
  26 /* cleanup for sockets */
  27 static apr_status_t clean_skt(void *data)
  28 {
  29     serf_connection_t *conn = data;
  30     apr_status_t status = APR_SUCCESS;
  31
  32     if (conn->skt) {
  33         serf__log_skt(SOCK_VERBOSE, __FILE__, conn->skt, "cleanup - ");
  34         status = apr_socket_close(conn->skt);
  35         conn->skt = NULL;
  36         serf__log_nopref(SOCK_VERBOSE, "closed socket, status %d\n", status);
  37     }
  38
  39     return status;
  40 }
  41
  42 static apr_status_t clean_resp(void *data)
  43 {
  44     serf_request_t *request = data;
  45
  46     /* The request's RESPOOL is being cleared.  */
  47
  48     /* If the response has allocated some buckets, then destroy them (since
  49        the bucket may hold resources other than memory in RESPOOL). Also
  50        make sure to set their fields to NULL so connection closure does
  51        not attempt to free them again.  */
  52     if (request->resp_bkt) {
  53         serf_bucket_destroy(request->resp_bkt);
  54         request->resp_bkt = NULL;
  55     }
  56     if (request->req_bkt) {
  57         serf_bucket_destroy(request->req_bkt);
  58         request->req_bkt = NULL;
  59     }
  60
  61     /* ### should we worry about debug stuff, like that performed in
  62        ### destroy_request()? should we worry about calling req->handler
  63        ### to notify this "cancellation" due to pool clearing?  */
  64
  65     /* This pool just got cleared/destroyed. Don't try to destroy the pool
  66        (again) when the request is canceled.  */
  67     request->respool = NULL;
  68
  69     return APR_SUCCESS;
  70 }
  71
  72 /* cleanup for conns */
  73 static apr_status_t clean_conn(void *data)
  74 {
  75     serf_connection_t *conn = data;
  76
  77     serf__log(CONN_VERBOSE, __FILE__, "cleaning up connection 0x%x\n",
  78               conn);
  79     serf_connection_close(conn);
  80
  81     return APR_SUCCESS;
  82 }
  83
  84 /* Update the pollset for this connection. We tweak the pollset based on
  85  * whether we want to read and/or write, given conditions within the
  86  * connection. If the connection is not (yet) in the pollset, then it
  87  * will be added.
  88  */
  89 apr_status_t serf__conn_update_pollset(serf_connection_t *conn)
  90 {
  91     serf_context_t *ctx = conn->ctx;
  92     apr_status_t status;
  93     apr_pollfd_t desc = { 0 };
  94
  95     if (!conn->skt) {
  96         return APR_SUCCESS;
  97     }
  98
  99     /* Remove the socket from the poll set. */
 100     desc.desc_type = APR_POLL_SOCKET;
 101     desc.desc.s = conn->skt;
 102     desc.reqevents = conn->reqevents;
 103
 104     status = ctx->pollset_rm(ctx->pollset_baton,
 105                              &desc, conn);
 106     if (status && !APR_STATUS_IS_NOTFOUND(status))
 107         return status;
 108
 109     /* Now put it back in with the correct read/write values. */
 110     desc.reqevents = APR_POLLHUP | APR_POLLERR;
 111     if (conn->requests &&
 112         conn->state != SERF_CONN_INIT) {
 113         /* If there are any outstanding events, then we want to read. */
 114         /* ### not true. we only want to read IF we have sent some data */
 115         desc.reqevents |= APR_POLLIN;
 116
 117         /* Don't write if OpenSSL told us that it needs to read data first. */
 118         if (conn->stop_writing != 1) {
 119
 120             /* If the connection is not closing down and
 121              *   has unwritten data or
 122              *   there are any requests that still have buckets to write out,
 123              *     then we want to write.
 124              */
 125             if (conn->vec_len &&
 126                 conn->state != SERF_CONN_CLOSING)
 127                 desc.reqevents |= APR_POLLOUT;
 128             else {
 129                 serf_request_t *request = conn->requests;
 130
 131                 if ((conn->probable_keepalive_limit &&
 132                      conn->completed_requests > conn->probable_keepalive_limit) ||
 133                     (conn->max_outstanding_requests &&
 134                      conn->completed_requests - conn->completed_responses >=
 135                      conn->max_outstanding_requests)) {
 136                         /* we wouldn't try to write any way right now. */
 137                     }
 138                 else {
 139                     while (request != NULL && request->req_bkt == NULL &&
 140                            request->written)
 141                         request = request->next;
 142                     if (request != NULL)
 143                         desc.reqevents |= APR_POLLOUT;
 144                 }
 145             }
 146         }
 147     }
 148
 149     /* If we can have async responses, always look for something to read. */
 150     if (conn->async_responses) {
 151         desc.reqevents |= APR_POLLIN;
 152     }
 153
 154     /* save our reqevents, so we can pass it in to remove later. */
 155     conn->reqevents = desc.reqevents;
 156
 157     /* Note: even if we don't want to read/write this socket, we still
 158      * want to poll it for hangups and errors.
 159      */
 160     return ctx->pollset_add(ctx->pollset_baton,
 161                             &desc, &conn->baton);
 162 }
 163
 164 #ifdef SERF_DEBUG_BUCKET_USE
 165
 166 /* Make sure all response buckets were drained. */
 167 static void check_buckets_drained(serf_connection_t *conn)
 168 {
 169     serf_request_t *request = conn->requests;
 170
 171     for ( ; request ; request = request->next ) {
 172         if (request->resp_bkt != NULL) {
 173             /* ### crap. can't do this. this allocator may have un-drained
 174              * ### REQUEST buckets.
 175              */
 176             /* serf_debug__entered_loop(request->resp_bkt->allocator); */
 177             /* ### for now, pretend we closed the conn (resets the tracking) */
 178             serf_debug__closed_conn(request->resp_bkt->allocator);
 179         }
 180     }
 181 }
 182
 183 #endif
 184
 185 static void destroy_ostream(serf_connection_t *conn)
 186 {
 187     if (conn->ostream_head != NULL) {
 188         serf_bucket_destroy(conn->ostream_head);
 189         conn->ostream_head = NULL;
 190         conn->ostream_tail = NULL;
 191     }
 192 }
 193
 194 static apr_status_t detect_eof(void *baton, serf_bucket_t *aggregate_bucket)
 195 {
 196     serf_connection_t *conn = baton;
 197     conn->hit_eof = 1;
 198     return APR_EAGAIN;
 199 }
 200
 201 static apr_status_t do_conn_setup(serf_connection_t *conn)
 202 {
 203     apr_status_t status;
 204     serf_bucket_t *ostream;
 205
 206     if (conn->ostream_head == NULL) {
 207         conn->ostream_head = serf_bucket_aggregate_create(conn->allocator);
 208     }
 209
 210     if (conn->ostream_tail == NULL) {
 211         conn->ostream_tail = serf__bucket_stream_create(conn->allocator,
 212                                                         detect_eof,
 213                                                         conn);
 214     }
 215
 216     ostream = conn->ostream_tail;
 217
 218     status = (*conn->setup)(conn->skt,
 219                             &conn->stream,
 220                             &ostream,
 221                             conn->setup_baton,
 222                             conn->pool);
 223     if (status) {
 224         /* extra destroy here since it wasn't added to the head bucket yet. */
 225         serf_bucket_destroy(conn->ostream_tail);
 226         destroy_ostream(conn);
 227         return status;
 228     }
 229
 230     serf_bucket_aggregate_append(conn->ostream_head,
 231                                  ostream);
 232
 233     return status;
 234 }
 235
 236 /* Set up the input and output stream buckets.
 237  When a tunnel over an http proxy is needed, create a socket bucket and
 238  empty aggregate bucket for sending and receiving unencrypted requests
 239  over the socket.
 240
 241  After the tunnel is there, or no tunnel was needed, ask the application
 242  to create the input and output buckets, which should take care of the
 243  [en/de]cryption.
 244  */
 245
 246 static apr_status_t prepare_conn_streams(serf_connection_t *conn,
 247                                          serf_bucket_t **istream,
 248                                          serf_bucket_t **ostreamt,
 249                                          serf_bucket_t **ostreamh)
 250 {
 251     apr_status_t status;
 252
 253     if (conn->stream == NULL) {
 254         conn->latency = apr_time_now() - conn->connect_time;
 255     }
 256
 257     /* Do we need a SSL tunnel first? */
 258     if (conn->state == SERF_CONN_CONNECTED) {
 259         /* If the connection does not have an associated bucket, then
 260          * call the setup callback to get one.
 261          */
 262         if (conn->stream == NULL) {
 263             status = do_conn_setup(conn);
 264             if (status) {
 265                 return status;
 266             }
 267         }
 268         *ostreamt = conn->ostream_tail;
 269         *ostreamh = conn->ostream_head;
 270         *istream = conn->stream;
 271     } else {
 272         /* SSL tunnel needed and not set up yet, get a direct unencrypted
 273          stream for this socket */
 274         if (conn->stream == NULL) {
 275             *istream = serf_bucket_socket_create(conn->skt,
 276                                                  conn->allocator);
 277         }
 278         /* Don't create the ostream bucket chain including the ssl_encrypt
 279          bucket yet. This ensure the CONNECT request is sent unencrypted
 280          to the proxy. */
 281         *ostreamt = *ostreamh = conn->ssltunnel_ostream;
 282     }
 283
 284     return APR_SUCCESS;
 285 }
 286
 287 /* Create and connect sockets for any connections which don't have them
 288  * yet. This is the core of our lazy-connect behavior.
 289  */
 290 apr_status_t serf__open_connections(serf_context_t *ctx)
 291 {
 292     int i;
 293
 294     for (i = ctx->conns->nelts; i--; ) {
 295         serf_connection_t *conn = GET_CONN(ctx, i);
 296         serf__authn_info_t *authn_info;
 297         apr_status_t status;
 298         apr_socket_t *skt;
 299
 300         conn->seen_in_pollset = 0;
 301
 302         if (conn->skt != NULL) {
 303 #ifdef SERF_DEBUG_BUCKET_USE
 304             check_buckets_drained(conn);
 305 #endif
 306             continue;
 307         }
 308
 309         /* Delay opening until we have something to deliver! */
 310         if (conn->requests == NULL) {
 311             continue;
 312         }
 313
 314         apr_pool_clear(conn->skt_pool);
 315         apr_pool_cleanup_register(conn->skt_pool, conn, clean_skt, clean_skt);
 316
 317         status = apr_socket_create(&skt, conn->address->family,
 318                                    SOCK_STREAM,
 319 #if APR_MAJOR_VERSION > 0
 320                                    APR_PROTO_TCP,
 321 #endif
 322                                    conn->skt_pool);
 323         serf__log(SOCK_VERBOSE, __FILE__,
 324                   "created socket for conn 0x%x, status %d\n", conn, status);
 325         if (status != APR_SUCCESS)
 326             return status;
 327
 328         /* Set the socket to be non-blocking */
 329         if ((status = apr_socket_timeout_set(skt, 0)) != APR_SUCCESS)
 330             return status;
 331
 332         /* Disable Nagle's algorithm */
 333         if ((status = apr_socket_opt_set(skt,
 334                                          APR_TCP_NODELAY, 1)) != APR_SUCCESS)
 335             return status;
 336
 337         /* Configured. Store it into the connection now. */
 338         conn->skt = skt;
 339
 340         /* Remember time when we started connecting to server to calculate
 341            network latency. */
 342         conn->connect_time = apr_time_now();
 343
 344         /* Now that the socket is set up, let's connect it. This should
 345          * return immediately.
 346          */
 347         status = apr_socket_connect(skt, conn->address);
 348         serf__log_skt(SOCK_VERBOSE, __FILE__, skt,
 349                       "connected socket for conn 0x%x, status %d\n",
 350                       conn, status);
 351         if (status != APR_SUCCESS) {
 352             if (!APR_STATUS_IS_EINPROGRESS(status))
 353                 return status;
 354         }
 355
 356         /* Flag our pollset as dirty now that we have a new socket. */
 357         conn->dirty_conn = 1;
 358         ctx->dirty_pollset = 1;
 359
 360         /* If the authentication was already started on another connection,
 361            prepare this connection (it might be possible to skip some
 362            part of the handshaking). */
 363         if (ctx->proxy_address) {
 364             authn_info = &ctx->proxy_authn_info;
 365             if (authn_info->scheme) {
 366                 authn_info->scheme->init_conn_func(authn_info->scheme, 407,
 367                                                    conn, conn->pool);
 368             }
 369         }
 370
 371         authn_info = serf__get_authn_info_for_server(conn);
 372         if (authn_info->scheme) {
 373             authn_info->scheme->init_conn_func(authn_info->scheme, 401,
 374                                                conn, conn->pool);
 375         }
 376
 377         /* Does this connection require a SSL tunnel over the proxy? */
 378         if (ctx->proxy_address && strcmp(conn->host_info.scheme, "https") == 0)
 379             serf__ssltunnel_connect(conn);
 380         else {
 381             serf_bucket_t *dummy1, *dummy2;
 382
 383             conn->state = SERF_CONN_CONNECTED;
 384
 385             status = prepare_conn_streams(conn, &conn->stream,
 386                                           &dummy1, &dummy2);
 387             if (status) {
 388                 return status;
 389             }
 390         }
 391     }
 392
 393     return APR_SUCCESS;
 394 }
 395
 396 static apr_status_t no_more_writes(serf_connection_t *conn,
 397                                    serf_request_t *request)
 398 {
 399     /* Note that we should hold new requests until we open our new socket. */
 400     conn->state = SERF_CONN_CLOSING;
 401     serf__log(CONN_VERBOSE, __FILE__, "stop writing on conn 0x%x\n",
 402               conn);
 403
 404     /* Clear our iovec. */
 405     conn->vec_len = 0;
 406
 407     /* Update the pollset to know we don't want to write on this socket any
 408      * more.
 409      */
 410     conn->dirty_conn = 1;
 411     conn->ctx->dirty_pollset = 1;
 412     return APR_SUCCESS;
 413 }
 414
 415 /* Read the 'Connection' header from the response. Return SERF_ERROR_CLOSING if
 416  * the header contains value 'close' indicating the server is closing the
 417  * connection right after this response.
 418  * Otherwise returns APR_SUCCESS.
 419  */
 420 static apr_status_t is_conn_closing(serf_bucket_t *response)
 421 {
 422     serf_bucket_t *hdrs;
 423     const char *val;
 424
 425     hdrs = serf_bucket_response_get_headers(response);
 426     val = serf_bucket_headers_get(hdrs, "Connection");
 427     if (val && strcasecmp("close", val) == 0)
 428         {
 429             return SERF_ERROR_CLOSING;
 430         }
 431
 432     return APR_SUCCESS;
 433 }
 434
 435 static void link_requests(serf_request_t **list, serf_request_t **tail,
 436                           serf_request_t *request)
 437 {
 438     if (*list == NULL) {
 439         *list = request;
 440         *tail = request;
 441     }
 442     else {
 443         (*tail)->next = request;
 444         *tail = request;
 445     }
 446 }
 447
 448 static apr_status_t destroy_request(serf_request_t *request)
 449 {
 450     serf_connection_t *conn = request->conn;
 451
 452     /* The request and response buckets are no longer needed,
 453        nor is the request's pool.  */
 454     if (request->resp_bkt) {
 455         serf_debug__closed_conn(request->resp_bkt->allocator);
 456         serf_bucket_destroy(request->resp_bkt);
 457         request->resp_bkt = NULL;
 458     }
 459     if (request->req_bkt) {
 460         serf_debug__closed_conn(request->req_bkt->allocator);
 461         serf_bucket_destroy(request->req_bkt);
 462         request->req_bkt = NULL;
 463     }
 464
 465     serf_debug__bucket_alloc_check(request->allocator);
 466     if (request->respool) {
 467         /* ### unregister the pool cleanup for self?  */
 468         apr_pool_destroy(request->respool);
 469     }
 470
 471     serf_bucket_mem_free(conn->allocator, request);
 472
 473     return APR_SUCCESS;
 474 }
 475
 476 static apr_status_t cancel_request(serf_request_t *request,
 477                                    serf_request_t **list,
 478                                    int notify_request)
 479 {
 480     /* If we haven't run setup, then we won't have a handler to call. */
 481     if (request->handler && notify_request) {
 482         /* We actually don't care what the handler returns.
 483          * We have bigger matters at hand.
 484          */
 485         (*request->handler)(request, NULL, request->handler_baton,
 486                             request->respool);
 487     }
 488
 489     if (*list == request) {
 490         *list = request->next;
 491     }
 492     else {
 493         serf_request_t *scan = *list;
 494
 495         while (scan->next && scan->next != request)
 496             scan = scan->next;
 497
 498         if (scan->next) {
 499             scan->next = scan->next->next;
 500         }
 501     }
 502
 503     return destroy_request(request);
 504 }
 505
 506 static apr_status_t remove_connection(serf_context_t *ctx,
 507                                       serf_connection_t *conn)
 508 {
 509     apr_pollfd_t desc = { 0 };
 510
 511     desc.desc_type = APR_POLL_SOCKET;
 512     desc.desc.s = conn->skt;
 513     desc.reqevents = conn->reqevents;
 514
 515     return ctx->pollset_rm(ctx->pollset_baton,
 516                            &desc, conn);
 517 }
 518
 519 /* A socket was closed, inform the application. */
 520 static void handle_conn_closed(serf_connection_t *conn, apr_status_t status)
 521 {
 522     (*conn->closed)(conn, conn->closed_baton, status,
 523                     conn->pool);
 524 }
 525
 526 static apr_status_t reset_connection(serf_connection_t *conn,
 527                                      int requeue_requests)
 528 {
 529     serf_context_t *ctx = conn->ctx;
 530     apr_status_t status;
 531     serf_request_t *old_reqs;
 532
 533     conn->probable_keepalive_limit = conn->completed_responses;
 534     conn->completed_requests = 0;
 535     conn->completed_responses = 0;
 536
 537     old_reqs = conn->requests;
 538
 539     conn->requests = NULL;
 540     conn->requests_tail = NULL;
 541
 542     /* Handle all outstanding requests. These have either not been written yet,
 543        or have been written but the expected reply wasn't received yet. */
 544     while (old_reqs) {
 545         /* If we haven't started to write the connection, bring it over
 546          * unchanged to our new socket.
 547          */
 548         if (requeue_requests && !old_reqs->written) {
 549             serf_request_t *req = old_reqs;
 550             old_reqs = old_reqs->next;
 551             req->next = NULL;
 552             link_requests(&conn->requests, &conn->requests_tail, req);
 553         }
 554         else {
 555             /* Request has been consumed, or we don't want to requeue the
 556                request. Either way, inform the application that the request
 557                is cancelled. */
 558             cancel_request(old_reqs, &old_reqs, requeue_requests);
 559         }
 560     }
 561
 562     /* Requests queue has been prepared for a new socket, close the old one. */
 563     if (conn->skt != NULL) {
 564         remove_connection(ctx, conn);
 565         status = apr_socket_close(conn->skt);
 566         serf__log_skt(SOCK_VERBOSE, __FILE__, conn->skt,
 567                       "closed socket, status %d\n", status);
 568         if (conn->closed != NULL) {
 569             handle_conn_closed(conn, status);
 570         }
 571         conn->skt = NULL;
 572     }
 573
 574     if (conn->stream != NULL) {
 575         serf_bucket_destroy(conn->stream);
 576         conn->stream = NULL;
 577     }
 578
 579     destroy_ostream(conn);
 580
 581     /* Don't try to resume any writes */
 582     conn->vec_len = 0;
 583
 584     conn->dirty_conn = 1;
 585     conn->ctx->dirty_pollset = 1;
 586     conn->state = SERF_CONN_INIT;
 587
 588     serf__log(CONN_VERBOSE, __FILE__, "reset connection 0x%x\n", conn);
 589
 590     conn->status = APR_SUCCESS;
 591
 592     /* Let our context know that we've 'reset' the socket already. */
 593     conn->seen_in_pollset |= APR_POLLHUP;
 594
 595     /* Found the connection. Closed it. All done. */
 596     return APR_SUCCESS;
 597 }
 598
 599 static apr_status_t socket_writev(serf_connection_t *conn)
 600 {
 601     apr_size_t written;
 602     apr_status_t status;
 603
 604     status = apr_socket_sendv(conn->skt, conn->vec,
 605                               conn->vec_len, &written);
 606     if (status && !APR_STATUS_IS_EAGAIN(status))
 607         serf__log_skt(SOCK_VERBOSE, __FILE__, conn->skt,
 608                       "socket_sendv error %d\n", status);
 609
 610     /* did we write everything? */
 611     if (written) {
 612         apr_size_t len = 0;
 613         int i;
 614
 615         serf__log_skt(SOCK_MSG_VERBOSE, __FILE__, conn->skt,
 616                       "--- socket_sendv:\n");
 617
 618         for (i = 0; i < conn->vec_len; i++) {
 619             len += conn->vec[i].iov_len;
 620             if (written < len) {
 621                 serf__log_nopref(SOCK_MSG_VERBOSE, "%.*s",
 622                                    conn->vec[i].iov_len - (len - written),
 623                                    conn->vec[i].iov_base);
 624                 if (i) {
 625                     memmove(conn->vec, &conn->vec[i],
 626                             sizeof(struct iovec) * (conn->vec_len - i));
 627                     conn->vec_len -= i;
 628                 }
 629                 conn->vec[0].iov_base = (char *)conn->vec[0].iov_base + (conn->vec[0].iov_len - (len - written));
 630                 conn->vec[0].iov_len = len - written;
 631                 break;
 632             } else {
 633                 serf__log_nopref(SOCK_MSG_VERBOSE, "%.*s",
 634                                    conn->vec[i].iov_len, conn->vec[i].iov_base);
 635             }
 636         }
 637         if (len == written) {
 638             conn->vec_len = 0;
 639         }
 640         serf__log_nopref(SOCK_MSG_VERBOSE, "-(%d)-\n", written);
 641
 642         /* Log progress information */
 643         serf__context_progress_delta(conn->ctx, 0, written);
 644     }
 645
 646     return status;
 647 }
 648
 649 static apr_status_t setup_request(serf_request_t *request)
 650 {
 651     serf_connection_t *conn = request->conn;
 652     apr_status_t status;
 653
 654     /* Now that we are about to serve the request, allocate a pool. */
 655     apr_pool_create(&request->respool, conn->pool);
 656     request->allocator = serf_bucket_allocator_create(request->respool,
 657                                                       NULL, NULL);
 658     apr_pool_cleanup_register(request->respool, request,
 659                               clean_resp, clean_resp);
 660
 661     /* Fill in the rest of the values for the request. */
 662     status = request->setup(request, request->setup_baton,
 663                             &request->req_bkt,
 664                             &request->acceptor,
 665                             &request->acceptor_baton,
 666                             &request->handler,
 667                             &request->handler_baton,
 668                             request->respool);
 669     return status;
 670 }
 671
 672 /* write data out to the connection */
 673 static apr_status_t write_to_connection(serf_connection_t *conn)
 674 {
 675     serf_request_t *request = conn->requests;
 676
 677     if (conn->probable_keepalive_limit &&
 678         conn->completed_requests > conn->probable_keepalive_limit) {
 679
 680         conn->dirty_conn = 1;
 681         conn->ctx->dirty_pollset = 1;
 682
 683         /* backoff for now. */
 684         return APR_SUCCESS;
 685     }
 686
 687     /* Find a request that has data which needs to be delivered. */
 688     while (request != NULL &&
 689            request->req_bkt == NULL && request->written)
 690         request = request->next;
 691
 692     /* assert: request != NULL || conn->vec_len */
 693
 694     /* Keep reading and sending until we run out of stuff to read, or
 695      * writing would block.
 696      */
 697     while (1) {
 698         int stop_reading = 0;
 699         apr_status_t status;
 700         apr_status_t read_status;
 701         serf_bucket_t *ostreamt, *ostreamh;
 702         int max_outstanding_requests = conn->max_outstanding_requests;
 703
 704         /* If we're setting up an ssl tunnel, we can't send real requests
 705            at yet, as they need to be encrypted and our encrypt buckets
 706            aren't created yet as we still need to read the unencrypted
 707            response of the CONNECT request. */
 708         if (conn->state != SERF_CONN_CONNECTED)
 709             max_outstanding_requests = 1;
 710
 711         if (max_outstanding_requests &&
 712             conn->completed_requests -
 713                 conn->completed_responses >= max_outstanding_requests) {
 714             /* backoff for now. */
 715             return APR_SUCCESS;
 716         }
 717
 718         /* If we have unwritten data, then write what we can. */
 719         while (conn->vec_len) {
 720             status = socket_writev(conn);
 721
 722             /* If the write would have blocked, then we're done. Don't try
 723              * to write anything else to the socket.
 724              */
 725             if (APR_STATUS_IS_EAGAIN(status))
 726                 return APR_SUCCESS;
 727             if (APR_STATUS_IS_EPIPE(status) ||
 728                 APR_STATUS_IS_ECONNRESET(status) ||
 729                 APR_STATUS_IS_ECONNABORTED(status))
 730                 return no_more_writes(conn, request);
 731             if (status)
 732                 return status;
 733         }
 734         /* ### can we have a short write, yet no EAGAIN? a short write
 735            ### would imply unwritten_len > 0 ... */
 736         /* assert: unwritten_len == 0. */
 737
 738         /* We may need to move forward to a request which has something
 739          * to write.
 740          */
 741         while (request != NULL &&
 742                request->req_bkt == NULL && request->written)
 743             request = request->next;
 744
 745         if (request == NULL) {
 746             /* No more requests (with data) are registered with the
 747              * connection. Let's update the pollset so that we don't
 748              * try to write to this socket again.
 749              */
 750             conn->dirty_conn = 1;
 751             conn->ctx->dirty_pollset = 1;
 752             return APR_SUCCESS;
 753         }
 754
 755         status = prepare_conn_streams(conn, &conn->stream, &ostreamt, &ostreamh);
 756         if (status) {
 757             return status;
 758         }
 759
 760         if (request->req_bkt == NULL) {
 761             read_status = setup_request(request);
 762             if (read_status) {
 763                 /* Something bad happened. Propagate any errors. */
 764                 return read_status;
 765             }
 766         }
 767
 768         if (!request->written) {
 769             request->written = 1;
 770             serf_bucket_aggregate_append(ostreamt, request->req_bkt);
 771         }
 772
 773         /* ### optimize at some point by using read_for_sendfile */
 774         /* TODO: now that read_iovec will effectively try to return as much
 775            data as available, we probably don't want to read ALL_AVAIL, but
 776            a lower number, like the size of one or a few TCP packets, the
 777            available TCP buffer size ... */
 778         read_status = serf_bucket_read_iovec(ostreamh,
 779                                              SERF_READ_ALL_AVAIL,
 780                                              IOV_MAX,
 781                                              conn->vec,
 782                                              &conn->vec_len);
 783
 784         if (!conn->hit_eof) {
 785             if (APR_STATUS_IS_EAGAIN(read_status)) {
 786                 /* We read some stuff, but should not try to read again. */
 787                 stop_reading = 1;
 788             }
 789             else if (read_status == SERF_ERROR_WAIT_CONN) {
 790                 /* The bucket told us that it can't provide more data until
 791                    more data is read from the socket. This normally happens
 792                    during a SSL handshake.
 793
 794                    We should avoid looking for writability for a while so
 795                    that (hopefully) something will appear in the bucket so
 796                    we can actually write something. otherwise, we could
 797                    end up in a CPU spin: socket wants something, but we
 798                    don't have anything (and keep returning EAGAIN)
 799                  */
 800                 conn->stop_writing = 1;
 801                 conn->dirty_conn = 1;
 802                 conn->ctx->dirty_pollset = 1;
 803             }
 804             else if (read_status && !APR_STATUS_IS_EOF(read_status)) {
 805                 /* Something bad happened. Propagate any errors. */
 806                 return read_status;
 807             }
 808         }
 809
 810         /* If we got some data, then deliver it. */
 811         /* ### what to do if we got no data?? is that a problem? */
 812         if (conn->vec_len > 0) {
 813             status = socket_writev(conn);
 814
 815             /* If we can't write any more, or an error occurred, then
 816              * we're done here.
 817              */
 818             if (APR_STATUS_IS_EAGAIN(status))
 819                 return APR_SUCCESS;
 820             if (APR_STATUS_IS_EPIPE(status))
 821                 return no_more_writes(conn, request);
 822             if (APR_STATUS_IS_ECONNRESET(status) ||
 823                 APR_STATUS_IS_ECONNABORTED(status)) {
 824                 return no_more_writes(conn, request);
 825             }
 826             if (status)
 827                 return status;
 828         }
 829
 830         if (read_status == SERF_ERROR_WAIT_CONN) {
 831             stop_reading = 1;
 832             conn->stop_writing = 1;
 833             conn->dirty_conn = 1;
 834             conn->ctx->dirty_pollset = 1;
 835         }
 836         else if (read_status && conn->hit_eof && conn->vec_len == 0) {
 837             /* If we hit the end of the request bucket and all of its data has
 838              * been written, then clear it out to signify that we're done
 839              * sending the request. On the next iteration through this loop:
 840              * - if there are remaining bytes they will be written, and as the
 841              * request bucket will be completely read it will be destroyed then.
 842              * - we'll see if there are other requests that need to be sent
 843              * ("pipelining").
 844              */
 845             conn->hit_eof = 0;
 846             serf_bucket_destroy(request->req_bkt);
 847             request->req_bkt = NULL;
 848
 849             /* If our connection has async responses enabled, we're not
 850              * going to get a reply back, so kill the request.
 851              */
 852             if (conn->async_responses) {
 853                 conn->requests = request->next;
 854                 destroy_request(request);
 855             }
 856
 857             conn->completed_requests++;
 858
 859             if (conn->probable_keepalive_limit &&
 860                 conn->completed_requests > conn->probable_keepalive_limit) {
 861                 /* backoff for now. */
 862                 stop_reading = 1;
 863             }
 864         }
 865
 866         if (stop_reading) {
 867             return APR_SUCCESS;
 868         }
 869     }
 870     /* NOTREACHED */
 871 }
 872
 873 /* A response message was received from the server, so call
 874    the handler as specified on the original request. */
 875 static apr_status_t handle_response(serf_request_t *request,
 876                                     apr_pool_t *pool)
 877 {
 878     apr_status_t status = APR_SUCCESS;
 879     int consumed_response = 0;
 880
 881     /* Only enable the new authentication framework if the program has
 882      * registered an authentication credential callback.
 883      *
 884      * This permits older Serf apps to still handle authentication
 885      * themselves by not registering credential callbacks.
 886      */
 887     if (request->conn->ctx->cred_cb) {
 888       status = serf__handle_auth_response(&consumed_response,
 889                                           request,
 890                                           request->resp_bkt,
 891                                           request->handler_baton,
 892                                           pool);
 893
 894       /* If there was an error reading the response (maybe there wasn't
 895          enough data available), don't bother passing the response to the
 896          application.
 897
 898          If the authentication was tried, but failed, pass the response
 899          to the application, maybe it can do better. */
 900       if (APR_STATUS_IS_EOF(status) ||
 901           APR_STATUS_IS_EAGAIN(status)) {
 902           return status;
 903       }
 904     }
 905
 906     if (!consumed_response) {
 907         return (*request->handler)(request,
 908                                    request->resp_bkt,
 909                                    request->handler_baton,
 910                                    pool);
 911     }
 912
 913     return status;
 914 }
 915
 916 /* An async response message was received from the server. */
 917 static apr_status_t handle_async_response(serf_connection_t *conn,
 918                                           apr_pool_t *pool)
 919 {
 920     apr_status_t status;
 921
 922     if (conn->current_async_response == NULL) {
 923         conn->current_async_response =
 924             (*conn->async_acceptor)(NULL, conn->stream,
 925                                     conn->async_acceptor_baton, pool);
 926     }
 927
 928     status = (*conn->async_handler)(NULL, conn->current_async_response,
 929                                     conn->async_handler_baton, pool);
 930
 931     if (APR_STATUS_IS_EOF(status)) {
 932         serf_bucket_destroy(conn->current_async_response);
 933         conn->current_async_response = NULL;
 934         status = APR_SUCCESS;
 935     }
 936
 937     return status;
 938 }
 939
 940
 941 apr_status_t
 942 serf__provide_credentials(serf_context_t *ctx,
 943                           char **username,
 944                           char **password,
 945                           serf_request_t *request, void *baton,
 946                           int code, const char *authn_type,
 947                           const char *realm,
 948                           apr_pool_t *pool)
 949 {
 950     serf_connection_t *conn = request->conn;
 951     serf_request_t *authn_req = request;
 952     apr_status_t status;
 953
 954     if (request->ssltunnel == 1 &&
 955         conn->state == SERF_CONN_SETUP_SSLTUNNEL) {
 956         /* This is a CONNECT request to set up an SSL tunnel over a proxy.
 957            This request is created by serf, so if the proxy requires
 958            authentication, we can't ask the application for credentials with
 959            this request.
 960
 961            Solution: setup the first request created by the application on
 962            this connection, and use that request and its handler_baton to
 963            call back to the application. */
 964
 965         authn_req = request->next;
 966         /* assert: app_request != NULL */
 967         if (!authn_req)
 968             return APR_EGENERAL;
 969
 970         if (!authn_req->req_bkt) {
 971             apr_status_t status;
 972
 973             status = setup_request(authn_req);
 974             /* If we can't setup a request, don't bother setting up the
 975                ssl tunnel. */
 976             if (status)
 977                 return status;
 978         }
 979     }
 980
 981     /* Ask the application. */
 982     status = (*ctx->cred_cb)(username, password,
 983                              authn_req, authn_req->handler_baton,
 984                              code, authn_type, realm, pool);
 985     if (status)
 986         return status;
 987
 988     return APR_SUCCESS;
 989 }
 990
 991 /* read data from the connection */
 992 static apr_status_t read_from_connection(serf_connection_t *conn)
 993 {
 994     apr_status_t status;
 995     apr_pool_t *tmppool;
 996     int close_connection = FALSE;
 997
 998     /* Whatever is coming in on the socket corresponds to the first request
 999      * on our chain.
1000      */
1001     serf_request_t *request = conn->requests;
1002
1003     /* If the stop_writing flag was set on the connection, reset it now because
1004        there is some data to read. */
1005     if (conn->stop_writing) {
1006         conn->stop_writing = 0;
1007         conn->dirty_conn = 1;
1008         conn->ctx->dirty_pollset = 1;
1009     }
1010
1011     /* assert: request != NULL */
1012
1013     if ((status = apr_pool_create(&tmppool, conn->pool)) != APR_SUCCESS)
1014         goto error;
1015
1016     /* Invoke response handlers until we have no more work. */
1017     while (1) {
1018         serf_bucket_t *dummy1, *dummy2;
1019
1020         apr_pool_clear(tmppool);
1021
1022         /* Only interested in the input stream here. */
1023         status = prepare_conn_streams(conn, &conn->stream, &dummy1, &dummy2);
1024         if (status) {
1025             goto error;
1026         }
1027
1028         /* We have a different codepath when we can have async responses. */
1029         if (conn->async_responses) {
1030             /* TODO What about socket errors? */
1031             status = handle_async_response(conn, tmppool);
1032             if (APR_STATUS_IS_EAGAIN(status)) {
1033                 status = APR_SUCCESS;
1034                 goto error;
1035             }
1036             if (status) {
1037                 goto error;
1038             }
1039             continue;
1040         }
1041
1042         /* We are reading a response for a request we haven't
1043          * written yet!
1044          *
1045          * This shouldn't normally happen EXCEPT:
1046          *
1047          * 1) when the other end has closed the socket and we're
1048          *    pending an EOF return.
1049          * 2) Doing the initial SSL handshake - we'll get EAGAIN
1050          *    as the SSL buckets will hide the handshake from us
1051          *    but not return any data.
1052          * 3) When the server sends us an SSL alert.
1053          *
1054          * In these cases, we should not receive any actual user data.
1055          *
1056          * 4) When the server sends a error response, like 408 Request timeout.
1057          *    This response should be passed to the application.
1058          *
1059          * If we see an EOF (due to either an expired timeout or the server
1060          * sending the SSL 'close notify' shutdown alert), we'll reset the
1061          * connection and open a new one.
1062          */
1063         if (request->req_bkt || !request->written) {
1064             const char *data;
1065             apr_size_t len;
1066
1067             status = serf_bucket_peek(conn->stream, &data, &len);
1068
1069             if (APR_STATUS_IS_EOF(status)) {
1070                 reset_connection(conn, 1);
1071                 status = APR_SUCCESS;
1072                 goto error;
1073             }
1074             else if (APR_STATUS_IS_EAGAIN(status) && !len) {
1075                 status = APR_SUCCESS;
1076                 goto error;
1077             } else if (status && !APR_STATUS_IS_EAGAIN(status)) {
1078                 /* Read error */
1079                 goto error;
1080             }
1081
1082             /* Unexpected response from the server */
1083
1084         }
1085
1086         /* If the request doesn't have a response bucket, then call the
1087          * acceptor to get one created.
1088          */
1089         if (request->resp_bkt == NULL) {
1090             request->resp_bkt = (*request->acceptor)(request, conn->stream,
1091                                                      request->acceptor_baton,
1092                                                      tmppool);
1093             apr_pool_clear(tmppool);
1094         }
1095
1096         status = handle_response(request, tmppool);
1097
1098         /* Some systems will not generate a HUP poll event so we have to
1099          * handle the ECONNRESET issue and ECONNABORT here.
1100          */
1101         if (APR_STATUS_IS_ECONNRESET(status) ||
1102             APR_STATUS_IS_ECONNABORTED(status) ||
1103             status == SERF_ERROR_REQUEST_LOST) {
1104             /* If the connection had ever been good, be optimistic & try again.
1105              * If it has never tried again (incl. a retry), fail.
1106              */
1107             if (conn->completed_responses) {
1108                 reset_connection(conn, 1);
1109                 status = APR_SUCCESS;
1110             }
1111             else if (status == SERF_ERROR_REQUEST_LOST) {
1112                 status = SERF_ERROR_ABORTED_CONNECTION;
1113             }
1114             goto error;
1115         }
1116
1117         /* If our response handler says it can't do anything more, we now
1118          * treat that as a success.
1119          */
1120         if (APR_STATUS_IS_EAGAIN(status)) {
1121             status = APR_SUCCESS;
1122             goto error;
1123         }
1124
1125         /* If we received APR_SUCCESS, run this loop again. */
1126         if (!status) {
1127             continue;
1128         }
1129
1130         close_connection = is_conn_closing(request->resp_bkt);
1131
1132         if (!APR_STATUS_IS_EOF(status) &&
1133             close_connection != SERF_ERROR_CLOSING) {
1134             /* Whether success, or an error, there is no more to do unless
1135              * this request has been completed.
1136              */
1137             goto error;
1138         }
1139
1140         /* The response has been fully-read, so that means the request has
1141          * either been fully-delivered (most likely), or that we don't need to
1142          * write the rest of it anymore, e.g. when a 408 Request timeout was
1143          $ received.
1144          * Remove it from our queue and loop to read another response.
1145          */
1146         conn->requests = request->next;
1147
1148         destroy_request(request);
1149
1150         request = conn->requests;
1151
1152         /* If we're truly empty, update our tail. */
1153         if (request == NULL) {
1154             conn->requests_tail = NULL;
1155         }
1156
1157         conn->completed_responses++;
1158
1159         /* We've to rebuild pollset since completed_responses is changed. */
1160         conn->dirty_conn = 1;
1161         conn->ctx->dirty_pollset = 1;
1162
1163         /* This means that we're being advised that the connection is done. */
1164         if (close_connection == SERF_ERROR_CLOSING) {
1165             reset_connection(conn, 1);
1166             if (APR_STATUS_IS_EOF(status))
1167                 status = APR_SUCCESS;
1168             goto error;
1169         }
1170
1171         /* The server is suddenly deciding to serve more responses than we've
1172          * seen before.
1173          *
1174          * Let our requests go.
1175          */
1176         if (conn->probable_keepalive_limit &&
1177             conn->completed_responses > conn->probable_keepalive_limit) {
1178             conn->probable_keepalive_limit = 0;
1179         }
1180
1181         /* If we just ran out of requests or have unwritten requests, then
1182          * update the pollset. We don't want to read from this socket any
1183          * more. We are definitely done with this loop, too.
1184          */
1185         if (request == NULL || !request->written) {
1186             conn->dirty_conn = 1;
1187             conn->ctx->dirty_pollset = 1;
1188             status = APR_SUCCESS;
1189             goto error;
1190         }
1191     }
1192
1193 error:
1194     apr_pool_destroy(tmppool);
1195     return status;
1196 }
1197
1198 /* process all events on the connection */
1199 apr_status_t serf__process_connection(serf_connection_t *conn,
1200                                       apr_int16_t events)
1201 {
1202     apr_status_t status;
1203
1204     /* POLLHUP/ERR should come after POLLIN so if there's an error message or
1205      * the like sitting on the connection, we give the app a chance to read
1206      * it before we trigger a reset condition.
1207      */
1208     if ((events & APR_POLLIN) != 0) {
1209         if ((status = read_from_connection(conn)) != APR_SUCCESS)
1210             return status;
1211
1212         /* If we decided to reset our connection, return now as we don't
1213          * want to write.
1214          */
1215         if ((conn->seen_in_pollset & APR_POLLHUP) != 0) {
1216             return APR_SUCCESS;
1217         }
1218     }
1219     if ((events & APR_POLLHUP) != 0) {
1220         /* The connection got reset by the server. On Windows this can happen
1221            when all data is read, so just cleanup the connection and open
1222            a new one.
1223            If we haven't had any successful responses on this connection,
1224            then error out as it is likely a server issue. */
1225         if (conn->completed_responses) {
1226             return reset_connection(conn, 1);
1227         }
1228         return SERF_ERROR_ABORTED_CONNECTION;
1229     }
1230     if ((events & APR_POLLERR) != 0) {
1231         /* We might be talking to a buggy HTTP server that doesn't
1232          * do lingering-close.  (httpd < 2.1.8 does this.)
1233          *
1234          * See:
1235          *
1236          * http://issues.apache.org/bugzilla/show_bug.cgi?id=35292
1237          */
1238         if (conn->completed_requests && !conn->probable_keepalive_limit) {
1239             return reset_connection(conn, 1);
1240         }
1241 #ifdef SO_ERROR
1242         /* If possible, get the error from the platform's socket layer and
1243            convert it to an APR status code. */
1244         {
1245             apr_os_sock_t osskt;
1246             if (!apr_os_sock_get(&osskt, conn->skt)) {
1247                 int error;
1248                 apr_socklen_t l = sizeof(error);
1249
1250                 if (!getsockopt(osskt, SOL_SOCKET, SO_ERROR, (char*)&error, &l))
1251                     return APR_FROM_OS_ERROR(error);
1252             }
1253         }
1254 #endif
1255         return APR_EGENERAL;
1256     }
1257     if ((events & APR_POLLOUT) != 0) {
1258         if ((status = write_to_connection(conn)) != APR_SUCCESS)
1259             return status;
1260     }
1261     return APR_SUCCESS;
1262 }
1263
1264 serf_connection_t *serf_connection_create(
1265     serf_context_t *ctx,
1266     apr_sockaddr_t *address,
1267     serf_connection_setup_t setup,
1268     void *setup_baton,
1269     serf_connection_closed_t closed,
1270     void *closed_baton,
1271     apr_pool_t *pool)
1272 {
1273     serf_connection_t *conn = apr_pcalloc(pool, sizeof(*conn));
1274
1275     conn->ctx = ctx;
1276     conn->status = APR_SUCCESS;
1277     /* Ignore server address if proxy was specified. */
1278     conn->address = ctx->proxy_address ? ctx->proxy_address : address;
1279     conn->setup = setup;
1280     conn->setup_baton = setup_baton;
1281     conn->closed = closed;
1282     conn->closed_baton = closed_baton;
1283     conn->pool = pool;
1284     conn->allocator = serf_bucket_allocator_create(pool, NULL, NULL);
1285     conn->stream = NULL;
1286     conn->ostream_head = NULL;
1287     conn->ostream_tail = NULL;
1288     conn->baton.type = SERF_IO_CONN;
1289     conn->baton.u.conn = conn;
1290     conn->hit_eof = 0;
1291     conn->state = SERF_CONN_INIT;
1292     conn->latency = -1; /* unknown */
1293
1294     /* Create a subpool for our connection. */
1295     apr_pool_create(&conn->skt_pool, conn->pool);
1296
1297     /* register a cleanup */
1298     apr_pool_cleanup_register(conn->pool, conn, clean_conn,
1299                               apr_pool_cleanup_null);
1300
1301     /* Add the connection to the context. */
1302     *(serf_connection_t **)apr_array_push(ctx->conns) = conn;
1303
1304     serf__log(CONN_VERBOSE, __FILE__, "created connection 0x%x\n",
1305               conn);
1306
1307     return conn;
1308 }
1309
1310 apr_status_t serf_connection_create2(
1311     serf_connection_t **conn,
1312     serf_context_t *ctx,
1313     apr_uri_t host_info,
1314     serf_connection_setup_t setup,
1315     void *setup_baton,
1316     serf_connection_closed_t closed,
1317     void *closed_baton,
1318     apr_pool_t *pool)
1319 {
1320     apr_status_t status = APR_SUCCESS;
1321     serf_connection_t *c;
1322     apr_sockaddr_t *host_address = NULL;
1323
1324     /* Set the port number explicitly, needed to create the socket later. */
1325     if (!host_info.port) {
1326         host_info.port = apr_uri_port_of_scheme(host_info.scheme);
1327     }
1328
1329     /* Only lookup the address of the server if no proxy server was
1330        configured. */
1331     if (!ctx->proxy_address) {
1332         status = apr_sockaddr_info_get(&host_address,
1333                                        host_info.hostname,
1334                                        APR_UNSPEC, host_info.port, 0, pool);
1335         if (status)
1336             return status;
1337     }
1338
1339     c = serf_connection_create(ctx, host_address, setup, setup_baton,
1340                                closed, closed_baton, pool);
1341
1342     /* We're not interested in the path following the hostname. */
1343     c->host_url = apr_uri_unparse(c->pool,
1344                                   &host_info,
1345                                   APR_URI_UNP_OMITPATHINFO);
1346
1347     /* Store the host info without the path on the connection. */
1348     (void)apr_uri_parse(c->pool, c->host_url, &(c->host_info));
1349     if (!c->host_info.port) {
1350         c->host_info.port = apr_uri_port_of_scheme(c->host_info.scheme);
1351     }
1352
1353     *conn = c;
1354
1355     return status;
1356 }
1357
1358 apr_status_t serf_connection_reset(
1359     serf_connection_t *conn)
1360 {
1361     return reset_connection(conn, 0);
1362 }
1363
1364
1365 apr_status_t serf_connection_close(
1366     serf_connection_t *conn)
1367 {
1368     int i;
1369     serf_context_t *ctx = conn->ctx;
1370     apr_status_t status;
1371
1372     for (i = ctx->conns->nelts; i--; ) {
1373         serf_connection_t *conn_seq = GET_CONN(ctx, i);
1374
1375         if (conn_seq == conn) {
1376             while (conn->requests) {
1377                 serf_request_cancel(conn->requests);
1378             }
1379             if (conn->skt != NULL) {
1380                 remove_connection(ctx, conn);
1381                 status = apr_socket_close(conn->skt);
1382                 serf__log_skt(SOCK_VERBOSE, __FILE__, conn->skt,
1383                               "closed socket, status %d\n",
1384                               status);
1385                 if (conn->closed != NULL) {
1386                     handle_conn_closed(conn, status);
1387                 }
1388                 conn->skt = NULL;
1389             }
1390             if (conn->stream != NULL) {
1391                 serf_bucket_destroy(conn->stream);
1392                 conn->stream = NULL;
1393             }
1394
1395             destroy_ostream(conn);
1396
1397             /* Remove the connection from the context. We don't want to
1398              * deal with it any more.
1399              */
1400             if (i < ctx->conns->nelts - 1) {
1401                 /* move later connections over this one. */
1402                 memmove(
1403                     &GET_CONN(ctx, i),
1404                     &GET_CONN(ctx, i + 1),
1405                     (ctx->conns->nelts - i - 1) * sizeof(serf_connection_t *));
1406             }
1407             --ctx->conns->nelts;
1408
1409             serf__log(CONN_VERBOSE, __FILE__, "closed connection 0x%x\n",
1410                       conn);
1411
1412             /* Found the connection. Closed it. All done. */
1413             return APR_SUCCESS;
1414         }
1415     }
1416
1417     /* We didn't find the specified connection. */
1418     /* ### doc talks about this w.r.t poll structures. use something else? */
1419     return APR_NOTFOUND;
1420 }
1421
1422
1423 void serf_connection_set_max_outstanding_requests(
1424     serf_connection_t *conn,
1425     unsigned int max_requests)
1426 {
1427     if (max_requests == 0)
1428         serf__log_skt(CONN_VERBOSE, __FILE__, conn->skt,
1429                       "Set max. nr. of outstanding requests for this "
1430                       "connection to unlimited.\n");
1431     else
1432         serf__log_skt(CONN_VERBOSE, __FILE__, conn->skt,
1433                       "Limit max. nr. of outstanding requests for this "
1434                       "connection to %u.\n", max_requests);
1435
1436     conn->max_outstanding_requests = max_requests;
1437 }
1438
1439
1440 void serf_connection_set_async_responses(
1441     serf_connection_t *conn,
1442     serf_response_acceptor_t acceptor,
1443     void *acceptor_baton,
1444     serf_response_handler_t handler,
1445     void *handler_baton)
1446 {
1447     conn->async_responses = 1;
1448     conn->async_acceptor = acceptor;
1449     conn->async_acceptor_baton = acceptor_baton;
1450     conn->async_handler = handler;
1451     conn->async_handler_baton = handler_baton;
1452 }
1453
1454 static serf_request_t *
1455 create_request(serf_connection_t *conn,
1456                serf_request_setup_t setup,
1457                void *setup_baton,
1458                int priority,
1459                int ssltunnel)
1460 {
1461     serf_request_t *request;
1462
1463     request = serf_bucket_mem_alloc(conn->allocator, sizeof(*request));
1464     request->conn = conn;
1465     request->setup = setup;
1466     request->setup_baton = setup_baton;
1467     request->handler = NULL;
1468     request->respool = NULL;
1469     request->req_bkt = NULL;
1470     request->resp_bkt = NULL;
1471     request->priority = priority;
1472     request->written = 0;
1473     request->ssltunnel = ssltunnel;
1474     request->next = NULL;
1475
1476     return request;
1477 }
1478
1479 serf_request_t *serf_connection_request_create(
1480     serf_connection_t *conn,
1481     serf_request_setup_t setup,
1482     void *setup_baton)
1483 {
1484     serf_request_t *request;
1485
1486     request = create_request(conn, setup, setup_baton,
1487                              0, /* priority */
1488                              0  /* ssl tunnel */);
1489
1490     /* Link the request to the end of the request chain. */
1491     link_requests(&conn->requests, &conn->requests_tail, request);
1492
1493     /* Ensure our pollset becomes writable in context run */
1494     conn->ctx->dirty_pollset = 1;
1495     conn->dirty_conn = 1;
1496
1497     return request;
1498 }
1499
1500 static serf_request_t *
1501 priority_request_create(serf_connection_t *conn,
1502                         int ssltunnelreq,
1503                         serf_request_setup_t setup,
1504                         void *setup_baton)
1505 {
1506     serf_request_t *request;
1507     serf_request_t *iter, *prev;
1508
1509     request = create_request(conn, setup, setup_baton,
1510                              1, /* priority */
1511                              ssltunnelreq);
1512
1513     /* Link the new request after the last written request. */
1514     iter = conn->requests;
1515     prev = NULL;
1516
1517     /* Find a request that has data which needs to be delivered. */
1518     while (iter != NULL && iter->req_bkt == NULL && iter->written) {
1519         prev = iter;
1520         iter = iter->next;
1521     }
1522
1523     /* A CONNECT request to setup an ssltunnel has absolute priority over all
1524        other requests on the connection, so:
1525        a. add it first to the queue
1526        b. ensure that other priority requests are added after the CONNECT
1527           request */
1528     if (!request->ssltunnel) {
1529         /* Advance to next non priority request */
1530         while (iter != NULL && iter->priority) {
1531             prev = iter;
1532             iter = iter->next;
1533         }
1534     }
1535
1536     if (prev) {
1537         request->next = iter;
1538         prev->next = request;
1539     } else {
1540         request->next = iter;
1541         conn->requests = request;
1542     }
1543
1544     /* Ensure our pollset becomes writable in context run */
1545     conn->ctx->dirty_pollset = 1;
1546     conn->dirty_conn = 1;
1547
1548     return request;
1549 }
1550
1551 serf_request_t *serf_connection_priority_request_create(
1552     serf_connection_t *conn,
1553     serf_request_setup_t setup,
1554     void *setup_baton)
1555 {
1556     return priority_request_create(conn,
1557                                    0, /* not a ssltunnel CONNECT request */
1558                                    setup, setup_baton);
1559 }
1560
1561 serf_request_t *serf__ssltunnel_request_create(serf_connection_t *conn,
1562                                                serf_request_setup_t setup,
1563                                                void *setup_baton)
1564 {
1565     return priority_request_create(conn,
1566                                    1, /* This is a ssltunnel CONNECT request */
1567                                    setup, setup_baton);
1568 }
1569
1570 apr_status_t serf_request_cancel(serf_request_t *request)
1571 {
1572     return cancel_request(request, &request->conn->requests, 0);
1573 }
1574
1575 apr_status_t serf_request_is_written(serf_request_t *request)
1576 {
1577     if (request->written && !request->req_bkt)
1578         return APR_SUCCESS;
1579
1580     return APR_EBUSY;
1581 }
1582
1583 apr_pool_t *serf_request_get_pool(const serf_request_t *request)
1584 {
1585     return request->respool;
1586 }
1587
1588
1589 serf_bucket_alloc_t *serf_request_get_alloc(
1590     const serf_request_t *request)
1591 {
1592     return request->allocator;
1593 }
1594
1595
1596 serf_connection_t *serf_request_get_conn(
1597     const serf_request_t *request)
1598 {
1599     return request->conn;
1600 }
1601
1602
1603 void serf_request_set_handler(
1604     serf_request_t *request,
1605     const serf_response_handler_t handler,
1606     const void **handler_baton)
1607 {
1608     request->handler = handler;
1609     request->handler_baton = handler_baton;
1610 }
1611
1612
1613 serf_bucket_t *serf_request_bucket_request_create(
1614     serf_request_t *request,
1615     const char *method,
1616     const char *uri,
1617     serf_bucket_t *body,
1618     serf_bucket_alloc_t *allocator)
1619 {
1620     serf_bucket_t *req_bkt, *hdrs_bkt;
1621     serf_connection_t *conn = request->conn;
1622     serf_context_t *ctx = conn->ctx;
1623     int ssltunnel;
1624
1625     ssltunnel = ctx->proxy_address &&
1626                 (strcmp(conn->host_info.scheme, "https") == 0);
1627
1628     req_bkt = serf_bucket_request_create(method, uri, body, allocator);
1629     hdrs_bkt = serf_bucket_request_get_headers(req_bkt);
1630
1631     /* Use absolute uri's in requests to a proxy. USe relative uri's in
1632        requests directly to a server or sent through an SSL tunnel. */
1633     if (ctx->proxy_address && conn->host_url &&
1634         !(ssltunnel && !request->ssltunnel)) {
1635
1636         serf_bucket_request_set_root(req_bkt, conn->host_url);
1637     }
1638
1639     if (conn->host_info.hostinfo)
1640         serf_bucket_headers_setn(hdrs_bkt, "Host",
1641                                  conn->host_info.hostinfo);
1642
1643     /* Setup server authorization headers, unless this is a CONNECT request. */
1644     if (!request->ssltunnel) {
1645         serf__authn_info_t *authn_info;
1646         authn_info = serf__get_authn_info_for_server(conn);
1647         if (authn_info->scheme)
1648             authn_info->scheme->setup_request_func(HOST, 0, conn, request,
1649                                                    method, uri,
1650                                                    hdrs_bkt);
1651     }
1652
1653     /* Setup proxy authorization headers.
1654        Don't set these headers on the requests to the server if we're using
1655        an SSL tunnel, only on the CONNECT request to setup the tunnel. */
1656     if (ctx->proxy_authn_info.scheme) {
1657         if (strcmp(conn->host_info.scheme, "https") == 0) {
1658             if (request->ssltunnel)
1659                 ctx->proxy_authn_info.scheme->setup_request_func(PROXY, 0, conn,
1660                                                                  request,
1661                                                                  method, uri,
1662                                                                  hdrs_bkt);
1663         } else {
1664             ctx->proxy_authn_info.scheme->setup_request_func(PROXY, 0, conn,
1665                                                              request,
1666                                                              method, uri,
1667                                                              hdrs_bkt);
1668         }
1669     }
1670
1671     return req_bkt;
1672 }
1673
1674 apr_interval_time_t serf_connection_get_latency(serf_connection_t *conn)
1675 {
1676     if (conn->ctx->proxy_address) {
1677         /* Detecting network latency for proxied connection is not implemented
1678            yet. */
1679         return -1;
1680     }
1681
1682     return conn->latency;
1683 }