Improved the Timeout API. Also resurrected the old Response API

Basically, this commit brings a brand new Response API to the party:
   - Response::send() can be used to send a fixed-size content and
     will use the classic Content-Length header
   - When content size is not known in advance, a Stream interface can
     be used to stream the response. A ResponseStream is obtaining by a
     call to the Response::stream member function. After that, headers
     will be immutable and data can be streamed a-la-ostream with
     operator<<

main.cc

@@ -32,12 +32,11 @@ class MyHandler : public Net::Http::Handler {
                     .add<Header::Server>("lys")
                     .add<Header::ContentType>(MIME(Text, Plain));
 
-                std::ostream os(response.rdbuf());
-                os << "PONG";
-
-                response.send(Net::Http::Code::Ok).then([](ssize_t bytes) {
-                    std::cout << "Sent total of " << bytes << " bytes" << std::endl;
-                }, Async::IgnoreException);
+                auto stream = response.stream(Net::Http::Code::Ok);
+                stream << "PO";
+                stream << flush;
+                stream << "NG";
+                stream << ends;
 
             }
         }

src/http.cc

@@ -21,12 +21,6 @@ namespace Net {
 
 namespace Http {
 
-template< class CharT, class Traits>
-std::basic_ostream<CharT, Traits>& crlf(std::basic_ostream<CharT, Traits>& os) {
-    static constexpr char CRLF[] = {0xD, 0xA};
-    os.write(CRLF, 2);
-}
-
 template<typename H, typename Stream, typename... Args>
 typename std::enable_if<Header::IsHeader<H>::value, void>::type
 writeHeader(Stream& stream, Args&& ...args) {
@@ -357,14 +351,79 @@ Request::peer() const {
     return p;
 }
 
+void
+ResponseStream::writeStatusLine() {
+    std::ostream os(&buf_);
+
+#define OUT(...) \
+    do { \
+        __VA_ARGS__; \
+        if (!os) { \
+            throw Error("Response exceeded buffer size"); \
+        } \
+    } while (0);
+
+    OUT(os << "HTTP/1.1 ");
+    OUT(os << static_cast<int>(code_));
+    OUT(os << ' ');
+    OUT(os << code_);
+    OUT(os << crlf);
+
+#undef OUT
+}
+
+void
+ResponseStream::writeHeaders() {
+    std::ostream os(&buf_);
+
+#define OUT(...) \
+    do { \
+        __VA_ARGS__; \
+        if (!os) { \
+            throw Error("Response exceeded buffer size"); \
+        } \
+    } while (0);
+
+    for (const auto& header: headers_.list()) {
+        OUT(os << header->name() << ": ");
+        OUT(header->write(os));
+        OUT(os << crlf);
+    }
+
+    OUT(writeHeader<Header::TransferEncoding>(os, Header::Encoding::Chunked));
+    OUT(os << crlf);
+
+#undef OUT
+
+}
+
+void
+ResponseStream::flush() {
+    io_->disarmTimer();
+    auto buf = buf_.buffer();
+    peer()->send(buf.data, buf.len);
+
+    buf_.clear();
+}
+
+void
+ResponseStream::ends() {
+    std::ostream os(&buf_);
+    os << "0" << crlf;
+    os << crlf;
+
+    if (!os) {
+        throw Error("Response exceeded buffer size");
+    }
+
+    flush();
+}
+
 Async::Promise<ssize_t>
-Response::putOnWire() const
+Response::putOnWire(const std::string& body)
 {
     try {
-        auto body = stream_.buffer();
-
-        NetworkStream stream(512 + body.len);
-        std::ostream os(&stream);
+        std::ostream os(&buf_);
 
 #define OUT(...) \
         do { \
@@ -386,18 +445,18 @@ Response::putOnWire() const
             OUT(os << crlf);
         }
 
-        OUT(writeHeader<Header::ContentLength>(os, body.len));
+        OUT(writeHeader<Header::ContentLength>(os, body.size()));
         OUT(os << crlf);
 
-        OUT(os.write(static_cast<const char *>(body.data), body.len));
+        os << body;
 
-        auto buf = stream.buffer();
+        auto buffer = buf_.buffer();
 
-        if (io_) {
-            io_->disarmTimer();
-        }
+        io_->disarmTimer();
+
+#undef OUT
 
-        return peer()->send(buf.data, buf.len);
+        return peer()->send(buffer.data, buffer.len);
 
     } catch (const std::runtime_error& e) {
         return Async::Promise<ssize_t>::rejected(e);

src/http.h

@@ -29,6 +29,12 @@ namespace Private {
     class BodyStep;
 }
 
+template< class CharT, class Traits>
+std::basic_ostream<CharT, Traits>& crlf(std::basic_ostream<CharT, Traits>& os) {
+    static constexpr char CRLF[] = {0xD, 0xA};
+    os.write(CRLF, 2);
+}
+
 // 4. HTTP Message
 class Message {
 public:
@@ -103,6 +109,136 @@ private:
 
 class Handler;
 class Timeout;
+class Response;
+
+class Timeout {
+public:
+
+    friend class Handler;
+
+    template<typename Duration>
+    void arm(Duration duration) {
+        Async::Promise<uint64_t> p([=](Async::Resolver& resolve, Async::Rejection& reject) {
+            io->armTimer(duration, resolve, reject);
+        });
+
+        p.then(
+            [=](uint64_t numWakeup) {
+                this->armed = false;
+                this->onTimeout(numWakeup);
+        },
+        [=](std::exception_ptr exc) {
+            std::rethrow_exception(exc);
+        });
+
+        armed = true;
+    }
+
+    void disarm() {
+        io->disarmTimer();
+        armed = false;
+    }
+
+    bool isArmed() const {
+        return armed;
+    }
+
+private:
+    Timeout(Tcp::IoWorker* io,
+            Handler* handler,
+            const std::shared_ptr<Tcp::Peer>& peer,
+            const Request& request)
+        : io(io)
+        , handler(handler)
+        , peer(peer)
+        , request(request)
+        , armed(false)
+    { }
+
+    void onTimeout(uint64_t numWakeup);
+
+    Handler* handler;
+    std::weak_ptr<Tcp::Peer> peer;
+    Request request;
+
+    Tcp::IoWorker *io;
+    bool armed;
+};
+
+class ResponseStream : private Message {
+public:
+    friend class Response;
+
+    template<typename T>
+    friend
+    ResponseStream& operator<<(ResponseStream& stream, const T& val);
+
+    const Header::Collection& headers() const {
+        return headers_;
+    }
+
+    Code code() const {
+        return code_;
+    }
+
+    void flush();
+    void ends();
+
+private:
+    ResponseStream(
+            Message&& other,
+            std::weak_ptr<Tcp::Peer> peer,
+            Tcp::IoWorker* io,
+            size_t streamSize)
+        : Message(std::move(other))
+        , peer_(std::move(peer))
+        , buf_(streamSize)
+        , io_(io)
+    {
+        writeStatusLine();
+        writeHeaders();
+    }
+
+    void writeStatusLine();
+    void writeHeaders();
+
+    std::shared_ptr<Tcp::Peer> peer() const {
+        if (peer_.expired())
+            throw std::runtime_error("Write failed: Broken pipe");
+
+        return peer_.lock();
+    }
+
+    std::weak_ptr<Tcp::Peer> peer_;
+    DynamicStreamBuf buf_;
+    Tcp::IoWorker* io_;
+};
+
+inline ResponseStream& ends(ResponseStream &stream) {
+    stream.ends();
+    return stream;
+}
+
+inline ResponseStream& flush(ResponseStream& stream) {
+    stream.flush();
+    return stream;
+}
+
+template<typename T>
+ResponseStream& operator<<(ResponseStream& stream, const T& val) {
+    Net::Size<T> size;
+
+    std::ostream os(&stream.buf_);
+    os << size(val) << crlf;
+    os << val << crlf;
+
+    return stream;
+}
+
+inline ResponseStream&
+operator<<(ResponseStream& stream, ResponseStream & (*func)(ResponseStream &)) {
+    return (*func)(stream);
+}
 
 // 6. Response
 class Response : private Message {
@@ -118,18 +254,17 @@ public:
     Response(Response&& other)
         : Message(std::move(other))
         , peer_(other.peer_)
-        , stream_(512)
+        , buf_(std::move(other.buf_))
         , io_(other.io_)
     { }
     Response& operator=(Response&& other) {
         Message::operator=(std::move(other));
-        peer_ = other.peer_;
+        peer_ = std::move(other.peer_);
         io_ = other.io_;
+        buf_ = std::move(other.buf_);
         return *this;
     }
 
-    Response(const Response& other) = default;
-
     const Header::Collection& headers() const {
         return headers_;
     }
@@ -152,7 +287,7 @@ public:
 
     Async::Promise<ssize_t> send(Code code) {
         code_ = code;
-        return putOnWire();
+        return putOnWire("");
     }
     Async::Promise<ssize_t> send(
             Code code,
@@ -169,24 +304,20 @@ public:
                 headers_.add(std::make_shared<Header::ContentType>(mime));
         }
 
-        std::ostream os(&stream_);
-        os << body;
-
-        if (!os)
-            return Async::Promise<ssize_t>::rejected(Error("Response exceeded buffer size"));
-
-        return putOnWire();
+        return putOnWire(body);
     }
 
+    ResponseStream stream(Code code, size_t streamSize = DefaultStreamSize) {
+        code_ = code;
 
-    std::streambuf *rdbuf() {
-        return &stream_;
+        return ResponseStream(std::move(*this), peer_, io_, streamSize);
     }
 
+
 private:
     Response(Tcp::IoWorker* io)
         : Message()
-        , stream_(512)
+        , buf_(DefaultStreamSize)
         , io_(io)
     { }
 
@@ -205,57 +336,13 @@ private:
         peer_ = peer;
     }
 
-    Async::Promise<ssize_t> putOnWire() const;
+    Async::Promise<ssize_t> putOnWire(const std::string& body);
 
     std::weak_ptr<Tcp::Peer> peer_;
-    NetworkStream stream_;
+    DynamicStreamBuf buf_;
     Tcp::IoWorker *io_;
 };
 
-class Timeout {
-public:
-
-    friend class Handler;
-
-    template<typename Duration>
-    void arm(Duration duration) {
-        Async::Promise<uint64_t> p([=](Async::Resolver& resolve, Async::Rejection& reject) {
-            io->armTimer(duration, resolve, reject);
-        });
-
-        p.then(
-            [=](uint64_t numWakeup) {
-                this->onTimeout(numWakeup);
-        },
-        [=](std::exception_ptr exc) {
-            std::rethrow_exception(exc);
-        });
-    }
-
-    void disarm() {
-        io->disarmTimer();
-    }
-
-private:
-    Timeout(Tcp::IoWorker* io,
-            Handler* handler,
-            const std::shared_ptr<Tcp::Peer>& peer,
-            const Request& request)
-        : io(io)
-        , handler(handler)
-        , peer(peer)
-        , request(request)
-    { }
-
-
-    void onTimeout(uint64_t numWakeup);
-
-    Handler* handler;
-    std::weak_ptr<Tcp::Peer> peer;
-    Request request;
-
-    Tcp::IoWorker *io;
-};
 
 namespace Private {
 

src/http_header.cc

@@ -31,6 +31,8 @@ const char* encodingString(Encoding encoding) {
         return "deflate";
     case Encoding::Identity:
         return "identity";
+    case Encoding::Chunked:
+        return "chunked";
     case Encoding::Unknown:
         return "unknown";
     }
@@ -362,7 +364,7 @@ Accept::write(std::ostream& os) const {
 }
 
 void
-ContentEncoding::parseRaw(const char* str, size_t len) {
+EncodingHeader::parseRaw(const char* str, size_t len) {
     // TODO: case-insensitive
     //
     if (!strncmp(str, "gzip", len)) {
@@ -377,13 +379,16 @@ ContentEncoding::parseRaw(const char* str, size_t len) {
     else if (!strncmp(str, "identity", len)) {
         encoding_ = Encoding::Identity;
     }
+    else if (!strncmp(str, "chunked", len)) {
+        encoding_ = Encoding::Chunked;
+    }
     else {
         encoding_ = Encoding::Unknown;
     }
 }
 
 void
-ContentEncoding::write(std::ostream& os) const {
+EncodingHeader::write(std::ostream& os) const {
     os << encodingString(encoding_);
 }
 

src/http_header.h

@@ -56,11 +56,13 @@ constexpr uint64_t hash(const char* str)
 #endif
 
 // 3.5 Content Codings
+// 3.6 Transfer Codings
 enum class Encoding {
     Gzip,
     Compress,
     Deflate,
     Identity,
+    Chunked,
     Unknown
 };
 
@@ -180,26 +182,45 @@ private:
     std::vector<Http::CacheDirective> directives_;
 };
 
+class EncodingHeader : public Header {
+public:
+
+    void parseRaw(const char* str, size_t len);
+    void write(std::ostream& os) const;
+
+    Encoding encoding() const {
+        return encoding_;
+    }
+
+protected:
+    EncodingHeader(Encoding encoding)
+        : encoding_(encoding)
+    { }
+
+private:
+    Encoding encoding_;
+};
 
-class ContentEncoding : public Header {
+class ContentEncoding : public EncodingHeader {
 public:
     NAME("Content-Encoding")
 
     ContentEncoding()
-       : encoding_(Encoding::Identity)
+       : EncodingHeader(Encoding::Identity)
     { }
 
     explicit ContentEncoding(Encoding encoding)
-       : encoding_(encoding)
+       : EncodingHeader(encoding)
     { }
+};
 
-    void parseRaw(const char* str, size_t len);
-    void write(std::ostream& os) const;
-
-    Encoding encoding() const { return encoding_; }
+class TransferEncoding : public EncodingHeader {
+public:
+    NAME("Transfer-Encoding")
 
-private:
-    Encoding encoding_;
+    explicit TransferEncoding(Encoding encoding)
+       : EncodingHeader(encoding)
+    { }
 };
 
 class ContentLength : public Header {

src/net.h

@@ -7,6 +7,7 @@
 #pragma once
 #include <string>
 #include <sys/socket.h>
+#include <cstring>
 #include <stdexcept>
 
 #ifndef _KERNEL_FASTOPEN
@@ -77,4 +78,67 @@ public:
     static Error system(const char* message);
 };
 
+template<typename T>
+struct Size { };
+
+template<typename T>
+size_t
+digitsCount(T val) {
+    size_t digits = 0;
+    while (val % 10) {
+        ++digits;
+
+        val /= 10;
+    }
+
+    return digits;
+}
+
+template<>
+struct Size<const char*> {
+    size_t operator()(const char *s) {
+        return std::strlen(s);
+    }
+};
+
+template<size_t N>
+struct Size<char[N]> {
+    constexpr size_t operator()(const char (&arr)[N]) {
+        // We omit the \0
+        return N - 1;
+    }
+};
+
+#define DEFINE_INTEGRAL_SIZE(Int) \
+    template<> \
+    struct Size<Int> { \
+        size_t operator()(Int val) { \
+            return digitsCount(val); \
+        } \
+    }
+
+DEFINE_INTEGRAL_SIZE(uint8_t);
+DEFINE_INTEGRAL_SIZE(int8_t);
+DEFINE_INTEGRAL_SIZE(uint16_t);
+DEFINE_INTEGRAL_SIZE(int16_t);
+DEFINE_INTEGRAL_SIZE(uint32_t);
+DEFINE_INTEGRAL_SIZE(int32_t);
+DEFINE_INTEGRAL_SIZE(uint64_t);
+DEFINE_INTEGRAL_SIZE(int64_t);
+
+template<>
+struct Size<bool> {
+    constexpr size_t operator()(bool) {
+        return 1;
+    }
+};
+
+template<>
+struct Size<char> {
+    constexpr size_t operator()(char) {
+        return 1;
+    }
+};
+
+
 } // namespace Net

src/stream.cc

@@ -7,8 +7,8 @@
 #include <algorithm>
 #include <iostream>
 
-NetworkStream::int_type
-NetworkStream::overflow(NetworkStream::int_type ch) {
+DynamicStreamBuf::int_type
+DynamicStreamBuf::overflow(DynamicStreamBuf::int_type ch) {
     if (!traits_type::eq_int_type(ch, traits_type::eof())) {
         const auto size = data_.size();
         if (size < maxSize_) {
@@ -23,7 +23,7 @@ NetworkStream::overflow(NetworkStream::int_type ch) {
 }
 
 void
-NetworkStream::reserve(size_t size)
+DynamicStreamBuf::reserve(size_t size)
 {
     if (size > maxSize_) size = maxSize_;
     const size_t oldSize = data_.size();

src/stream.h

@@ -113,7 +113,7 @@ private:
     size_t size;
 };
 
-class NetworkStream : public StreamBuf<char> {
+class DynamicStreamBuf : public StreamBuf<char> {
 public:
     struct Buffer {
         Buffer(const void *data, size_t len)
@@ -129,7 +129,7 @@ public:
     typedef typename Base::traits_type traits_type;
     typedef typename Base::int_type int_type;
 
-    NetworkStream(
+    DynamicStreamBuf(
             size_t size,
             size_t maxSize = std::numeric_limits<uint32_t>::max())
         : maxSize_(maxSize)
@@ -137,17 +137,17 @@ public:
         reserve(size);
     }
 
-    NetworkStream(const NetworkStream& other) = delete;
-    NetworkStream& operator=(const NetworkStream& other) = delete;
+    DynamicStreamBuf(const DynamicStreamBuf& other) = delete;
+    DynamicStreamBuf& operator=(const DynamicStreamBuf& other) = delete;
 
-    NetworkStream(NetworkStream&& other)
+    DynamicStreamBuf(DynamicStreamBuf&& other)
        : maxSize_(other.maxSize_)
        , data_(std::move(other.data_)) {
            setp(other.pptr(), other.epptr());
            other.setp(nullptr, nullptr);
     }
 
-    NetworkStream& operator=(NetworkStream&& other) {
+    DynamicStreamBuf& operator=(DynamicStreamBuf&& other) {
         maxSize_ = other.maxSize_;
         data_ = std::move(other.data_);
         setp(other.pptr(), other.epptr());
@@ -159,6 +159,11 @@ public:
         return Buffer(data_.data(), pptr() - &data_[0]);
     }
 
+    void clear() {
+        data_.clear();
+        this->setp(&data_[0], &data_[0] + data_.capacity());
+    }
+
 protected:
     int_type overflow(int_type ch);