Fixes #114: added support for async client requests. Also includes some cleanup, support for parallel requests, and client reconnect
This commit is contained in:
eidheim
parent
f5ba673350
commit
e1aebba344
4 changed files with 471 additions and 213 deletions
488
client_http.hpp
488
client_http.hpp
|
|
@ -2,13 +2,16 @@
|
|||
#define CLIENT_HTTP_HPP
|
||||
|
||||
#include <unordered_map>
|
||||
#include <map>
|
||||
#include <vector>
|
||||
#include <random>
|
||||
#include <mutex>
|
||||
#include <type_traits>
|
||||
|
||||
#ifdef USE_STANDALONE_ASIO
|
||||
#include <boost/utility/string_ref.hpp>
|
||||
#include <boost/thread/locks.hpp>
|
||||
#include <boost/thread/shared_mutex.hpp>
|
||||
|
||||
#ifdef USE_STANDALONE_ASIO
|
||||
#include <asio.hpp>
|
||||
#include <type_traits>
|
||||
#include <system_error>
|
||||
|
|
@ -20,7 +23,6 @@ namespace SimpleWeb {
|
|||
}
|
||||
#else
|
||||
#include <boost/asio.hpp>
|
||||
#include <boost/utility/string_ref.hpp>
|
||||
#include <boost/algorithm/string/predicate.hpp>
|
||||
#include <boost/functional/hash.hpp>
|
||||
namespace SimpleWeb {
|
||||
|
|
@ -62,14 +64,14 @@ namespace SimpleWeb {
|
|||
# endif
|
||||
|
||||
namespace SimpleWeb {
|
||||
using Header = std::unordered_multimap<std::string, std::string, CaseInsensitiveHash, CaseInsensitiveEqual>;
|
||||
|
||||
template <class socket_type>
|
||||
class Client;
|
||||
|
||||
template <class socket_type>
|
||||
class ClientBase {
|
||||
public:
|
||||
virtual ~ClientBase() {}
|
||||
|
||||
class Response {
|
||||
friend class ClientBase<socket_type>;
|
||||
friend class Client<socket_type>;
|
||||
|
|
@ -78,7 +80,7 @@ namespace SimpleWeb {
|
|||
|
||||
std::istream content;
|
||||
|
||||
std::unordered_multimap<std::string, std::string, CaseInsensitiveHash, CaseInsensitiveEqual> header;
|
||||
Header header;
|
||||
|
||||
private:
|
||||
asio::streambuf content_buffer;
|
||||
|
|
@ -99,69 +101,268 @@ namespace SimpleWeb {
|
|||
std::string proxy_server;
|
||||
};
|
||||
|
||||
protected:
|
||||
class RequestCallback {
|
||||
public:
|
||||
bool stop=false;
|
||||
std::mutex stop_mutex;
|
||||
};
|
||||
|
||||
class Connection {
|
||||
public:
|
||||
Connection(const std::string &host, unsigned short port, const Config &config, std::unique_ptr<socket_type> &&socket) :
|
||||
host(host), port(port), config(config), socket(std::move(socket)) {
|
||||
if(config.proxy_server.empty())
|
||||
query=std::unique_ptr<asio::ip::tcp::resolver::query>(new asio::ip::tcp::resolver::query(host, std::to_string(port)));
|
||||
else {
|
||||
auto proxy_host_port=parse_host_port(config.proxy_server, 8080);
|
||||
query=std::unique_ptr<asio::ip::tcp::resolver::query>(new asio::ip::tcp::resolver::query(proxy_host_port.first, std::to_string(proxy_host_port.second)));
|
||||
}
|
||||
}
|
||||
std::string host;
|
||||
unsigned short port;
|
||||
Config config;
|
||||
|
||||
std::unique_ptr<socket_type> socket;
|
||||
bool in_use=false;
|
||||
bool reconnecting=false;
|
||||
|
||||
std::unique_ptr<asio::ip::tcp::resolver::query> query;
|
||||
};
|
||||
|
||||
class Session {
|
||||
public:
|
||||
Session(const std::shared_ptr<asio::io_service> &io_service, const std::shared_ptr<Connection> &connection, std::unique_ptr<asio::streambuf> &&request_buffer) :
|
||||
io_service(io_service), connection(connection), request_buffer(std::move(request_buffer)), response(new Response()) {}
|
||||
std::shared_ptr<asio::io_service> io_service;
|
||||
std::shared_ptr<Connection> connection;
|
||||
std::unique_ptr<asio::streambuf> request_buffer;
|
||||
std::shared_ptr<Response> response;
|
||||
std::function<void(const error_code&)> callback;
|
||||
};
|
||||
|
||||
public:
|
||||
/// Set before calling request
|
||||
Config config;
|
||||
|
||||
std::shared_ptr<Response> request(const std::string& request_type, const std::string& path="/", boost::string_ref content="",
|
||||
const std::map<std::string, std::string>& header=std::map<std::string, std::string>()) {
|
||||
auto write_buffer=create_request_header(request_type, path, header);
|
||||
/// If you have your own asio::io_service, store its pointer here before calling request().
|
||||
/// When using asynchronous requests, running the io_service is up to the programmer.
|
||||
std::shared_ptr<asio::io_service> io_service;
|
||||
|
||||
virtual ~ClientBase() {
|
||||
boost::unique_lock<boost::shared_mutex> lock(*stop_request_callback_mutex);
|
||||
*stop_request_callback=true;
|
||||
}
|
||||
|
||||
/// Synchronous request. The io_service is run within this function.
|
||||
std::shared_ptr<Response> request(const std::string& method, const std::string& path=std::string("/"), boost::string_ref content="", const Header& header=Header()) {
|
||||
auto session=std::make_shared<Session>(io_service, get_connection(), create_request_header(method, path, header));
|
||||
std::shared_ptr<Response> response;
|
||||
session->callback=[this, &response, session](const error_code &ec) {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(*connections_mutex);
|
||||
session->connection->in_use=false;
|
||||
}
|
||||
response=session->response;
|
||||
if(ec)
|
||||
throw system_error(ec);
|
||||
};
|
||||
|
||||
std::ostream write_stream(write_buffer.get());
|
||||
std::ostream write_stream(session->request_buffer.get());
|
||||
if(content.size()>0)
|
||||
write_stream << "Content-Length: " << content.size() << "\r\n";
|
||||
write_stream << "\r\n" << content;
|
||||
|
||||
request_write(write_buffer);
|
||||
Client<socket_type>::connect(session);
|
||||
|
||||
io_service->reset();
|
||||
io_service->run();
|
||||
|
||||
return request_read();
|
||||
return response;
|
||||
}
|
||||
|
||||
std::shared_ptr<Response> request(const std::string& request_type, const std::string& path, std::iostream& content,
|
||||
const std::map<std::string, std::string>& header=std::map<std::string, std::string>()) {
|
||||
auto write_buffer=create_request_header(request_type, path, header);
|
||||
/// Synchronous request. The io_service is run within this function.
|
||||
std::shared_ptr<Response> request(const std::string& method, const std::string& path, std::iostream& content, const Header& header=Header()) {
|
||||
auto session=std::make_shared<Session>(io_service, get_connection(), create_request_header(method, path, header));
|
||||
std::shared_ptr<Response> response;
|
||||
session->callback=[this, &response, session](const error_code &ec) {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(*connections_mutex);
|
||||
session->connection->in_use=false;
|
||||
}
|
||||
response=session->response;
|
||||
if(ec)
|
||||
throw system_error(ec);
|
||||
};
|
||||
|
||||
content.seekp(0, std::ios::end);
|
||||
auto content_length=content.tellp();
|
||||
content.seekp(0, std::ios::beg);
|
||||
std::ostream write_stream(write_buffer.get());
|
||||
std::ostream write_stream(session->request_buffer.get());
|
||||
if(content_length>0)
|
||||
write_stream << "Content-Length: " << content_length << "\r\n";
|
||||
write_stream << "\r\n";
|
||||
if(content_length>0)
|
||||
write_stream << content.rdbuf();
|
||||
|
||||
request_write(write_buffer);
|
||||
Client<socket_type>::connect(session);
|
||||
|
||||
io_service->reset();
|
||||
io_service->run();
|
||||
|
||||
return request_read();
|
||||
return response;
|
||||
}
|
||||
|
||||
void close() {
|
||||
error_code ec;
|
||||
socket->lowest_layer().shutdown(asio::ip::tcp::socket::shutdown_both, ec);
|
||||
socket->lowest_layer().close(ec);
|
||||
/// Asynchronous request where setting and/or running Client's io_service is required.
|
||||
/// The request callback is not run if the calling Client object has been destroyed.
|
||||
void request(const std::string &method, const std::string &path, boost::string_ref content, const Header& header,
|
||||
std::function<void(std::shared_ptr<Response>, const error_code&)> &&request_callback_) {
|
||||
auto session=std::make_shared<Session>(io_service, get_connection(), create_request_header(method, path, header));
|
||||
auto request_callback=std::make_shared<std::function<void(std::shared_ptr<Response>, const error_code&)>>(std::move(request_callback_));
|
||||
auto connections_mutex=this->connections_mutex;
|
||||
auto stop_request_callback=this->stop_request_callback;
|
||||
auto stop_request_callback_mutex=this->stop_request_callback_mutex;
|
||||
session->callback=[session, request_callback, connections_mutex, stop_request_callback, stop_request_callback_mutex](const error_code &ec) {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(*connections_mutex);
|
||||
session->connection->in_use=false;
|
||||
}
|
||||
|
||||
boost::shared_lock<boost::shared_mutex> lock(*stop_request_callback_mutex);
|
||||
if(*stop_request_callback)
|
||||
return;
|
||||
|
||||
if(*request_callback)
|
||||
(*request_callback)(session->response, ec);
|
||||
};
|
||||
|
||||
std::ostream write_stream(session->request_buffer.get());
|
||||
if(content.size()>0)
|
||||
write_stream << "Content-Length: " << content.size() << "\r\n";
|
||||
write_stream << "\r\n" << content;
|
||||
|
||||
Client<socket_type>::connect(session);
|
||||
}
|
||||
|
||||
/// Asynchronous request where setting and/or running Client's io_service is required.
|
||||
/// The request callback is not run if the calling Client object has been destroyed.
|
||||
void request(const std::string &method, const std::string &path, boost::string_ref content,
|
||||
std::function<void(std::shared_ptr<Response>, const error_code&)> &&request_callback) {
|
||||
request(method, path, content, Header(), std::move(request_callback));
|
||||
}
|
||||
|
||||
/// Asynchronous request where setting and/or running Client's io_service is required.
|
||||
/// The request callback is not run if the calling Client object has been destroyed.
|
||||
void request(const std::string &method, const std::string &path,
|
||||
std::function<void(std::shared_ptr<Response>, const error_code&)> &&request_callback) {
|
||||
request(method, path, std::string(), Header(), std::move(request_callback));
|
||||
}
|
||||
|
||||
/// Asynchronous request where setting and/or running Client's io_service is required.
|
||||
/// The request callback is not run if the calling Client object has been destroyed.
|
||||
void request(const std::string &method, std::function<void(std::shared_ptr<Response>, const error_code&)> &&request_callback) {
|
||||
request(method, std::string("/"), std::string(), Header(), std::move(request_callback));
|
||||
}
|
||||
|
||||
/// Asynchronous request where setting and/or running Client's io_service is required.
|
||||
/// The request callback is not run if the calling Client object has been destroyed.
|
||||
void request(const std::string &method, const std::string &path, std::iostream& content, const Header& header,
|
||||
std::function<void(std::shared_ptr<Response>, const error_code&)> &&request_callback_) {
|
||||
auto session=std::make_shared<Session>(io_service, get_connection(), create_request_header(method, path, header));
|
||||
auto request_callback=std::make_shared<std::function<void(std::shared_ptr<Response>, const error_code&)>>(std::move(request_callback_));
|
||||
auto connections_mutex=this->connections_mutex;
|
||||
auto stop_request_callback=this->stop_request_callback;
|
||||
auto stop_request_callback_mutex=this->stop_request_callback_mutex;
|
||||
session->callback=[session, request_callback, connections_mutex, stop_request_callback, stop_request_callback_mutex](const error_code &ec) {
|
||||
{
|
||||
std::lock_guard<std::mutex> lock(*connections_mutex);
|
||||
session->connection->in_use=false;
|
||||
}
|
||||
|
||||
boost::shared_lock<boost::shared_mutex> lock(*stop_request_callback_mutex);
|
||||
if(*stop_request_callback)
|
||||
return;
|
||||
|
||||
if(*request_callback)
|
||||
(*request_callback)(session->response, ec);
|
||||
};
|
||||
|
||||
content.seekp(0, std::ios::end);
|
||||
auto content_length=content.tellp();
|
||||
content.seekp(0, std::ios::beg);
|
||||
std::ostream write_stream(session->request_buffer.get());
|
||||
if(content_length>0)
|
||||
write_stream << "Content-Length: " << content_length << "\r\n";
|
||||
write_stream << "\r\n";
|
||||
if(content_length>0)
|
||||
write_stream << content.rdbuf();
|
||||
|
||||
Client<socket_type>::connect(session);
|
||||
}
|
||||
|
||||
void request(const std::string &method, const std::string &path, std::iostream& content,
|
||||
std::function<void(std::shared_ptr<Response>, const error_code&)> &&request_callback) {
|
||||
request(method, path, content, Header(), std::move(request_callback));
|
||||
}
|
||||
|
||||
protected:
|
||||
std::shared_ptr<asio::io_service> io_service;
|
||||
|
||||
std::unique_ptr<socket_type> socket;
|
||||
|
||||
std::string host;
|
||||
unsigned short port;
|
||||
|
||||
ClientBase(const std::string& host_port, unsigned short default_port) : io_service(new asio::io_service()) {
|
||||
|
||||
std::vector<std::shared_ptr<Connection>> connections;
|
||||
std::shared_ptr<std::mutex> connections_mutex;
|
||||
|
||||
std::shared_ptr<bool> stop_request_callback;
|
||||
std::shared_ptr<boost::shared_mutex> stop_request_callback_mutex;
|
||||
|
||||
ClientBase(const std::string& host_port, unsigned short default_port) : io_service(new asio::io_service()),
|
||||
connections_mutex(new std::mutex()), stop_request_callback(new bool(false)), stop_request_callback_mutex(new boost::shared_mutex()) {
|
||||
auto parsed_host_port=parse_host_port(host_port, default_port);
|
||||
host=parsed_host_port.first;
|
||||
port=parsed_host_port.second;
|
||||
}
|
||||
|
||||
std::pair<std::string, unsigned short> parse_host_port(const std::string &host_port, unsigned short default_port) {
|
||||
std::shared_ptr<Connection> get_connection() {
|
||||
std::shared_ptr<Connection> connection;
|
||||
std::lock_guard<std::mutex> lock(*connections_mutex);
|
||||
for(auto it=connections.begin();it!=connections.end();) {
|
||||
if((*it)->in_use)
|
||||
++it;
|
||||
else if(!connection) {
|
||||
connection=*it;
|
||||
++it;
|
||||
}
|
||||
else
|
||||
it=connections.erase(it);
|
||||
}
|
||||
if(!connection) {
|
||||
connection=create_connection();
|
||||
connections.emplace_back(connection);
|
||||
}
|
||||
connection->reconnecting=false;
|
||||
connection->in_use=true;
|
||||
return connection;
|
||||
}
|
||||
|
||||
virtual std::shared_ptr<Connection> create_connection()=0;
|
||||
|
||||
std::unique_ptr<asio::streambuf> create_request_header(const std::string& method, const std::string& path, const Header& header) const {
|
||||
auto corrected_path=path;
|
||||
if(corrected_path=="")
|
||||
corrected_path="/";
|
||||
if(!config.proxy_server.empty() && std::is_same<socket_type, asio::ip::tcp::socket>::value)
|
||||
corrected_path="http://"+host+':'+std::to_string(port)+corrected_path;
|
||||
|
||||
std::unique_ptr<asio::streambuf> request_buffer(new asio::streambuf());
|
||||
std::ostream write_stream(request_buffer.get());
|
||||
write_stream << method << " " << corrected_path << " HTTP/1.1\r\n";
|
||||
write_stream << "Host: " << host << "\r\n";
|
||||
for(auto& h: header)
|
||||
write_stream << h.first << ": " << h.second << "\r\n";
|
||||
return request_buffer;
|
||||
}
|
||||
|
||||
static std::pair<std::string, unsigned short> parse_host_port(const std::string &host_port, unsigned short default_port) {
|
||||
std::pair<std::string, unsigned short> parsed_host_port;
|
||||
size_t host_end=host_port.find(':');
|
||||
if(host_end==std::string::npos) {
|
||||
|
|
@ -175,25 +376,22 @@ namespace SimpleWeb {
|
|||
return parsed_host_port;
|
||||
}
|
||||
|
||||
virtual void connect()=0;
|
||||
|
||||
std::shared_ptr<asio::deadline_timer> get_timeout_timer(size_t timeout=0) {
|
||||
static std::shared_ptr<asio::deadline_timer> get_timeout_timer(const std::shared_ptr<Session> &session, size_t timeout=0) {
|
||||
if(timeout==0)
|
||||
timeout=config.timeout;
|
||||
timeout=session->connection->config.timeout;
|
||||
if(timeout==0)
|
||||
return nullptr;
|
||||
|
||||
auto timer=std::make_shared<asio::deadline_timer>(*io_service);
|
||||
auto timer=std::make_shared<asio::deadline_timer>(*session->io_service);
|
||||
timer->expires_from_now(boost::posix_time::seconds(timeout));
|
||||
timer->async_wait([this](const error_code& ec) {
|
||||
if(!ec) {
|
||||
close();
|
||||
}
|
||||
timer->async_wait([session](const error_code& ec) {
|
||||
if(!ec)
|
||||
close(session);
|
||||
});
|
||||
return timer;
|
||||
}
|
||||
|
||||
void parse_response_header(const std::shared_ptr<Response> &response) const {
|
||||
static void parse_response_header(const std::shared_ptr<Response> &response) {
|
||||
std::string line;
|
||||
getline(response->content, line);
|
||||
size_t version_end=line.find(' ');
|
||||
|
|
@ -202,7 +400,7 @@ namespace SimpleWeb {
|
|||
response->http_version=line.substr(5, version_end-5);
|
||||
if((version_end+1)<line.size())
|
||||
response->status_code=line.substr(version_end+1, line.size()-(version_end+1)-1);
|
||||
|
||||
|
||||
getline(response->content, line);
|
||||
size_t param_end;
|
||||
while((param_end=line.find(':'))!=std::string::npos) {
|
||||
|
|
@ -213,145 +411,145 @@ namespace SimpleWeb {
|
|||
if(value_start<line.size())
|
||||
response->header.insert(std::make_pair(line.substr(0, param_end), line.substr(value_start, line.size()-value_start-1)));
|
||||
}
|
||||
|
||||
|
||||
getline(response->content, line);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
std::shared_ptr<asio::streambuf> create_request_header(const std::string& request_type, const std::string& path, const std::map<std::string, std::string>& header) const {
|
||||
auto corrected_path=path;
|
||||
if(corrected_path=="")
|
||||
corrected_path="/";
|
||||
if(!config.proxy_server.empty() && std::is_same<socket_type, asio::ip::tcp::socket>::value)
|
||||
corrected_path="http://"+host+':'+std::to_string(port)+corrected_path;
|
||||
|
||||
auto write_buffer=std::make_shared<asio::streambuf>();
|
||||
std::ostream write_stream(write_buffer.get());
|
||||
write_stream << request_type << " " << corrected_path << " HTTP/1.1\r\n";
|
||||
write_stream << "Host: " << host << "\r\n";
|
||||
for(auto& h: header)
|
||||
write_stream << h.first << ": " << h.second << "\r\n";
|
||||
return write_buffer;
|
||||
}
|
||||
|
||||
void request_write(const std::shared_ptr<asio::streambuf> &write_buffer) {
|
||||
connect();
|
||||
|
||||
auto timer=get_timeout_timer();
|
||||
asio::async_write(*socket, *write_buffer, [this, write_buffer, timer](const error_code &ec, size_t /*bytes_transferred*/) {
|
||||
static void write(const std::shared_ptr<Session> &session) {
|
||||
auto timer=get_timeout_timer(session);
|
||||
asio::async_write(*session->connection->socket, session->request_buffer->data(), [session, timer](const error_code &ec, size_t /*bytes_transferred*/) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(ec) {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
if(!ec)
|
||||
read(session);
|
||||
else {
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
std::shared_ptr<Response> request_read() {
|
||||
std::shared_ptr<Response> response(new Response());
|
||||
|
||||
auto timer=get_timeout_timer();
|
||||
asio::async_read_until(*socket, response->content_buffer, "\r\n\r\n",
|
||||
[this, response, timer](const error_code& ec, size_t bytes_transferred) {
|
||||
static void read(const std::shared_ptr<Session> &session) {
|
||||
auto timer=get_timeout_timer(session);
|
||||
asio::async_read_until(*session->connection->socket, session->response->content_buffer, "\r\n\r\n",
|
||||
[session, timer](const error_code& ec, size_t bytes_transferred) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec) {
|
||||
size_t num_additional_bytes=response->content_buffer.size()-bytes_transferred;
|
||||
session->connection->reconnecting=false;
|
||||
|
||||
parse_response_header(response);
|
||||
|
||||
auto header_it=response->header.find("Content-Length");
|
||||
if(header_it!=response->header.end()) {
|
||||
size_t num_additional_bytes=session->response->content_buffer.size()-bytes_transferred;
|
||||
|
||||
parse_response_header(session->response);
|
||||
|
||||
auto header_it=session->response->header.find("Content-Length");
|
||||
if(header_it!=session->response->header.end()) {
|
||||
auto content_length=stoull(header_it->second);
|
||||
if(content_length>num_additional_bytes) {
|
||||
auto timer=get_timeout_timer();
|
||||
asio::async_read(*socket, response->content_buffer, asio::transfer_exactly(content_length-num_additional_bytes),
|
||||
[this, response, timer](const error_code& ec, size_t /*bytes_transferred*/) {
|
||||
auto timer=get_timeout_timer(session);
|
||||
asio::async_read(*session->connection->socket, session->response->content_buffer, asio::transfer_exactly(content_length-num_additional_bytes),
|
||||
[session, timer](const error_code& ec, size_t /*bytes_transferred*/) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(ec) {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
if(!ec)
|
||||
session->callback(ec);
|
||||
else {
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
else
|
||||
session->callback(ec);
|
||||
}
|
||||
else if((header_it=response->header.find("Transfer-Encoding"))!=response->header.end() && header_it->second=="chunked") {
|
||||
else if((header_it=session->response->header.find("Transfer-Encoding"))!=session->response->header.end() && header_it->second=="chunked") {
|
||||
auto tmp_streambuf=std::make_shared<asio::streambuf>();
|
||||
request_read_chunked(response, tmp_streambuf);
|
||||
read_chunked(session, tmp_streambuf);
|
||||
}
|
||||
else if(response->http_version<"1.1" || ((header_it=response->header.find("Connection"))!=response->header.end() && header_it->second=="close")) {
|
||||
auto timer=get_timeout_timer();
|
||||
asio::async_read(*socket, response->content_buffer, [this, response, timer](const error_code& ec, size_t /*bytes_transferred*/) {
|
||||
else if(session->response->http_version<"1.1" || ((header_it=session->response->header.find("Session"))!=session->response->header.end() && header_it->second=="close")) {
|
||||
auto timer=get_timeout_timer(session);
|
||||
asio::async_read(*session->connection->socket, session->response->content_buffer, [session, timer](const error_code& ec, size_t /*bytes_transferred*/) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(ec) {
|
||||
close();
|
||||
if(ec!=asio::error::eof)
|
||||
throw system_error(ec);
|
||||
timer->cancel();
|
||||
if(!ec)
|
||||
session->callback(ec);
|
||||
else {
|
||||
close(session);
|
||||
if(ec==asio::error::eof) {
|
||||
error_code ec;
|
||||
session->callback(ec);
|
||||
}
|
||||
else
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
else
|
||||
session->callback(ec);
|
||||
}
|
||||
else {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
if(!session->connection->reconnecting) {
|
||||
session->connection->reconnecting=true;
|
||||
close(session);
|
||||
Client<socket_type>::connect(session);
|
||||
}
|
||||
else {
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
}
|
||||
});
|
||||
io_service->reset();
|
||||
io_service->run();
|
||||
|
||||
return response;
|
||||
}
|
||||
|
||||
void request_read_chunked(const std::shared_ptr<Response> &response, const std::shared_ptr<asio::streambuf> &tmp_streambuf) {
|
||||
auto timer=get_timeout_timer();
|
||||
asio::async_read_until(*socket, response->content_buffer, "\r\n", [this, response, tmp_streambuf, timer](const error_code& ec, size_t bytes_transferred) {
|
||||
static void read_chunked(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &tmp_streambuf) {
|
||||
auto timer=get_timeout_timer(session);
|
||||
asio::async_read_until(*session->connection->socket, session->response->content_buffer, "\r\n", [session, tmp_streambuf, timer](const error_code& ec, size_t bytes_transferred) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec) {
|
||||
std::string line;
|
||||
getline(response->content, line);
|
||||
getline(session->response->content, line);
|
||||
bytes_transferred-=line.size()+1;
|
||||
line.pop_back();
|
||||
std::streamsize length=stol(line, 0, 16);
|
||||
|
||||
auto num_additional_bytes=static_cast<std::streamsize>(response->content_buffer.size()-bytes_transferred);
|
||||
auto num_additional_bytes=static_cast<std::streamsize>(session->response->content_buffer.size()-bytes_transferred);
|
||||
|
||||
auto post_process=[this, response, tmp_streambuf, length] {
|
||||
auto post_process=[session, tmp_streambuf, length] {
|
||||
std::ostream tmp_stream(tmp_streambuf.get());
|
||||
if(length>0) {
|
||||
std::vector<char> buffer(static_cast<size_t>(length));
|
||||
response->content.read(&buffer[0], length);
|
||||
session->response->content.read(&buffer[0], length);
|
||||
tmp_stream.write(&buffer[0], length);
|
||||
}
|
||||
|
||||
//Remove "\r\n"
|
||||
response->content.get();
|
||||
response->content.get();
|
||||
session->response->content.get();
|
||||
session->response->content.get();
|
||||
|
||||
if(length>0)
|
||||
request_read_chunked(response, tmp_streambuf);
|
||||
read_chunked(session, tmp_streambuf);
|
||||
else {
|
||||
std::ostream response_stream(&response->content_buffer);
|
||||
std::ostream response_stream(&session->response->content_buffer);
|
||||
response_stream << tmp_stream.rdbuf();
|
||||
error_code ec;
|
||||
session->callback(ec);
|
||||
}
|
||||
};
|
||||
|
||||
if((2+length)>num_additional_bytes) {
|
||||
auto timer=get_timeout_timer();
|
||||
asio::async_read(*socket, response->content_buffer, asio::transfer_exactly(2+length-num_additional_bytes),
|
||||
[this, response, post_process, timer](const error_code& ec, size_t /*bytes_transferred*/) {
|
||||
auto timer=get_timeout_timer(session);
|
||||
asio::async_read(*session->connection->socket, session->response->content_buffer, asio::transfer_exactly(2+length-num_additional_bytes),
|
||||
[session, post_process, timer](const error_code& ec, size_t /*bytes_transferred*/) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec) {
|
||||
if(!ec)
|
||||
post_process();
|
||||
}
|
||||
else {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
|
@ -359,11 +557,17 @@ namespace SimpleWeb {
|
|||
post_process();
|
||||
}
|
||||
else {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
static void close(const std::shared_ptr<Session> &session) {
|
||||
error_code ec;
|
||||
session->connection->socket->lowest_layer().shutdown(asio::ip::tcp::socket::shutdown_both, ec);
|
||||
session->connection->socket->lowest_layer().close(ec);
|
||||
}
|
||||
};
|
||||
|
||||
template<class socket_type>
|
||||
|
|
@ -374,46 +578,46 @@ namespace SimpleWeb {
|
|||
template<>
|
||||
class Client<HTTP> : public ClientBase<HTTP> {
|
||||
public:
|
||||
Client(const std::string& server_port_path) : ClientBase<HTTP>::ClientBase(server_port_path, 80) {
|
||||
socket=std::unique_ptr<HTTP>(new HTTP(*io_service));
|
||||
}
|
||||
friend ClientBase<HTTP>;
|
||||
|
||||
Client(const std::string& server_port_path) : ClientBase<HTTP>::ClientBase(server_port_path, 80) {}
|
||||
|
||||
protected:
|
||||
void connect() {
|
||||
if(!socket->lowest_layer().is_open()) {
|
||||
std::unique_ptr<asio::ip::tcp::resolver::query> query;
|
||||
if(config.proxy_server.empty())
|
||||
query=std::unique_ptr<asio::ip::tcp::resolver::query>(new asio::ip::tcp::resolver::query(host, std::to_string(port)));
|
||||
else {
|
||||
auto proxy_host_port=parse_host_port(config.proxy_server, 8080);
|
||||
query=std::unique_ptr<asio::ip::tcp::resolver::query>(new asio::ip::tcp::resolver::query(proxy_host_port.first, std::to_string(proxy_host_port.second)));
|
||||
}
|
||||
|
||||
auto resolver=std::make_shared<asio::ip::tcp::resolver>(*io_service);
|
||||
resolver->async_resolve(*query, [this, resolver](const error_code &ec, asio::ip::tcp::resolver::iterator it){
|
||||
std::shared_ptr<Connection> create_connection() override {
|
||||
return std::make_shared<Connection>(host, port, config, std::unique_ptr<HTTP>(new HTTP(*io_service)));
|
||||
}
|
||||
|
||||
static void connect(const std::shared_ptr<Session> &session) {
|
||||
if(!session->connection->socket->lowest_layer().is_open()) {
|
||||
auto resolver=std::make_shared<asio::ip::tcp::resolver>(*session->io_service);
|
||||
auto timer=get_timeout_timer(session, session->connection->config.timeout_connect);
|
||||
resolver->async_resolve(*session->connection->query, [session, timer, resolver](const error_code &ec, asio::ip::tcp::resolver::iterator it){
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec) {
|
||||
auto timer=get_timeout_timer(config.timeout_connect);
|
||||
asio::async_connect(*socket, it, [this, resolver, timer](const error_code &ec, asio::ip::tcp::resolver::iterator /*it*/){
|
||||
auto timer=get_timeout_timer(session, session->connection->config.timeout_connect);
|
||||
asio::async_connect(*session->connection->socket, it, [session, timer, resolver](const error_code &ec, asio::ip::tcp::resolver::iterator /*it*/){
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec) {
|
||||
asio::ip::tcp::no_delay option(true);
|
||||
this->socket->set_option(option);
|
||||
session->connection->socket->set_option(option);
|
||||
write(session);
|
||||
}
|
||||
else {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
else {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
io_service->reset();
|
||||
io_service->run();
|
||||
}
|
||||
else
|
||||
write(session);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
|
|
|||
142
client_https.hpp
142
client_https.hpp
|
|
@ -15,6 +15,8 @@ namespace SimpleWeb {
|
|||
template<>
|
||||
class Client<HTTPS> : public ClientBase<HTTPS> {
|
||||
public:
|
||||
friend ClientBase<HTTPS>;
|
||||
|
||||
Client(const std::string& server_port_path, bool verify_certificate=true,
|
||||
const std::string& cert_file=std::string(), const std::string& private_key_file=std::string(),
|
||||
const std::string& verify_file=std::string()) :
|
||||
|
|
@ -36,95 +38,95 @@ namespace SimpleWeb {
|
|||
context.set_verify_mode(asio::ssl::verify_peer);
|
||||
else
|
||||
context.set_verify_mode(asio::ssl::verify_none);
|
||||
|
||||
socket=std::unique_ptr<HTTPS>(new HTTPS(*io_service, context));
|
||||
}
|
||||
|
||||
protected:
|
||||
asio::ssl::context context;
|
||||
|
||||
void connect() {
|
||||
if(!socket->lowest_layer().is_open()) {
|
||||
std::unique_ptr<asio::ip::tcp::resolver::query> query;
|
||||
if(config.proxy_server.empty())
|
||||
query=std::unique_ptr<asio::ip::tcp::resolver::query>(new asio::ip::tcp::resolver::query(host, std::to_string(port)));
|
||||
else {
|
||||
auto proxy_host_port=parse_host_port(config.proxy_server, 8080);
|
||||
query=std::unique_ptr<asio::ip::tcp::resolver::query>(new asio::ip::tcp::resolver::query(proxy_host_port.first, std::to_string(proxy_host_port.second)));
|
||||
}
|
||||
auto resolver=std::make_shared<asio::ip::tcp::resolver>(*io_service);
|
||||
resolver->async_resolve(*query, [this, resolver] (const error_code &ec, asio::ip::tcp::resolver::iterator it){
|
||||
std::shared_ptr<Connection> create_connection() override {
|
||||
return std::make_shared<Connection>(host, port, config, std::unique_ptr<HTTPS>(new HTTPS(*io_service, context)));
|
||||
}
|
||||
|
||||
static void connect(const std::shared_ptr<Session> &session) {
|
||||
if(!session->connection->socket->lowest_layer().is_open()) {
|
||||
auto resolver=std::make_shared<asio::ip::tcp::resolver>(*session->io_service);
|
||||
resolver->async_resolve(*session->connection->query, [session, resolver] (const error_code &ec, asio::ip::tcp::resolver::iterator it){
|
||||
if(!ec) {
|
||||
auto timer=get_timeout_timer(config.timeout_connect);
|
||||
asio::async_connect(socket->lowest_layer(), it, [this, resolver, timer] (const error_code &ec, asio::ip::tcp::resolver::iterator /*it*/){
|
||||
auto timer=get_timeout_timer(session, session->connection->config.timeout_connect);
|
||||
asio::async_connect(session->connection->socket->lowest_layer(), it, [session, resolver, timer] (const error_code &ec, asio::ip::tcp::resolver::iterator /*it*/){
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec) {
|
||||
asio::ip::tcp::no_delay option(true);
|
||||
this->socket->lowest_layer().set_option(option);
|
||||
session->connection->socket->lowest_layer().set_option(option);
|
||||
|
||||
if(!session->connection->config.proxy_server.empty()) {
|
||||
auto write_buffer=std::make_shared<asio::streambuf>();
|
||||
std::ostream write_stream(write_buffer.get());
|
||||
auto host_port=session->connection->host+':'+std::to_string(session->connection->port);
|
||||
write_stream << "CONNECT "+host_port+" HTTP/1.1\r\n" << "Host: " << host_port << "\r\n\r\n";
|
||||
auto timer=get_timeout_timer(session, session->connection->config.timeout_connect);
|
||||
asio::async_write(session->connection->socket->next_layer(), *write_buffer, [session, write_buffer, timer](const error_code &ec, size_t /*bytes_transferred*/) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec) {
|
||||
std::shared_ptr<Response> response(new Response());
|
||||
auto timer=get_timeout_timer(session, session->connection->config.timeout_connect);
|
||||
asio::async_read_until(session->connection->socket->next_layer(), response->content_buffer, "\r\n\r\n", [session, response, timer](const error_code& ec, size_t /*bytes_transferred*/) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec) {
|
||||
parse_response_header(response);
|
||||
if (response->status_code.empty() || response->status_code.compare(0, 3, "200") != 0) {
|
||||
close(session);
|
||||
session->callback(make_error_code::make_error_code(errc::permission_denied));
|
||||
}
|
||||
else
|
||||
handshake(session);
|
||||
}
|
||||
else {
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
else {
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
else
|
||||
handshake(session);
|
||||
}
|
||||
else {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
else {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
io_service->reset();
|
||||
io_service->run();
|
||||
|
||||
if(!config.proxy_server.empty()) {
|
||||
auto write_buffer=std::make_shared<asio::streambuf>();
|
||||
std::ostream write_stream(write_buffer.get());
|
||||
auto host_port=host+':'+std::to_string(port);
|
||||
write_stream << "CONNECT "+host_port+" HTTP/1.1\r\n" << "Host: " << host_port << "\r\n\r\n";
|
||||
auto timer=get_timeout_timer();
|
||||
asio::async_write(socket->next_layer(), *write_buffer, [this, write_buffer, timer](const error_code &ec, size_t /*bytes_transferred*/) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(ec) {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
}
|
||||
});
|
||||
io_service->reset();
|
||||
io_service->run();
|
||||
|
||||
std::shared_ptr<Response> response(new Response());
|
||||
timer=get_timeout_timer();
|
||||
asio::async_read_until(socket->next_layer(), response->content_buffer, "\r\n\r\n", [this, response, timer](const error_code& ec, size_t /*bytes_transferred*/) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(ec) {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
}
|
||||
});
|
||||
io_service->reset();
|
||||
io_service->run();
|
||||
parse_response_header(response);
|
||||
if (response->status_code.empty() || response->status_code.compare(0, 3, "200") != 0) {
|
||||
close();
|
||||
throw make_error_code::make_error_code(errc::permission_denied);
|
||||
}
|
||||
}
|
||||
|
||||
auto timer=get_timeout_timer();
|
||||
this->socket->async_handshake(asio::ssl::stream_base::client, [this, timer](const error_code& ec) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(ec) {
|
||||
close();
|
||||
throw system_error(ec);
|
||||
}
|
||||
});
|
||||
io_service->reset();
|
||||
io_service->run();
|
||||
}
|
||||
else
|
||||
write(session);
|
||||
}
|
||||
|
||||
static void handshake(const std::shared_ptr<Session> &session) {
|
||||
auto timer=get_timeout_timer(session, session->connection->config.timeout_connect);
|
||||
session->connection->socket->async_handshake(asio::ssl::stream_base::client, [session, timer](const error_code& ec) {
|
||||
if(timer)
|
||||
timer->cancel();
|
||||
if(!ec)
|
||||
write(session);
|
||||
else {
|
||||
close(session);
|
||||
session->callback(ec);
|
||||
}
|
||||
});
|
||||
}
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -33,6 +33,12 @@ int main() {
|
|||
*response << "HTTP/1.1 200 OK\r\nContent-Length: " << number.length() << "\r\n\r\n" << number;
|
||||
};
|
||||
|
||||
server.resource["^/header$"]["GET"]=[](shared_ptr<HttpServer::Response> response, shared_ptr<HttpServer::Request> request) {
|
||||
auto content=request->header.find("test1")->second+request->header.find("test2")->second;
|
||||
|
||||
*response << "HTTP/1.1 200 OK\r\nContent-Length: " << content.length() << "\r\n\r\n" << content;
|
||||
};
|
||||
|
||||
thread server_thread([&server](){
|
||||
//Start server
|
||||
server.start();
|
||||
|
|
@ -89,6 +95,50 @@ int main() {
|
|||
output << r->content.rdbuf();
|
||||
assert(output.str()=="A string");
|
||||
}
|
||||
|
||||
{
|
||||
stringstream output;
|
||||
auto r=client.request("GET", "/header", "", {{"test1", "test"}, {"test2", "ing"}});
|
||||
output << r->content.rdbuf();
|
||||
assert(output.str()=="testing");
|
||||
}
|
||||
}
|
||||
|
||||
{
|
||||
HttpClient client("localhost:8080");
|
||||
bool call=false;
|
||||
client.request("GET", "/match/123", [&call](shared_ptr<HttpClient::Response> response, const SimpleWeb::error_code &ec) {
|
||||
assert(!ec);
|
||||
stringstream output;
|
||||
output << response->content.rdbuf();
|
||||
assert(output.str()=="123");
|
||||
call=true;
|
||||
});
|
||||
client.io_service->run();
|
||||
assert(call);
|
||||
|
||||
{
|
||||
vector<bool> calls(100);
|
||||
vector<thread> threads;
|
||||
for(size_t c=0;c<100;++c) {
|
||||
calls[c]=false;
|
||||
threads.emplace_back([c, &client, &calls] {
|
||||
client.request("GET", "/match/123", [c, &calls](shared_ptr<HttpClient::Response> response, const SimpleWeb::error_code &ec) {
|
||||
assert(!ec);
|
||||
stringstream output;
|
||||
output << response->content.rdbuf();
|
||||
assert(output.str()=="123");
|
||||
calls[c]=true;
|
||||
});
|
||||
});
|
||||
}
|
||||
for(auto &thread: threads)
|
||||
thread.join();
|
||||
client.io_service->reset();
|
||||
client.io_service->run();
|
||||
for(auto call: calls)
|
||||
assert(call);
|
||||
}
|
||||
}
|
||||
|
||||
server.stop();
|
||||
|
|
|
|||
|
|
@ -55,7 +55,9 @@ class ClientTest : public ClientBase<HTTP> {
|
|||
public:
|
||||
ClientTest(const std::string& server_port_path) : ClientBase<HTTP>::ClientBase(server_port_path, 80) {}
|
||||
|
||||
void connect() {}
|
||||
std::shared_ptr<Connection> create_connection() override {
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
void constructor_parse_test1() {
|
||||
assert(host=="test.org");
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue