simgear/io/HTTPClient.cxx

   1 /**
   2  * \file HTTPClient.cxx - simple HTTP client engine for SimHear
   3  */
   4
   5 // Written by James Turner
   6 //
   7 // Copyright (C) 2013  James Turner  <zakalawe@mac.com>
   8 //
   9 // This library is free software; you can redistribute it and/or
  10 // modify it under the terms of the GNU Library General Public
  11 // License as published by the Free Software Foundation; either
  12 // version 2 of the License, or (at your option) any later version.
  13 //
  14 // This library is distributed in the hope that it will be useful,
  15 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  16 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  17 // Library General Public License for more details.
  18 //
  19 // You should have received a copy of the GNU General Public License
  20 // along with this program; if not, write to the Free Software
  21 // Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  22 //
  23
  24 #include "HTTPClient.hxx"
  25 #include "HTTPFileRequest.hxx"
  26
  27 #include <sstream>
  28 #include <cassert>
  29 #include <cstdlib> // rand()
  30 #include <list>
  31 #include <errno.h>
  32 #include <map>
  33 #include <stdexcept>
  34
  35 #include <boost/foreach.hpp>
  36 #include <boost/algorithm/string/case_conv.hpp>
  37
  38 #include <simgear/io/sg_netChat.hxx>
  39 #include <simgear/io/HTTPContentDecode.hxx>
  40 #include <simgear/misc/strutils.hxx>
  41 #include <simgear/compiler.h>
  42 #include <simgear/debug/logstream.hxx>
  43 #include <simgear/timing/timestamp.hxx>
  44 #include <simgear/structure/exception.hxx>
  45
  46 #if defined( HAVE_VERSION_H ) && HAVE_VERSION_H
  47 #include "version.h"
  48 #else
  49 #  if !defined(SIMGEAR_VERSION)
  50 #    define SIMGEAR_VERSION "simgear-development"
  51 #  endif
  52 #endif
  53
  54 namespace simgear
  55 {
  56
  57 namespace HTTP
  58 {
  59
  60 extern const int DEFAULT_HTTP_PORT = 80;
  61 const char* CONTENT_TYPE_URL_ENCODED = "application/x-www-form-urlencoded";
  62 const unsigned int MAX_INFLIGHT_REQUESTS = 32;
  63
  64 class Connection;
  65 typedef std::multimap<std::string, Connection*> ConnectionDict;
  66 typedef std::list<Request_ptr> RequestList;
  67
  68 static bool isFailureStatus(int httpStatus)
  69 {
  70   int majorCode = httpStatus / 100;
  71   return (majorCode != 2);
  72 }
  73
  74 class Client::ClientPrivate
  75 {
  76 public:
  77     std::string userAgent;
  78     std::string proxy;
  79     int proxyPort;
  80     std::string proxyAuth;
  81     NetChannelPoller poller;
  82     unsigned int maxConnections;
  83
  84     RequestList pendingRequests;
  85
  86 // connections by host (potentially more than one)
  87     ConnectionDict connections;
  88
  89     SGTimeStamp timeTransferSample;
  90     unsigned int bytesTransferred;
  91     unsigned int lastTransferRate;
  92     uint64_t totalBytesDownloaded;
  93 };
  94
  95 class Connection : public NetChat
  96 {
  97 public:
  98     Connection(Client* pr) :
  99         client(pr),
 100         state(STATE_CLOSED),
 101         port(DEFAULT_HTTP_PORT)
 102     {
 103     }
 104
 105     virtual ~Connection()
 106     {
 107     }
 108
 109     virtual void handleBufferRead (NetBuffer& buffer)
 110     {
 111       if( !activeRequest || !activeRequest->isComplete() )
 112         return NetChat::handleBufferRead(buffer);
 113
 114       // Request should be aborted (signaled by setting its state to complete).
 115
 116       // force the state to GETTING_BODY, to simplify logic in
 117       // responseComplete and handleClose
 118       state = STATE_GETTING_BODY;
 119       responseComplete();
 120     }
 121
 122     void setServer(const std::string& h, short p)
 123     {
 124         host = h;
 125         port = p;
 126     }
 127
 128     // socket-level errors
 129     virtual void handleError(int error)
 130     {
 131         if (error == ENOENT) {
 132         // name lookup failure
 133             // we won't have an active request yet, so the logic below won't
 134             // fire to actually call setFailure. Let's fail all of the requests
 135             BOOST_FOREACH(Request_ptr req, sentRequests) {
 136                 req->setFailure(error, "hostname lookup failure");
 137             }
 138
 139             BOOST_FOREACH(Request_ptr req, queuedRequests) {
 140                 req->setFailure(error, "hostname lookup failure");
 141             }
 142
 143         // name lookup failure, abandon all requests on this connection
 144             sentRequests.clear();
 145             queuedRequests.clear();
 146         }
 147
 148         NetChat::handleError(error);
 149         if (activeRequest) {
 150             SG_LOG(SG_IO, SG_INFO, "HTTP socket error");
 151             activeRequest->setFailure(error, "socket error");
 152             activeRequest = NULL;
 153             _contentDecoder.reset();
 154         }
 155
 156         state = STATE_SOCKET_ERROR;
 157     }
 158
 159     virtual void handleClose()
 160     {
 161         NetChat::handleClose();
 162
 163     // closing of the connection from the server side when getting the body,
 164         bool canCloseState = (state == STATE_GETTING_BODY);
 165         if (canCloseState && activeRequest) {
 166         // force state here, so responseComplete can avoid closing the
 167         // socket again
 168             state =  STATE_CLOSED;
 169             responseComplete();
 170         } else {
 171             if (activeRequest) {
 172                 activeRequest->setFailure(500, "server closed connection");
 173                 // remove the failed request from sentRequests, so it does
 174                 // not get restored
 175                 RequestList::iterator it = std::find(sentRequests.begin(),
 176                     sentRequests.end(), activeRequest);
 177                 if (it != sentRequests.end()) {
 178                     sentRequests.erase(it);
 179                 }
 180                 activeRequest = NULL;
 181                 _contentDecoder.reset();
 182             }
 183
 184             state = STATE_CLOSED;
 185         }
 186
 187       if (sentRequests.empty()) {
 188         return;
 189       }
 190
 191     // restore sent requests to the queue, so they will be re-sent
 192     // when the connection opens again
 193       queuedRequests.insert(queuedRequests.begin(),
 194                               sentRequests.begin(), sentRequests.end());
 195       sentRequests.clear();
 196     }
 197
 198     void handleTimeout()
 199     {
 200         NetChat::handleError(ETIMEDOUT);
 201         if (activeRequest) {
 202             SG_LOG(SG_IO, SG_DEBUG, "HTTP socket timeout");
 203             activeRequest->setFailure(ETIMEDOUT, "socket timeout");
 204             activeRequest = NULL;
 205             _contentDecoder.reset();
 206         }
 207
 208         state = STATE_SOCKET_ERROR;
 209     }
 210
 211     void queueRequest(const Request_ptr& r)
 212     {
 213         queuedRequests.push_back(r);
 214         tryStartNextRequest();
 215     }
 216
 217     void beginResponse()
 218     {
 219         assert(!sentRequests.empty());
 220         assert(state == STATE_WAITING_FOR_RESPONSE);
 221
 222         activeRequest = sentRequests.front();
 223         activeRequest->responseStart(buffer);
 224         if (isFailureStatus(activeRequest->responseCode())) {
 225           handleError(EIO);
 226           return;
 227         }
 228
 229       state = STATE_GETTING_HEADERS;
 230       buffer.clear();
 231       if (activeRequest->responseCode() == 204) {
 232         noMessageBody = true;
 233       } else if (activeRequest->method() == "HEAD") {
 234         noMessageBody = true;
 235       } else {
 236         noMessageBody = false;
 237       }
 238
 239       bodyTransferSize = -1;
 240       chunkedTransfer = false;
 241       _contentDecoder.reset();
 242     }
 243
 244     void tryStartNextRequest()
 245     {
 246       while( !queuedRequests.empty()
 247           && queuedRequests.front()->isComplete() )
 248         queuedRequests.pop_front();
 249
 250       if (queuedRequests.empty()) {
 251         idleTime.stamp();
 252         return;
 253       }
 254
 255       if (sentRequests.size() > MAX_INFLIGHT_REQUESTS) {
 256         return;
 257       }
 258
 259       if (state == STATE_CLOSED) {
 260           if (!connectToHost()) {
 261               return;
 262           }
 263
 264           setTerminator("\r\n");
 265           state = STATE_IDLE;
 266       }
 267
 268       Request_ptr r = queuedRequests.front();
 269       r->requestStart();
 270
 271       std::stringstream headerData;
 272       std::string path = r->path();
 273       assert(!path.empty());
 274       std::string query = r->query();
 275       std::string bodyData;
 276
 277       if (!client->proxyHost().empty()) {
 278           path = r->scheme() + "://" + r->host() + r->path();
 279       }
 280
 281       if (r->bodyType() == CONTENT_TYPE_URL_ENCODED) {
 282           headerData << r->method() << " " << path << " HTTP/1.1\r\n";
 283           bodyData = query.substr(1); // URL-encode, drop the leading '?'
 284           headerData << "Content-Type:" << CONTENT_TYPE_URL_ENCODED << "\r\n";
 285           headerData << "Content-Length:" << bodyData.size() << "\r\n";
 286       } else {
 287           headerData << r->method() << " " << path << query << " HTTP/1.1\r\n";
 288           if( r->hasBodyData() )
 289           {
 290             headerData << "Content-Length:" << r->bodyLength() << "\r\n";
 291             headerData << "Content-Type:" << r->bodyType() << "\r\n";
 292           }
 293       }
 294
 295       headerData << "Host: " << r->hostAndPort() << "\r\n";
 296       headerData << "User-Agent:" << client->userAgent() << "\r\n";
 297       headerData << "Accept-Encoding: deflate, gzip\r\n";
 298       if (!client->proxyAuth().empty()) {
 299           headerData << "Proxy-Authorization: " << client->proxyAuth() << "\r\n";
 300       }
 301
 302       BOOST_FOREACH(const StringMap::value_type& h, r->requestHeaders()) {
 303           headerData << h.first << ": " << h.second << "\r\n";
 304       }
 305
 306       headerData << "\r\n"; // final CRLF to terminate the headers
 307       if (!bodyData.empty()) {
 308           headerData << bodyData;
 309       }
 310
 311       bool ok = push(headerData.str().c_str());
 312       if (!ok) {
 313           SG_LOG(SG_IO, SG_WARN, "HTTPClient: over-stuffed the socket");
 314           // we've over-stuffed the socket, give up for now, let things
 315           // drain down before trying to start any more requests.
 316           return;
 317       }
 318
 319       if( r->hasBodyData() )
 320         for(size_t body_bytes_sent = 0; body_bytes_sent < r->bodyLength();)
 321         {
 322           char buf[4096];
 323           size_t len = r->getBodyData(buf, body_bytes_sent, 4096);
 324           if( len )
 325           {
 326             if( !bufferSend(buf, len) )
 327             {
 328               SG_LOG(SG_IO,
 329                      SG_WARN,
 330                      "overflow the HTTP::Connection output buffer");
 331               state = STATE_SOCKET_ERROR;
 332               return;
 333             }
 334             body_bytes_sent += len;
 335           }
 336           else
 337           {
 338             SG_LOG(SG_IO,
 339                    SG_WARN,
 340                    "HTTP asynchronous request body generation is unsupported");
 341             break;
 342           }
 343         }
 344
 345       //   SG_LOG(SG_IO, SG_INFO, "did start request:" << r->url() <<
 346       //       "\n\t @ " << reinterpret_cast<void*>(r.ptr()) <<
 347       //      "\n\t on connection " << this);
 348       // successfully sent, remove from queue, and maybe send the next
 349       queuedRequests.pop_front();
 350       sentRequests.push_back(r);
 351       state = STATE_WAITING_FOR_RESPONSE;
 352
 353       // pipelining, let's maybe send the next request right away
 354       tryStartNextRequest();
 355     }
 356
 357     virtual void collectIncomingData(const char* s, int n)
 358     {
 359         idleTime.stamp();
 360         client->receivedBytes(static_cast<unsigned int>(n));
 361
 362         if(   (state == STATE_GETTING_BODY)
 363            || (state == STATE_GETTING_CHUNKED_BYTES) )
 364           _contentDecoder.receivedBytes(s, n);
 365         else
 366           buffer.append(s, n);
 367     }
 368
 369     virtual void foundTerminator(void)
 370     {
 371         idleTime.stamp();
 372         switch (state) {
 373         case STATE_WAITING_FOR_RESPONSE:
 374             beginResponse();
 375             break;
 376
 377         case STATE_GETTING_HEADERS:
 378             processHeader();
 379             buffer.clear();
 380             break;
 381
 382         case STATE_GETTING_BODY:
 383             responseComplete();
 384             break;
 385
 386         case STATE_GETTING_CHUNKED:
 387             processChunkHeader();
 388             break;
 389
 390         case STATE_GETTING_CHUNKED_BYTES:
 391             setTerminator("\r\n");
 392             state = STATE_GETTING_CHUNKED;
 393             buffer.clear();
 394             break;
 395
 396
 397         case STATE_GETTING_TRAILER:
 398             processTrailer();
 399             buffer.clear();
 400             break;
 401
 402         case STATE_IDLE:
 403             SG_LOG(SG_IO, SG_WARN, "HTTP got data in IDLE state, bad server?");
 404
 405         default:
 406             break;
 407         }
 408     }
 409
 410     bool hasIdleTimeout() const
 411     {
 412         if (state != STATE_IDLE) {
 413             return false;
 414         }
 415
 416         assert(sentRequests.empty());
 417         return idleTime.elapsedMSec() > 1000 * 10; // ten seconds
 418     }
 419
 420     bool hasErrorTimeout() const
 421     {
 422       if (state == STATE_IDLE) {
 423         return false;
 424       }
 425
 426       return idleTime.elapsedMSec() > (1000 * 30); // 30 seconds
 427     }
 428
 429     bool hasError() const
 430     {
 431         return (state == STATE_SOCKET_ERROR);
 432     }
 433
 434     bool shouldStartNext() const
 435     {
 436       return !queuedRequests.empty() && (sentRequests.size() < MAX_INFLIGHT_REQUESTS);
 437     }
 438
 439     bool isActive() const
 440     {
 441         return !queuedRequests.empty() || !sentRequests.empty();
 442     }
 443 private:
 444     bool connectToHost()
 445     {
 446         SG_LOG(SG_IO, SG_DEBUG, "HTTP connecting to " << host << ":" << port);
 447
 448         if (!open()) {
 449             SG_LOG(SG_ALL, SG_WARN, "HTTP::Connection: connectToHost: open() failed");
 450             return false;
 451         }
 452
 453         if (connect(host.c_str(), port) != 0) {
 454             return false;
 455         }
 456
 457         return true;
 458     }
 459
 460
 461     void processHeader()
 462     {
 463         std::string h = strutils::simplify(buffer);
 464         if (h.empty()) { // blank line terminates headers
 465             headersComplete();
 466             return;
 467         }
 468
 469         int colonPos = buffer.find(':');
 470         if (colonPos < 0) {
 471             SG_LOG(SG_IO, SG_WARN, "malformed HTTP response header:" << h);
 472             return;
 473         }
 474
 475         std::string key = strutils::simplify(buffer.substr(0, colonPos));
 476         std::string lkey = boost::to_lower_copy(key);
 477         std::string value = strutils::strip(buffer.substr(colonPos + 1));
 478
 479         // only consider these if getting headers (as opposed to trailers
 480         // of a chunked transfer)
 481         if (state == STATE_GETTING_HEADERS) {
 482             if (lkey == "content-length") {
 483
 484                 int sz = strutils::to_int(value);
 485                 if (bodyTransferSize <= 0) {
 486                     bodyTransferSize = sz;
 487                 }
 488                 activeRequest->setResponseLength(sz);
 489             } else if (lkey == "transfer-length") {
 490                 bodyTransferSize = strutils::to_int(value);
 491             } else if (lkey == "transfer-encoding") {
 492                 processTransferEncoding(value);
 493             } else if (lkey == "content-encoding") {
 494                 _contentDecoder.setEncoding(value);
 495             }
 496         }
 497
 498         activeRequest->responseHeader(lkey, value);
 499     }
 500
 501     void processTransferEncoding(const std::string& te)
 502     {
 503         if (te == "chunked") {
 504             chunkedTransfer = true;
 505         } else {
 506             SG_LOG(SG_IO, SG_WARN, "unsupported transfer encoding:" << te);
 507             // failure
 508         }
 509     }
 510
 511     void processChunkHeader()
 512     {
 513         if (buffer.empty()) {
 514             // blank line after chunk data
 515             return;
 516         }
 517
 518         int chunkSize = 0;
 519         int semiPos = buffer.find(';');
 520         if (semiPos >= 0) {
 521             // extensions ignored for the moment
 522             chunkSize = strutils::to_int(buffer.substr(0, semiPos), 16);
 523         } else {
 524             chunkSize = strutils::to_int(buffer, 16);
 525         }
 526
 527         buffer.clear();
 528         if (chunkSize == 0) {  //  trailer start
 529             state = STATE_GETTING_TRAILER;
 530             return;
 531         }
 532
 533         state = STATE_GETTING_CHUNKED_BYTES;
 534         setByteCount(chunkSize);
 535     }
 536
 537     void processTrailer()
 538     {
 539         if (buffer.empty()) {
 540             // end of trailers
 541             responseComplete();
 542             return;
 543         }
 544
 545     // process as a normal header
 546         processHeader();
 547     }
 548
 549     void headersComplete()
 550     {
 551         activeRequest->responseHeadersComplete();
 552         _contentDecoder.initWithRequest(activeRequest);
 553
 554         if (chunkedTransfer) {
 555             state = STATE_GETTING_CHUNKED;
 556         } else if (noMessageBody || (bodyTransferSize == 0)) {
 557             // force the state to GETTING_BODY, to simplify logic in
 558             // responseComplete and handleClose
 559             state = STATE_GETTING_BODY;
 560             responseComplete();
 561         } else {
 562             setByteCount(bodyTransferSize); // may be -1, that's fine
 563             state = STATE_GETTING_BODY;
 564         }
 565     }
 566
 567     void responseComplete()
 568     {
 569         Request_ptr completedRequest = activeRequest;
 570         _contentDecoder.finish();
 571
 572         assert(sentRequests.front() == activeRequest);
 573         sentRequests.pop_front();
 574         bool doClose = activeRequest->closeAfterComplete();
 575         activeRequest = NULL;
 576
 577         if ((state == STATE_GETTING_BODY) || (state == STATE_GETTING_TRAILER)) {
 578             if (doClose) {
 579           // this will bring us into handleClose() above, which updates
 580           // state to STATE_CLOSED
 581               close();
 582
 583           // if we have additional requests waiting, try to start them now
 584               tryStartNextRequest();
 585             }
 586         }
 587
 588         if (state != STATE_CLOSED)  {
 589             state = sentRequests.empty() ? STATE_IDLE : STATE_WAITING_FOR_RESPONSE;
 590         }
 591
 592     // notify request after we change state, so this connection is idle
 593     // if completion triggers other requests (which is likely)
 594         //   SG_LOG(SG_IO, SG_INFO, "*** responseComplete:" << activeRequest->url());
 595         completedRequest->responseComplete();
 596         client->requestFinished(this);
 597
 598         setTerminator("\r\n");
 599     }
 600
 601     enum ConnectionState {
 602         STATE_IDLE = 0,
 603         STATE_WAITING_FOR_RESPONSE,
 604         STATE_GETTING_HEADERS,
 605         STATE_GETTING_BODY,
 606         STATE_GETTING_CHUNKED,
 607         STATE_GETTING_CHUNKED_BYTES,
 608         STATE_GETTING_TRAILER,
 609         STATE_SOCKET_ERROR,
 610         STATE_CLOSED             ///< connection should be closed now
 611     };
 612
 613     Client* client;
 614     Request_ptr activeRequest;
 615     ConnectionState state;
 616     std::string host;
 617     short port;
 618     std::string buffer;
 619     int bodyTransferSize;
 620     SGTimeStamp idleTime;
 621     bool chunkedTransfer;
 622     bool noMessageBody;
 623
 624     RequestList queuedRequests;
 625     RequestList sentRequests;
 626
 627     ContentDecoder _contentDecoder;
 628 };
 629
 630 Client::Client() :
 631     d(new ClientPrivate)
 632 {
 633     d->proxyPort = 0;
 634     d->maxConnections = 4;
 635     d->bytesTransferred = 0;
 636     d->lastTransferRate = 0;
 637     d->timeTransferSample.stamp();
 638     d->totalBytesDownloaded = 0;
 639
 640     setUserAgent("SimGear-" SG_STRINGIZE(SIMGEAR_VERSION));
 641 }
 642
 643 Client::~Client()
 644 {
 645 }
 646
 647 void Client::setMaxConnections(unsigned int maxCon)
 648 {
 649     if (maxCon < 1) {
 650         throw sg_range_exception("illegal HTTP::Client::setMaxConnections value");
 651     }
 652
 653     d->maxConnections = maxCon;
 654 }
 655
 656 void Client::update(int waitTimeout)
 657 {
 658     if (!d->poller.hasChannels() && (waitTimeout > 0)) {
 659         SGTimeStamp::sleepForMSec(waitTimeout);
 660     } else {
 661         d->poller.poll(waitTimeout);
 662     }
 663
 664     bool waitingRequests = !d->pendingRequests.empty();
 665     ConnectionDict::iterator it = d->connections.begin();
 666     for (; it != d->connections.end(); ) {
 667         Connection* con = it->second;
 668         if (con->hasIdleTimeout() ||
 669             con->hasError() ||
 670             con->hasErrorTimeout() ||
 671             (!con->isActive() && waitingRequests))
 672         {
 673             if (con->hasErrorTimeout()) {
 674                 // tell the connection we're timing it out
 675                 con->handleTimeout();
 676             }
 677
 678         // connection has been idle for a while, clean it up
 679         // (or if we have requests waiting for a different host,
 680         // or an error condition
 681             ConnectionDict::iterator del = it++;
 682             delete del->second;
 683             d->connections.erase(del);
 684         } else {
 685             if (it->second->shouldStartNext()) {
 686                 it->second->tryStartNextRequest();
 687             }
 688             ++it;
 689         }
 690     } // of connection iteration
 691
 692     if (waitingRequests && (d->connections.size() < d->maxConnections)) {
 693         RequestList waiting(d->pendingRequests);
 694         d->pendingRequests.clear();
 695
 696         // re-submit all waiting requests in order; this takes care of
 697         // finding multiple pending items targetted to the same (new)
 698         // connection
 699         BOOST_FOREACH(Request_ptr req, waiting) {
 700             makeRequest(req);
 701         }
 702     }
 703 }
 704
 705 void Client::makeRequest(const Request_ptr& r)
 706 {
 707     if( r->isComplete() )
 708       return;
 709
 710     if( r->url().find("://") == std::string::npos ) {
 711         r->setFailure(EINVAL, "malformed URL");
 712         return;
 713     }
 714
 715     if( r->url().find("http://") != 0 ) {
 716         r->setFailure(EINVAL, "only HTTP protocol is supported");
 717         return;
 718     }
 719
 720     std::string host = r->host();
 721     int port = r->port();
 722     if (!d->proxy.empty()) {
 723         host = d->proxy;
 724         port = d->proxyPort;
 725     }
 726
 727     Connection* con = NULL;
 728     std::stringstream ss;
 729     ss << host << "-" << port;
 730     std::string connectionId = ss.str();
 731     bool havePending = !d->pendingRequests.empty();
 732     bool atConnectionsLimit = d->connections.size() >= d->maxConnections;
 733     ConnectionDict::iterator consEnd = d->connections.end();
 734
 735     // assign request to an existing Connection.
 736     // various options exist here, examined in order
 737     ConnectionDict::iterator it = d->connections.find(connectionId);
 738     if (atConnectionsLimit && (it == consEnd)) {
 739         // maximum number of connections active, queue this request
 740         // when a connection goes inactive, we'll start this one
 741         d->pendingRequests.push_back(r);
 742         return;
 743     }
 744
 745     // scan for an idle Connection to the same host (likely if we're
 746     // retrieving multiple resources from the same host in quick succession)
 747     // if we have pending requests (waiting for a free Connection), then
 748     // force new requests on this id to always use the first Connection
 749     // (instead of the random selection below). This ensures that when
 750     // there's pressure on the number of connections to keep alive, one
 751     // host can't DoS every other.
 752     int count = 0;
 753     for (; (it != consEnd) && (it->first == connectionId); ++it, ++count) {
 754         if (havePending || !it->second->isActive()) {
 755             con = it->second;
 756             break;
 757         }
 758     }
 759
 760     if (!con && atConnectionsLimit) {
 761         // all current connections are busy (active), and we don't
 762         // have free connections to allocate, so let's assign to
 763         // an existing one randomly. Ideally we'd used whichever one will
 764         // complete first but we don't have that info.
 765         int index = rand() % count;
 766         for (it = d->connections.find(connectionId); index > 0; --index) { ; }
 767         con = it->second;
 768     }
 769
 770     // allocate a new connection object
 771     if (!con) {
 772         con = new Connection(this);
 773         con->setServer(host, port);
 774         d->poller.addChannel(con);
 775         d->connections.insert(d->connections.end(),
 776             ConnectionDict::value_type(connectionId, con));
 777     }
 778
 779     con->queueRequest(r);
 780 }
 781
 782 //------------------------------------------------------------------------------
 783 FileRequestRef Client::save( const std::string& url,
 784                              const std::string& filename )
 785 {
 786   FileRequestRef req = new FileRequest(url, filename);
 787   makeRequest(req);
 788   return req;
 789 }
 790
 791 //------------------------------------------------------------------------------
 792 MemoryRequestRef Client::load(const std::string& url)
 793 {
 794   MemoryRequestRef req = new MemoryRequest(url);
 795   makeRequest(req);
 796   return req;
 797 }
 798
 799 void Client::requestFinished(Connection* con)
 800 {
 801
 802 }
 803
 804 void Client::setUserAgent(const std::string& ua)
 805 {
 806     d->userAgent = ua;
 807 }
 808
 809 const std::string& Client::userAgent() const
 810 {
 811     return d->userAgent;
 812 }
 813
 814 const std::string& Client::proxyHost() const
 815 {
 816     return d->proxy;
 817 }
 818
 819 const std::string& Client::proxyAuth() const
 820 {
 821     return d->proxyAuth;
 822 }
 823
 824 void Client::setProxy( const std::string& proxy,
 825                        int port,
 826                        const std::string& auth )
 827 {
 828     d->proxy = proxy;
 829     d->proxyPort = port;
 830     d->proxyAuth = auth;
 831 }
 832
 833 bool Client::hasActiveRequests() const
 834 {
 835     ConnectionDict::const_iterator it = d->connections.begin();
 836     for (; it != d->connections.end(); ++it) {
 837         if (it->second->isActive()) return true;
 838     }
 839
 840     return false;
 841 }
 842
 843 void Client::receivedBytes(unsigned int count)
 844 {
 845     d->bytesTransferred += count;
 846     d->totalBytesDownloaded += count;
 847 }
 848
 849 unsigned int Client::transferRateBytesPerSec() const
 850 {
 851     unsigned int e = d->timeTransferSample.elapsedMSec();
 852     if (e > 400) {
 853         // too long a window, ignore
 854         d->timeTransferSample.stamp();
 855         d->bytesTransferred = 0;
 856         d->lastTransferRate = 0;
 857         return 0;
 858     }
 859
 860     if (e < 100) { // avoid really narrow windows
 861         return d->lastTransferRate;
 862     }
 863
 864     unsigned int ratio = (d->bytesTransferred * 1000) / e;
 865     // run a low-pass filter
 866     unsigned int smoothed = ((400 - e) * d->lastTransferRate) + (e * ratio);
 867     smoothed /= 400;
 868
 869     d->timeTransferSample.stamp();
 870     d->bytesTransferred = 0;
 871     d->lastTransferRate = smoothed;
 872     return smoothed;
 873 }
 874
 875 uint64_t Client::totalBytesDownloaded() const
 876 {
 877     return d->totalBytesDownloaded;
 878 }
 879
 880 } // of namespace HTTP
 881
 882 } // of namespace simgear