Commit 19894130 authored by knowledge4igor's avatar knowledge4igor

Client code refactoring

parent 673cf423
...@@ -5,11 +5,6 @@ ...@@ -5,11 +5,6 @@
*/ */
#pragma once #pragma once
#include <atomic>
#include <deque>
#include <sys/types.h>
#include <sys/socket.h>
#include <pistache/async.h> #include <pistache/async.h>
#include <pistache/os.h> #include <pistache/os.h>
...@@ -18,6 +13,18 @@ ...@@ -18,6 +13,18 @@
#include <pistache/reactor.h> #include <pistache/reactor.h>
#include <pistache/view.h> #include <pistache/view.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <atomic>
#include <chrono>
#include <deque>
#include <functional>
#include <memory>
#include <mutex>
#include <string>
#include <unordered_map>
namespace Pistache { namespace Pistache {
namespace Http { namespace Http {
...@@ -30,7 +37,7 @@ struct Connection : public std::enable_shared_from_this<Connection> { ...@@ -30,7 +37,7 @@ struct Connection : public std::enable_shared_from_this<Connection> {
friend class ConnectionPool; friend class ConnectionPool;
typedef std::function<void()> OnDone; using OnDone = std::function<void()>;
Connection() Connection()
: fd(-1) : fd(-1)
...@@ -147,7 +154,7 @@ public: ...@@ -147,7 +154,7 @@ public:
void init(size_t maxConnsPerHost); void init(size_t maxConnsPerHost);
std::shared_ptr<Connection> pickConnection(const std::string& domain); std::shared_ptr<Connection> pickConnection(const std::string& domain);
void releaseConnection(const std::shared_ptr<Connection>& connection); static void releaseConnection(const std::shared_ptr<Connection>& connection);
size_t usedConnections(const std::string& domain) const; size_t usedConnections(const std::string& domain) const;
size_t idleConnections(const std::string& domain) const; size_t idleConnections(const std::string& domain) const;
...@@ -187,8 +194,6 @@ public: ...@@ -187,8 +194,6 @@ public:
, timeouts() , timeouts()
{ } { }
typedef std::function<void()> OnResponseParsed;
void onReady(const Aio::FdSet& fds) override; void onReady(const Aio::FdSet& fds) override;
void registerPoller(Polling::Epoll& poller) override; void registerPoller(Polling::Epoll& poller) override;
...@@ -198,7 +203,7 @@ public: ...@@ -198,7 +203,7 @@ public:
Async::Promise<ssize_t> asyncSendRequest( Async::Promise<ssize_t> asyncSendRequest(
const std::shared_ptr<Connection>& connection, const std::shared_ptr<Connection>& connection,
std::shared_ptr<TimerPool::Entry> timer, std::shared_ptr<TimerPool::Entry> timer,
const Buffer& buffer); std::string buffer);
private: private:
...@@ -230,12 +235,12 @@ private: ...@@ -230,12 +235,12 @@ private:
Async::Resolver resolve, Async::Rejection reject, Async::Resolver resolve, Async::Rejection reject,
std::shared_ptr<Connection> connection, std::shared_ptr<Connection> connection,
std::shared_ptr<TimerPool::Entry> timer, std::shared_ptr<TimerPool::Entry> timer,
const Buffer& buffer) std::string buf)
: resolve(std::move(resolve)) : resolve(std::move(resolve))
, reject(std::move(reject)) , reject(std::move(reject))
, connection(std::move(connection)) , connection(std::move(connection))
, timer(std::move(timer)) , timer(std::move(timer))
, buffer(buffer) , buffer(std::move(buf))
{ {
} }
...@@ -243,8 +248,7 @@ private: ...@@ -243,8 +248,7 @@ private:
Async::Rejection reject; Async::Rejection reject;
std::shared_ptr<Connection> connection; std::shared_ptr<Connection> connection;
std::shared_ptr<TimerPool::Entry> timer; std::shared_ptr<TimerPool::Entry> timer;
Buffer buffer; std::string buffer;
}; };
...@@ -301,7 +305,7 @@ public: ...@@ -301,7 +305,7 @@ public:
Async::Promise<Response> send(); Async::Promise<Response> send();
private: private:
RequestBuilder(Client* const client) explicit RequestBuilder(Client* const client)
: client_(client) : client_(client)
, request_() , request_()
, timeout_(std::chrono::milliseconds(0)) , timeout_(std::chrono::milliseconds(0))
......
...@@ -4,14 +4,17 @@ ...@@ -4,14 +4,17 @@
Implementation of the Http client Implementation of the Http client
*/ */
#include <algorithm> #include <pistache/client.h>
#include <pistache/http.h>
#include <pistache/stream.h>
#include <sys/sendfile.h> #include <sys/sendfile.h>
#include <netdb.h> #include <netdb.h>
#include <pistache/client.h> #include <algorithm>
#include <pistache/http.h> #include <memory>
#include <pistache/stream.h> #include <sstream>
#include <string>
namespace Pistache { namespace Pistache {
...@@ -51,42 +54,29 @@ struct ExceptionPrinter { ...@@ -51,42 +54,29 @@ struct ExceptionPrinter {
}; };
namespace { namespace {
#define OUT(...) \ template<typename H, typename... Args>
do { \ void writeHeader(std::stringstream& streamBuf, Args&& ...args) {
__VA_ARGS__; \
if (!os) return false; \
} while (0)
template<typename H, typename Stream, typename... Args>
typename std::enable_if<Http::Header::IsHeader<H>::value, Stream&>::type
writeHeader(Stream& stream, Args&& ...args) {
using Http::crlf; using Http::crlf;
H header(std::forward<Args>(args)...); H header(std::forward<Args>(args)...);
stream << H::Name << ": "; streamBuf << H::Name << ": ";
header.write(stream); header.write(streamBuf);
stream << crlf;
return stream; streamBuf << crlf;
} }
bool writeHeaders(const Http::Header::Collection& headers, DynamicStreamBuf& buf) { void writeHeaders(std::stringstream& streamBuf, const Http::Header::Collection& headers) {
using Http::crlf; using Http::crlf;
std::ostream os(&buf);
for (const auto& header: headers.list()) { for (const auto& header: headers.list()) {
OUT(os << header->name() << ": "); streamBuf << header->name() << ": ";
OUT(header->write(os)); header->write(streamBuf);
OUT(os << crlf); streamBuf << crlf;
} }
return true;
} }
bool writeCookies(const Http::CookieJar& cookies, DynamicStreamBuf& buf) { void writeCookies(std::stringstream& streamBuf, const Http::CookieJar& cookies) {
using Http::crlf; using Http::crlf;
std::vector<std::pair<std::string, std::string>> cookiesInfo; std::vector<std::pair<std::string, std::string>> cookiesInfo;
...@@ -94,24 +84,19 @@ namespace { ...@@ -94,24 +84,19 @@ namespace {
cookiesInfo.emplace_back(cookie.name, cookie.value); cookiesInfo.emplace_back(cookie.name, cookie.value);
} }
std::ostream os(&buf); streamBuf << "Cookie: ";
OUT(os << "Cookie: ");
for (size_t i = 0; i < cookiesInfo.size(); ++i) { for (size_t i = 0; i < cookiesInfo.size(); ++i) {
if (i > 0) { if (i > 0) {
OUT(os << "; "); streamBuf << "; ";
} }
OUT(os << cookiesInfo[i].first << "=" << cookiesInfo[i].second); streamBuf << cookiesInfo[i].first << "=" << cookiesInfo[i].second;
} }
OUT(os << crlf); streamBuf << crlf;
return true;
} }
bool writeRequest(const Http::Request& request, DynamicStreamBuf &buf) { void writeRequest(std::stringstream& streamBuf, const Http::Request& request) {
using Http::crlf; using Http::crlf;
std::ostream os(&buf);
auto res = request.resource(); auto res = request.resource();
auto s = splitUrl(res); auto s = splitUrl(res);
auto body = request.body(); auto body = request.body();
...@@ -121,32 +106,26 @@ namespace { ...@@ -121,32 +106,26 @@ namespace {
auto pathStr = path.toString(); auto pathStr = path.toString();
OUT(os << request.method() << " "); streamBuf << request.method() << " ";
if (pathStr[0] != '/') if (pathStr[0] != '/')
OUT(os << '/'); streamBuf << '/';
OUT(os << pathStr); streamBuf << pathStr;
OUT(os << " HTTP/1.1" << crlf); streamBuf << " HTTP/1.1" << crlf;
if (!writeCookies(request.cookies(), buf)) return false; writeCookies(streamBuf, request.cookies());
if (!writeHeaders(request.headers(), buf)) return false; writeHeaders(streamBuf, request.headers());
if (!writeHeader<Http::Header::UserAgent>(os, UA)) return false; writeHeader<Http::Header::UserAgent>(streamBuf, UA);
if (!writeHeader<Http::Header::Host>(os, host.toString())) return false; writeHeader<Http::Header::Host>(streamBuf, host.toString());
if (!body.empty()) { if (!body.empty()) {
if (!writeHeader<Http::Header::ContentLength>(os, body.size())) return false; writeHeader<Http::Header::ContentLength>(streamBuf, body.size());
} }
OUT(os << crlf); streamBuf << crlf;
if (!body.empty()) { if (!body.empty()) {
OUT(os << body); streamBuf << body;
} }
return true;
} }
#undef OUT
} }
void void
...@@ -214,17 +193,15 @@ Async::Promise<ssize_t> ...@@ -214,17 +193,15 @@ Async::Promise<ssize_t>
Transport::asyncSendRequest( Transport::asyncSendRequest(
const std::shared_ptr<Connection>& connection, const std::shared_ptr<Connection>& connection,
std::shared_ptr<TimerPool::Entry> timer, std::shared_ptr<TimerPool::Entry> timer,
const Buffer& buffer) { std::string buffer) {
return Async::Promise<ssize_t>([&](Async::Resolver& resolve, Async::Rejection& reject) { return Async::Promise<ssize_t>([&](Async::Resolver& resolve, Async::Rejection& reject) {
auto ctx = context(); auto ctx = context();
RequestEntry req(std::move(resolve), std::move(reject), connection, std::move(timer), std::move(buffer));
if (std::this_thread::get_id() != ctx.thread()) { if (std::this_thread::get_id() != ctx.thread()) {
RequestEntry req(std::move(resolve), std::move(reject), connection, std::move(timer), buffer.detach());
auto *e = requestsQueue.allocEntry(std::move(req)); auto *e = requestsQueue.allocEntry(std::move(req));
requestsQueue.push(e); requestsQueue.push(e);
} else { } else {
RequestEntry req(std::move(resolve), std::move(reject), connection, std::move(timer), buffer);
asyncSendRequestImpl(req); asyncSendRequestImpl(req);
} }
}); });
...@@ -235,12 +212,7 @@ void ...@@ -235,12 +212,7 @@ void
Transport::asyncSendRequestImpl( Transport::asyncSendRequestImpl(
const RequestEntry& req, WriteStatus status) const RequestEntry& req, WriteStatus status)
{ {
auto buffer = req.buffer; const auto& buffer = req.buffer;
auto cleanUp = [&]() {
if (buffer.isOwned) delete[] buffer.data;
};
auto conn = req.connection; auto conn = req.connection;
auto fd = conn->fd; auto fd = conn->fd;
...@@ -248,8 +220,8 @@ Transport::asyncSendRequestImpl( ...@@ -248,8 +220,8 @@ Transport::asyncSendRequestImpl(
ssize_t totalWritten = 0; ssize_t totalWritten = 0;
for (;;) { for (;;) {
ssize_t bytesWritten = 0; ssize_t bytesWritten = 0;
ssize_t len = buffer.len - totalWritten; ssize_t len = buffer.size() - totalWritten;
auto ptr = buffer.data + totalWritten; auto ptr = buffer.data() + totalWritten;
bytesWritten = ::send(fd, ptr, len, 0); bytesWritten = ::send(fd, ptr, len, 0);
if (bytesWritten < 0) { if (bytesWritten < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) { if (errno == EAGAIN || errno == EWOULDBLOCK) {
...@@ -259,7 +231,6 @@ Transport::asyncSendRequestImpl( ...@@ -259,7 +231,6 @@ Transport::asyncSendRequestImpl(
reactor()->modifyFd(key(), fd, NotifyOn::Write, Polling::Mode::Edge); reactor()->modifyFd(key(), fd, NotifyOn::Write, Polling::Mode::Edge);
} }
else { else {
cleanUp();
req.reject(Error::system("Could not send request")); req.reject(Error::system("Could not send request"));
} }
break; break;
...@@ -267,7 +238,6 @@ Transport::asyncSendRequestImpl( ...@@ -267,7 +238,6 @@ Transport::asyncSendRequestImpl(
else { else {
totalWritten += bytesWritten; totalWritten += bytesWritten;
if (totalWritten == len) { if (totalWritten == len) {
cleanUp();
if (req.timer) { if (req.timer) {
timeouts.insert( timeouts.insert(
std::make_pair(req.timer->fd, conn)); std::make_pair(req.timer->fd, conn));
...@@ -562,12 +532,12 @@ Connection::performImpl( ...@@ -562,12 +532,12 @@ Connection::performImpl(
Async::Rejection reject, Async::Rejection reject,
Connection::OnDone onDone) { Connection::OnDone onDone) {
DynamicStreamBuf buf(128); std::stringstream streamBuf;
writeRequest(streamBuf, request);
if (!writeRequest(request, buf)) if (!streamBuf)
reject(std::runtime_error("Could not write request")); reject(std::runtime_error("Could not write request"));
std::string buffer = streamBuf.str();
auto buffer = buf.buffer();
std::shared_ptr<TimerPool::Entry> timer(nullptr); std::shared_ptr<TimerPool::Entry> timer(nullptr);
if (timeout.count() > 0) { if (timeout.count() > 0) {
timer = timerPool_.pickTimer(); timer = timerPool_.pickTimer();
...@@ -582,7 +552,7 @@ Connection::performImpl( ...@@ -582,7 +552,7 @@ Connection::performImpl(
auto rejectMover = make_copy_mover(std::move(reject)); auto rejectMover = make_copy_mover(std::move(reject));
transport_->asyncSendRequest(shared_from_this(), timer, buffer).then( transport_->asyncSendRequest(shared_from_this(), timer, std::move(buffer)).then(
[=](ssize_t) mutable { [=](ssize_t) mutable {
inflightRequests.push_back(RequestEntry(std::move(resolveMover), std::move(rejectMover), std::move(timer), std::move(onDone))); inflightRequests.push_back(RequestEntry(std::move(resolveMover), std::move(rejectMover), std::move(timer), std::move(onDone)));
}, },
......
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