Commit aa02f308 authored by hyperxor's avatar hyperxor

Issue #673

parent d2a024d8
......@@ -43,19 +43,16 @@ struct Connection : public std::enable_shared_from_this<Connection> {
RequestData(
Async::Resolver resolve, Async::Rejection reject,
const Http::Request& request,
std::chrono::milliseconds timeout,
OnDone onDone)
: resolve(std::move(resolve))
, reject(std::move(reject))
, request(request)
, timeout(timeout)
, onDone(std::move(onDone))
{ }
Async::Resolver resolve;
Async::Rejection reject;
Http::Request request;
std::chrono::milliseconds timeout;
OnDone onDone;
};
......@@ -79,17 +76,14 @@ struct Connection : public std::enable_shared_from_this<Connection> {
Async::Promise<Response> perform(
const Http::Request& request,
std::chrono::milliseconds timeout,
OnDone onDone);
Async::Promise<Response> asyncPerform(
const Http::Request& request,
std::chrono::milliseconds timeout,
OnDone onDone);
void performImpl(
const Http::Request& request,
std::chrono::milliseconds timeout,
Async::Resolver resolve,
Async::Rejection reject,
OnDone onDone);
......@@ -299,7 +293,6 @@ public:
RequestBuilder& cookie(const Cookie& cookie);
RequestBuilder& body(const std::string& val);
RequestBuilder& body(std::string&& val);
RequestBuilder& timeout(std::chrono::milliseconds val);
Async::Promise<Response> send();
......@@ -308,13 +301,11 @@ private:
explicit RequestBuilder(Client* const client)
: client_(client)
, request_()
, timeout_(std::chrono::milliseconds(0))
{ }
Client* const client_;
Request request_;
std::chrono::milliseconds timeout_;
};
......@@ -373,9 +364,7 @@ private:
RequestBuilder prepareRequest(const std::string& resource, Http::Method method);
Async::Promise<Response> doRequest(
Http::Request request,
std::chrono::milliseconds timeout);
Async::Promise<Response> doRequest(Http::Request request);
void processRequestQueue();
......
......@@ -73,7 +73,8 @@ public:
friend class Private::BodyStep;
friend class Private::ParserBase;
Message();
Message() = default;
explicit Message(Version version);
Message(const Message& other) = default;
Message& operator=(const Message& other) = default;
......@@ -81,15 +82,24 @@ public:
Message(Message&& other) = default;
Message& operator=(Message&& other) = default;
Version version() const;
Code code() const;
std::string body() const;
const CookieJar& cookies() const;
CookieJar& cookies();
const Header::Collection& headers() const;
Header::Collection& headers();
protected:
Version version_;
Version version_ = Version::Http11;
Code code_;
std::string body_;
CookieJar cookies_;
Header::Collection headers_;
};
namespace Uri {
......@@ -147,7 +157,7 @@ public:
// @Todo: try to remove the need for friend-ness here
friend class Client;
Request();
Request() = default;
Request(const Request& other) = default;
Request& operator=(const Request& other) = default;
......@@ -155,17 +165,11 @@ public:
Request(Request&& other) = default;
Request& operator=(Request&& other) = default;
Version version() const;
Method method() const;
std::string resource() const;
std::string body() const;
const Header::Collection& headers() const;
const Uri::Query& query() const;
const CookieJar& cookies() const;
/* @Investigate: this is disabled because of a lock in the shared_ptr / weak_ptr
implementation of libstdc++. Under contention, we experience a performance
drop of 5x with that lock
......@@ -184,6 +188,8 @@ public:
address_ = address;
}
std::chrono::milliseconds timeout() const;
private:
#ifdef LIBSTDCPP_SMARTPTR_LOCK_FIXME
void associatePeer(const std::shared_ptr<Tcp::Peer>& peer) {
......@@ -203,6 +209,7 @@ private:
std::weak_ptr<Tcp::Peer> peer_;
#endif
Address address_;
std::chrono::milliseconds timeout_ = std::chrono::milliseconds(0);
};
class Handler;
......@@ -338,18 +345,6 @@ public:
return sz;
}
const Header::Collection& headers() const {
return headers_;
}
const CookieJar& cookies() const {
return cookies_;
}
Code code() const {
return code_;
}
void flush();
void ends();
......@@ -408,46 +403,12 @@ public:
friend class Private::Parser<Http::Response>;
Response() = default;
explicit Response(Version version)
: Message()
{
version_ = version;
}
explicit Response(Version version);
Response(const Response& other) = default;
Response& operator=(const Response& other) = default;
Response(Response&& other) = default;
Response& operator=(Response&& other) = default;
const Header::Collection& headers() const {
return headers_;
}
Header::Collection& headers() {
return headers_;
}
const CookieJar& cookies() const {
return cookies_;
}
CookieJar& cookies() {
return cookies_;
}
Code code() const {
return code_;
}
std::string body() const {
return body_;
}
Version version() const {
return version_;
}
};
class ResponseWriter : public Response {
......
......@@ -509,28 +509,25 @@ void Connection::handleTimeout() {
}
Async::Promise<Response> Connection::perform(const Http::Request &request,
std::chrono::milliseconds timeout,
Connection::OnDone onDone) {
return Async::Promise<Response>(
[=](Async::Resolver &resolve, Async::Rejection &reject) {
performImpl(request, timeout, std::move(resolve), std::move(reject),
performImpl(request, std::move(resolve), std::move(reject),
std::move(onDone));
});
}
Async::Promise<Response>
Connection::asyncPerform(const Http::Request &request,
std::chrono::milliseconds timeout,
Connection::OnDone onDone) {
return Async::Promise<Response>(
[=](Async::Resolver &resolve, Async::Rejection &reject) {
requestsQueue.push(RequestData(std::move(resolve), std::move(reject),
request, timeout, std::move(onDone)));
request, std::move(onDone)));
});
}
void Connection::performImpl(const Http::Request &request,
std::chrono::milliseconds timeout,
Async::Resolver resolve, Async::Rejection reject,
Connection::OnDone onDone) {
......@@ -541,6 +538,7 @@ void Connection::performImpl(const Http::Request &request,
std::string buffer = streamBuf.str();
std::shared_ptr<TimerPool::Entry> timer(nullptr);
auto timeout = request.timeout();
if (timeout.count() > 0) {
timer = timerPool_.pickTimer();
timer->arm(timeout);
......@@ -557,7 +555,7 @@ void Connection::processRequestQueue() {
if (!req)
break;
performImpl(req->request, req->timeout, std::move(req->resolve),
performImpl(req->request, std::move(req->resolve),
std::move(req->reject), std::move(req->onDone));
}
}
......@@ -680,12 +678,12 @@ RequestBuilder &RequestBuilder::body(std::string &&val) {
}
RequestBuilder &RequestBuilder::timeout(std::chrono::milliseconds val) {
timeout_ = val;
request_.timeout_ = val;
return *this;
}
Async::Promise<Response> RequestBuilder::send() {
return client_->doRequest(request_, timeout_);
return client_->doRequest(request_);
}
Client::Options &Client::Options::threads(int val) {
......@@ -755,27 +753,25 @@ RequestBuilder Client::prepareRequest(const std::string &resource,
return builder;
}
Async::Promise<Response> Client::doRequest(Http::Request request,
std::chrono::milliseconds timeout) {
Async::Promise<Response> Client::doRequest(Http::Request request) {
// request.headers_.add<Header::Connection>(ConnectionControl::KeepAlive);
request.headers_.remove<Header::UserAgent>();
auto resource = request.resource();
auto resourceData = request.resource();
auto s = splitUrl(resource);
auto conn = pool.pickConnection(s.first);
auto resource = splitUrl(resourceData);
auto conn = pool.pickConnection(resource.first);
if (conn == nullptr) {
// TODO: C++14 - use capture move for s
return Async::Promise<Response>(
[this, s, request, timeout](Async::Resolver &resolve,
Async::Rejection &reject) {
Guard guard(queuesLock);
auto data = std::make_shared<Connection::RequestData>(
std::move(resolve), std::move(reject), request, timeout, nullptr);
auto &queue = requestsQueues[s.first];
if (!queue.enqueue(data))
data->reject(std::runtime_error("Queue is full"));
[this, resource = std::move(resource), request]
(Async::Resolver &resolve, Async::Rejection &reject) {
Guard guard(queuesLock);
auto data = std::make_shared<Connection::RequestData>(
std::move(resolve), std::move(reject), std::move(request), nullptr);
auto &queue = requestsQueues[resource.first];
if (!queue.enqueue(data))
data->reject(std::runtime_error("Queue is full"));
});
} else {
......@@ -788,25 +784,25 @@ Async::Promise<Response> Client::doRequest(Http::Request request,
}
if (!conn->isConnected()) {
std::weak_ptr<Connection> weakConn = conn;
auto res = conn->asyncPerform(request, timeout, [this, weakConn]() {
auto conn = weakConn.lock();
if (conn) {
pool.releaseConnection(conn);
processRequestQueue();
}
});
conn->connect(helpers::httpAddr(s.first));
return res;
std::weak_ptr<Connection> weakConn = conn;
auto res = conn->asyncPerform(request, [this, weakConn]() {
auto conn = weakConn.lock();
if (conn) {
pool.releaseConnection(conn);
processRequestQueue();
}
});
conn->connect(helpers::httpAddr(resource.first));
return res;
}
std::weak_ptr<Connection> weakConn = conn;
return conn->perform(request, timeout, [this, weakConn]() {
auto conn = weakConn.lock();
if (conn) {
pool.releaseConnection(conn);
processRequestQueue();
}
return conn->perform(request, [this, weakConn]() {
auto conn = weakConn.lock();
if (conn) {
pool.releaseConnection(conn);
processRequestQueue();
}
});
}
}
......@@ -831,7 +827,7 @@ void Client::processRequestQueue() {
break;
}
conn->performImpl(data->request, data->timeout, std::move(data->resolve),
conn->performImpl(data->request, std::move(data->resolve),
std::move(data->reject), [this, conn]() {
pool.releaseConnection(conn);
processRequestQueue();
......
......@@ -489,9 +489,6 @@ void ParserBase::reset() {
} // namespace Private
Message::Message()
: version_(Version::Http11), code_(), body_(), cookies_(), headers_() {}
namespace Uri {
Query::Query() : params() {}
......@@ -529,24 +526,40 @@ bool Query::has(const std::string &name) const {
} // namespace Uri
Request::Request() = default;
Message::Message(Version version)
: version_(version)
{}
Version Request::version() const { return version_; }
Version Message::version() const { return version_; }
Method Request::method() const { return method_; }
Code Message::code() const { return code_; }
std::string Request::resource() const { return resource_; }
std::string Message::body() const { return body_; }
std::string Request::body() const { return body_; }
const Header::Collection &Message::headers() const { return headers_; }
const Header::Collection &Request::headers() const { return headers_; }
Header::Collection &Message::headers() { return headers_; }
const Uri::Query &Request::query() const { return query_; }
const CookieJar &Message::cookies() const { return cookies_; }
const CookieJar &Request::cookies() const { return cookies_; }
CookieJar &Message::cookies() { return cookies_; }
Method Request::method() const { return method_; }
std::string Request::resource() const { return resource_; }
const Uri::Query &Request::query() const { return query_; }
const Address &Request::address() const { return address_; }
std::chrono::milliseconds Request::timeout() const {
return timeout_;
}
Response::Response(Version version)
: Message(version)
{}
#ifdef LIBSTDCPP_SMARTPTR_LOCK_FIXME
std::shared_ptr<Tcp::Peer> Request::peer() const {
auto p = peer_.lock();
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment