Unverified Commit 16fbe079 authored by Dennis Jenkins's avatar Dennis Jenkins Committed by GitHub

Merge pull request #674 from hyperxor/issue_673

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