FIX | some bugs

Author: xiadnoring

main.cpp

@@ -261,13 +261,13 @@ int main () {
 
         router.POST ("/upload2", [] (manapi::net::http::request &req, manapi::net::http::response &resp)
             -> manapi::future<> {
-            resp.header(manapi::net::http::HEADER.CONTENT_LENGTH, req.header(manapi::net::http::HEADER.CONTENT_LENGTH));
+            //resp.header(manapi::net::http::HEADER.CONTENT_LENGTH, req.header(manapi::net::http::HEADER.CONTENT_LENGTH));
             co_return resp.callback_stream([&resp, &req] (manapi::net::http::response::resp_stream_cb cb) -> manapi::future<> {
-                std::size_t sum = 0;
-                co_await req.callback_async([&sum, cb = std::move(cb)] (const char *buffer, ssize_t size, bool fin) mutable
+
+                co_await req.callback_async([cb = std::move(cb)] (const char *buffer, ssize_t size, bool fin) mutable
                     -> manapi::future<ssize_t> {
-                    sum += size;
-                    std::cout << sum << " " << size << " " << fin << "\n";
+                    //sum += size;
+                    //std::cout << sum << " " << size << " " << fin << "\n";
                     co_return co_await cb (buffer, size, fin);
                 });
             });

src/components/FormData.cpp

@@ -121,7 +121,7 @@ manapi::future<void> manapi::net::formdata_recv::get(onparam_cb_t cb) {
             case CONTENT_TYPE_MULTIPART_FORM_DATA: {
                 co_await this->onrecv_cb_ (this->worker_, this->conn_, this->req_,
                     [this] (const char *buffer, ssize_t size, bool fin) -> manapi::future<ssize_t> {
-                        return this->onrecv_multipart_(buffer, size);
+                        co_return co_await this->onrecv_multipart_(buffer, size);
                 });
                 break;
             }
@@ -480,7 +480,7 @@ manapi::future<ssize_t> manapi::net::formdata_recv::onrecv_multipart_(const char
                         if (!n1) {
                             auto const beyond = (pos - n1);
                             auto const copy = beyond - this->ctx_->n2;
-
+                            assert((copy >= 0));
                             if (this->ondata_cb_) {
                                 ssize_t res = 0;
 

src/http/base_http.cpp

@@ -346,9 +346,8 @@ manapi::future<void> manapi::net::http::internal::send_response_proxy(uq_handle_
         auto rhs = co_await cdata->worker->fwrite (cdata->conn, buffer,
             size, content_length <= size);
 
-        if (rhs < 0) {
+        if (rhs < 0)
             co_return -1;
-        }
 
         content_length -= rhs;
         co_return rhs;
@@ -452,22 +451,28 @@ void manapi::net::http::internal::send_response_sync_cb(uq_handle_data_t cdata,
             }
 
             if (result && *result >= 0) {
-                const ssize_t reserved = res->config()->buffer_size;
                 auto cb_sync = std::make_unique<http::response::resp_callback_sync>(std::move(res->callback_sync()));
-                manapi::async::run ([res = std::move(res), cb_sync = std::move(cb_sync), reserved, cdata = std::move(cdata)] () mutable
+                manapi::async::run ([res = std::move(res), cb_sync = std::move(cb_sync), cdata = std::move(cdata)] () mutable
                     -> manapi::future<> {
-                        // TODO: speed up
-                        std::string buffer;
-                        buffer.reserve(reserved);
+                        auto buffer = manapi::async::current()->memory_fabric().buffer (res->config()->buffer_size);
 
                         bool finish = false;
+                        std::size_t cursor = 0;
 
                         while (!finish) {
-                            auto rhs = cb_sync->operator()(buffer.data(), reserved, finish);
+                            auto rhs = cb_sync->operator()(buffer.data() + cursor, buffer.size() - cursor, finish);
                             if (rhs < 0) {
                                 THROW_MANAPIHTTP_EXCEPTION2(ERR_INVALID_ARGUMENT, "The callback returned an invalid length");
                             }
-                            co_await cdata->worker->fwrite(cdata->conn, buffer.data(), rhs, finish);
+                            rhs = co_await cdata->worker->write(cdata->conn, buffer.data() + cursor, rhs, finish);
+
+                            if (rhs <= 0)
+                                co_return;
+
+                            cursor += rhs;
+
+                            if (buffer.size() == cursor)
+                                cursor = 0;
                         }
                     });
             }
@@ -526,25 +531,27 @@ void manapi::net::http::internal::send_response_async_cb(uq_handle_data_t cdata,
             }
 
             if (result && *result >= 0) {
-                auto reserved = res->config()->buffer_size;
                 auto cb_async = std::make_unique<http::response::resp_callback_async>(std::move(res->callback_async()));
-                manapi::async::run ([reserved, cb_async = std::move(cb_async), cdata = std::move(cdata)] () mutable
+                manapi::async::run ([res = std::move(res), cb_async = std::move(cb_async), cdata = std::move(cdata)] () mutable
                     -> manapi::future<> {
-                        // TODO: speed up
-                        std::string buffer;
-                        buffer.reserve(reserved);
-
+                        auto buffer = manapi::async::current()->memory_fabric().buffer(res->config()->buffer_size);
 
+                        std::size_t cursor = 0;
                         bool finish = false;
 
                         while (!finish) {
-                            auto rhs = co_await cb_async->operator()(buffer.data(), reserved, finish);
-                            if (rhs < 0) {
+                            auto rhs = co_await cb_async->operator()(buffer.data() + cursor, buffer.size() - cursor, finish);
+                            if (rhs < 0)
                                 THROW_MANAPIHTTP_EXCEPTION2(ERR_INVALID_ARGUMENT, "The callback returned an invalid length");
-                            }
-                            if (rhs != co_await cdata->worker->fwrite(cdata->conn, buffer.data(), rhs, finish)) {
-                                THROW_MANAPIHTTP_EXCEPTION2(ERR_ABORTED, "Write failed");
-                            }
+
+                            rhs = co_await cdata->worker->write(cdata->conn, buffer.data() + cursor, rhs, finish);
+                            if (rhs <= 0)
+                                co_return;
+
+                            cursor += rhs;
+
+                            if (cursor == buffer.size())
+                                cursor = 0;
                         }
                     });
             }

src/worker/TCP.cpp

@@ -507,6 +507,8 @@ int manapi::net::worker::TCP::event_flags(const shared_conn & conn, int flags) {
 
         flush_read_ (conn, data);
 
+    }
+    if (status & ev::READ) {
         if (!data->watcher->is_active()) {
             assert(!data->watcher->read_start());
         }

src/worker/TLS.cpp

@@ -175,7 +175,8 @@ int manapi::net::worker::TLS::event_flags(const shared_conn & conn, int flags) n
 
         flush_read_ (conn, data);
 
-        if (!(status & (CONN_CLOSED|CONN_REMOVED)) && !data->watcher->is_active()) {
+        if ((status & (CONN_READ|CONN_CLOSED|CONN_REMOVED)) == CONN_READ
+            && !data->watcher->is_active()) {
             data->watcher->read_start();
         }
     }