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
  25 #include "HTTPClient.hxx"
  26 #include "HTTPFileRequest.hxx"
  27
  28 #include <sstream>
  29 #include <cassert>
  30 #include <cstdlib> // rand()
  31 #include <list>
  32 #include <errno.h>
  33 #include <map>
  34 #include <stdexcept>
  35
  36 #include <boost/foreach.hpp>
  37 #include <boost/algorithm/string/case_conv.hpp>
  38
  39 #include <simgear/simgear_config.h>
  40
  41 #if defined(ENABLE_CURL)
  42   #include <curl/multi.h>
  43 #else
  44     #include <simgear/io/HTTPContentDecode.hxx>
  45 #endif
  46
  47 #include <simgear/io/sg_netChat.hxx>
  48
  49 #include <simgear/misc/strutils.hxx>
  50 #include <simgear/compiler.h>
  51 #include <simgear/debug/logstream.hxx>
  52 #include <simgear/timing/timestamp.hxx>
  53 #include <simgear/structure/exception.hxx>
  54
  55 #if defined( HAVE_VERSION_H ) && HAVE_VERSION_H
  56 #include "version.h"
  57 #else
  58 #  if !defined(SIMGEAR_VERSION)
  59 #    define SIMGEAR_VERSION "simgear-development"
  60 #  endif
  61 #endif
  62
  63 namespace simgear
  64 {
  65
  66 namespace HTTP
  67 {
  68
  69 extern const int DEFAULT_HTTP_PORT = 80;
  70 const char* CONTENT_TYPE_URL_ENCODED = "application/x-www-form-urlencoded";
  71 const unsigned int MAX_INFLIGHT_REQUESTS = 32;
  72
  73 class Connection;
  74 typedef std::multimap<std::string, Connection*> ConnectionDict;
  75 typedef std::list<Request_ptr> RequestList;
  76
  77 class Client::ClientPrivate
  78 {
  79 public:
  80 #if defined(ENABLE_CURL)
  81     CURLM* curlMulti;
  82     bool haveActiveRequests;
  83
  84     void createCurlMulti()
  85     {
  86         curlMulti = curl_multi_init();
  87         // see https://curl.haxx.se/libcurl/c/CURLMOPT_PIPELINING.html
  88         // we request HTTP 1.1 pipelining
  89         curl_multi_setopt(curlMulti, CURLMOPT_PIPELINING, CURLPIPE_HTTP1);
  90         curl_multi_setopt(curlMulti, CURLMOPT_MAX_TOTAL_CONNECTIONS, (long) maxConnections);
  91         curl_multi_setopt(curlMulti, CURLMOPT_MAX_PIPELINE_LENGTH,
  92                           (long) MAX_INFLIGHT_REQUESTS);
  93
  94     }
  95 #else
  96     NetChannelPoller poller;
  97 // connections by host (potentially more than one)
  98     ConnectionDict connections;
  99 #endif
 100
 101     std::string userAgent;
 102     std::string proxy;
 103     int proxyPort;
 104     std::string proxyAuth;
 105     unsigned int maxConnections;
 106
 107     RequestList pendingRequests;
 108
 109
 110
 111     SGTimeStamp timeTransferSample;
 112     unsigned int bytesTransferred;
 113     unsigned int lastTransferRate;
 114     uint64_t totalBytesDownloaded;
 115 };
 116
 117 #if !defined(ENABLE_CURL)
 118 class Connection : public NetChat
 119 {
 120 public:
 121     Connection(Client* pr, const std::string& conId) :
 122         client(pr),
 123         state(STATE_CLOSED),
 124         port(DEFAULT_HTTP_PORT),
 125         connectionId(conId)
 126     {
 127     }
 128
 129     virtual ~Connection()
 130     {
 131     }
 132
 133     virtual void handleBufferRead (NetBuffer& buffer)
 134     {
 135       if( !activeRequest || !activeRequest->isComplete() )
 136         return NetChat::handleBufferRead(buffer);
 137
 138       // Request should be aborted (signaled by setting its state to complete).
 139
 140       // force the state to GETTING_BODY, to simplify logic in
 141       // responseComplete and handleClose
 142       state = STATE_GETTING_BODY;
 143       responseComplete();
 144     }
 145
 146     void setServer(const std::string& h, short p)
 147     {
 148         host = h;
 149         port = p;
 150     }
 151
 152     // socket-level errors
 153     virtual void handleError(int error)
 154     {
 155         const char* errStr = strerror(error);
 156         SG_LOG(SG_IO, SG_WARN, "HTTP Connection handleError:" << error << " ("
 157                << errStr << ")");
 158
 159         debugDumpRequests();
 160
 161         if (!activeRequest)
 162         {
 163         // connection level failure, eg name lookup or routing
 164         // we won't have an active request yet, so let's fail all of the
 165         // requests since we presume it's a systematic failure for
 166         // the host in question
 167             BOOST_FOREACH(Request_ptr req, sentRequests) {
 168                 req->setFailure(error, errStr);
 169             }
 170
 171             BOOST_FOREACH(Request_ptr req, queuedRequests) {
 172                 req->setFailure(error, errStr);
 173             }
 174
 175             sentRequests.clear();
 176             queuedRequests.clear();
 177         }
 178
 179         NetChat::handleError(error);
 180         if (activeRequest) {
 181             activeRequest->setFailure(error, errStr);
 182             activeRequest = NULL;
 183             _contentDecoder.reset();
 184         }
 185
 186         state = STATE_SOCKET_ERROR;
 187     }
 188
 189     void handleTimeout()
 190     {
 191         handleError(ETIMEDOUT);
 192     }
 193
 194     virtual void handleClose()
 195     {
 196         NetChat::handleClose();
 197
 198     // closing of the connection from the server side when getting the body,
 199         bool canCloseState = (state == STATE_GETTING_BODY);
 200         if (canCloseState && activeRequest) {
 201         // force state here, so responseComplete can avoid closing the
 202         // socket again
 203             state =  STATE_CLOSED;
 204             responseComplete();
 205         } else {
 206             if (state == STATE_WAITING_FOR_RESPONSE) {
 207                 assert(!sentRequests.empty());
 208                 sentRequests.front()->setFailure(500, "server closed connection unexpectedly");
 209                 // no active request, but don't restore the front sent one
 210                 sentRequests.erase(sentRequests.begin());
 211             }
 212
 213             if (activeRequest) {
 214                 activeRequest->setFailure(500, "server closed connection");
 215                 // remove the failed request from sentRequests, so it does
 216                 // not get restored
 217                 RequestList::iterator it = std::find(sentRequests.begin(),
 218                     sentRequests.end(), activeRequest);
 219                 if (it != sentRequests.end()) {
 220                     sentRequests.erase(it);
 221                 }
 222                 activeRequest = NULL;
 223                 _contentDecoder.reset();
 224             }
 225
 226             state = STATE_CLOSED;
 227         }
 228
 229       if (sentRequests.empty()) {
 230         return;
 231       }
 232
 233     // restore sent requests to the queue, so they will be re-sent
 234     // when the connection opens again
 235       queuedRequests.insert(queuedRequests.begin(),
 236                               sentRequests.begin(), sentRequests.end());
 237       sentRequests.clear();
 238     }
 239
 240     void queueRequest(const Request_ptr& r)
 241     {
 242         queuedRequests.push_back(r);
 243         tryStartNextRequest();
 244     }
 245
 246     void beginResponse()
 247     {
 248         assert(!sentRequests.empty());
 249         assert(state == STATE_WAITING_FOR_RESPONSE);
 250
 251         activeRequest = sentRequests.front();
 252         try {
 253             activeRequest->responseStart(buffer);
 254         } catch (sg_exception& e) {
 255             handleError(EIO);
 256             return;
 257         }
 258
 259       state = STATE_GETTING_HEADERS;
 260       buffer.clear();
 261       if (activeRequest->responseCode() == 204) {
 262         noMessageBody = true;
 263       } else if (activeRequest->method() == "HEAD") {
 264         noMessageBody = true;
 265       } else {
 266         noMessageBody = false;
 267       }
 268
 269       bodyTransferSize = -1;
 270       chunkedTransfer = false;
 271       _contentDecoder.reset();
 272     }
 273
 274     void tryStartNextRequest()
 275     {
 276       while( !queuedRequests.empty()
 277           && queuedRequests.front()->isComplete() )
 278         queuedRequests.pop_front();
 279
 280       if (queuedRequests.empty()) {
 281         idleTime.stamp();
 282         return;
 283       }
 284
 285       if (sentRequests.size() > MAX_INFLIGHT_REQUESTS) {
 286         return;
 287       }
 288
 289       if (state == STATE_CLOSED) {
 290           if (!connectToHost()) {
 291
 292               return;
 293           }
 294
 295           setTerminator("\r\n");
 296           state = STATE_IDLE;
 297       }
 298
 299       Request_ptr r = queuedRequests.front();
 300       r->requestStart();
 301
 302       std::stringstream headerData;
 303       std::string path = r->path();
 304       assert(!path.empty());
 305       std::string query = r->query();
 306       std::string bodyData;
 307
 308       if (!client->proxyHost().empty()) {
 309           path = r->scheme() + "://" + r->host() + r->path();
 310       }
 311
 312       if (r->bodyType() == CONTENT_TYPE_URL_ENCODED) {
 313           headerData << r->method() << " " << path << " HTTP/1.1\r\n";
 314           bodyData = query.substr(1); // URL-encode, drop the leading '?'
 315           headerData << "Content-Type:" << CONTENT_TYPE_URL_ENCODED << "\r\n";
 316           headerData << "Content-Length:" << bodyData.size() << "\r\n";
 317       } else {
 318           headerData << r->method() << " " << path << query << " HTTP/1.1\r\n";
 319           if( r->hasBodyData() )
 320           {
 321             headerData << "Content-Length:" << r->bodyLength() << "\r\n";
 322             headerData << "Content-Type:" << r->bodyType() << "\r\n";
 323           }
 324       }
 325
 326       headerData << "Host: " << r->hostAndPort() << "\r\n";
 327       headerData << "User-Agent:" << client->userAgent() << "\r\n";
 328       headerData << "Accept-Encoding: deflate, gzip\r\n";
 329       if (!client->proxyAuth().empty()) {
 330           headerData << "Proxy-Authorization: " << client->proxyAuth() << "\r\n";
 331       }
 332
 333       BOOST_FOREACH(const StringMap::value_type& h, r->requestHeaders()) {
 334           headerData << h.first << ": " << h.second << "\r\n";
 335       }
 336
 337       headerData << "\r\n"; // final CRLF to terminate the headers
 338       if (!bodyData.empty()) {
 339           headerData << bodyData;
 340       }
 341
 342       bool ok = push(headerData.str().c_str());
 343       if (!ok) {
 344           SG_LOG(SG_IO, SG_WARN, "HTTPClient: over-stuffed the socket");
 345           // we've over-stuffed the socket, give up for now, let things
 346           // drain down before trying to start any more requests.
 347           return;
 348       }
 349
 350       if( r->hasBodyData() )
 351         for(size_t body_bytes_sent = 0; body_bytes_sent < r->bodyLength();)
 352         {
 353           char buf[4096];
 354           size_t len = r->getBodyData(buf, body_bytes_sent, 4096);
 355           if( len )
 356           {
 357             if( !bufferSend(buf, len) )
 358             {
 359               SG_LOG(SG_IO,
 360                      SG_WARN,
 361                      "overflow the HTTP::Connection output buffer");
 362               state = STATE_SOCKET_ERROR;
 363               return;
 364             }
 365             body_bytes_sent += len;
 366           }
 367           else
 368           {
 369             SG_LOG(SG_IO,
 370                    SG_WARN,
 371                    "HTTP asynchronous request body generation is unsupported");
 372             break;
 373           }
 374         }
 375
 376         SG_LOG(SG_IO, SG_DEBUG, "con:" << connectionId << " did start request:" << r->url());
 377       // successfully sent, remove from queue, and maybe send the next
 378       queuedRequests.pop_front();
 379       sentRequests.push_back(r);
 380       state = STATE_WAITING_FOR_RESPONSE;
 381
 382       // pipelining, let's maybe send the next request right away
 383       tryStartNextRequest();
 384     }
 385
 386     virtual void collectIncomingData(const char* s, int n)
 387     {
 388         idleTime.stamp();
 389         client->receivedBytes(static_cast<unsigned int>(n));
 390
 391         if(   (state == STATE_GETTING_BODY)
 392            || (state == STATE_GETTING_CHUNKED_BYTES) )
 393           _contentDecoder.receivedBytes(s, n);
 394         else
 395           buffer.append(s, n);
 396     }
 397
 398     virtual void foundTerminator(void)
 399     {
 400         idleTime.stamp();
 401         switch (state) {
 402         case STATE_WAITING_FOR_RESPONSE:
 403             beginResponse();
 404             break;
 405
 406         case STATE_GETTING_HEADERS:
 407             processHeader();
 408             buffer.clear();
 409             break;
 410
 411         case STATE_GETTING_BODY:
 412             responseComplete();
 413             break;
 414
 415         case STATE_GETTING_CHUNKED:
 416             processChunkHeader();
 417             break;
 418
 419         case STATE_GETTING_CHUNKED_BYTES:
 420             setTerminator("\r\n");
 421             state = STATE_GETTING_CHUNKED;
 422             buffer.clear();
 423             break;
 424
 425
 426         case STATE_GETTING_TRAILER:
 427             processTrailer();
 428             buffer.clear();
 429             break;
 430
 431         case STATE_IDLE:
 432             SG_LOG(SG_IO, SG_WARN, "HTTP got data in IDLE state, bad server?");
 433
 434         default:
 435             break;
 436         }
 437     }
 438
 439     bool hasIdleTimeout() const
 440     {
 441         if ((state != STATE_IDLE) && (state != STATE_CLOSED)) {
 442             return false;
 443         }
 444
 445         assert(sentRequests.empty());
 446         bool isTimedOut = (idleTime.elapsedMSec() > (1000 * 10)); // 10 seconds
 447         return isTimedOut;
 448     }
 449
 450     bool hasErrorTimeout() const
 451     {
 452       if ((state == STATE_IDLE) || (state == STATE_CLOSED)) {
 453         return false;
 454       }
 455
 456         bool isTimedOut = (idleTime.elapsedMSec() > (1000 * 30)); // 30 seconds
 457         return isTimedOut;
 458     }
 459
 460     bool hasError() const
 461     {
 462         return (state == STATE_SOCKET_ERROR);
 463     }
 464
 465     bool shouldStartNext() const
 466     {
 467       return !queuedRequests.empty() && (sentRequests.size() < MAX_INFLIGHT_REQUESTS);
 468     }
 469
 470     bool isActive() const
 471     {
 472         return !queuedRequests.empty() || !sentRequests.empty();
 473     }
 474
 475     void debugDumpRequests() const
 476     {
 477         SG_LOG(SG_IO, SG_DEBUG, "requests for:" << host << ":" << port << " (conId=" << connectionId
 478                << "; state=" << state << ")");
 479         if (activeRequest) {
 480             SG_LOG(SG_IO, SG_DEBUG, "\tactive:" << activeRequest->url());
 481         } else {
 482             SG_LOG(SG_IO, SG_DEBUG, "\tNo active request");
 483         }
 484
 485         BOOST_FOREACH(Request_ptr req, sentRequests) {
 486             SG_LOG(SG_IO, SG_DEBUG, "\tsent:" << req->url());
 487         }
 488
 489         BOOST_FOREACH(Request_ptr req, queuedRequests) {
 490             SG_LOG(SG_IO, SG_DEBUG, "\tqueued:" << req->url());
 491         }
 492     }
 493 private:
 494     bool connectToHost()
 495     {
 496         SG_LOG(SG_IO, SG_DEBUG, "HTTP connecting to " << host << ":" << port);
 497
 498         if (!open()) {
 499             SG_LOG(SG_IO, SG_WARN, "HTTP::Connection: connectToHost: open() failed");
 500             return false;
 501         }
 502
 503         if (connect(host.c_str(), port) != 0) {
 504             SG_LOG(SG_IO, SG_WARN, "HTTP::Connection: connectToHost: connect() failed");
 505             return false;
 506         }
 507
 508         return true;
 509     }
 510
 511
 512     void processHeader()
 513     {
 514         std::string h = strutils::simplify(buffer);
 515         if (h.empty()) { // blank line terminates headers
 516             headersComplete();
 517             return;
 518         }
 519
 520         int colonPos = buffer.find(':');
 521         if (colonPos < 0) {
 522             SG_LOG(SG_IO, SG_WARN, "malformed HTTP response header:" << h);
 523             return;
 524         }
 525
 526         std::string key = strutils::simplify(buffer.substr(0, colonPos));
 527         std::string lkey = boost::to_lower_copy(key);
 528         std::string value = strutils::strip(buffer.substr(colonPos + 1));
 529
 530         // only consider these if getting headers (as opposed to trailers
 531         // of a chunked transfer)
 532         if (state == STATE_GETTING_HEADERS) {
 533             if (lkey == "content-length") {
 534
 535                 int sz = strutils::to_int(value);
 536                 if (bodyTransferSize <= 0) {
 537                     bodyTransferSize = sz;
 538                 }
 539                 activeRequest->setResponseLength(sz);
 540             } else if (lkey == "transfer-length") {
 541                 bodyTransferSize = strutils::to_int(value);
 542             } else if (lkey == "transfer-encoding") {
 543                 processTransferEncoding(value);
 544             } else if (lkey == "content-encoding") {
 545                 _contentDecoder.setEncoding(value);
 546             }
 547         }
 548
 549         activeRequest->responseHeader(lkey, value);
 550     }
 551
 552     void processTransferEncoding(const std::string& te)
 553     {
 554         if (te == "chunked") {
 555             chunkedTransfer = true;
 556         } else {
 557             SG_LOG(SG_IO, SG_WARN, "unsupported transfer encoding:" << te);
 558             // failure
 559         }
 560     }
 561
 562     void processChunkHeader()
 563     {
 564         if (buffer.empty()) {
 565             // blank line after chunk data
 566             return;
 567         }
 568
 569         int chunkSize = 0;
 570         int semiPos = buffer.find(';');
 571         if (semiPos >= 0) {
 572             // extensions ignored for the moment
 573             chunkSize = strutils::to_int(buffer.substr(0, semiPos), 16);
 574         } else {
 575             chunkSize = strutils::to_int(buffer, 16);
 576         }
 577
 578         buffer.clear();
 579         if (chunkSize == 0) {  //  trailer start
 580             state = STATE_GETTING_TRAILER;
 581             return;
 582         }
 583
 584         state = STATE_GETTING_CHUNKED_BYTES;
 585         setByteCount(chunkSize);
 586     }
 587
 588     void processTrailer()
 589     {
 590         if (buffer.empty()) {
 591             // end of trailers
 592             responseComplete();
 593             return;
 594         }
 595
 596     // process as a normal header
 597         processHeader();
 598     }
 599
 600     void headersComplete()
 601     {
 602         activeRequest->responseHeadersComplete();
 603         _contentDecoder.initWithRequest(activeRequest);
 604
 605         if (chunkedTransfer) {
 606             state = STATE_GETTING_CHUNKED;
 607         } else if (noMessageBody || (bodyTransferSize == 0)) {
 608             // force the state to GETTING_BODY, to simplify logic in
 609             // responseComplete and handleClose
 610             state = STATE_GETTING_BODY;
 611             responseComplete();
 612         } else {
 613             setByteCount(bodyTransferSize); // may be -1, that's fine
 614             state = STATE_GETTING_BODY;
 615         }
 616     }
 617
 618     void responseComplete()
 619     {
 620         Request_ptr completedRequest = activeRequest;
 621         _contentDecoder.finish();
 622
 623         assert(sentRequests.front() == activeRequest);
 624         sentRequests.pop_front();
 625         bool doClose = activeRequest->closeAfterComplete();
 626         activeRequest = NULL;
 627
 628         if ((state == STATE_GETTING_BODY) || (state == STATE_GETTING_TRAILER)) {
 629             if (doClose) {
 630           // this will bring us into handleClose() above, which updates
 631           // state to STATE_CLOSED
 632               close();
 633
 634           // if we have additional requests waiting, try to start them now
 635               tryStartNextRequest();
 636             }
 637         }
 638
 639         if (state != STATE_CLOSED)  {
 640             state = sentRequests.empty() ? STATE_IDLE : STATE_WAITING_FOR_RESPONSE;
 641         }
 642
 643     // notify request after we change state, so this connection is idle
 644     // if completion triggers other requests (which is likely)
 645         completedRequest->responseComplete();
 646         client->requestFinished(this);
 647
 648         setTerminator("\r\n");
 649     }
 650
 651     enum ConnectionState {
 652         STATE_IDLE = 0,
 653         STATE_WAITING_FOR_RESPONSE,
 654         STATE_GETTING_HEADERS,
 655         STATE_GETTING_BODY,
 656         STATE_GETTING_CHUNKED,
 657         STATE_GETTING_CHUNKED_BYTES,
 658         STATE_GETTING_TRAILER,
 659         STATE_SOCKET_ERROR,
 660         STATE_CLOSED             ///< connection should be closed now
 661     };
 662
 663     Client* client;
 664     Request_ptr activeRequest;
 665     ConnectionState state;
 666     std::string host;
 667     short port;
 668     std::string buffer;
 669     int bodyTransferSize;
 670     SGTimeStamp idleTime;
 671     bool chunkedTransfer;
 672     bool noMessageBody;
 673
 674     RequestList queuedRequests;
 675     RequestList sentRequests;
 676
 677     ContentDecoder _contentDecoder;
 678     std::string connectionId;
 679 };
 680 #endif // of !ENABLE_CURL
 681
 682 Client::Client() :
 683     d(new ClientPrivate)
 684 {
 685     d->proxyPort = 0;
 686     d->maxConnections = 4;
 687     d->bytesTransferred = 0;
 688     d->lastTransferRate = 0;
 689     d->timeTransferSample.stamp();
 690     d->totalBytesDownloaded = 0;
 691
 692     setUserAgent("SimGear-" SG_STRINGIZE(SIMGEAR_VERSION));
 693 #if defined(ENABLE_CURL)
 694     static bool didInitCurlGlobal = false;
 695     if (!didInitCurlGlobal) {
 696       curl_global_init(CURL_GLOBAL_ALL);
 697       didInitCurlGlobal = true;
 698     }
 699
 700     d->createCurlMulti();
 701 #endif
 702 }
 703
 704 Client::~Client()
 705 {
 706 #if defined(ENABLE_CURL)
 707   curl_multi_cleanup(d->curlMulti);
 708 #endif
 709 }
 710
 711 void Client::setMaxConnections(unsigned int maxCon)
 712 {
 713     if (maxCon < 1) {
 714         throw sg_range_exception("illegal HTTP::Client::setMaxConnections value");
 715     }
 716
 717     d->maxConnections = maxCon;
 718 #if defined(ENABLE_CURL)
 719     curl_multi_setopt(d->curlMulti, CURLMOPT_MAX_TOTAL_CONNECTIONS, (long) maxCon);
 720 #endif
 721 }
 722
 723 void Client::update(int waitTimeout)
 724 {
 725 #if defined(ENABLE_CURL)
 726     int remainingActive, messagesInQueue;
 727     curl_multi_perform(d->curlMulti, &remainingActive);
 728     d->haveActiveRequests = (remainingActive > 0);
 729
 730     CURLMsg* msg;
 731     while ((msg = curl_multi_info_read(d->curlMulti, &messagesInQueue))) {
 732       if (msg->msg == CURLMSG_DONE) {
 733         Request* req;
 734         CURL *e = msg->easy_handle;
 735         curl_easy_getinfo(e, CURLINFO_PRIVATE, &req);
 736
 737         long responseCode;
 738         curl_easy_getinfo(e, CURLINFO_RESPONSE_CODE, &responseCode);
 739
 740         if (msg->data.result == 0) {
 741           req->responseComplete();
 742         } else {
 743           fprintf(stderr, "Result: %d - %s\n",
 744                 msg->data.result, curl_easy_strerror(msg->data.result));
 745           req->setFailure(msg->data.result, curl_easy_strerror(msg->data.result));
 746         }
 747
 748         curl_multi_remove_handle(d->curlMulti, e);
 749
 750         // balance the reference we take in makeRequest
 751         SGReferenced::put(req);
 752         curl_easy_cleanup(e);
 753       }
 754       else {
 755         SG_LOG(SG_IO, SG_ALERT, "CurlMSG:" << msg->msg);
 756       }
 757     } // of curl message processing loop
 758 #else
 759     if (!d->poller.hasChannels() && (waitTimeout > 0)) {
 760         SGTimeStamp::sleepForMSec(waitTimeout);
 761     } else {
 762         d->poller.poll(waitTimeout);
 763     }
 764
 765     bool waitingRequests = !d->pendingRequests.empty();
 766     ConnectionDict::iterator it = d->connections.begin();
 767     for (; it != d->connections.end(); ) {
 768         Connection* con = it->second;
 769         if (con->hasIdleTimeout() ||
 770             con->hasError() ||
 771             con->hasErrorTimeout() ||
 772             (!con->isActive() && waitingRequests))
 773         {
 774             if (con->hasErrorTimeout()) {
 775                 // tell the connection we're timing it out
 776                 con->handleTimeout();
 777             }
 778
 779         // connection has been idle for a while, clean it up
 780         // (or if we have requests waiting for a different host,
 781         // or an error condition
 782             ConnectionDict::iterator del = it++;
 783             delete del->second;
 784             d->connections.erase(del);
 785         } else {
 786             if (it->second->shouldStartNext()) {
 787                 it->second->tryStartNextRequest();
 788             }
 789             ++it;
 790         }
 791     } // of connection iteration
 792
 793     if (waitingRequests && (d->connections.size() < d->maxConnections)) {
 794         RequestList waiting(d->pendingRequests);
 795         d->pendingRequests.clear();
 796
 797         // re-submit all waiting requests in order; this takes care of
 798         // finding multiple pending items targetted to the same (new)
 799         // connection
 800         BOOST_FOREACH(Request_ptr req, waiting) {
 801             makeRequest(req);
 802         }
 803     }
 804 #endif
 805 }
 806
 807 void Client::makeRequest(const Request_ptr& r)
 808 {
 809     if( r->isComplete() )
 810       return;
 811
 812     if( r->url().find("://") == std::string::npos ) {
 813         r->setFailure(EINVAL, "malformed URL");
 814         return;
 815     }
 816
 817 #if defined(ENABLE_CURL)
 818     CURL* curlRequest = curl_easy_init();
 819     curl_easy_setopt(curlRequest, CURLOPT_URL, r->url().c_str());
 820
 821     // manually increase the ref count of the request
 822     SGReferenced::get(r.get());
 823     curl_easy_setopt(curlRequest, CURLOPT_PRIVATE, r.get());
 824     // disable built-in libCurl progress feedback
 825     curl_easy_setopt(curlRequest, CURLOPT_NOPROGRESS, 1);
 826
 827     curl_easy_setopt(curlRequest, CURLOPT_WRITEFUNCTION, requestWriteCallback);
 828     curl_easy_setopt(curlRequest, CURLOPT_WRITEDATA, r.get());
 829     curl_easy_setopt(curlRequest, CURLOPT_HEADERFUNCTION, requestHeaderCallback);
 830     curl_easy_setopt(curlRequest, CURLOPT_HEADERDATA, r.get());
 831
 832     curl_easy_setopt(curlRequest, CURLOPT_USERAGENT, d->userAgent.c_str());
 833     curl_easy_setopt(curlRequest, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);
 834
 835     if (!d->proxy.empty()) {
 836       curl_easy_setopt(curlRequest, CURLOPT_PROXY, d->proxy.c_str());
 837       curl_easy_setopt(curlRequest, CURLOPT_PROXYPORT, d->proxyPort);
 838
 839       if (!d->proxyAuth.empty()) {
 840         curl_easy_setopt(curlRequest, CURLOPT_PROXYAUTH, CURLAUTH_BASIC);
 841         curl_easy_setopt(curlRequest, CURLOPT_PROXYUSERPWD, d->proxyAuth.c_str());
 842       }
 843     }
 844
 845     std::string method = boost::to_lower_copy(r->method());
 846     if (method == "get") {
 847       curl_easy_setopt(curlRequest, CURLOPT_HTTPGET, 1);
 848     } else if (method == "put") {
 849       curl_easy_setopt(curlRequest, CURLOPT_PUT, 1);
 850       curl_easy_setopt(curlRequest, CURLOPT_UPLOAD, 1);
 851     } else if (method == "post") {
 852       // see http://curl.haxx.se/libcurl/c/CURLOPT_POST.html
 853       curl_easy_setopt(curlRequest, CURLOPT_HTTPPOST, 1);
 854
 855       std::string q = r->query().substr(1);
 856       curl_easy_setopt(curlRequest, CURLOPT_COPYPOSTFIELDS, q.c_str());
 857
 858       // reset URL to exclude query pieces
 859       std::string urlWithoutQuery = r->url();
 860       std::string::size_type queryPos = urlWithoutQuery.find('?');
 861       urlWithoutQuery.resize(queryPos);
 862       curl_easy_setopt(curlRequest, CURLOPT_URL, urlWithoutQuery.c_str());
 863     } else {
 864       curl_easy_setopt(curlRequest, CURLOPT_CUSTOMREQUEST, r->method().c_str());
 865     }
 866
 867     struct curl_slist* headerList = NULL;
 868     if (r->hasBodyData() && (method != "post")) {
 869       curl_easy_setopt(curlRequest, CURLOPT_UPLOAD, 1);
 870       curl_easy_setopt(curlRequest, CURLOPT_INFILESIZE, r->bodyLength());
 871       curl_easy_setopt(curlRequest, CURLOPT_READFUNCTION, requestReadCallback);
 872       curl_easy_setopt(curlRequest, CURLOPT_READDATA, r.get());
 873       std::string h = "Content-Type:" + r->bodyType();
 874       headerList = curl_slist_append(headerList, h.c_str());
 875     }
 876
 877     StringMap::const_iterator it;
 878     for (it = r->requestHeaders().begin(); it != r->requestHeaders().end(); ++it) {
 879       std::string h = it->first + ": " + it->second;
 880       headerList = curl_slist_append(headerList, h.c_str());
 881     }
 882
 883     if (headerList != NULL) {
 884       curl_easy_setopt(curlRequest, CURLOPT_HTTPHEADER, headerList);
 885     }
 886
 887     curl_multi_add_handle(d->curlMulti, curlRequest);
 888     d->haveActiveRequests = true;
 889
 890 // FIXME - premature?
 891     r->requestStart();
 892
 893 #else
 894     if( r->url().find("http://") != 0 ) {
 895         r->setFailure(EINVAL, "only HTTP protocol is supported");
 896         return;
 897     }
 898
 899     std::string host = r->host();
 900     int port = r->port();
 901     if (!d->proxy.empty()) {
 902         host = d->proxy;
 903         port = d->proxyPort;
 904     }
 905
 906     Connection* con = NULL;
 907     std::stringstream ss;
 908     ss << host << "-" << port;
 909     std::string connectionId = ss.str();
 910     bool havePending = !d->pendingRequests.empty();
 911     bool atConnectionsLimit = d->connections.size() >= d->maxConnections;
 912     ConnectionDict::iterator consEnd = d->connections.end();
 913
 914     // assign request to an existing Connection.
 915     // various options exist here, examined in order
 916     ConnectionDict::iterator it = d->connections.find(connectionId);
 917     if (atConnectionsLimit && (it == consEnd)) {
 918         // maximum number of connections active, queue this request
 919         // when a connection goes inactive, we'll start this one
 920         d->pendingRequests.push_back(r);
 921         return;
 922     }
 923
 924     // scan for an idle Connection to the same host (likely if we're
 925     // retrieving multiple resources from the same host in quick succession)
 926     // if we have pending requests (waiting for a free Connection), then
 927     // force new requests on this id to always use the first Connection
 928     // (instead of the random selection below). This ensures that when
 929     // there's pressure on the number of connections to keep alive, one
 930     // host can't DoS every other.
 931     int count = 0;
 932     for (; (it != consEnd) && (it->first == connectionId); ++it, ++count) {
 933         if (havePending || !it->second->isActive()) {
 934             con = it->second;
 935             break;
 936         }
 937     }
 938
 939     if (!con && atConnectionsLimit) {
 940         // all current connections are busy (active), and we don't
 941         // have free connections to allocate, so let's assign to
 942         // an existing one randomly. Ideally we'd used whichever one will
 943         // complete first but we don't have that info.
 944         int index = rand() % count;
 945         for (it = d->connections.find(connectionId); index > 0; --index) { ; }
 946         con = it->second;
 947     }
 948
 949     // allocate a new connection object
 950     if (!con) {
 951         con = new Connection(this, connectionId);
 952         con->setServer(host, port);
 953         d->poller.addChannel(con);
 954         d->connections.insert(d->connections.end(),
 955             ConnectionDict::value_type(connectionId, con));
 956     }
 957
 958     con->queueRequest(r);
 959 #endif
 960 }
 961
 962 //------------------------------------------------------------------------------
 963 FileRequestRef Client::save( const std::string& url,
 964                              const std::string& filename )
 965 {
 966   FileRequestRef req = new FileRequest(url, filename);
 967   makeRequest(req);
 968   return req;
 969 }
 970
 971 //------------------------------------------------------------------------------
 972 MemoryRequestRef Client::load(const std::string& url)
 973 {
 974   MemoryRequestRef req = new MemoryRequest(url);
 975   makeRequest(req);
 976   return req;
 977 }
 978
 979 void Client::requestFinished(Connection* con)
 980 {
 981
 982 }
 983
 984 void Client::setUserAgent(const std::string& ua)
 985 {
 986     d->userAgent = ua;
 987 }
 988
 989 const std::string& Client::userAgent() const
 990 {
 991     return d->userAgent;
 992 }
 993
 994 const std::string& Client::proxyHost() const
 995 {
 996     return d->proxy;
 997 }
 998
 999 const std::string& Client::proxyAuth() const
1000 {
1001     return d->proxyAuth;
1002 }
1003
1004 void Client::setProxy( const std::string& proxy,
1005                        int port,
1006                        const std::string& auth )
1007 {
1008     d->proxy = proxy;
1009     d->proxyPort = port;
1010     d->proxyAuth = auth;
1011 }
1012
1013 bool Client::hasActiveRequests() const
1014 {
1015   #if defined(ENABLE_CURL)
1016     return d->haveActiveRequests;
1017   #else
1018     ConnectionDict::const_iterator it = d->connections.begin();
1019     for (; it != d->connections.end(); ++it) {
1020         if (it->second->isActive()) return true;
1021     }
1022
1023     return false;
1024 #endif
1025 }
1026
1027 void Client::receivedBytes(unsigned int count)
1028 {
1029     d->bytesTransferred += count;
1030     d->totalBytesDownloaded += count;
1031 }
1032
1033 unsigned int Client::transferRateBytesPerSec() const
1034 {
1035     unsigned int e = d->timeTransferSample.elapsedMSec();
1036     if (e > 400) {
1037         // too long a window, ignore
1038         d->timeTransferSample.stamp();
1039         d->bytesTransferred = 0;
1040         d->lastTransferRate = 0;
1041         return 0;
1042     }
1043
1044     if (e < 100) { // avoid really narrow windows
1045         return d->lastTransferRate;
1046     }
1047
1048     unsigned int ratio = (d->bytesTransferred * 1000) / e;
1049     // run a low-pass filter
1050     unsigned int smoothed = ((400 - e) * d->lastTransferRate) + (e * ratio);
1051     smoothed /= 400;
1052
1053     d->timeTransferSample.stamp();
1054     d->bytesTransferred = 0;
1055     d->lastTransferRate = smoothed;
1056     return smoothed;
1057 }
1058
1059 uint64_t Client::totalBytesDownloaded() const
1060 {
1061     return d->totalBytesDownloaded;
1062 }
1063
1064 size_t Client::requestWriteCallback(char *ptr, size_t size, size_t nmemb, void *userdata)
1065 {
1066   size_t byteSize = size * nmemb;
1067
1068   Request* req = static_cast<Request*>(userdata);
1069   req->processBodyBytes(ptr, byteSize);
1070   return byteSize;
1071 }
1072
1073 size_t Client::requestReadCallback(char *ptr, size_t size, size_t nmemb, void *userdata)
1074 {
1075   size_t maxBytes = size * nmemb;
1076   Request* req = static_cast<Request*>(userdata);
1077   size_t actualBytes = req->getBodyData(ptr, 0, maxBytes);
1078   return actualBytes;
1079 }
1080
1081 size_t Client::requestHeaderCallback(char *rawBuffer, size_t size, size_t nitems, void *userdata)
1082 {
1083   size_t byteSize = size * nitems;
1084   Request* req = static_cast<Request*>(userdata);
1085   std::string h = strutils::simplify(std::string(rawBuffer, byteSize));
1086
1087   if (req->readyState() == HTTP::Request::OPENED) {
1088     req->responseStart(h);
1089     return byteSize;
1090   }
1091
1092   if (h.empty()) {
1093       // got a 100-continue reponse; restart
1094       if (req->responseCode() == 100) {
1095           req->setReadyState(HTTP::Request::OPENED);
1096           return byteSize;
1097       }
1098
1099     req->responseHeadersComplete();
1100     return byteSize;
1101   }
1102
1103   if (req->responseCode() == 100) {
1104       return byteSize; // skip headers associated with 100-continue status
1105   }
1106
1107   size_t colonPos = h.find(':');
1108   if (colonPos == std::string::npos) {
1109       SG_LOG(SG_IO, SG_WARN, "malformed HTTP response header:" << h);
1110       return byteSize;
1111   }
1112
1113   std::string key = strutils::simplify(h.substr(0, colonPos));
1114   std::string lkey = boost::to_lower_copy(key);
1115   std::string value = strutils::strip(h.substr(colonPos + 1));
1116
1117   req->responseHeader(lkey, value);
1118   return byteSize;
1119 }
1120
1121 void Client::debugDumpRequests()
1122 {
1123 #if defined(ENABLE_CURL)
1124
1125 #else
1126     SG_LOG(SG_IO, SG_INFO, "== HTTP connection dump");
1127     ConnectionDict::iterator it = d->connections.begin();
1128     for (; it != d->connections.end(); ++it) {
1129         it->second->debugDumpRequests();
1130     }
1131     SG_LOG(SG_IO, SG_INFO, "==");
1132 #endif
1133 }
1134
1135 void Client::clearAllConnections()
1136 {
1137 #if defined(ENABLE_CURL)
1138     curl_multi_cleanup(d->curlMulti);
1139     d->createCurlMulti();
1140 #else
1141     ConnectionDict::iterator it = d->connections.begin();
1142     for (; it != d->connections.end(); ++it) {
1143         delete it->second;
1144     }
1145     d->connections.clear();
1146 #endif
1147 }
1148
1149 } // of namespace HTTP
1150
1151 } // of namespace simgear