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