iostream/http: Fix output_stream::write(temporary_buffer) overload

Previously, calling the 'write(temporary_buffer<char>)' overload on an
output_stream<char> obtained from http::reply::write_body
would cause an abort.

This commit fixes this by providing a helper function (in the data_sink_impl
base class) named 'fallback_put', which classes that inherit from this
base class can use to implement the 'put(net::packet)' overload.
Also, this commit lets http_chunked_data_sink_impl and
http_content_length_data_sink_impl make use of this helper.

fallback_put(net::packet) works simply by extracting the temporary_buffers
contained in the packet, and calling the 'put(temporary_buffer<char>)'
overload for each temporary_buffer.

Note that we create this 'fallback_put' function instead of defining the
virtual member 'put(net::packet)' function in the data_sink_impl class,
to avoid 'put(net::packet)' and 'put(temporary_buffer<char>)' calling
each other leading to infinite recursion.

This commit also adds a unit test to tests/unit/httpd_test.cc

Thanks to Pavel (xemul) for his help.

Fixes #1701

include/seastar/core/iostream.hh

@@ -36,6 +36,7 @@
 #pragma once
 
 #include <seastar/core/future.hh>
+#include <seastar/core/coroutine.hh>
 #include <seastar/core/temporary_buffer.hh>
 #include <seastar/core/scattered_message.hh>
 #include <seastar/util/std-compat.hh>
@@ -143,6 +144,20 @@ public:
     virtual void on_batch_flush_error() noexcept {
         assert(false && "Data sink must implement on_batch_flush_error() method");
     }
+
+protected:
+    // This is a helper function that classes that inherit from data_sink_impl
+    // can use to implement the put overload for net::packet.
+    // Unfortunately, we currently cannot define this function as
+    // 'virtual future<> put(net::packet)', because we would get infinite
+    // recursion between this function and
+    // 'virtual future<> put(temporary_buffer<char>)'.
+    future<> fallback_put(net::packet data) {
+        auto buffers = data.release();
+        for (temporary_buffer<char>& buf : buffers) {
+            co_await this->put(std::move(buf));
+        }
+    }
 };
 
 class data_sink {

src/http/common.cc

@@ -119,7 +119,9 @@ class http_chunked_data_sink_impl : public data_sink_impl {
 public:
     http_chunked_data_sink_impl(output_stream<char>& out) : _out(out) {
     }
-    virtual future<> put(net::packet data)  override { abort(); }
+    virtual future<> put(net::packet data) override {
+        return data_sink_impl::fallback_put(std::move(data));
+    }
     using data_sink_impl::put;
     virtual future<> put(temporary_buffer<char> buf) override {
         if (buf.size() == 0) {
@@ -166,7 +168,9 @@ class http_content_length_data_sink_impl : public data_sink_impl {
         // at the very beginning, 0 bytes were written
         _bytes_written = 0;
     }
-    virtual future<> put(net::packet data)  override { abort(); }
+    virtual future<> put(net::packet data) override {
+        return data_sink_impl::fallback_put(std::move(data));
+    }
     using data_sink_impl::put;
     virtual future<> put(temporary_buffer<char> buf) override {
         if (buf.size() == 0 || _bytes_written == _limit) {

tests/unit/httpd_test.cc

@@ -781,6 +781,22 @@ SEASTAR_TEST_CASE(json_stream) {
     });
 }
 
+// See issue https://github.com/scylladb/seastar/issues/1701
+SEASTAR_TEST_CASE(dont_abort) {
+    return test_client_server::run_test(
+            [](seastar::output_stream<char>&& stream_) -> future<> {
+        auto stream = std::move(stream_);
+        co_await stream.write(seastar::temporary_buffer<char>{3});
+        co_await stream.close();
+    }
+    , [](size_t s, http_consumer& h) {
+        BOOST_REQUIRE_EQUAL(h._size, 3);
+        return false;
+    });
+
+}
+
+
 class json_test_handler : public handler_base {
     std::function<future<>(output_stream<char> &&)> _write_func;
 public: