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