lib/cpp/src/transport/THttpClient.cpp

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements. See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership. The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied. See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

#include <cstdlib>
#include <sstream>

#include "THttpClient.h"
#include "TSocket.h"

namespace apache { namespace thrift { namespace transport {

using namespace std;

/**
 * Http client implementation.
 *
 */

// Yeah, yeah, hacky to put these here, I know.
static const char* CRLF = "\r\n";
static const int CRLF_LEN = 2;

THttpClient::THttpClient(boost::shared_ptr<TTransport> transport, string host, string path) :
  transport_(transport),
  host_(host),
  path_(path),
  readHeaders_(true),
  chunked_(false),
  chunkedDone_(false),
  chunkSize_(0),
  contentLength_(0),
  httpBuf_(NULL),
  httpPos_(0),
  httpBufLen_(0),
  httpBufSize_(1024) {
  init();
}

THttpClient::THttpClient(string host, int port, string path) :
  host_(host),
  path_(path),
  readHeaders_(true),
  chunked_(false),
  chunkedDone_(false),
  chunkSize_(0),
  contentLength_(0),
  httpBuf_(NULL),
  httpPos_(0),
  httpBufLen_(0),
  httpBufSize_(1024) {
  transport_ = boost::shared_ptr<TTransport>(new TSocket(host, port));
  init();
}

void THttpClient::init() {
  httpBuf_ = (char*)std::malloc(httpBufSize_+1);
  if (httpBuf_ == NULL) {
    throw TTransportException("Out of memory.");
  }
  httpBuf_[httpBufLen_] = '\0';
}

THttpClient::~THttpClient() {
  if (httpBuf_ != NULL) {
    std::free(httpBuf_);
  }
}

uint32_t THttpClient::read(uint8_t* buf, uint32_t len) {
  if (readBuffer_.available_read() == 0) {
    readBuffer_.resetBuffer();
    uint32_t got = readMoreData();
    if (got == 0) {
      return 0;
    }
  }
  return readBuffer_.read(buf, len);
}

void THttpClient::readEnd() {
  // Read any pending chunked data (footers etc.)
  if (chunked_) {
    while (!chunkedDone_) {
      readChunked();
    }
  }
}

uint32_t THttpClient::readMoreData() {
  // Get more data!
  refill();

  if (readHeaders_) {
    readHeaders();
  }

  if (chunked_) {
    return readChunked();
  } else {
    return readContent(contentLength_);
  }
}

uint32_t THttpClient::readChunked() {
  uint32_t length = 0;

  char* line = readLine();
  uint32_t chunkSize = parseChunkSize(line);
  if (chunkSize == 0) {
    readChunkedFooters();
  } else {
    // Read data content
    length += readContent(chunkSize);
    // Read trailing CRLF after content
    readLine();
  }
  return length;
}

void THttpClient::readChunkedFooters() {
  // End of data, read footer lines until a blank one appears
  while (true) {
    char* line = readLine();
    if (strlen(line) == 0) {
      chunkedDone_ = true;
      break;
    }
  }
}

uint32_t THttpClient::parseChunkSize(char* line) {
  char* semi = strchr(line, ';');
  if (semi != NULL) {
    *semi = '\0';
  }
  int size = 0;
  sscanf(line, "%x", &size);
  return (uint32_t)size;
}

uint32_t THttpClient::readContent(uint32_t size) {
  uint32_t need = size;
  while (need > 0) {
    uint32_t avail = httpBufLen_ - httpPos_;
    if (avail == 0) {
      // We have given all the data, reset position to head of the buffer
      httpPos_ = 0;
      httpBufLen_ = 0;
      refill();

      // Now have available however much we read
      avail = httpBufLen_;
    }
    uint32_t give = avail;
    if (need < give) {
      give = need;
    }
    readBuffer_.write((uint8_t*)(httpBuf_+httpPos_), give);
    httpPos_ += give;
    need -= give;
  }
  return size;
}

char* THttpClient::readLine() {
  while (true) {
    char* eol = NULL;

    eol = strstr(httpBuf_+httpPos_, CRLF);

    // No CRLF yet?
    if (eol == NULL) {
      // Shift whatever we have now to front and refill
      shift();
      refill();
    } else {
      // Return pointer to next line
      *eol = '\0';
      char* line = httpBuf_+httpPos_;
      httpPos_ = (eol-httpBuf_) + CRLF_LEN;
      return line;
    }
  }

}

void THttpClient::shift() {
  if (httpBufLen_ > httpPos_) {
    // Shift down remaining data and read more
    uint32_t length = httpBufLen_ - httpPos_;
    memmove(httpBuf_, httpBuf_+httpPos_, length);
    httpBufLen_ = length;
  } else {
    httpBufLen_ = 0;
  }
  httpPos_ = 0;
  httpBuf_[httpBufLen_] = '\0';
}

void THttpClient::refill() {
  uint32_t avail = httpBufSize_ - httpBufLen_;
  if (avail <= (httpBufSize_ / 4)) {
    httpBufSize_ *= 2;
    httpBuf_ = (char*)std::realloc(httpBuf_, httpBufSize_+1);
    if (httpBuf_ == NULL) {
      throw TTransportException("Out of memory.");
    }
  }

  // Read more data
  uint32_t got = transport_->read((uint8_t*)(httpBuf_+httpBufLen_), httpBufSize_-httpBufLen_);
  httpBufLen_ += got;
  httpBuf_[httpBufLen_] = '\0';

  if (got == 0) {
    throw TTransportException("Could not refill buffer");
  }
}

void THttpClient::readHeaders() {
  // Initialize headers state variables
  contentLength_ = 0;
  chunked_ = false;
  chunkedDone_ = false;
  chunkSize_ = 0;

  // Control state flow
  bool statusLine = true;
  bool finished = false;

  // Loop until headers are finished
  while (true) {
    char* line = readLine();

    if (strlen(line) == 0) {
      if (finished) {
        readHeaders_ = false;
        return;
      } else {
        // Must have been an HTTP 100, keep going for another status line
        statusLine = true;
      }
    } else {
      if (statusLine) {
        statusLine = false;
        finished = parseStatusLine(line);
      } else {
        parseHeader(line);
      }
    }
  }
}

bool THttpClient::parseStatusLine(char* status) {
  char* http = status;

  char* code = strchr(http, ' ');
  if (code == NULL) {
    throw TTransportException(string("Bad Status: ") + status);
  }

  *code = '\0';
  while (*(code++) == ' ');

  char* msg = strchr(code, ' ');
  if (msg == NULL) {
    throw TTransportException(string("Bad Status: ") + status);
  }
  *msg = '\0';

  if (strcmp(code, "200") == 0) {
    // HTTP 200 = OK, we got the response
    return true;
  } else if (strcmp(code, "100") == 0) {
    // HTTP 100 = continue, just keep reading
    return false;
  } else {
    throw TTransportException(string("Bad Status: ") + status);
  }
}

void THttpClient::parseHeader(char* header) {
  char* colon = strchr(header, ':');
  if (colon == NULL) {
    return;
  }
  uint32_t sz = colon - header;
  char* value = colon+1;

  if (strncmp(header, "Transfer-Encoding", sz) == 0) {
    if (strstr(value, "chunked") != NULL) {
      chunked_ = true;
    }
  } else if (strncmp(header, "Content-Length", sz) == 0) {
    chunked_ = false;
    contentLength_ = atoi(value);
  }
}

void THttpClient::write(const uint8_t* buf, uint32_t len) {
  writeBuffer_.write(buf, len);
}

void THttpClient::flush() {
  // Fetch the contents of the write buffer
  uint8_t* buf;
  uint32_t len;
  writeBuffer_.getBuffer(&buf, &len);

  // Construct the HTTP header
  std::ostringstream h;
  h <<
    "POST " << path_ << " HTTP/1.1" << CRLF <<
    "Host: " << host_ << CRLF <<
    "Content-Type: application/x-thrift" << CRLF <<
    "Content-Length: " << len << CRLF <<
    "Accept: application/x-thrift" << CRLF <<
    "User-Agent: C++/THttpClient" << CRLF <<
    CRLF;
  string header = h.str();

  // Write the header, then the data, then flush
  transport_->write((const uint8_t*)header.c_str(), header.size());
  transport_->write(buf, len);
  transport_->flush();

  // Reset the buffer and header variables
  writeBuffer_.resetBuffer();
  readHeaders_ = true;
}

}}} // apache::thrift::transport