Refactored a little bit the API. Peer is not a reference anymore but

instead is a shared_ptr

main.cc

@@ -8,28 +8,28 @@
 using namespace std;
 
 class MyHandler : public Net::Http::Handler {
-    void onRequest(const Net::Http::Request& req, Net::Tcp::Peer& peer) {
-        if (req.resource == "/ping") {
-            if (req.method == Net::Http::Method::Get) {
+    void onRequest(const Net::Http::Request& req, Net::Http::Response response) {
+        if (req.resource() == "/ping") {
+            if (req.method() == Net::Http::Method::Get) {
 
                 using namespace Net::Http;
 
-                Net::Http::Response response(Net::Http::Code::Ok, "PONG");
-                response.headers
+                response.headers()
                     .add(std::make_shared<Header::Server>("lys"))
                     .add(std::make_shared<Header::ContentType>(MIME(Text, Plain)));
 
-                response.writeTo(peer);
+                response.send(Net::Http::Code::Ok, "PONG");
 
             }
         }
-        else if (req.resource == "/echo") {
-            if (req.method == Net::Http::Method::Post) {
-                Net::Http::Response response(Net::Http::Code::Ok, req.body);
-
-                response.writeTo(peer);
+        else if (req.resource() == "/echo") {
+            if (req.method() == Net::Http::Method::Post) {
+                response.send(Net::Http::Code::Ok, req.body());
             }
         }
+        else if (req.resource() == "/exception") {
+            throw std::runtime_error("Exception thrown in the handler");
+        }
     }
 };
 

src/http.cc

@@ -32,12 +32,12 @@ static constexpr const char* ParserData = "__Parser";
 namespace Private {
 
     void
-    Parser::Step::raise(const char* msg, Code code /* = Code::Bad_Request */) {
+    Step::raise(const char* msg, Code code /* = Code::Bad_Request */) {
         throw HttpError(code, msg);
     }
 
-    Parser::State
-    Parser::RequestLineStep::apply(StreamCursor& cursor) {
+    State
+    RequestLineStep::apply(StreamCursor& cursor) {
         StreamCursor::Revert revert(cursor);
 
         // Method
@@ -59,7 +59,7 @@ namespace Private {
         bool found = false;
         for (const auto& method: Methods) {
             if (match_raw(method.str, method.len, cursor)) {
-                request->method = method.repr;
+                request->method_ = method.repr;
                 found = true;
                 break;
             }
@@ -83,7 +83,7 @@ namespace Private {
             if (!cursor.advance(1)) return State::Again;
         }
 
-        request->resource = std::string(cursor.offset(start), cursor.diff(start) + 1);
+        request->resource_ = std::string(cursor.offset(start), cursor.diff(start) + 1);
         if ((n = cursor.next()) == StreamCursor::Eof) return State::Again;
 
         if (n != ' ')
@@ -100,10 +100,10 @@ namespace Private {
 
         const size_t diff = cursor.diff(start);
         if (strncmp(cursor.offset(start), "HTTP/1.0", diff) == 0) {
-            request->version = Version::Http10;
+            request->version_ = Version::Http10;
         }
         else if (strncmp(cursor.offset(start), "HTTP/1.1", diff) == 0) {
-            request->version = Version::Http11;
+            request->version_ = Version::Http11;
         }
         else {
             raise("Encountered invalid HTTP version");
@@ -116,8 +116,8 @@ namespace Private {
 
     }
 
-    Parser::State
-    Parser::HeadersStep::apply(StreamCursor& cursor) {
+    State
+    HeadersStep::apply(StreamCursor& cursor) {
         StreamCursor::Revert revert(cursor);
 
         while (!cursor.eol()) {
@@ -147,11 +147,11 @@ namespace Private {
             if (Header::Registry::isRegistered(name)) {
                 std::shared_ptr<Header::Header> header = Header::Registry::makeHeader(name);
                 header->parseRaw(cursor.offset(start), cursor.diff(start));
-                request->headers.add(header);
+                request->headers_.add(header);
             }
             else {
                 std::string value(cursor.offset(start), cursor.diff(start));
-                request->headers.addRaw(Header::Raw(std::move(name), std::move(value)));
+                request->headers_.addRaw(Header::Raw(std::move(name), std::move(value)));
             }
 
             // CRLF
@@ -161,13 +161,13 @@ namespace Private {
         }
 
         revert.ignore();
-        return Parser::State::Next;
+        return State::Next;
     }
 
-    Parser::State
-    Parser::BodyStep::apply(StreamCursor& cursor) {
-        auto cl = request->headers.tryGet<Header::ContentLength>();
-        if (!cl) return Parser::State::Done;
+    State
+    BodyStep::apply(StreamCursor& cursor) {
+        auto cl = request->headers_.tryGet<Header::ContentLength>();
+        if (!cl) return State::Done;
 
         auto contentLength = cl->value();
         // We already started to read some bytes but we got an incomplete payload
@@ -181,14 +181,14 @@ namespace Private {
             if (!cursor.advance(remaining)) {
                 const size_t available = cursor.remaining();
 
-                request->body.append(cursor.offset(start), available);
+                request->body_.append(cursor.offset(start), available);
                 bytesRead += available;
 
                 cursor.advance(available);
                 return State::Again;
             }
             else {
-                request->body.append(cursor.offset(), remaining);
+                request->body_.append(cursor.offset(), remaining);
             }
 
         }
@@ -196,7 +196,7 @@ namespace Private {
         else {
             if (!cursor.advance(2)) return State::Again;
 
-            request->body.reserve(contentLength);
+            request->body_.reserve(contentLength);
 
             auto start = cursor;
 
@@ -204,21 +204,21 @@ namespace Private {
             if (!cursor.advance(contentLength)) {
                 const size_t available = cursor.remaining();
 
-                request->body.append(cursor.offset(start), available);
+                request->body_.append(cursor.offset(start), available);
                 bytesRead += available;
 
                 cursor.advance(available);
                 return State::Again;
             }
 
-            request->body.append(cursor.offset(start), contentLength);
+            request->body_.append(cursor.offset(start), contentLength);
         }
 
         bytesRead = 0;
-        return Parser::State::Done;
+        return State::Done;
     }
 
-    Parser::State
+    State
     Parser::parse() {
         State state = State::Again;
         do {
@@ -245,88 +245,147 @@ namespace Private {
 
         currentStep = 0;
 
-        request.headers.clear();
-        request.body.clear();
-        request.resource.clear();
+        request.headers_.clear();
+        request.body_.clear();
+        request.resource_.clear();
     }
 
 } // namespace Private
 
 Message::Message()
-    : version(Version::Http11)
+    : version_(Version::Http11)
 { }
 
 Request::Request()
     : Message()
 { }
 
-Response::Response(int code, std::string body)
-{
-    this->body = std::move(body);
-    code_ = code;
+Version
+Request::version() const {
+    return version_;
+}
+
+Method
+Request::method() const {
+    return method_;
+}
+
+std::string
+Request::resource() const {
+    return resource_;
+}
+
+std::string
+Request::body() const {
+    return body_;
 }
 
-Response::Response(Code code, std::string body)
+const Header::Collection&
+Request::headers() const {
+    return headers_;
+}
+
+Response::Response()
     : Message()
+{ }
+
+void
+Response::associatePeer(const std::shared_ptr<Tcp::Peer>& peer)
 {
-    this->body = std::move(body);
-    code_ = static_cast<int>(code);
+    if (peer_.use_count() > 0)
+        throw std::runtime_error("A peer was already associated to the response");
+
+    peer_ = peer;
 }
 
-void
-Response::writeTo(Tcp::Peer& peer)
+ssize_t
+Response::send(Code code) {
+    return send(code, "");
+}
+
+ssize_t
+Response::send(Code code, const std::string& body, const Mime::MediaType& mime)
 {
-    int fd = peer.fd();
+    int fd = peer()->fd();
 
     char buffer[Const::MaxBuffer];
     Io::OutArrayBuf obuf(buffer, Io::Init::ZeroOut);
 
     std::ostream stream(&obuf);
     stream << "HTTP/1.1 ";
-    stream << code_;
+    stream << static_cast<int>(code);
     stream << ' ';
-    stream << static_cast<Code>(code_);
+    stream << code;
     stream << crlf;
 
-    for (const auto& header: headers.list()) {
+    for (const auto& header: headers_.list()) {
         stream << header->name() << ": ";
         header->write(stream);
         stream << crlf;
     }
 
-    stream << "Content-Length: " << body.size() << crlf;
-    stream << crlf;
+    if (!body.empty()) {
+        stream << "Content-Length: " << body.size() << crlf;
+        stream << crlf;
+
+        stream << body;
+    }
+
+    ssize_t bytes = ::send(fd, buffer, obuf.len(), 0);
+    return bytes;
+}
+
+void
+Response::setMime(const Mime::MediaType& mime)
+{
+    headers_.add(std::make_shared<Header::ContentType>(mime));
+}
+
+Header::Collection&
+Response::headers() {
+    return headers_;
+}
+
+const Header::Collection&
+Response::headers() const {
+    return headers_;
+}
 
-    stream << body;
+std::shared_ptr<Tcp::Peer>
+Response::peer() const {
+    if (peer_.expired()) {
+        throw std::runtime_error("Broken pipe");
+    }
 
-    ssize_t bytes = send(fd, buffer, obuf.len(), 0);
+    return peer_.lock();
 }
 
 void
-Handler::onInput(const char* buffer, size_t len, Tcp::Peer& peer) {
+Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer) {
     try {
         auto& parser = getParser(peer);
-       // scope (failure), {
-       //     parser.reset();
-       // };
         if (!parser.feed(buffer, len)) {
             parser.reset();
             throw HttpError(Code::Request_Entity_Too_Large, "Request exceeded maximum buffer size");
         }
 
         auto state = parser.parse();
-        if (state == Private::Parser::State::Done) {
-            onRequest(parser.request, peer);
+        if (state == Private::State::Done) {
+            Response response;
+            response.associatePeer(peer);
+            onRequest(parser.request, std::move(response));
             parser.reset();
         }
     } catch (const HttpError &err) {
-        Response response(err.code(), err.reason());
-        response.writeTo(peer);
+        Response response;
+        response.associatePeer(peer);
+        response.send(static_cast<Code>(err.code()), err.reason());
         getParser(peer).reset();
     }
     catch (const std::exception& e) {
-        Response response(Code::Internal_Server_Error, e.what());
-        response.writeTo(peer);
+        Response response;
+        response.associatePeer(peer);
+        response.send(Code::Internal_Server_Error, e.what());
         getParser(peer).reset();
     }
 }
@@ -336,17 +395,17 @@ Handler::onOutput() {
 }
 
 void
-Handler::onConnection(Tcp::Peer& peer) {
-    peer.putData(ParserData, std::make_shared<Private::Parser>());
+Handler::onConnection(const std::shared_ptr<Tcp::Peer>& peer) {
+    peer->putData(ParserData, std::make_shared<Private::Parser>());
 }
 
 void
-Handler::onDisconnection(Tcp::Peer& peer) {
+Handler::onDisconnection(const shared_ptr<Tcp::Peer>& peer) {
 }
 
 Private::Parser&
-Handler::getParser(Tcp::Peer& peer) const {
-    return *peer.getData<Private::Parser>(ParserData);
+Handler::getParser(const std::shared_ptr<Tcp::Peer>& peer) const {
+    return *peer->getData<Private::Parser>(ParserData);
 }
 
 Endpoint::Endpoint()

src/http.h

@@ -14,88 +14,127 @@
 #include "http_headers.h"
 #include "http_defs.h"
 #include "stream.h"
+#include "mime.h"
 
 namespace Net {
 
 namespace Http {
 
+namespace Private {
+    class Parser;
+    class RequestLineStep;
+    class HeadersStep;
+    class BodyStep;
+}
+
 // 4. HTTP Message
 class Message {
 public:
     Message();
-    Version version;
+    Version version_;
 
-    Header::Collection headers;
-    std::string body;
+    Header::Collection headers_;
+    std::string body_;
 };
 
 // 5. Request
-class Request : public Message {
+class Request : private Message {
 public:
+    friend class Private::RequestLineStep;
+    friend class Private::HeadersStep;
+    friend class Private::BodyStep;
+    friend class Private::Parser;
+
     Request();
 
-    Method method;
-    std::string resource;
+    Version version() const;
+    Method method() const;
+    std::string resource() const;
+
+    std::string body() const;
+
+    const Header::Collection& headers() const;
+
+private:
+    Method method_;
+    std::string resource_;
 };
 
+class Handler;
+
 // 6. Response
-class Response : public Message {
+class Response : private Message {
 public:
-    Response(int code, std::string body);
-    Response(Code code, std::string body);
+    friend class Handler;
+
+    Response(const Response& other) = delete;
+    Response& operator=(const Response& other) = delete;
+
+    Response(Response&& other) = default;
+    Response& operator=(Response&& other) = default;
 
-    void writeTo(Tcp::Peer& peer);
-    std::string mimeType;
+    Header::Collection& headers();
+    const Header::Collection& headers() const;
+
+    void setMime(const Mime::MediaType& mime);
+
+    ssize_t send(Code code);
+    ssize_t send(Code code, const std::string& body, const Mime::MediaType &mime = Mime::MediaType());
 
 private:
-    int code_;
+    Response();
+
+    std::shared_ptr<Tcp::Peer> peer() const;
+
+    void associatePeer(const std::shared_ptr<Tcp::Peer>& peer);
+    std::weak_ptr<Tcp::Peer> peer_;
 };
 
 namespace Private {
 
-    struct Parser {
+    enum class State { Again, Next, Done };
 
-        enum class State { Again, Next, Done };
+    struct Step {
+        Step(Request* request)
+            : request(request)
+        { }
 
-        struct Step {
-            Step(Request* request)
-                : request(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;
+    };
 
-            Request *request;
-        };
+    struct RequestLineStep : public Step {
+        RequestLineStep(Request* request)
+            : Step(request)
+        { }
 
-        struct RequestLineStep : public Step {
-            RequestLineStep(Request* request)
-                : Step(request)
-            { }
+        State apply(StreamCursor& cursor);
+    };
 
-            State apply(StreamCursor& cursor);
-        };
+    struct HeadersStep : public Step {
+        HeadersStep(Request* request)
+            : Step(request)
+        { }
 
-        struct HeadersStep : public Step {
-            HeadersStep(Request* request)
-                : Step(request)
-            { }
+        State apply(StreamCursor& cursor);
+    };
 
-            State apply(StreamCursor& cursor);
-        };
+    struct BodyStep : public Step {
+        BodyStep(Request* request)
+            : Step(request)
+            , bytesRead(0)
+        { }
 
-        struct BodyStep : public Step {
-            BodyStep(Request* request)
-                : Step(request)
-                , bytesRead(0)
-            { }
+        State apply(StreamCursor& cursor);
 
-            State apply(StreamCursor& cursor);
+    private:
+        size_t bytesRead;
+    };
 
-        private:
-            size_t bytesRead;
-        };
+    struct Parser {
 
         Parser()
             : contentLength(-1)
@@ -127,8 +166,6 @@ namespace Private {
 
         State parse();
 
-       // Buffer buffer;
-       // Cursor cursor;
         ArrayStreamBuf<Const::MaxBuffer> buffer;
         StreamCursor cursor;
 
@@ -147,16 +184,16 @@ namespace Private {
 
 class Handler : public Net::Tcp::Handler {
 public:
-    void onInput(const char* buffer, size_t len, Tcp::Peer& peer);
+    void onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer);
     void onOutput();
 
-    void onConnection(Tcp::Peer& peer);
-    void onDisconnection(Tcp::Peer& peer);
+    void onConnection(const std::shared_ptr<Tcp::Peer>& peer);
+    void onDisconnection(const std::shared_ptr<Tcp::Peer>& peer);
 
-    virtual void onRequest(const Request& request, Tcp::Peer& peer) = 0;
+    virtual void onRequest(const Request& request, Response response) = 0;
 
 private:
-    Private::Parser& getParser(Tcp::Peer& peer) const;
+    Private::Parser& getParser(const std::shared_ptr<Tcp::Peer>& peer) const;
 };
 
 class Endpoint {

src/listener.cc

@@ -118,8 +118,8 @@ IoWorker::pin(const CpuSet& set) {
     }
 }
 
-std::shared_ptr<Peer>
-IoWorker::getPeer(Fd fd) const
+std::shared_ptr<Peer>&
+IoWorker::getPeer(Fd fd)
 {
     std::unique_lock<std::mutex> guard(peersMutex);
     auto it = peers.find(fd);
@@ -130,8 +130,8 @@ IoWorker::getPeer(Fd fd) const
     return it->second;
 }
 
-std::shared_ptr<Peer>
-IoWorker::getPeer(Polling::Tag tag) const
+std::shared_ptr<Peer>&
+IoWorker::getPeer(Polling::Tag tag)
 {
     return getPeer(tag.value());
 }
@@ -153,11 +153,11 @@ IoWorker::handleIncoming(const std::shared_ptr<Peer>& peer) {
         if (bytes == -1) {
             if (errno == EAGAIN || errno == EWOULDBLOCK) {
                 if (totalBytes > 0) {
-                    handler_->onInput(buffer, totalBytes, *peer);
+                    handler_->onInput(buffer, totalBytes, peer);
                 }
             } else {
                 if (errno == ECONNRESET) {
-                    handler_->onDisconnection(*peer);
+                    handler_->onDisconnection(peer);
                     close(fd);
                 }
                 else {
@@ -167,7 +167,7 @@ IoWorker::handleIncoming(const std::shared_ptr<Peer>& peer) {
             break;
         }
         else if (bytes == 0) {
-            handler_->onDisconnection(*peer);
+            handler_->onDisconnection(peer);
             close(fd);
             break;
         }
@@ -192,7 +192,7 @@ IoWorker::handleNewPeer(const std::shared_ptr<Peer>& peer)
         peers.insert(std::make_pair(fd, peer));
     }
 
-    handler_->onConnection(*peer);
+    handler_->onConnection(peer);
     poller.addFd(fd, NotifyOn::Read, Polling::Tag(fd), Polling::Mode::Edge);
 }
 
@@ -226,7 +226,8 @@ IoWorker::run() {
                     }
                 } else {
                     if (event.flags.hasFlag(NotifyOn::Read)) {
-                        handleIncoming(getPeer(event.tag));
+                        auto& peer = getPeer(event.tag);
+                        handleIncoming(peer);
                     }
                 }
             }
@@ -247,11 +248,11 @@ Handler::~Handler()
 { }
 
 void
-Handler::onConnection(Tcp::Peer& peer) {
+Handler::onConnection(const std::shared_ptr<Tcp::Peer>& peer) {
 }
 
 void
-Handler::onDisconnection(Tcp::Peer& peer) {
+Handler::onDisconnection(const std::shared_ptr<Tcp::Peer>& peer) {
 }
 
 Listener::Listener()

src/listener.h

@@ -61,8 +61,8 @@ private:
 
     CpuSet pins;
 
-    std::shared_ptr<Peer> getPeer(Fd fd) const;
-    std::shared_ptr<Peer> getPeer(Polling::Tag tag) const;
+    std::shared_ptr<Peer>& getPeer(Fd fd);
+    std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
 
     void handleIncoming(const std::shared_ptr<Peer>& peer);
     void handleMailbox();
@@ -108,11 +108,11 @@ public:
     Handler();
     ~Handler();
 
-    virtual void onInput(const char *buffer, size_t len, Tcp::Peer& peer) = 0;
+    virtual void onInput(const char *buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer) = 0;
     virtual void onOutput() = 0;
 
-    virtual void onConnection(Tcp::Peer& peer);
-    virtual void onDisconnection(Tcp::Peer& peer);
+    virtual void onConnection(const std::shared_ptr<Tcp::Peer>& peer);
+    virtual void onDisconnection(const std::shared_ptr<Tcp::Peer>& peer);
 
 };