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