Commit ce01689c authored by octal's avatar octal

Experimental http client

parent 3a154192
......@@ -18,154 +18,121 @@ namespace Net {
namespace Http {
class Connection;
class Transport : public Io::Handler {
public:
PROTOTYPE_OF(Io::Handler, Transport)
void onReady(const Io::FdSet& fds);
void registerPoller(Polling::Epoll& poller);
Async::Promise<void>
asyncConnect(Fd fd, const struct sockaddr* address, socklen_t addr_len);
template<typename Buf>
Async::Promise<ssize_t> asyncWrite(Fd fd, const Buf& buffer, int flags = 0) {
// If the I/O operation has been initiated from an other thread, we queue it and we'll process
// it in our own thread so that we make sure that every I/O operation happens in the right thread
if (std::this_thread::get_id() != io()->thread()) {
return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
BufferHolder holder(buffer);
auto detached = holder.detach();
OnHoldWrite write(std::move(resolve), std::move(reject), detached, flags);
write.peerFd = fd;
auto *e = writesQueue.allocEntry(std::move(write));
writesQueue.push(e);
});
}
return Async::Promise<ssize_t>([&](Async::Resolver& resolve, Async::Rejection& reject) {
auto it = toWrite.find(fd);
if (it != std::end(toWrite)) {
reject(Net::Error("Multiple writes on the same fd"));
return;
}
class ConnectionPool;
class Transport;
asyncWriteImpl(fd, flags, BufferHolder(buffer), std::move(resolve), std::move(reject));
struct Connection {
});
}
friend class ConnectionPool;
void addInFlight(Fd fd, Async::Resolver&& resolve, Async::Rejection&& reject) {
InflightRequest req(std::move(resolve), std::move(reject));
inflightRequests.insert(
std::make_pair(fd, std::move(req)));
}
typedef std::function<void()> OnDone;
private:
struct RequestData {
struct Event {
enum class Type {
Connection
};
RequestData(
Async::Resolver resolve, Async::Rejection reject,
const Http::Request& request,
std::string host,
OnDone onDone)
: resolve(std::move(resolve))
, reject(std::move(reject))
, request(request)
, host(std::move(host))
, onDone(std::move(onDone))
{ }
Async::Resolver resolve;
Async::Rejection reject;
virtual Type type() const = 0;
std::string host;
Http::Request request;
OnDone onDone;
};
struct ConnectionEvent : public Event {
ConnectionEvent(const Connection* connection);
Type type() const { return Type::Connection; }
const Connection* connection_;
enum State : uint32_t {
Idle,
Used
};
enum WriteStatus {
FirstTry,
Retry
enum ConnectionState {
NotConnected,
Connecting,
Connected
};
struct BufferHolder {
enum Type { Raw, File };
Connection()
: fd(-1)
, connectionState_(NotConnected)
{
state_.store(static_cast<uint32_t>(State::Idle));
}
explicit BufferHolder(const Buffer& buffer)
: type(Raw)
, u(buffer)
{
size_ = buffer.len;
}
void connect(Net::Address addr);
void close();
bool isConnected() const;
bool hasTransport() const;
void associateTransport(const std::shared_ptr<Transport>& transport);
explicit BufferHolder(const FileBuffer& buffer)
: type(File)
, u(buffer.fd())
{
size_ = buffer.size();
}
Async::Promise<Response> perform(
const Http::Request& request,
std::string host,
OnDone onDone);
bool isFile() const { return type == File; }
bool isRaw() const { return type == Raw; }
size_t size() const { return size_; }
void performImpl(
const Http::Request& request,
std::string host,
Async::Resolver resolve,
Async::Rejection reject,
OnDone onDone);
Fd fd() const {
if (!isFile())
throw std::runtime_error("Tried to retrieve fd of a non-filebuffer");
Fd fd;
return u.fd;
private:
void processRequestQueue();
}
std::atomic<uint32_t> state_;
ConnectionState connectionState_;
std::shared_ptr<Transport> transport_;
Queue<RequestData> requestsQueue;
};
Buffer raw() const {
if (!isRaw())
throw std::runtime_error("Tried to retrieve raw data of a non-buffer");
struct ConnectionPool {
void init(size_t max = 1);
return u.raw;
}
std::shared_ptr<Connection> pickConnection();
void releaseConnection(const std::shared_ptr<Connection>& connection);
BufferHolder detach(size_t offset = 0) const {
if (!isRaw())
return BufferHolder(u.fd, size_);
size_t availableCount() const;
if (u.raw.isOwned)
return BufferHolder(u.raw);
private:
std::atomic<uint32_t> usedCount;
std::vector<std::shared_ptr<Connection>> connections;
};
auto detached = u.raw.detach(offset);
return BufferHolder(detached);
}
class Transport : public Io::Handler {
public:
private:
BufferHolder(Fd fd, size_t size)
: u(fd)
, size_(size)
, type(File)
{ }
PROTOTYPE_OF(Io::Handler, Transport)
union U {
Buffer raw;
Fd fd;
typedef std::function<void()> OnResponseParsed;
U(Buffer buffer) : raw(buffer) { }
U(Fd fd) : fd(fd) { }
} u;
size_t size_;
Type type;
};
void onReady(const Io::FdSet& fds);
void registerPoller(Polling::Epoll& poller);
struct OnHoldWrite {
OnHoldWrite(Async::Resolver resolve, Async::Rejection reject,
BufferHolder buffer, int flags = 0)
: resolve(std::move(resolve))
, reject(std::move(reject))
, buffer(std::move(buffer))
, flags(flags)
, peerFd(-1)
{ }
Async::Promise<void>
asyncConnect(Fd fd, const struct sockaddr* address, socklen_t addr_len);
Async::Resolver resolve;
Async::Rejection reject;
BufferHolder buffer;
int flags;
Fd peerFd;
void asyncSendRequest(
Fd fd,
const Buffer& buffer,
Async::Resolver resolve,
Async::Rejection reject,
OnResponseParsed onParsed);
private:
enum WriteStatus {
FirstTry,
Retry
};
struct PendingConnection {
......@@ -188,99 +155,57 @@ private:
struct InflightRequest {
InflightRequest(
Async::Resolver resolve, Async::Rejection reject)
: resolve(std::move(resolve))
Async::Resolver resolve, Async::Rejection reject,
Fd fd,
const Buffer& buffer,
OnResponseParsed onParsed = nullptr)
: resolve_(std::move(resolve))
, reject(std::move(reject))
, fd(fd)
, buffer(buffer)
, onParsed(onParsed)
{
parser.reset(new Private::Parser<Http::Response>());
}
Async::Resolver resolve;
Async::Rejection reject;
std::shared_ptr<Private::Parser<Http::Response>> parser;
};
PollableQueue<OnHoldWrite> writesQueue;
PollableQueue<PendingConnection> connectionsQueue;
void feed(const char* buffer, size_t totalBytes) {
if (!parser)
parser.reset(new Private::Parser<Http::Response>());
std::unordered_map<Fd, OnHoldWrite> toWrite;
std::unordered_map<Fd, PendingConnection> pendingConnections;
std::unordered_map<Fd, InflightRequest> inflightRequests;
void asyncWriteImpl(Fd fd, OnHoldWrite& entry, WriteStatus status = FirstTry);
void asyncWriteImpl(
Fd fd, int flags, const BufferHolder& buffer,
Async::Resolver resolve, Async::Rejection reject,
WriteStatus status = FirstTry);
void handleWriteQueue();
void handleConnectionQueue();
void handleIncoming(Fd fd);
void handleResponsePacket(Fd fd, const char* buffer, size_t totalBytes);
};
parser->feed(buffer, totalBytes);
}
class ConnectionPool;
void resolve(Http::Response response) {
if (onParsed)
onParsed();
resolve_(std::move(response));
}
struct Connection {
Async::Resolver resolve_;
Async::Rejection reject;
Fd fd;
Buffer buffer;
friend class ConnectionPool;
OnResponseParsed onParsed;
enum State : uint32_t {
Idle,
Used,
Connecting,
Connected
std::shared_ptr<Private::Parser<Http::Response>> parser;
};
Connection()
: fd(-1)
{
}
void connect(Net::Address addr);
bool isConnected() const;
bool hasTransport() const;
void associateTransport(const std::shared_ptr<Transport>& transport);
Async::Promise<Response> perform(const Http::Request& request);
Fd fd;
PollableQueue<InflightRequest> requestsQueue;
PollableQueue<PendingConnection> connectionsQueue;
private:
void performImpl(
const Http::Request& request,
Async::Resolver resolve,
Async::Rejection reject);
void processRequestQueue();
std::unordered_map<Fd, PendingConnection> pendingConnections;
std::unordered_map<Fd, std::deque<InflightRequest>> inflightRequests;
struct RequestData {
RequestData(Async::Resolver resolve, Async::Rejection reject, const Http::Request& request)
: resolve(std::move(resolve))
, reject(std::move(reject))
, request(request)
{ }
Async::Resolver resolve;
Async::Rejection reject;
void asyncSendRequestImpl(InflightRequest& req, WriteStatus status = FirstTry);
Http::Request request;
};
void handleRequestsQueue();
void handleConnectionQueue();
void handleIncoming(Fd fd);
void handleResponsePacket(Fd fd, const char* buffer, size_t totalBytes);
std::atomic<uint32_t> state_;
std::shared_ptr<Transport> transport_;
Queue<RequestData> requestsQueue;
};
struct ConnectionPool {
void init(size_t max = 1);
std::shared_ptr<Connection> pickConnection();
void returnConnection(const std::shared_ptr<Connection>& connection);
private:
std::vector<std::shared_ptr<Connection>> connections;
};
namespace Default {
constexpr int Threads = 1;
......@@ -321,14 +246,21 @@ public:
const Http::Request& request,
std::chrono::milliseconds timeout = std::chrono::milliseconds(0));
void shutdown();
private:
Io::ServiceGroup io_;
std::string url_;
std::string host_;
Net::Address addr_;
std::shared_ptr<Transport> transport_;
ConnectionPool pool;
std::shared_ptr<Transport> transport_;
std::atomic<uint64_t> ioIndex;
Queue<Connection::RequestData> requestsQueue;
void processRequestQueue();
};
......
......@@ -15,6 +15,8 @@
#include <sys/socket.h>
#include <netdb.h>
#define unsafe
#define TRY(...) \
do { \
auto ret = __VA_ARGS__; \
......
......@@ -97,6 +97,7 @@ public:
friend class Private::Parser<Http::Request>;
friend class RequestBuilder;
friend class Client;
Request(const Request& other) = default;
Request& operator=(const Request& other) = default;
......
......@@ -314,7 +314,8 @@ public:
Host()
{ }
explicit Host(const std::string& host, Net::Port port = 80)
explicit Host(const std::string& host);
explicit Host(const std::string& host, Net::Port port)
: host_(host)
, port_(port)
{ }
......@@ -364,7 +365,11 @@ public:
void parse(const std::string& data);
void write(std::ostream& os) const;
std::string ua() const { return ua_; }
void setAgent(std::string ua) {
ua_ = std::move(ua);
}
std::string agent() const { return ua_; }
private:
std::string ua_;
......
......@@ -61,6 +61,14 @@ public:
return add(std::make_shared<H>(std::forward<Args>(args)...));
}
template<typename H>
typename std::enable_if<
IsHeader<H>::value, bool
>::type
remove() {
return remove(H::Name);
}
std::shared_ptr<const Header> get(const std::string& name) const;
std::shared_ptr<Header> get(const std::string& name);
Raw getRaw(const std::string& name) const;
......@@ -80,6 +88,8 @@ public:
std::vector<std::shared_ptr<Header>> list() const;
bool remove(const std::string& name);
void clear();
private:
......
......@@ -339,22 +339,32 @@ int main(int argc, char *argv[]) {
#if 1
int main() {
Net::Http::Client client("http://www.foaas.com");
Net::Http::Client client("http://supnetwork.org:9080");
auto opts = Net::Http::Client::options()
.threads(1)
.maxConnections(1);
.maxConnections(20);
using namespace Net::Http;
constexpr size_t Requests = 1000;
std::atomic<int> responsesReceived(0);
client.init(opts);
client.get(client
.request("/off/octal/nask")
.header<Header::ContentType>(MIME(Text, Plain))
.cookie(Cookie("FOO", "bar")))
.then([](const Http::Response& response) {
std::cout << "code = " << response.code() << std::endl;
std::cout << "body = " << response.body() << std::endl;
}, Async::NoExcept);
std::this_thread::sleep_for(std::chrono::seconds(1));
for (int i = 0; i < Requests; ++i) {
client.get(client
.request("/ping")
.cookie(Cookie("FOO", "bar")))
.then([&](const Http::Response& response) {
responsesReceived.fetch_add(1);
//std::cout << "code = " << response.code() << std::endl;
// std::cout << "body = " << response.body() << std::endl;
}, Async::NoExcept);
}
std::cout << "Sent " << Requests << " requests" << std::endl;
for (;;) {
std::this_thread::sleep_for(std::chrono::seconds(1));
std::cout << "Received " << responsesReceived.load() << " responses" << std::endl;
}
client.shutdown();
}
#endif
......@@ -59,10 +59,10 @@ namespace {
}
bool writeRequest(const Http::Request& request, DynamicStreamBuf &buf) {
bool writeRequest(const Http::Request& request, std::string host, DynamicStreamBuf &buf) {
std::ostream os(&buf);
OUT(os << "GET ");
OUT(os << request.method() << " ");
OUT(os << request.resource());
OUT(os << " HTTP/1.1" << crlf);
......@@ -70,7 +70,7 @@ namespace {
if (!writeHeaders(request.headers(), buf)) return false;
if (!writeHeader<Header::UserAgent>(os, "restpp/0.1")) return false;
if (!writeHeader<Header::Host>(os, "foaas.com")) return false;
if (!writeHeader<Header::Host>(os, std::move(host))) return false;
OUT(os << crlf);
return true;
......@@ -81,18 +81,14 @@ namespace {
#undef OUT
}
Transport::ConnectionEvent::ConnectionEvent(const Connection* connection)
: connection_(connection)
{ }
void
Transport::onReady(const Io::FdSet& fds) {
for (const auto& entry: fds) {
if (entry.getTag() == connectionsQueue.tag()) {
handleConnectionQueue();
}
else if (entry.getTag() == writesQueue.tag()) {
handleWriteQueue();
else if (entry.getTag() == requestsQueue.tag()) {
handleRequestsQueue();
}
else if (entry.isReadable()) {
......@@ -112,14 +108,19 @@ Transport::onReady(const Io::FdSet& fds) {
continue;
}
#if 0
auto writeIt = toWrite.find(fd);
if (writeIt != std::end(toWrite)) {
/* @Bug: should not need modifyFd, investigate why I can't use
* registerFd
*/
io()->modifyFd(fd, NotifyOn::Read, Polling::Mode::Edge);
auto& write = writeIt->second;
asyncWriteImpl(fd, write, Retry);
continue;
}
#endif
throw std::runtime_error("Unknown fd");
}
......@@ -128,7 +129,7 @@ Transport::onReady(const Io::FdSet& fds) {
void
Transport::registerPoller(Polling::Epoll& poller) {
writesQueue.bind(poller);
requestsQueue.bind(poller);
connectionsQueue.bind(poller);
}
......@@ -143,50 +144,54 @@ Transport::asyncConnect(Fd fd, const struct sockaddr* address, socklen_t addr_le
}
void
Transport::asyncWriteImpl(Fd fd, Transport::OnHoldWrite& entry, WriteStatus status) {
asyncWriteImpl(fd, entry.flags, entry.buffer, std::move(entry.resolve), std::move(entry.reject), status);
Transport::asyncSendRequest(
Fd fd,
const Buffer& buffer,
Async::Resolver resolve,
Async::Rejection reject,
OnResponseParsed onParsed) {
if (std::this_thread::get_id() != io()->thread()) {
InflightRequest req(std::move(resolve), std::move(reject), fd, buffer.detach(), std::move(onParsed));
auto detached = buffer.detach();
auto *e = requestsQueue.allocEntry(std::move(req));
requestsQueue.push(e);
} else {
InflightRequest req(std::move(resolve), std::move(reject), fd, buffer, std::move(onParsed));
asyncSendRequestImpl(req);
}
}
void
Transport::asyncWriteImpl(
Fd fd, int flags, const BufferHolder& buffer,
Async::Resolver resolve, Async::Rejection reject, WriteStatus status)
Transport::asyncSendRequestImpl(
InflightRequest& req, WriteStatus status)
{
auto cleanUp = [&]() {
if (buffer.isRaw()) {
auto raw = buffer.raw();
if (raw.isOwned) delete[] raw.data;
}
auto buffer = req.buffer;
if (status == Retry)
toWrite.erase(fd);
auto cleanUp = [&]() {
if (buffer.isOwned) delete[] buffer.data;
};
auto fd = req.fd;
ssize_t totalWritten = 0;
for (;;) {
ssize_t bytesWritten = 0;
auto len = buffer.size() - totalWritten;
if (buffer.isRaw()) {
auto raw = buffer.raw();
auto ptr = raw.data + totalWritten;
bytesWritten = ::send(fd, ptr, len, flags);
} else {
auto file = buffer.fd();
off_t offset = totalWritten;
bytesWritten = ::sendfile(fd, file, &offset, len);
}
auto len = buffer.len - totalWritten;
auto ptr = buffer.data + totalWritten;
bytesWritten = ::send(fd, ptr, len, 0);
if (bytesWritten < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
if (status == FirstTry) {
toWrite.insert(
std::make_pair(fd,
OnHoldWrite(std::move(resolve), std::move(reject), buffer.detach(totalWritten), flags)));
throw std::runtime_error("Unimplemented, fix me!");
}
io()->modifyFd(fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
}
else {
cleanUp();
reject(Net::Error::system("Could not write data"));
req.reject(Net::Error::system("Could not send request"));
}
break;
}
......@@ -194,7 +199,8 @@ Transport::asyncWriteImpl(
totalWritten += bytesWritten;
if (totalWritten == len) {
cleanUp();
resolve(totalWritten);
auto& queue = inflightRequests[fd];
queue.push_back(std::move(req));
break;
}
}
......@@ -202,14 +208,14 @@ Transport::asyncWriteImpl(
}
void
Transport::handleWriteQueue() {
Transport::handleRequestsQueue() {
// Let's drain the queue
for (;;) {
std::unique_ptr<PollableQueue<OnHoldWrite>::Entry> entry(writesQueue.pop());
std::unique_ptr<PollableQueue<InflightRequest>::Entry> entry(requestsQueue.pop());
if (!entry) break;
auto &write = entry->data();
asyncWriteImpl(write.peerFd, write);
auto &req = entry->data();
asyncSendRequestImpl(req);
}
}
......@@ -279,10 +285,12 @@ Transport::handleResponsePacket(Fd fd, const char* buffer, size_t totalBytes) {
if (it == std::end(inflightRequests))
throw std::runtime_error("Received response for a non-inflight request");
auto &req = it->second;
req.parser->feed(buffer, totalBytes);
auto &queue = it->second;
auto &req = queue.front();
req.feed(buffer, totalBytes);
if (req.parser->parse() == Private::State::Done) {
req.resolve(std::move(req.parser->response));
queue.pop_front();
}
}
......@@ -316,13 +324,13 @@ Connection::connect(Net::Address addr)
make_non_blocking(sfd);
state_.store(static_cast<uint32_t>(State::Connecting));
connectionState_ = Connecting;
transport_->asyncConnect(sfd, addr->ai_addr, addr->ai_addrlen)
.then([=]() {
connectionState_ = Connected;
fd = sfd;
transport_->io()->modifyFd(fd, NotifyOn::Read);
state_.store(static_cast<uint32_t>(State::Connected));
processRequestQueue();
}, Async::Throw);
break;
......@@ -335,7 +343,12 @@ Connection::connect(Net::Address addr)
bool
Connection::isConnected() const {
return static_cast<State>(state_.load()) == State::Connected;
return connectionState_ == Connected;
}
void
Connection::close() {
connectionState_ = NotConnected;
}
void
......@@ -352,49 +365,66 @@ Connection::hasTransport() const {
}
Async::Promise<Response>
Connection::perform(const Http::Request& request) {
Connection::perform(
const Http::Request& request,
std::string host,
OnDone onDone) {
return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) {
if (!isConnected()) {
auto* entry = requestsQueue.allocEntry(RequestData(std::move(resolve), std::move(reject), request));
auto* entry = requestsQueue.allocEntry(
RequestData(std::move(resolve), std::move(reject), request, std::move(host), std::move(onDone)));
requestsQueue.push(entry);
} else {
performImpl(request, std::move(resolve), std::move(reject));
performImpl(request, std::move(host), std::move(resolve), std::move(reject), std::move(onDone));
}
});
}
#if 0
struct OnRequestWriteCompleted {
OnRequestWriteCompleted(const std::shared_ptr<Transport>& transport, Fd fd, Async::Resolver resolve, Async::Rejection reject)
OnRequestWriteCompleted(
const std::shared_ptr<Transport>& transport,
Fd fd,
Async::Resolver resolve, Async::Rejection reject,
Connection::OnDone onDone)
: transport(transport)
, fd(fd)
, resolve(std::move(resolve))
, reject(std::move(reject))
, onDone(std::move(onDone))
{ }
void operator()(size_t bytes) {
transport->addInFlight(fd, std::move(resolve), std::move(reject));
onDone();
}
std::shared_ptr<Transport> transport;
Fd fd;
Async::Resolver resolve;
Async::Rejection reject;
Connection::OnDone onDone;
};
#endif
void
Connection::performImpl(
const Http::Request& request,
std::string host,
Async::Resolver resolve,
Async::Rejection reject) {
Async::Rejection reject,
OnDone onDone) {
DynamicStreamBuf buf(128);
if (!writeRequest(request, buf))
if (!writeRequest(request, std::move(host), buf))
reject(std::runtime_error("Could not write request"));
auto buffer = buf.buffer();
OnRequestWriteCompleted onCompleted(transport_, fd, std::move(resolve), std::move(reject));
transport_->asyncWrite(fd, buffer).then(std::move(onCompleted) , Async::Throw);
#if 0
OnRequestWriteCompleted onCompleted(transport_, fd, std::move(resolve), std::move(reject), std::move(onDone));
#endif
transport_->asyncSendRequest(fd, buffer, std::move(resolve), std::move(reject), std::move(onDone));
}
void
......@@ -404,7 +434,8 @@ Connection::processRequestQueue() {
if (!entry) break;
auto &req = entry->data();
performImpl(req.request, std::move(req.resolve), std::move(req.reject));
performImpl(req.request,
std::move(req.host), std::move(req.resolve), std::move(req.reject), std::move(req.onDone));
}
}
......@@ -415,20 +446,19 @@ ConnectionPool::init(size_t max)
for (size_t i = 0; i < max; ++i) {
connections.push_back(std::make_shared<Connection>());
}
usedCount.store(0);
}
std::shared_ptr<Connection>
ConnectionPool::pickConnection() {
for (auto& conn: connections) {
auto& state = conn->state_;
if (static_cast<Connection::State>(state.load()) == Connection::State::Idle) {
auto curState = static_cast<uint32_t>(Connection::State::Idle);
auto newState = Connection::State::Used;
if (state.compare_exchange_strong(
curState,
static_cast<uint32_t>(Connection::State::Used))) {
return conn;
}
auto curState = static_cast<uint32_t>(Connection::State::Idle);
auto newState = static_cast<uint32_t>(Connection::State::Used);
if (state.compare_exchange_strong(curState, newState)) {
usedCount.fetch_add(1);
return conn;
}
}
......@@ -436,8 +466,14 @@ ConnectionPool::pickConnection() {
}
void
ConnectionPool::returnConnection(const std::shared_ptr<Connection>& connection) {
ConnectionPool::releaseConnection(const std::shared_ptr<Connection>& connection) {
connection->state_.store(static_cast<uint32_t>(Connection::State::Idle));
usedCount.fetch_add(-1);
}
size_t
ConnectionPool::availableCount() const {
return connections.size() - usedCount.load();
}
Client::Options&
......@@ -460,7 +496,13 @@ Client::Options::keepAlive(bool val) {
Client::Client(const std::string& base)
: url_(base)
, ioIndex(0)
{ }
{
host_ = url_;
constexpr const char *Http = "http://";
constexpr size_t Size = sizeof("http://") - 1;
if (!host_.compare(0, Size, Http))
host_.erase(0, Size);
}
Client::Options
Client::options() {
......@@ -497,6 +539,11 @@ Client::init(const Client::Options& options) {
pool.init(options.maxConnections_);
}
void
Client::shutdown() {
io_.shutdown();
}
RequestBuilder
Client::request(std::string val) {
RequestBuilder builder;
......@@ -507,9 +554,20 @@ Client::request(std::string val) {
Async::Promise<Http::Response>
Client::get(const Http::Request& request, std::chrono::milliseconds timeout)
{
unsafe {
auto &req = const_cast<Http::Request &>(request);
req.method_ = Http::Method::Get;
req.headers_.remove<Header::UserAgent>();
}
auto conn = pool.pickConnection();
if (conn == nullptr) {
std::cout << "No connection available yet, bailing-out" << std::endl;
return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) {
auto entry = requestsQueue.allocEntry(
Connection::RequestData(std::move(resolve), std::move(reject), request, host_, nullptr));
requestsQueue.push(entry);
});
}
else {
......@@ -520,14 +578,42 @@ Client::get(const Http::Request& request, std::chrono::milliseconds timeout)
auto transport = std::static_pointer_cast<Transport>(service->handler());
conn->associateTransport(transport);
conn->connect(addr_);
}
return conn->perform(request);
if (!conn->isConnected())
conn->connect(addr_);
return conn->perform(request, host_, [=]() {
pool.releaseConnection(conn);
processRequestQueue();
});
}
}
void
Client::processRequestQueue() {
for (;;) {
auto conn = pool.pickConnection();
if (!conn) break;
std::unique_ptr<Queue<Connection::RequestData>::Entry> entry(requestsQueue.pop());
if (!entry) {
pool.releaseConnection(conn);
break;
}
auto& req = entry->data();
conn->performImpl(
req.request, std::move(req.host),
std::move(req.resolve), std::move(req.reject),
[=]() {
pool.releaseConnection(conn);
processRequestQueue();
});
}
}
} // namespace Http
} // namespace Net
......@@ -297,6 +297,10 @@ Expect::write(std::ostream& os) const {
}
}
Host::Host(const std::string& data) {
parse(data);
}
void
Host::parse(const std::string& data) {
auto pos = data.find(':');
......
......@@ -157,6 +157,20 @@ Collection::list() const {
return ret;
}
bool
Collection::remove(const std::string& name) {
auto tit = headers.find(name);
if (tit == std::end(headers)) {
auto rit = rawHeaders.find(name);
if (rit == std::end(rawHeaders)) return false;
rawHeaders.erase(rit);
return true;
}
headers.erase(tit);
return true;
}
void
Collection::clear() {
headers.clear();
......
......@@ -225,7 +225,7 @@ TEST(headers_test, user_agent) {
Header::UserAgent ua;
ua.parse("CERN-LineMode/2.15 libwww/2.17b3");
ASSERT_EQ(ua.ua(), "CERN-LineMode/2.15 libwww/2.17b3");
ASSERT_EQ(ua.agent(), "CERN-LineMode/2.15 libwww/2.17b3");
}
TEST(headers_test, content_encoding) {
......
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