Commit cbcc982b authored by octal's avatar octal

Started implementation of the http client

parent a425e933
...@@ -11,6 +11,7 @@ ...@@ -11,6 +11,7 @@
#include <functional> #include <functional>
#include <memory> #include <memory>
#include <atomic> #include <atomic>
#include <vector>
#include "optional.h" #include "optional.h"
#include "typeid.h" #include "typeid.h"
......
/*
Mathieu Stefani, 29 janvier 2016
The Http client
*/
#pragma once
#include "async.h"
#include "os.h"
#include "http.h"
#include "io.h"
#include <atomic>
#include <sys/types.h>
#include <sys/socket.h>
#include <deque>
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;
}
std::cout << "I am in the same thread, wouhou" << std::endl;
asyncWriteImpl(fd, flags, BufferHolder(buffer), std::move(resolve), std::move(reject));
});
}
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)));
}
private:
struct Event {
enum class Type {
Connection
};
virtual Type type() const = 0;
};
struct ConnectionEvent : public Event {
ConnectionEvent(const Connection* connection);
Type type() const { return Type::Connection; }
const Connection* connection_;
};
enum WriteStatus {
FirstTry,
Retry
};
struct BufferHolder {
enum Type { Raw, File };
explicit BufferHolder(const Buffer& buffer)
: type(Raw)
, u(buffer)
{
size_ = buffer.len;
}
explicit BufferHolder(const FileBuffer& buffer)
: type(File)
, u(buffer.fd())
{
size_ = buffer.size();
}
bool isFile() const { return type == File; }
bool isRaw() const { return type == Raw; }
size_t size() const { return size_; }
Fd fd() const {
if (!isFile())
throw std::runtime_error("Tried to retrieve fd of a non-filebuffer");
return u.fd;
}
Buffer raw() const {
if (!isRaw())
throw std::runtime_error("Tried to retrieve raw data of a non-buffer");
return u.raw;
}
BufferHolder detach(size_t offset = 0) const {
if (!isRaw())
return BufferHolder(u.fd, size_);
if (u.raw.isOwned)
return BufferHolder(u.raw);
auto detached = u.raw.detach(offset);
return BufferHolder(detached);
}
private:
BufferHolder(Fd fd, size_t size)
: u(fd)
, size_(size)
, type(File)
{ }
union U {
Buffer raw;
Fd fd;
U(Buffer buffer) : raw(buffer) { }
U(Fd fd) : fd(fd) { }
} u;
size_t size_;
Type type;
};
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::Resolver resolve;
Async::Rejection reject;
BufferHolder buffer;
int flags;
Fd peerFd;
};
struct PendingConnection {
PendingConnection(
Async::Resolver resolve, Async::Rejection reject,
Fd fd, const struct sockaddr* addr, socklen_t addr_len)
: resolve(std::move(resolve))
, reject(std::move(reject))
, fd(fd)
, addr(addr)
, addr_len(addr_len)
{ }
Async::Resolver resolve;
Async::Rejection reject;
Fd fd;
const struct sockaddr* addr;
socklen_t addr_len;
};
struct InflightRequest {
InflightRequest(
Async::Resolver resolve, Async::Rejection reject)
: resolve(std::move(resolve))
, reject(std::move(reject))
{
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;
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);
};
class ConnectionPool;
struct Connection {
friend class ConnectionPool;
enum State : uint32_t {
Idle,
Used,
Connecting,
Connected
};
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;
private:
void performImpl(
const Http::Request& request,
Async::Resolver resolve,
Async::Rejection reject);
void processRequestQueue();
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;
Http::Request request;
};
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;
constexpr int MaxConnections = 8;
constexpr bool KeepAlive = true;
}
class Client {
public:
struct Options {
friend class Client;
Options()
: threads_(Default::Threads)
, maxConnections_(Default::MaxConnections)
, keepAlive_(Default::KeepAlive)
{ }
Options& threads(int val);
Options& maxConnections(int val);
Options& keepAlive(bool val);
private:
int threads_;
int maxConnections_;
bool keepAlive_;
};
Client(const std::string &base);
static Options options();
void init(const Options& options);
RequestBuilder newRequest(std::string resource);
Async::Promise<Response> get(
const Http::Request& request,
std::chrono::milliseconds timeout = std::chrono::milliseconds(0));
private:
Io::ServiceGroup io_;
std::string url_;
Net::Address addr_;
std::shared_ptr<Transport> transport_;
ConnectionPool pool;
std::atomic<uint64_t> ioIndex;
};
} // namespace Http
} // namespace Net
...@@ -26,7 +26,8 @@ namespace Net { ...@@ -26,7 +26,8 @@ namespace Net {
namespace Http { namespace Http {
namespace Private { namespace Private {
class Parser; class ParserBase;
template<typename T> struct Parser;
class RequestLineStep; class RequestLineStep;
class HeadersStep; class HeadersStep;
class BodyStep; class BodyStep;
...@@ -72,20 +73,34 @@ namespace Uri { ...@@ -72,20 +73,34 @@ namespace Uri {
Optional<std::string> get(const std::string& name) const; Optional<std::string> get(const std::string& name) const;
bool has(const std::string& name) const; bool has(const std::string& name) const;
void clear() {
params.clear();
}
private: private:
std::unordered_map<std::string, std::string> params; std::unordered_map<std::string, std::string> params;
}; };
} // namespace Uri } // namespace Uri
class RequestBuilder;
// 5. Request // 5. Request
class Request : private Message { class Request : private Message {
public: public:
friend class Private::RequestLineStep; friend class Private::RequestLineStep;
friend class Private::HeadersStep; friend class Private::HeadersStep;
friend class Private::BodyStep; friend class Private::BodyStep;
friend class Private::Parser; friend class Private::ParserBase;
friend class Private::Parser<Http::Request>;
friend class RequestBuilder;
Request(const Request& other) = default;
Request& operator=(const Request& other) = default;
Request(Request&& other) = default;
Request& operator=(Request&& other) = default;
friend class Handler;
Version version() const; Version version() const;
Method method() const; Method method() const;
...@@ -126,7 +141,33 @@ private: ...@@ -126,7 +141,33 @@ private:
std::string resource_; std::string resource_;
Uri::Query query_; Uri::Query query_;
#ifdef LIBSTDCPP_SMARTPTR_LOCK_FIXME
std::weak_ptr<Tcp::Peer> peer_; std::weak_ptr<Tcp::Peer> peer_;
#endif
};
class RequestBuilder {
public:
RequestBuilder& resource(std::string val);
RequestBuilder& params(const Uri::Query& query);
RequestBuilder& header(const std::shared_ptr<Header::Header>& header);
template<typename H, typename... Args>
typename
std::enable_if<
Header::IsHeader<H>::value, RequestBuilder&
>::type
header(Args&& ...args) {
return header(std::make_shared<H>(std::forward<Args>(args)...));
}
RequestBuilder& cookie(const Cookie& cookie);
RequestBuilder& body(std::string val);
const Request& request() const { return request_; }
operator Request() const { return request_; }
private:
Request request_;
}; };
class Handler; class Handler;
...@@ -271,10 +312,12 @@ operator<<(ResponseStream& stream, ResponseStream & (*func)(ResponseStream &)) { ...@@ -271,10 +312,12 @@ operator<<(ResponseStream& stream, ResponseStream & (*func)(ResponseStream &)) {
} }
// 6. Response // 6. Response
class Response : private Message { class Response : public Message {
public: public:
friend class Handler; friend class Handler;
friend class Timeout; friend class Timeout;
friend class Private::ParserBase;
friend class Private::Parser<Response>;
friend Async::Promise<ssize_t> serveFile(Response&, const char *, const Mime::MediaType&); friend Async::Promise<ssize_t> serveFile(Response&, const char *, const Mime::MediaType&);
...@@ -384,6 +427,12 @@ public: ...@@ -384,6 +427,12 @@ public:
} }
private: private:
Response()
: Message()
, buf_(0)
, transport_(nullptr)
{ }
Response(Tcp::Transport* transport) Response(Tcp::Transport* transport)
: Message() : Message()
, buf_(DefaultStreamSize) , buf_(DefaultStreamSize)
...@@ -421,15 +470,15 @@ namespace Private { ...@@ -421,15 +470,15 @@ namespace Private {
enum class State { Again, Next, Done }; enum class State { Again, Next, Done };
struct Step { struct Step {
Step(Request* request) Step(Message* request)
: request(request) : message(request)
{ } { }
virtual State apply(StreamCursor& cursor) = 0; virtual State apply(StreamCursor& cursor) = 0;
void raise(const char* msg, Code code = Code::Bad_Request); void raise(const char* msg, Code code = Code::Bad_Request);
Request *request; Message *message;
}; };
struct RequestLineStep : public Step { struct RequestLineStep : public Step {
...@@ -440,8 +489,16 @@ namespace Private { ...@@ -440,8 +489,16 @@ namespace Private {
State apply(StreamCursor& cursor); State apply(StreamCursor& cursor);
}; };
struct ResponseLineStep : public Step {
ResponseLineStep(Response* response)
: Step(response)
{ }
State apply(StreamCursor& cursor);
};
struct HeadersStep : public Step { struct HeadersStep : public Step {
HeadersStep(Request* request) HeadersStep(Message* request)
: Step(request) : Step(request)
{ } { }
...@@ -449,7 +506,7 @@ namespace Private { ...@@ -449,7 +506,7 @@ namespace Private {
}; };
struct BodyStep : public Step { struct BodyStep : public Step {
BodyStep(Request* request) BodyStep(Message* request)
: Step(request) : Step(request)
, bytesRead(0) , bytesRead(0)
{ } { }
...@@ -460,22 +517,51 @@ namespace Private { ...@@ -460,22 +517,51 @@ namespace Private {
size_t bytesRead; size_t bytesRead;
}; };
struct Parser { struct ParserBase {
ParserBase()
: currentStep(0)
, cursor(&buffer)
{
}
Parser() ParserBase(const char* data, size_t len)
: contentLength(-1) : currentStep(0)
, currentStep(0)
, cursor(&buffer) , cursor(&buffer)
{ {
}
ParserBase(const ParserBase& other) = delete;
ParserBase(ParserBase&& other) = default;
bool feed(const char* data, size_t len);
virtual void reset();
State parse();
ArrayStreamBuf<Const::MaxBuffer> buffer;
StreamCursor cursor;
protected:
static constexpr size_t StepsCount = 3;
std::array<std::unique_ptr<Step>, StepsCount> allSteps;
size_t currentStep;
};
template<typename Message> struct Parser;
template<> struct Parser<Http::Request> : public ParserBase {
Parser()
: ParserBase()
{
allSteps[0].reset(new RequestLineStep(&request)); allSteps[0].reset(new RequestLineStep(&request));
allSteps[1].reset(new HeadersStep(&request)); allSteps[1].reset(new HeadersStep(&request));
allSteps[2].reset(new BodyStep(&request)); allSteps[2].reset(new BodyStep(&request));
} }
Parser(const char* data, size_t len) Parser(const char* data, size_t len)
: contentLength(-1) : ParserBase()
, currentStep(0)
, cursor(&buffer)
{ {
allSteps[0].reset(new RequestLineStep(&request)); allSteps[0].reset(new RequestLineStep(&request));
allSteps[1].reset(new HeadersStep(&request)); allSteps[1].reset(new HeadersStep(&request));
...@@ -484,26 +570,38 @@ namespace Private { ...@@ -484,26 +570,38 @@ namespace Private {
feed(data, len); feed(data, len);
} }
Parser(const Parser& other) = delete; void reset() {
Parser(Parser&& other) = default; ParserBase::reset();
bool feed(const char* data, size_t len);
void reset();
State parse();
ArrayStreamBuf<Const::MaxBuffer> buffer; request.headers_.clear();
StreamCursor cursor; request.body_.clear();
request.resource_.clear();
request.query_.clear();
}
Request request; Request request;
};
private: template<> struct Parser<Http::Response> : public ParserBase {
static constexpr size_t StepsCount = 3; Parser()
: ParserBase()
{
allSteps[0].reset(new ResponseLineStep(&response));
allSteps[1].reset(new HeadersStep(&response));
allSteps[2].reset(new BodyStep(&response));
}
std::array<std::unique_ptr<Step>, StepsCount> allSteps; Parser(const char* data, size_t len)
size_t currentStep; : ParserBase()
{
allSteps[0].reset(new ResponseLineStep(&response));
allSteps[1].reset(new HeadersStep(&response));
allSteps[2].reset(new BodyStep(&response));
feed(data, len);
}
ssize_t contentLength; Response response;
}; };
} // namespace Private } // namespace Private
...@@ -520,7 +618,7 @@ public: ...@@ -520,7 +618,7 @@ public:
virtual void onTimeout(const Request& request, Response response); virtual void onTimeout(const Request& request, Response response);
private: private:
Private::Parser& getParser(const std::shared_ptr<Tcp::Peer>& peer) const; Private::Parser<Http::Request>& getParser(const std::shared_ptr<Tcp::Peer>& peer) const;
}; };
} // namespace Http } // namespace Http
......
...@@ -349,6 +349,10 @@ public: ...@@ -349,6 +349,10 @@ public:
NAME("User-Agent") NAME("User-Agent")
UserAgent() { } UserAgent() { }
explicit UserAgent(const char* ua)
: ua_(ua)
{ }
explicit UserAgent(const std::string& ua) : explicit UserAgent(const std::string& ua) :
ua_(ua) ua_(ua)
{ } { }
......
...@@ -101,9 +101,12 @@ namespace Io { ...@@ -101,9 +101,12 @@ namespace Io {
void registerFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level); void registerFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level);
void registerFdOneShot(Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level); void registerFdOneShot(Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level);
void modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level); void modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode = Polling::Mode::Level);
void registerFd(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode = Polling::Mode::Level);
void registerFdOneShot(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode = Polling::Mode::Level);
void modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode = Polling::Mode::Level);
void init(const std::shared_ptr<Handler>& handler); void init(const std::shared_ptr<Handler>& handler);
void run(); void run();
void shutdown(); void shutdown();
...@@ -168,7 +171,7 @@ namespace Io { ...@@ -168,7 +171,7 @@ namespace Io {
void shutdown(); void shutdown();
std::shared_ptr<Service> service(Fd fd) const; std::shared_ptr<Service> service(Fd fd) const;
std::shared_ptr<Service> service(size_t index) const;
std::vector<Async::Promise<rusage>> load() const; std::vector<Async::Promise<rusage>> load() const;
size_t size() const { size_t size() const {
......
...@@ -4,6 +4,7 @@ ...@@ -4,6 +4,7 @@
#include "cookie.h" #include "cookie.h"
#include "router.h" #include "router.h"
#include "endpoint.h" #include "endpoint.h"
#include "client.h"
#include <iostream> #include <iostream>
#include <cstring> #include <cstring>
#include <algorithm> #include <algorithm>
...@@ -287,6 +288,7 @@ private: ...@@ -287,6 +288,7 @@ private:
}; };
#endif #endif
/*
int main(int argc, char *argv[]) { int main(int argc, char *argv[]) {
Net::Port port(9080); Net::Port port(9080);
...@@ -333,3 +335,23 @@ int main(int argc, char *argv[]) { ...@@ -333,3 +335,23 @@ int main(int argc, char *argv[]) {
server->shutdown(); server->shutdown();
monitor.shutdown(); monitor.shutdown();
} }
*/
#if 1
int main() {
Net::Http::Client client("http://www.foaas.com");
auto opts = Net::Http::Client::options()
.threads(1)
.maxConnections(1);
using namespace Net::Http;
client.init(opts);
client.get(
client
.newRequest("/off/octal/nask")
.header<Header::ContentType>(MIME(Text, Plain))
.cookie(Cookie("FOO", "bar")));
std::this_thread::sleep_for(std::chrono::seconds(1));
}
#endif
/*
Mathieu Stefani, 29 janvier 2016
Implementation of the Http client
*/
#include "client.h"
#include "stream.h"
#include <sys/sendfile.h>
#include <netdb.h>
using namespace Polling;
namespace Net {
namespace Http {
namespace {
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) return false; \
} while (0)
template<typename H, typename Stream, typename... Args>
typename std::enable_if<Header::IsHeader<H>::value, Stream&>::type
writeHeader(Stream& stream, Args&& ...args) {
H header(std::forward<Args>(args)...);
stream << H::Name << ": ";
header.write(stream);
stream << crlf;
return stream;
}
bool writeHeaders(const Header::Collection& headers, DynamicStreamBuf& buf) {
std::ostream os(&buf);
for (const auto& header: headers.list()) {
OUT(os << header->name() << ": ");
OUT(header->write(os));
OUT(os << crlf);
}
return true;
}
bool writeCookies(const CookieJar& cookies, DynamicStreamBuf& buf) {
std::ostream os(&buf);
for (const auto& cookie: cookies) {
OUT(os << "Cookie: ");
OUT(cookie.write(os));
OUT(os << crlf);
}
return true;
}
bool writeRequest(const Http::Request& request, DynamicStreamBuf &buf) {
std::ostream os(&buf);
OUT(os << "GET ");
OUT(os << request.resource());
OUT(os << " HTTP/1.1" << crlf);
if (!writeCookies(request.cookies(), buf)) return false;
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;
OUT(os << crlf);
return true;
}
#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.isReadable()) {
auto tag = entry.getTag();
auto fd = tag.value();
handleIncoming(fd);
}
else if (entry.isWritable()) {
auto tag = entry.getTag();
auto fd = tag.value();
auto connIt = pendingConnections.find(fd);
if (connIt != std::end(pendingConnections)) {
auto& conn = connIt->second;
conn.resolve();
pendingConnections.erase(fd);
continue;
}
auto writeIt = toWrite.find(fd);
if (writeIt != std::end(toWrite)) {
io()->modifyFd(fd, NotifyOn::Read, Polling::Mode::Edge);
auto& write = writeIt->second;
asyncWriteImpl(fd, write, Retry);
continue;
}
throw std::runtime_error("Unknown fd");
}
}
}
void
Transport::registerPoller(Polling::Epoll& poller) {
writesQueue.bind(poller);
connectionsQueue.bind(poller);
}
Async::Promise<void>
Transport::asyncConnect(Fd fd, const struct sockaddr* address, socklen_t addr_len)
{
return Async::Promise<void>([=](Async::Resolver& resolve, Async::Rejection& reject) {
PendingConnection conn(std::move(resolve), std::move(reject), fd, address, addr_len);
auto *entry = connectionsQueue.allocEntry(std::move(conn));
connectionsQueue.push(entry);
});
}
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);
}
void
Transport::asyncWriteImpl(
Fd fd, int flags, const BufferHolder& buffer,
Async::Resolver resolve, Async::Rejection reject, WriteStatus status)
{
auto cleanUp = [&]() {
if (buffer.isRaw()) {
auto raw = buffer.raw();
if (raw.isOwned) delete[] raw.data;
}
if (status == Retry)
toWrite.erase(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);
}
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)));
}
io()->modifyFd(fd, NotifyOn::Read | NotifyOn::Write, Polling::Mode::Edge);
}
else {
cleanUp();
reject(Net::Error::system("Could not write data"));
}
break;
}
else {
totalWritten += bytesWritten;
if (totalWritten == len) {
cleanUp();
resolve(totalWritten);
break;
}
}
}
}
void
Transport::handleWriteQueue() {
// Let's drain the queue
for (;;) {
std::unique_ptr<PollableQueue<OnHoldWrite>::Entry> entry(writesQueue.pop());
if (!entry) break;
auto &write = entry->data();
asyncWriteImpl(write.peerFd, write);
}
}
void
Transport::handleConnectionQueue() {
for (;;) {
std::unique_ptr<PollableQueue<PendingConnection>::Entry> entry(connectionsQueue.pop());
if (!entry) break;
auto &conn = entry->data();
int res = ::connect(conn.fd, conn.addr, conn.addr_len);
if (res == -1) {
if (errno == EINPROGRESS) {
io()->registerFdOneShot(conn.fd, NotifyOn::Write);
pendingConnections.insert(
std::make_pair(conn.fd, std::move(conn)));
}
else {
conn.reject(Error::system("Failed to connect"));
}
}
}
}
void
Transport::handleIncoming(Fd fd) {
std::cout << "Handling incoming on fd " << fd << std::endl;
char buffer[Const::MaxBuffer];
memset(buffer, 0, sizeof buffer);
ssize_t totalBytes = 0;
for (;;) {
ssize_t bytes;
bytes = recv(fd, buffer + totalBytes, Const::MaxBuffer - totalBytes, 0);
if (bytes == -1) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
if (totalBytes > 0) {
handleResponsePacket(fd, buffer, totalBytes);
}
} else {
if (errno == ECONNRESET) {
}
else {
throw std::runtime_error(strerror(errno));
}
}
break;
}
else if (bytes == 0) {
break;
}
else {
totalBytes += bytes;
if (totalBytes >= Const::MaxBuffer) {
std::cerr << "Too long packet" << std::endl;
break;
}
}
}
}
void
Transport::handleResponsePacket(Fd fd, const char* buffer, size_t totalBytes) {
auto it = inflightRequests.find(fd);
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);
if (req.parser->parse() == Private::State::Done) {
std::cout << "DOne parsing!" << std::endl;
}
}
void
Connection::connect(Net::Address addr)
{
struct addrinfo hints;
struct addrinfo *addrs;
memset(&hints, 0, sizeof(struct addrinfo));
hints.ai_family = AF_UNSPEC; /* Allow IPv4 or IPv6 */
hints.ai_socktype = SOCK_STREAM; /* Stream socket */
hints.ai_flags = 0;
hints.ai_protocol = 0;
auto host = addr.host();
/* We rely on the fact that a string literal is an lvalue const char[N] */
static constexpr size_t MaxPortLen = sizeof("65535");
char port[MaxPortLen];
std::fill(port, port + MaxPortLen, 0);
std::snprintf(port, MaxPortLen, "%d", static_cast<uint16_t>(addr.port()));
TRY(::getaddrinfo(host.c_str(), port, &hints, &addrs));
int sfd = -1;
for (struct addrinfo *addr = addrs; addr; addr = addr->ai_next) {
sfd = ::socket(addr->ai_family, addr->ai_socktype, addr->ai_protocol);
if (sfd < 0) continue;
make_non_blocking(sfd);
state_.store(static_cast<uint32_t>(State::Connecting));
transport_->asyncConnect(sfd, addr->ai_addr, addr->ai_addrlen)
.then([=]() {
fd = sfd;
transport_->io()->modifyFd(fd, NotifyOn::Read);
state_.store(static_cast<uint32_t>(State::Connected));
processRequestQueue();
}, Async::Throw);
break;
}
if (sfd < 0)
throw std::runtime_error("Failed to connect");
}
bool
Connection::isConnected() const {
return static_cast<State>(state_.load()) == State::Connected;
}
void
Connection::associateTransport(const std::shared_ptr<Transport>& transport) {
if (transport_)
throw std::runtime_error("A transport has already been associated to the connection");
transport_ = transport;
}
bool
Connection::hasTransport() const {
return transport_ != nullptr;
}
Async::Promise<Response>
Connection::perform(const Http::Request& request) {
return Async::Promise<Response>([=](Async::Resolver& resolve, Async::Rejection& reject) {
if (!isConnected()) {
auto* entry = requestsQueue.allocEntry(RequestData(std::move(resolve), std::move(reject), request));
requestsQueue.push(entry);
} else {
performImpl(request, std::move(resolve), std::move(reject));
}
});
}
struct OnRequestWriteCompleted {
OnRequestWriteCompleted(const std::shared_ptr<Transport>& transport, Fd fd, Async::Resolver resolve, Async::Rejection reject)
: transport(transport)
, fd(fd)
, resolve(std::move(resolve))
, reject(std::move(reject))
{ }
void operator()(size_t bytes) const {
transport->addInFlight(fd, std::move(resolve), std::move(reject));
}
mutable std::shared_ptr<Transport> transport;
Fd fd;
mutable Async::Resolver resolve;
mutable Async::Rejection reject;
};
void
Connection::performImpl(
const Http::Request& request,
Async::Resolver resolve,
Async::Rejection reject) {
DynamicStreamBuf buf(128);
if (!writeRequest(request, 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);
}
void
Connection::processRequestQueue() {
for (;;) {
std::unique_ptr<Queue<RequestData>::Entry> entry(requestsQueue.pop());
if (!entry) break;
auto &req = entry->data();
performImpl(req.request, std::move(req.resolve), std::move(req.reject));
}
}
void
ConnectionPool::init(size_t max)
{
for (size_t i = 0; i < max; ++i) {
connections.push_back(std::make_shared<Connection>());
}
}
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;
}
}
}
return nullptr;
}
void
ConnectionPool::returnConnection(const std::shared_ptr<Connection>& connection) {
connection->state_.store(static_cast<uint32_t>(Connection::State::Idle));
}
Client::Options&
Client::Options::threads(int val) {
threads_ = val;
return *this;
}
Client::Options&
Client::Options::maxConnections(int val) {
maxConnections_ = val;
return *this;
}
Client::Options&
Client::Options::keepAlive(bool val) {
keepAlive_ = val;
}
Client::Client(const std::string& base)
: url_(base)
, ioIndex(0)
{ }
Client::Options
Client::options() {
return Client::Options();
}
void
Client::init(const Client::Options& options) {
transport_.reset(new Transport);
io_.init(options.threads_, transport_);
io_.start();
constexpr const char *Http = "http://";
constexpr size_t Size = sizeof("http://") - 1;
std::string url(url_);
if (!url.compare(0, Size, Http)) {
url.erase(0, Size);
}
auto pos = url.find(':');
Port port(80);
std::string host;
if (pos != std::string::npos) {
host = url.substr(0, pos);
port = std::stol(url.substr(pos + 1));
}
else {
host = url;
}
addr_ = Net::Address(host, port);
pool.init(options.maxConnections_);
}
RequestBuilder
Client::newRequest(std::string val) {
RequestBuilder builder;
builder.resource(std::move(val));
return builder;
}
Async::Promise<Http::Response>
Client::get(const Http::Request& request, std::chrono::milliseconds timeout)
{
auto conn = pool.pickConnection();
if (conn == nullptr) {
std::cout << "No connection available yet, bailing-out" << std::endl;
}
else {
if (!conn->hasTransport()) {
auto index = ioIndex.fetch_add(1) % io_.size();
auto service = io_.service(index);
auto transport = std::static_pointer_cast<Transport>(service->handler());
conn->associateTransport(transport);
conn->connect(addr_);
}
conn->perform(request);
}
return Async::Promise<Http::Response>::rejected(std::runtime_error("Unimplemented"));
}
} // namespace Http
} // namespace Net
...@@ -129,6 +129,8 @@ namespace Private { ...@@ -129,6 +129,8 @@ namespace Private {
#undef METHOD #undef METHOD
}; };
auto request = static_cast<Request *>(message);
bool found = false; bool found = false;
for (const auto& method: Methods) { for (const auto& method: Methods) {
if (match_raw(method.str, method.len, cursor)) { if (match_raw(method.str, method.len, cursor)) {
...@@ -220,6 +222,47 @@ namespace Private { ...@@ -220,6 +222,47 @@ namespace Private {
} }
State
ResponseLineStep::apply(StreamCursor& cursor) {
StreamCursor::Revert revert(cursor);
if (match_raw("HTTP/1.1", sizeof("HTTP/1.1") - 1, cursor)) {
std::cout << "Matched http/1.1" << std::endl;
//response->version = Version::Http11;
}
else if (match_raw("HTTP/1.0", sizeof("HTTP/1.0") - 1, cursor)) {
}
else {
raise("Encountered invalid HTTP version");
}
int n;
// SP
if ((n = cursor.current()) != StreamCursor::Eof && n != ' ')
raise("Expected SPACE after http version");
if (!cursor.advance(1)) return State::Again;
StreamCursor::Token codeToken(cursor);
if (!match_until(' ', cursor))
return State::Again;
auto code = codeToken.text();
std::cout << "Code = " << code << std::endl;
if (!cursor.advance(1)) return State::Again;
StreamCursor::Token textToken(cursor);
while (!cursor.eol()) cursor.advance(1);
auto text = textToken.text();
std::cout << "Text = "<< text << std::endl;
if (!cursor.advance(2)) return State::Again;
revert.ignore();
return State::Next;
}
State State
HeadersStep::apply(StreamCursor& cursor) { HeadersStep::apply(StreamCursor& cursor) {
StreamCursor::Revert revert(cursor); StreamCursor::Revert revert(cursor);
...@@ -249,7 +292,7 @@ namespace Private { ...@@ -249,7 +292,7 @@ namespace Private {
} }
if (name == "Cookie") { if (name == "Cookie") {
request->cookies_.add( message->cookies_.add(
Cookie::fromRaw(cursor.offset(start), cursor.diff(start)) Cookie::fromRaw(cursor.offset(start), cursor.diff(start))
); );
} }
...@@ -257,11 +300,11 @@ namespace Private { ...@@ -257,11 +300,11 @@ namespace Private {
else if (Header::Registry::isRegistered(name)) { else if (Header::Registry::isRegistered(name)) {
std::shared_ptr<Header::Header> header = Header::Registry::makeHeader(name); std::shared_ptr<Header::Header> header = Header::Registry::makeHeader(name);
header->parseRaw(cursor.offset(start), cursor.diff(start)); header->parseRaw(cursor.offset(start), cursor.diff(start));
request->headers_.add(header); message->headers_.add(header);
} }
else { else {
std::string value(cursor.offset(start), cursor.diff(start)); std::string value(cursor.offset(start), cursor.diff(start));
request->headers_.addRaw(Header::Raw(std::move(name), std::move(value))); message->headers_.addRaw(Header::Raw(std::move(name), std::move(value)));
} }
// CRLF // CRLF
...@@ -276,7 +319,7 @@ namespace Private { ...@@ -276,7 +319,7 @@ namespace Private {
State State
BodyStep::apply(StreamCursor& cursor) { BodyStep::apply(StreamCursor& cursor) {
auto cl = request->headers_.tryGet<Header::ContentLength>(); auto cl = message->headers_.tryGet<Header::ContentLength>();
if (!cl) return State::Done; if (!cl) return State::Done;
auto contentLength = cl->value(); auto contentLength = cl->value();
...@@ -292,7 +335,7 @@ namespace Private { ...@@ -292,7 +335,7 @@ namespace Private {
// for that right now // for that right now
if (available < remaining) { if (available < remaining) {
cursor.advance(available); cursor.advance(available);
request->body_ += token.text(); message->body_ += token.text();
bytesRead += available; bytesRead += available;
...@@ -300,7 +343,7 @@ namespace Private { ...@@ -300,7 +343,7 @@ namespace Private {
} }
else { else {
cursor.advance(remaining); cursor.advance(remaining);
request->body_ += token.text(); message->body_ += token.text();
} }
} }
...@@ -308,20 +351,20 @@ namespace Private { ...@@ -308,20 +351,20 @@ namespace Private {
else { else {
if (!cursor.advance(2)) return State::Again; if (!cursor.advance(2)) return State::Again;
request->body_.reserve(contentLength); message->body_.reserve(contentLength);
StreamCursor::Token token(cursor); StreamCursor::Token token(cursor);
const size_t available = cursor.remaining(); const size_t available = cursor.remaining();
// We have an incomplete body, read what we can // We have an incomplete body, read what we can
if (available < contentLength) { if (available < contentLength) {
cursor.advance(available); cursor.advance(available);
request->body_ += token.text(); message->body_ += token.text();
bytesRead += available; bytesRead += available;
return State::Again; return State::Again;
} }
cursor.advance(contentLength); cursor.advance(contentLength);
request->body_ = token.text(); message->body_ = token.text();
} }
bytesRead = 0; bytesRead = 0;
...@@ -329,7 +372,7 @@ namespace Private { ...@@ -329,7 +372,7 @@ namespace Private {
} }
State State
Parser::parse() { ParserBase::parse() {
State state = State::Again; State state = State::Again;
do { do {
Step *step = allSteps[currentStep].get(); Step *step = allSteps[currentStep].get();
...@@ -344,20 +387,17 @@ namespace Private { ...@@ -344,20 +387,17 @@ namespace Private {
} }
bool bool
Parser::feed(const char* data, size_t len) { ParserBase::feed(const char* data, size_t len) {
return buffer.feed(data, len); return buffer.feed(data, len);
} }
void void
Parser::reset() { ParserBase::reset() {
buffer.reset(); buffer.reset();
cursor.reset(); cursor.reset();
currentStep = 0; currentStep = 0;
request.headers_.clear();
request.body_.clear();
request.resource_.clear();
} }
} // namespace Private } // namespace Private
...@@ -446,6 +486,35 @@ Request::peer() const { ...@@ -446,6 +486,35 @@ Request::peer() const {
} }
#endif #endif
RequestBuilder&
RequestBuilder::resource(std::string val) {
request_.resource_ = std::move(val);
return *this;
}
RequestBuilder&
RequestBuilder::params(const Uri::Query& params) {
request_.query_ = params;
return *this;
}
RequestBuilder&
RequestBuilder::header(const std::shared_ptr<Header::Header>& header) {
request_.headers_.add(header);
return *this;
}
RequestBuilder&
RequestBuilder::cookie(const Cookie& cookie) {
request_.cookies_.add(cookie);
return *this;
}
RequestBuilder&
RequestBuilder::body(std::string val) {
request_.body_ = std::move(val);
}
ResponseStream::ResponseStream( ResponseStream::ResponseStream(
Message&& other, Message&& other,
std::weak_ptr<Tcp::Peer> peer, std::weak_ptr<Tcp::Peer> peer,
...@@ -641,7 +710,7 @@ Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer ...@@ -641,7 +710,7 @@ Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer
void void
Handler::onConnection(const std::shared_ptr<Tcp::Peer>& peer) { Handler::onConnection(const std::shared_ptr<Tcp::Peer>& peer) {
peer->putData(ParserData, std::make_shared<Private::Parser>()); peer->putData(ParserData, std::make_shared<Private::Parser<Http::Request>>());
} }
void void
...@@ -663,9 +732,9 @@ Timeout::onTimeout(uint64_t numWakeup) { ...@@ -663,9 +732,9 @@ Timeout::onTimeout(uint64_t numWakeup) {
} }
Private::Parser& Private::Parser<Http::Request>&
Handler::getParser(const std::shared_ptr<Tcp::Peer>& peer) const { Handler::getParser(const std::shared_ptr<Tcp::Peer>& peer) const {
return *peer->getData<Private::Parser>(ParserData); return *peer->getData<Private::Parser<Http::Request>>(ParserData);
} }
......
...@@ -50,6 +50,21 @@ Service::modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode) { ...@@ -50,6 +50,21 @@ Service::modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Mode mode) {
poller.rearmFd(fd, interest, Polling::Tag(fd), mode); poller.rearmFd(fd, interest, Polling::Tag(fd), mode);
} }
void
Service::registerFd(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode) {
poller.addFd(fd, interest, tag, mode);
}
void
Service::registerFdOneShot(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode) {
poller.addFdOneShot(fd, interest, tag, mode);
}
void
Service::modifyFd(Fd fd, Polling::NotifyOn interest, Polling::Tag tag, Polling::Mode mode) {
poller.rearmFd(fd, interest, tag, mode);
}
void void
Service::init(const std::shared_ptr<Handler>& handler) { Service::init(const std::shared_ptr<Handler>& handler) {
handler_ = handler; handler_ = handler;
...@@ -247,6 +262,14 @@ ServiceGroup::service(Fd fd) const { ...@@ -247,6 +262,14 @@ ServiceGroup::service(Fd fd) const {
auto& wrk = workers_[worker]; auto& wrk = workers_[worker];
return wrk->service(); return wrk->service();
} }
std::shared_ptr<Service>
ServiceGroup::service(size_t index) const {
if (index >= workers_.size())
throw std::out_of_range("index out of range");
return workers_[index]->service();
}
ServiceGroup::Worker::Worker() { ServiceGroup::Worker::Worker() {
service_.reset(new Service); service_.reset(new Service);
......
...@@ -178,6 +178,7 @@ Listener::bind(const Address& address) { ...@@ -178,6 +178,7 @@ Listener::bind(const Address& address) {
} }
TRY(::listen(fd, backlog_)); TRY(::listen(fd, backlog_));
break;
} }
make_non_blocking(fd); make_non_blocking(fd);
......
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