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