aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--apps/higet/higet.cc74
-rw-r--r--apps/http-proxy/src/ATSConnector.cc92
-rw-r--r--apps/http-proxy/src/ATSConnector.h13
-rw-r--r--apps/http-proxy/src/HTTP1.xMessageFastParser.cc19
-rw-r--r--apps/http-proxy/src/HTTP1.xMessageFastParser.h8
-rw-r--r--libtransport/src/hicn/transport/http/response.cc46
-rw-r--r--libtransport/src/hicn/transport/http/response.h6
7 files changed, 213 insertions, 45 deletions
diff --git a/apps/higet/higet.cc b/apps/higet/higet.cc
index df34d5c14..fcb0cc540 100644
--- a/apps/higet/higet.cc
+++ b/apps/higet/higet.cc
@@ -17,6 +17,9 @@
#include <fstream>
#include <map>
+#include <experimental/algorithm>
+#include <experimental/functional>
+
#ifndef ASIO_STANDALONE
#define ASIO_STANDALONE
#include <asio.hpp>
@@ -41,12 +44,16 @@ typedef struct {
class ReadBytesCallbackImplementation
: public transport::http::HTTPClientConnection::ReadBytesCallback {
+ static std::string chunk_separator;
+
public:
ReadBytesCallbackImplementation(std::string file_name, long yet_downloaded)
: file_name_(file_name),
temp_file_name_(file_name_ + ".temp"),
yet_downloaded_(yet_downloaded),
byte_downloaded_(yet_downloaded),
+ chunked_(false),
+ chunk_size_(0),
work_(std::make_unique<asio::io_service::work>(io_service_)),
thread_(
std::make_unique<std::thread>([this]() { io_service_.run(); })) {
@@ -71,21 +78,72 @@ class ReadBytesCallbackImplementation
auto buffer_ptr = buffer.release();
io_service_.post([this, buffer_ptr]() {
auto buffer = std::unique_ptr<utils::MemBuf>(buffer_ptr);
+ std::unique_ptr<utils::MemBuf> payload;
if (!first_chunk_read_) {
transport::http::HTTPResponse http_response(std::move(buffer));
- auto payload = http_response.getPayload();
+ payload = http_response.getPayload();
auto header = http_response.getHeaders();
+ content_size_ = yet_downloaded_;
std::map<std::string, std::string>::iterator it =
header.find("Content-Length");
if (it != header.end()) {
- content_size_ = yet_downloaded_ + std::stol(it->second);
+ content_size_ += std::stol(it->second);
+ } else {
+ it = header.find("Transfer-Encoding");
+ if (it != header.end() && it->second.compare("chunked") == 0) {
+ chunked_ = true;
+ }
}
- out_->write((char *)payload->data(), payload->length());
first_chunk_read_ = true;
- byte_downloaded_ += payload->length();
} else {
- out_->write((char *)buffer->data(), buffer->length());
- byte_downloaded_ += buffer->length();
+ payload = std::move(buffer);
+ }
+
+ if (chunked_) {
+ if (chunk_size_ > 0) {
+ out_->write((char *)payload->data(), chunk_size_);
+ payload->trimStart(chunk_size_);
+
+ if (payload->length() >= chunk_separator.size()) {
+ payload->trimStart(chunk_separator.size());
+ }
+ }
+
+ while (payload->length() > 0) {
+ // read next chunk size
+ const char *begin = (const char *)payload->data();
+ const char *end = (const char *)payload->tail();
+
+ using std::experimental::make_boyer_moore_searcher;
+ auto it = std::experimental::search(
+ begin, end,
+ make_boyer_moore_searcher(chunk_separator.begin(),
+ chunk_separator.end()));
+ if (it != end) {
+ chunk_size_ = std::stoul(begin, 0, 16);
+ content_size_ += chunk_size_;
+ payload->trimStart(it + chunk_separator.size() - begin);
+
+ std::size_t to_write;
+ if (payload->length() >= chunk_size_) {
+ to_write = chunk_size_;
+ } else {
+ to_write = payload->length();
+ chunk_size_ -= payload->length();
+ }
+
+ out_->write((char *)payload->data(), to_write);
+ byte_downloaded_ += to_write;
+ payload->trimStart(to_write);
+
+ if (payload->length() >= chunk_separator.size()) {
+ payload->trimStart(chunk_separator.size());
+ }
+ }
+ }
+ } else {
+ out_->write((char *)payload->data(), payload->length());
+ byte_downloaded_ += payload->length();
}
if (file_name_ != "-") {
@@ -174,11 +232,15 @@ class ReadBytesCallbackImplementation
long content_size_;
bool first_chunk_read_ = false;
long byte_downloaded_ = 0;
+ bool chunked_;
+ std::size_t chunk_size_;
asio::io_service io_service_;
std::unique_ptr<asio::io_service::work> work_;
std::unique_ptr<std::thread> thread_;
};
+std::string ReadBytesCallbackImplementation::chunk_separator = "\r\n";
+
long checkFileStatus(std::string file_name) {
struct stat stat_buf;
std::string temp_file_name_ = file_name + ".temp";
diff --git a/apps/http-proxy/src/ATSConnector.cc b/apps/http-proxy/src/ATSConnector.cc
index f656a68cb..a9b889941 100644
--- a/apps/http-proxy/src/ATSConnector.cc
+++ b/apps/http-proxy/src/ATSConnector.cc
@@ -33,6 +33,9 @@ ATSConnector::ATSConnector(asio::io_service &io_service,
timer_(io_service),
is_reconnection_(false),
data_available_(false),
+ content_length_(0),
+ is_last_chunk_(false),
+ chunked_(false),
receive_callback_(receive_callback),
on_reconnect_callback_(on_reconnect_callback) {
input_buffer_.prepare(buffer_size + 2048);
@@ -96,23 +99,26 @@ void ATSConnector::doWrite() {
});
} // namespace transport
-void ATSConnector::handleRead(std::error_code ec, std::size_t length,
- std::size_t size) {
+void ATSConnector::handleRead(std::error_code ec, std::size_t length) {
if (TRANSPORT_EXPECT_TRUE(!ec)) {
- size -= length;
+ content_length_ -= length;
const uint8_t *buffer =
asio::buffer_cast<const uint8_t *>(input_buffer_.data());
- receive_callback_(buffer, input_buffer_.size(), !size, false);
+ receive_callback_(buffer, input_buffer_.size(), !content_length_, false);
input_buffer_.consume(input_buffer_.size());
- if (!size) {
- doReadHeader();
+ if (!content_length_) {
+ if (!chunked_ || is_last_chunk_) {
+ doReadHeader();
+ } else {
+ doReadChunkedHeader();
+ }
} else {
- auto to_read = size >= buffer_size ? buffer_size : size;
- asio::async_read(
- socket_, input_buffer_, asio::transfer_exactly(to_read),
- std::bind(&ATSConnector::handleRead, this, std::placeholders::_1,
- std::placeholders::_2, size));
+ auto to_read =
+ content_length_ >= buffer_size ? buffer_size : content_length_;
+ asio::async_read(socket_, input_buffer_, asio::transfer_exactly(to_read),
+ std::bind(&ATSConnector::handleRead, this,
+ std::placeholders::_1, std::placeholders::_2));
}
} else if (ec == asio::error::eof) {
input_buffer_.consume(input_buffer_.size());
@@ -129,20 +135,47 @@ void ATSConnector::doReadBody(std::size_t body_size,
? (buffer_size - input_buffer_.size())
: bytes_to_read;
+ is_last_chunk_ = chunked_ && body_size == 5;
+
if (to_read > 0) {
- asio::async_read(
- socket_, input_buffer_, asio::transfer_exactly(to_read),
- std::bind(&ATSConnector::handleRead, this, std::placeholders::_1,
- std::placeholders::_2, bytes_to_read));
+ content_length_ = bytes_to_read;
+ asio::async_read(socket_, input_buffer_, asio::transfer_exactly(to_read),
+ std::bind(&ATSConnector::handleRead, this,
+ std::placeholders::_1, std::placeholders::_2));
} else {
const uint8_t *buffer =
asio::buffer_cast<const uint8_t *>(input_buffer_.data());
- receive_callback_(buffer, body_size, !to_read, false);
+ receive_callback_(buffer, body_size, chunked_ ? is_last_chunk_ : !to_read,
+ false);
input_buffer_.consume(body_size);
- doReadHeader();
+
+ if (!chunked_ || is_last_chunk_) {
+ doReadHeader();
+ } else {
+ doReadChunkedHeader();
+ }
}
}
+void ATSConnector::doReadChunkedHeader() {
+ asio::async_read_until(
+ socket_, input_buffer_, "\r\n",
+ [this](std::error_code ec, std::size_t length) {
+ if (TRANSPORT_EXPECT_TRUE(!ec)) {
+ const uint8_t *buffer =
+ asio::buffer_cast<const uint8_t *>(input_buffer_.data());
+ std::size_t chunk_size =
+ std::stoul(reinterpret_cast<const char *>(buffer), 0, 16) + 2 +
+ length;
+ auto additional_bytes = input_buffer_.size();
+ doReadBody(chunk_size, additional_bytes);
+ } else {
+ input_buffer_.consume(input_buffer_.size());
+ tryReconnection();
+ }
+ });
+}
+
void ATSConnector::doReadHeader() {
asio::async_read_until(
socket_, input_buffer_, "\r\n\r\n",
@@ -150,14 +183,31 @@ void ATSConnector::doReadHeader() {
if (TRANSPORT_EXPECT_TRUE(!ec)) {
const uint8_t *buffer =
asio::buffer_cast<const uint8_t *>(input_buffer_.data());
- std::size_t size = HTTPMessageFastParser::hasBody(buffer, length);
+ auto headers = HTTPMessageFastParser::getHeaders(buffer, length);
- auto additional_bytes = input_buffer_.size() - length;
+ // Try to get content length, if available
+ auto it = headers.find(HTTPMessageFastParser::content_length);
+ std::size_t size = 0;
+ if (it != headers.end()) {
+ size = std::stoull(it->second);
+ chunked_ = false;
+ } else {
+ it = headers.find(HTTPMessageFastParser::transfer_encoding);
+ if (it != headers.end() &&
+ it->second.compare(HTTPMessageFastParser::chunked) == 0) {
+ chunked_ = true;
+ }
+ }
- receive_callback_(buffer, length, !size, true);
+ receive_callback_(buffer, length, !size && !chunked_, true);
+ auto additional_bytes = input_buffer_.size() - length;
input_buffer_.consume(length);
- doReadBody(size, additional_bytes);
+ if (!chunked_) {
+ doReadBody(size, additional_bytes);
+ } else {
+ doReadChunkedHeader();
+ }
} else {
input_buffer_.consume(input_buffer_.size());
tryReconnection();
diff --git a/apps/http-proxy/src/ATSConnector.h b/apps/http-proxy/src/ATSConnector.h
index dbec30353..8d91b7b7b 100644
--- a/apps/http-proxy/src/ATSConnector.h
+++ b/apps/http-proxy/src/ATSConnector.h
@@ -65,12 +65,17 @@ class ATSConnector {
void doReadBody(std::size_t body_size, std::size_t additional_bytes);
+ // void handleReadChunked(std::error_code ec, std::size_t length,
+ // std::size_t size);
+
+ void doReadChunkedHeader();
+
void doWrite();
bool checkConnected();
private:
- void handleRead(std::error_code ec, std::size_t length, std::size_t bytes);
+ void handleRead(std::error_code ec, std::size_t length);
void tryReconnection();
void startConnectionTimer();
void handleDeadline(const std::error_code &ec);
@@ -88,6 +93,12 @@ class ATSConnector {
bool is_reconnection_;
bool data_available_;
+ std::size_t content_length_;
+
+ // Chunked encoding
+ bool is_last_chunk_;
+ bool chunked_;
+
ContentReceivedCallback receive_callback_;
OnReconnect on_reconnect_callback_;
diff --git a/apps/http-proxy/src/HTTP1.xMessageFastParser.cc b/apps/http-proxy/src/HTTP1.xMessageFastParser.cc
index a03871649..729eb3aeb 100644
--- a/apps/http-proxy/src/HTTP1.xMessageFastParser.cc
+++ b/apps/http-proxy/src/HTTP1.xMessageFastParser.cc
@@ -15,17 +15,36 @@
#include "HTTP1.xMessageFastParser.h"
+#include <hicn/transport/http/response.h>
+
#include <experimental/algorithm>
#include <experimental/functional>
#include <iostream>
std::string HTTPMessageFastParser::numbers = "0123456789";
std::string HTTPMessageFastParser::content_length = "Content-Length";
+std::string HTTPMessageFastParser::transfer_encoding = "Transfer-Encoding";
+std::string HTTPMessageFastParser::chunked = "chunked";
std::string HTTPMessageFastParser::cache_control = "Cache-Control";
std::string HTTPMessageFastParser::mpd = "mpd";
std::string HTTPMessageFastParser::connection = "Connection";
std::string HTTPMessageFastParser::separator = "\r\n\r\n";
+HTTPHeaders HTTPMessageFastParser::getHeaders(const uint8_t *headers,
+ std::size_t length) {
+ HTTPHeaders ret;
+ std::string http_version;
+ std::string status_code;
+ std::string status_string;
+
+ if (transport::http::HTTPResponse::parseHeaders(headers, length, ret, http_version,
+ status_code, status_string)) {
+ return ret;
+ }
+
+ throw std::runtime_error("Error parsing response headers.");
+}
+
std::size_t HTTPMessageFastParser::hasBody(const uint8_t *headers,
std::size_t length) {
const char *buffer = reinterpret_cast<const char *>(headers);
diff --git a/apps/http-proxy/src/HTTP1.xMessageFastParser.h b/apps/http-proxy/src/HTTP1.xMessageFastParser.h
index 10a70c3e9..79dbce19d 100644
--- a/apps/http-proxy/src/HTTP1.xMessageFastParser.h
+++ b/apps/http-proxy/src/HTTP1.xMessageFastParser.h
@@ -18,15 +18,21 @@
#include <algorithm>
#include <string>
+#include <hicn/transport/http/message.h>
+
+using transport::http::HTTPHeaders;
+
class HTTPMessageFastParser {
public:
+ static HTTPHeaders getHeaders(const uint8_t* headers, std::size_t length);
static std::size_t hasBody(const uint8_t* headers, std::size_t length);
static bool isMpdRequest(const uint8_t* headers, std::size_t length);
static uint32_t parseCacheControl(const uint8_t* headers, std::size_t length);
- private:
static std::string numbers;
static std::string content_length;
+ static std::string transfer_encoding;
+ static std::string chunked;
static std::string cache_control;
static std::string connection;
static std::string mpd;
diff --git a/libtransport/src/hicn/transport/http/response.cc b/libtransport/src/hicn/transport/http/response.cc
index a2bc47e6b..ba0acd1ac 100644
--- a/libtransport/src/hicn/transport/http/response.cc
+++ b/libtransport/src/hicn/transport/http/response.cc
@@ -41,16 +41,33 @@ void HTTPResponse::appendResponseChunk(
}
bool HTTPResponse::parseHeaders(std::unique_ptr<utils::MemBuf> &&buffer) {
+ auto ret =
+ HTTPResponse::parseHeaders(buffer->data(), buffer->length(), headers_,
+ http_version_, status_code_, status_string_);
+
+ if (ret) {
+ buffer->trimStart(ret);
+ payload_ = std::move(buffer);
+ return true;
+ }
+
+ return false;
+}
+
+std::size_t HTTPResponse::parseHeaders(const uint8_t *buffer, std::size_t size,
+ HTTPHeaders &headers,
+ std::string &http_version,
+ std::string &status_code,
+ std::string &status_string) {
const char *crlf2 = "\r\n\r\n";
- const char *begin = (const char *)buffer->data();
- const char *end = begin + buffer->length();
+ const char *begin = (const char *)buffer;
+ const char *end = begin + size;
auto it =
std::experimental::search(begin, end,
std::experimental::make_boyer_moore_searcher(
crlf2, crlf2 + strlen(crlf2)));
if (it != end) {
- buffer->trimStart(it + strlen(crlf2) - begin);
std::stringstream ss;
ss.str(std::string(begin, it));
@@ -58,29 +75,28 @@ bool HTTPResponse::parseHeaders(std::unique_ptr<utils::MemBuf> &&buffer) {
getline(ss, line);
std::istringstream line_s(line);
std::string _http_version;
- std::string http_version;
line_s >> _http_version;
std::size_t separator;
if ((separator = _http_version.find('/')) != std::string::npos) {
if (_http_version.substr(0, separator) != "HTTP") {
- return false;
+ return 0;
}
- http_version_ =
+ http_version =
line.substr(separator + 1, _http_version.length() - separator - 1);
} else {
- return false;
+ return 0;
}
- std::string status_code, status_string;
+ std::string _status_string;
- line_s >> status_code_;
- line_s >> status_string;
+ line_s >> status_code;
+ line_s >> _status_string;
auto _it = std::search(line.begin(), line.end(), status_string.begin(),
status_string.end());
- status_string_ = std::string(_it, line.end() - 1);
+ status_string = std::string(_it, line.end() - 1);
std::size_t param_end;
std::size_t value_start;
@@ -92,19 +108,17 @@ bool HTTPResponse::parseHeaders(std::unique_ptr<utils::MemBuf> &&buffer) {
value_start++;
}
if (value_start < line.size()) {
- headers_[line.substr(0, param_end)] =
+ headers[line.substr(0, param_end)] =
line.substr(value_start, line.size() - value_start - 1);
}
}
} else {
- return false;
+ return 0;
}
}
}
- payload_ = std::move(buffer);
-
- return true;
+ return it + strlen(crlf2) - begin;
}
void HTTPResponse::parse(std::unique_ptr<utils::MemBuf> &&response) {
diff --git a/libtransport/src/hicn/transport/http/response.h b/libtransport/src/hicn/transport/http/response.h
index 7ef655059..bab41acb8 100644
--- a/libtransport/src/hicn/transport/http/response.h
+++ b/libtransport/src/hicn/transport/http/response.h
@@ -42,6 +42,12 @@ class HTTPResponse : public HTTPMessage {
bool parseHeaders(std::unique_ptr<utils::MemBuf> &&buffer);
+ static std::size_t parseHeaders(const uint8_t *buffer, std::size_t size,
+ HTTPHeaders &headers,
+ std::string &http_version,
+ std::string &status_code,
+ std::string &status_string);
+
private:
std::string status_code_;
std::string status_string_;