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