Add asynchronous chunked content provider for streaming responses

docs/guides/streaming.md

@@ -0,0 +1,123 @@
+A response body whose size is not known in advance, or which is simply too large
+to fit in memory, can be produced on demand and sent using
+[chunked transfer encoding](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Transfer-Encoding).
+
+Call `#!cpp response.set_chunked_content_provider(<provider>, <mime-type>)` with a
+callable that produces the body one piece at a time. Crow sets
+`Transfer-Encoding: chunked`, omits `Content-Length`, and calls the provider
+repeatedly while writing the response.
+
+## The provider
+
+```cpp
+bool provider(std::string& chunk);
+```
+
+Fill `chunk` with the next piece of the body and return `#!cpp true` while more
+data is coming, `#!cpp false` on the last invocation. Leaving `chunk` empty is
+allowed as an occasional occurrence and sends nothing. A provider that has no
+data yet should block until data is available (or finish the transfer): the
+provider is called again immediately, so returning `#!cpp true` with an empty
+chunk in a tight loop spins the connection thread needlessly.
+
+### Example
+
+```cpp
+auto app = crow::SimpleApp();
+
+CROW_ROUTE(app, "/numbers")
+([](const crow::request&, crow::response& res) {
+    int remaining = 100;
+    res.set_chunked_content_provider(
+      [remaining](std::string& chunk) mutable -> bool {
+          if (remaining == 0)
+              return false;
+          chunk = std::to_string(100 - remaining) + '\n';
+          --remaining;
+          return true;
+      },
+      "text/plain");
+    res.end();
+});
+```
+
+## Aborting the transfer
+
+A provider that discovers midway that the body cannot be finished (the source of
+the data failed, for example) should not let the response end normally: without
+`Content-Length`, the terminating frame is the only thing that tells the client
+the body is complete. For this case the provider can return
+`#!cpp crow::chunk_result` instead of `#!cpp bool`:
+
+```cpp
+crow::chunk_result provider(std::string& chunk);
+```
+
+Return `#!cpp crow::chunk_result::more` while more data is coming,
+`#!cpp crow::chunk_result::done` on the last invocation, or
+`#!cpp crow::chunk_result::abort` to stop the transfer. On `abort` Crow closes
+the connection without sending the terminating frame, so the client sees a
+truncated body instead of a seemingly complete one.
+
+```cpp
+CROW_ROUTE(app, "/file")
+([](const crow::request&, crow::response& res) {
+    auto file = open_source_somehow();
+    res.set_chunked_content_provider(
+      [file](std::string& chunk) mutable -> crow::chunk_result {
+          if (!file->read(chunk))
+              return crow::chunk_result::abort; // reading failed: truncate the body
+          return chunk.empty() ? crow::chunk_result::done : crow::chunk_result::more;
+      },
+      "application/octet-stream");
+    res.end();
+});
+```
+
+## Completion handler
+
+To find out how the transfer ended (to release the source of the data, or to log
+a failure), set a handler that is called once after the body has been written:
+
+```cpp
+res.set_chunked_completion_handler([](bool clean) {
+    if (!clean)
+        CROW_LOG_WARNING << "chunked transfer did not finish cleanly";
+});
+```
+
+`clean` is `#!cpp true` when the provider finished normally
+(`#!cpp crow::chunk_result::done`, or `#!cpp false` from the `bool` provider)
+and every write succeeded; it is `#!cpp false` when the provider aborted or a
+write error occurred. The handler runs on the connection's thread, after the
+last write and before the response is finalized. For a `HEAD` request the
+provider is never called, but the handler still runs (with `clean == true`)
+when the response ends, so it is a reliable place to release the source of
+the data. The handler should not throw: an exception that escapes it is
+logged and swallowed.
+
+## Notes
+
+!!! note
+
+    The provider runs on the connection's thread while the response is being
+    written, so a provider that blocks keeps that thread busy for the whole
+    transfer.
+
+!!! note
+
+    The connection deadline is cancelled for the duration of the transfer.
+    Without that, a body that takes longer to produce than the timeout would be
+    cut short by the connection being closed.
+
+!!! note
+
+    A response to a `HEAD` request never calls the provider: the headers are sent
+    and the body is skipped. The completion handler still runs, with
+    `clean == true`.
+
+!!! note
+
+    A write error in the middle of the transfer is treated like `abort` as far
+    as the connection is concerned: the terminating frame is not sent and the
+    connection is closed instead of being reused for keep-alive.

include/crow/http_connection.h

@@ -275,6 +275,10 @@ namespace crow
             {
                 do_write_static();
             }
+            else if (res.is_chunked_type())
+            {
+                do_write_chunked();
+            }
             else
             {
                 do_write_general();
@@ -331,6 +335,158 @@ namespace crow
             parser_.clear();
         }
 
+        /// Format a chunk size the way chunked transfer encoding wants it: lowercase hex.
+        static std::string chunk_size_to_hex(std::size_t value)
+        {
+            static const char digits[] = "0123456789abcdef";
+            if (value == 0)
+            {
+                return "0";
+            }
+            std::string out;
+            while (value != 0)
+            {
+                out.insert(out.begin(), digits[value & 0xF]);
+                value >>= 4;
+            }
+            return out;
+        }
+
+        void do_write_chunked()
+        {
+            error_code ec;
+            asio::write(adaptor_.socket(), buffers_, ec); // Write the response start / headers
+            if (ec)
+            {
+                CROW_LOG_ERROR << ec << " - buffer write error happened while sending response start / headers. Writing stopped premature.";
+            }
+
+            // Producing the body may take arbitrarily long, so the connection must not be
+            // closed by the deadline while chunks are still on their way.
+            cancel_deadline_timer();
+
+            // do_write_sync() clears the response on every write, so the provider and the
+            // completion handler have to be taken out of it before the loop starts.
+            auto provider = std::move(res.chunk_provider_ex_);
+            res.chunk_provider_ex_ = nullptr;
+            if (!provider && res.chunk_provider_)
+            {
+                provider = [bool_provider = std::move(res.chunk_provider_)](std::string& chunk) {
+                    return bool_provider(chunk) ? response::chunk_result::more : response::chunk_result::done;
+                };
+            }
+            res.chunk_provider_ = nullptr;
+            auto completion_handler = std::move(res.chunk_complete_);
+            res.chunk_complete_ = nullptr;
+
+            std::string chunk;
+            std::string chunk_header;
+            std::vector<asio::const_buffer> buffers{3};
+            auto result = provider ? response::chunk_result::more : response::chunk_result::done;
+            while (result == response::chunk_result::more && !ec)
+            {
+                chunk.clear();
+                // An exception from the provider must not escape into the Asio stack; it is
+                // treated as an abort: no terminating frame, forced close, completion(false).
+                try
+                {
+                    result = provider(chunk);
+                }
+                catch (const std::exception& e)
+                {
+                    CROW_LOG_ERROR << "An uncaught exception occurred in the chunk provider: " << e.what();
+                    result = response::chunk_result::abort;
+                }
+                catch (...)
+                {
+                    CROW_LOG_ERROR << "An uncaught exception occurred in the chunk provider.";
+                    result = response::chunk_result::abort;
+                }
+                if (result == response::chunk_result::abort || chunk.empty())
+                {
+                    continue;
+                }
+
+                chunk_header = chunk_size_to_hex(chunk.size());
+                chunk_header += crlf;
+                buffers[0] = asio::const_buffer(chunk_header.data(), chunk_header.size());
+                buffers[1] = asio::const_buffer(chunk.data(), chunk.size());
+                buffers[2] = asio::const_buffer(crlf.data(), crlf.size());
+                ec = do_write_sync(buffers);
+                if (ec)
+                {
+                    CROW_LOG_ERROR << ec << " - buffer write error happened while sending a chunk. Writing stopped premature.";
+                }
+            }
+
+            // The terminating frame marks the body as complete, so it is only sent when the
+            // provider finished cleanly and every previous write succeeded.
+            if (result == response::chunk_result::done && !ec)
+            {
+                static const std::string last_chunk = "0\r\n\r\n";
+                std::vector<asio::const_buffer> tail{1};
+                tail[0] = asio::const_buffer(last_chunk.data(), last_chunk.size());
+                ec = do_write_sync(tail);
+                if (ec)
+                {
+                    CROW_LOG_ERROR << ec << " - buffer write error happened while sending the last chunk.";
+                }
+            }
+
+            // A write failure leaves the message framing just as incomplete as an explicit
+            // abort (the terminating frame never made it out), so the connection policy is
+            // the same for both: force the close and never reuse the socket for keep-alive.
+            const bool aborted = (result == response::chunk_result::abort);
+            const bool force_close = aborted || static_cast<bool>(ec);
+            if (force_close)
+            {
+                // Close the connection forcefully, without the terminating frame, so that the
+                // client sees a truncated body instead of a seemingly complete one.
+                adaptor_.shutdown_readwrite();
+                adaptor_.close();
+                CROW_LOG_DEBUG << this << " from write (chunked, " << (aborted ? "aborted" : "write error") << ")";
+            }
+
+            if (completion_handler)
+            {
+                // An exception from the handler must not escape into the Asio stack or skip
+                // the cleanup below; it is logged and swallowed.
+                try
+                {
+                    completion_handler(result == response::chunk_result::done && !ec);
+                }
+                catch (const std::exception& e)
+                {
+                    CROW_LOG_ERROR << "An uncaught exception occurred in the chunked completion handler: " << e.what();
+                }
+                catch (...)
+                {
+                    CROW_LOG_ERROR << "An uncaught exception occurred in the chunked completion handler.";
+                }
+            }
+
+            if (close_connection_ && !force_close)
+            {
+                adaptor_.shutdown_readwrite();
+                adaptor_.close();
+                CROW_LOG_DEBUG << this << " from write (chunked)";
+            }
+
+            res.end();
+            res.clear();
+            buffers_.clear();
+            parser_.clear();
+
+            // The deadline was cancelled for the duration of the transfer, so a kept-alive
+            // connection has to be put back into reading state explicitly.
+            if (!force_close && !close_connection_ && need_to_start_read_after_complete_)
+            {
+                need_to_start_read_after_complete_ = false;
+                start_deadline();
+                do_read();
+            }
+        }
+
         void do_write_general()
         {
             error_code ec;