Unverified Commit 79374629 authored by Dennis Jenkins's avatar Dennis Jenkins Committed by GitHub

Merge pull request #374 from knowledge4igor/fix_connection_problem_in_client_plus_refactoring

Fix Connection problem in Client + code refactoring + one more test
parents c012e91c a494bd7d
......@@ -159,6 +159,10 @@ namespace Async {
}
};
struct EmptyCall {
void operator()(size_t) const { }
};
struct Core;
class Request {
......@@ -924,6 +928,8 @@ namespace Async {
static constexpr Private::NoExcept NoExcept{};
static constexpr Private::Throw Throw{};
static constexpr Private::EmptyCall EmptyCall{};
namespace details {
/*
......
......@@ -18,7 +18,6 @@
#include <atomic>
#include <chrono>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
......@@ -41,8 +40,7 @@ struct Connection : public std::enable_shared_from_this<Connection> {
Connection()
: fd(-1)
, inflightCount(0)
, responsesReceived(0)
, requestEntry(nullptr)
, connectionState_(NotConnected)
{
state_.store(static_cast<uint32_t>(State::Idle));
......@@ -108,11 +106,6 @@ struct Connection : public std::enable_shared_from_this<Connection> {
std::string dump() const;
private:
std::atomic<int> inflightCount;
std::atomic<int> responsesReceived;
struct sockaddr_in saddr;
void processRequestQueue();
struct RequestEntry {
......@@ -132,15 +125,14 @@ private:
OnDone onDone;
};
struct sockaddr_in saddr;
std::unique_ptr<RequestEntry> requestEntry;
std::atomic<uint32_t> state_;
ConnectionState connectionState_;
std::shared_ptr<Transport> transport_;
Queue<RequestData> requestsQueue;
std::deque<RequestEntry> inflightRequests;
TimerPool timerPool_;
Private::Parser<Http::Response> parser_;
};
class ConnectionPool {
......
......@@ -415,53 +415,54 @@ Connection::hasTransport() const {
void
Connection::handleResponsePacket(const char* buffer, size_t totalBytes) {
parser_.feed(buffer, totalBytes);
if (parser_.parse() == Private::State::Done) {
if (!inflightRequests.empty()) {
auto req = std::move(inflightRequests.front());
inflightRequests.pop_back();
if (req.timer) {
req.timer->disarm();
timerPool_.releaseTimer(req.timer);
Private::Parser<Http::Response> parser;
parser.feed(buffer, totalBytes);
if (parser.parse() == Private::State::Done) {
if (requestEntry) {
if (requestEntry->timer) {
requestEntry->timer->disarm();
timerPool_.releaseTimer(requestEntry->timer);
}
req.resolve(std::move(parser_.response));
if (req.onDone)
req.onDone();
}
requestEntry->resolve(std::move(parser.response));
if (requestEntry->onDone)
requestEntry->onDone();
parser_.reset();
requestEntry.reset(nullptr);
}
} else {
// TODO: Do more specific error
requestEntry->reject(Error("Response problem"));
}
}
void
Connection::handleError(const char* error) {
if (!inflightRequests.empty()) {
auto req = std::move(inflightRequests.front());
if (req.timer) {
req.timer->disarm();
timerPool_.releaseTimer(req.timer);
if (requestEntry) {
if (requestEntry->timer) {
requestEntry->timer->disarm();
timerPool_.releaseTimer(requestEntry->timer);
}
req.reject(Error(error));
if (req.onDone)
req.onDone();
requestEntry->reject(Error(error));
if (requestEntry->onDone)
requestEntry->onDone();
requestEntry.reset(nullptr);
}
}
void
Connection::handleTimeout() {
if (!inflightRequests.empty()) {
auto req = std::move(inflightRequests.front());
inflightRequests.pop_back();
if (requestEntry) {
timerPool_.releaseTimer(requestEntry->timer);
timerPool_.releaseTimer(req.timer);
if (req.onDone)
req.onDone();
if (requestEntry->onDone)
requestEntry->onDone();
/* @API: create a TimeoutException */
req.reject(std::runtime_error("Timeout"));
requestEntry->reject(std::runtime_error("Timeout"));
requestEntry.reset(nullptr);
}
}
......@@ -487,43 +488,6 @@ Connection::perform(
});
}
/**
* This class is used to emulate the generalized lambda capture feature from C++14
* whereby a given object can be moved inside a lambda, directly from the capture-list
*
* So instead, it will use the exact same semantic than auto_ptr (don't beat me for that),
* meaning that it will move the value on copy
*/
template<typename T>
struct MoveOnCopy {
MoveOnCopy(T val)
: val(std::move(val))
{ }
MoveOnCopy(MoveOnCopy& other)
: val(std::move(other.val))
{ }
MoveOnCopy& operator=(MoveOnCopy& other) {
val = std::move(other.val);
return *this;
}
MoveOnCopy(MoveOnCopy&& other) = default;
MoveOnCopy& operator=(MoveOnCopy&& other) = default;
operator T&&() {
return std::move(val);
}
T val;
};
template<typename T>
MoveOnCopy<T> make_copy_mover(T arg) {
return MoveOnCopy<T>(std::move(arg));
}
void
Connection::performImpl(
const Http::Request& request,
......@@ -545,18 +509,11 @@ Connection::performImpl(
}
auto rejectClone = reject.clone();
auto rejectCloneMover = make_copy_mover(std::move(rejectClone));
// Move the resolver and rejecter inside the lambda
auto resolveMover = make_copy_mover(std::move(resolve));
auto rejectMover = make_copy_mover(std::move(reject));
requestEntry.reset(new RequestEntry(std::move(resolve), std::move(reject), timer, std::move(onDone)));
transport_->asyncSendRequest(shared_from_this(), timer, std::move(buffer)).then(
[=](ssize_t) mutable {
inflightRequests.push_back(RequestEntry(std::move(resolveMover), std::move(rejectMover), std::move(timer), std::move(onDone)));
},
[=](std::exception_ptr e) { rejectCloneMover.val(e); });
Async::EmptyCall,
[&](std::exception_ptr e) { rejectClone(e); });
}
void
......
......@@ -9,7 +9,8 @@
using namespace Pistache;
struct HelloHandler : public Http::Handler {
struct HelloHandler : public Http::Handler
{
HTTP_PROTOTYPE(HelloHandler)
void onRequest(const Http::Request& /*request*/, Http::ResponseWriter writer) override
......@@ -18,8 +19,20 @@ struct HelloHandler : public Http::Handler {
}
};
TEST(http_client_test, one_client_with_one_request) {
const std::string address = "localhost:9079";
struct DelayHandler : public Http::Handler
{
HTTP_PROTOTYPE(DelayHandler)
void onRequest(const Http::Request& /*request*/, Http::ResponseWriter writer) override
{
std::this_thread::sleep_for(std::chrono::seconds(4));
writer.send(Http::Code::Ok, "Hello, World!");
}
};
TEST(http_client_test, one_client_with_one_request)
{
const std::string address = "localhost:9100";
Http::Endpoint server(address);
auto flags = Tcp::Options::InstallSignalHandler | Tcp::Options::ReuseAddr;
......@@ -31,18 +44,16 @@ TEST(http_client_test, one_client_with_one_request) {
Http::Client client;
client.init();
std::vector<Async::Promise<Http::Response>> responses;
auto rb = client.get(address);
auto response = rb.send();
bool done = false;
response.then([&](Http::Response rsp) {
if (rsp.code() == Http::Code::Ok)
done = true;
}, Async::IgnoreException);
responses.push_back(std::move(response));
response.then([&](Http::Response rsp)
{
if (rsp.code() == Http::Code::Ok)
done = true;
}, Async::IgnoreException);
auto sync = Async::whenAll(responses.begin(), responses.end());
Async::Barrier<std::vector<Http::Response>> barrier(sync);
Async::Barrier<Http::Response> barrier(response);
barrier.wait_for(std::chrono::seconds(5));
server.shutdown();
......@@ -52,7 +63,7 @@ TEST(http_client_test, one_client_with_one_request) {
}
TEST(http_client_test, one_client_with_multiple_requests) {
const std::string address = "localhost:9080";
const std::string address = "localhost:9101";
Http::Endpoint server(address);
auto flags = Tcp::Options::InstallSignalHandler | Tcp::Options::ReuseAddr;
......@@ -90,7 +101,7 @@ TEST(http_client_test, one_client_with_multiple_requests) {
}
TEST(http_client_test, multiple_clients_with_one_request) {
const std::string address = "localhost:9081";
const std::string address = "localhost:9102";
Http::Endpoint server(address);
auto flags = Tcp::Options::InstallSignalHandler | Tcp::Options::ReuseAddr;
......@@ -144,3 +155,38 @@ TEST(http_client_test, multiple_clients_with_one_request) {
ASSERT_TRUE(response_counter == CLIENT_SIZE);
}
TEST(http_client_test, timeout_reject)
{
const std::string address = "localhost:9103";
Http::Endpoint server(address);
auto flags = Tcp::Options::InstallSignalHandler | Tcp::Options::ReuseAddr;
auto server_opts = Http::Endpoint::options().flags(flags);
server.init(server_opts);
server.setHandler(Http::make_handler<DelayHandler>());
server.serveThreaded();
Http::Client client;
client.init();
auto rb = client.get(address).timeout(std::chrono::milliseconds(1000));
auto response = rb.send();
bool is_reject = false;
response.then([&is_reject](Http::Response /*rsp*/)
{
is_reject = false;
},
[&is_reject](std::exception_ptr /*exc*/)
{
is_reject = true;
});
Async::Barrier<Http::Response> barrier(response);
barrier.wait_for(std::chrono::seconds(5));
server.shutdown();
client.shutdown();
ASSERT_TRUE(is_reject);
}
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