Skip to content

Commit e2dfd66

Browse files
xiadnoringxiadnoring
committed
UPD | chunked
1 parent 14e00a6 commit e2dfd66

20 files changed

+339
-159
lines changed

‎include/ManapiHttpRequest.hpp

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,9 @@
1818
namespace manapi::net::http {
1919
class request {
2020
public:
21+
using onrecv_sync_cb = std::move_only_function<ssize_t(const char *buffer, ssize_t size, bool fin)>;
22+
using onrecv_async_cb = std::move_only_function<manapi::future<ssize_t>(const char *buffer, ssize_t size, bool fin)>;
23+
2124
request(std::unique_ptr<manapi::net::http::manapi_socket_information> ip_data, manapi::net::http::request_data_t *request_data, manapi::net::worker::shared_conn *conn, worker::shared_worker worker, const http_handler_function *handler);
2225

2326
~request();
@@ -44,9 +47,9 @@ namespace manapi::net::http {
4447

4548
future<> form (formdata_recv::onparam_cb_t cb);
4649

47-
future<void> callback_sync (std::move_only_function<ssize_t(const char *buffer, ssize_t size)> callback);
50+
future<void> callback_sync (onrecv_sync_cb callback);
4851

49-
future<void> callback_async (std::move_only_function<manapi::future<ssize_t>(const char *buffer, ssize_t size)> callback);
52+
future<void> callback_async (onrecv_async_cb callback);
5053

5154
future<void> file (std::string filepath);
5255

@@ -72,8 +75,8 @@ namespace manapi::net::http {
7275

7376
[[nodiscard]] bool propagation () const;
7477
private:
75-
static future<void> read_body_ (worker::base *worker, worker::shared_conn *conn, request_data_t *req, std::move_only_function<ssize_t(const char *, ssize_t )> handler);
76-
static future<void> read_async_body_ (worker::base *worker, worker::shared_conn *conn, request_data_t *req, std::move_only_function<manapi::future<ssize_t>(const char *, ssize_t )> handler);
78+
static future<void> read_body_ (worker::base *worker, worker::shared_conn *conn, request_data_t *req, onrecv_sync_cb handler);
79+
static future<void> read_async_body_ (worker::base *worker, worker::shared_conn *conn, request_data_t *req, onrecv_async_cb handler);
7780

7881
std::unique_ptr<std::map<std::string, std::string>> get_params_;
7982

‎include/ManapiHttpResponse.hpp

Lines changed: 15 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -20,19 +20,18 @@ namespace manapi::net::http {
2020
};
2121

2222
struct custom_data_deleter_t {
23-
void operator()(custom_data_t *n) {
24-
if (n && n->clean)
25-
n->clean(n->src);
26-
}
23+
void operator()(custom_data_t *n);
2724
};
2825

2926
class response {
3027
public:
3128
using resp_callback_sync = std::move_only_function<ssize_t(char *buffer, ssize_t size, bool&)>;
3229
using resp_callback_async = std::move_only_function<manapi::future<ssize_t>(char *buffer, ssize_t size, bool&)>;
30+
using resp_stream_cb = std::move_only_function<manapi::future<ssize_t>(const void *buffer, ssize_t size, bool)>;
31+
using resp_stream = std::move_only_function<manapi::future<>(resp_stream_cb cb)>;
3332
using resp_proxy_setup_cb = std::move_only_function<void(class manapi::net::fetch &)>;
3433

35-
response (manapi::net::http::request_data_t *request_data, int status, http::config *config);
34+
response (manapi::net::http::request_data_t *request_data, int status, http::config *config, std::unique_ptr<http::request> req);
3635

3736
~response ();
3837

@@ -59,12 +58,14 @@ namespace manapi::net::http {
5958
#ifdef MANAPIHTTP_FETCH_SUPPORT
6059
void proxy (std::string url);
6160

62-
void proxy (std::string url, std::move_only_function<void(class fetch &)> cb);
61+
void proxy (std::string url, resp_proxy_setup_cb cb);
6362
#endif
6463

65-
void callback_sync (std::move_only_function<ssize_t(char *, ssize_t, bool&)> cb);
64+
void callback_sync (resp_callback_sync cb);
6665

67-
void callback_async (std::move_only_function<manapi::future<ssize_t>(char *, ssize_t, bool&)> cb);
66+
void callback_async (resp_callback_async cb);
67+
68+
void callback_stream (resp_stream cb);
6869

6970
[[nodiscard]] int status_code () const;
7071

@@ -126,7 +127,11 @@ namespace manapi::net::http {
126127

127128
resp_callback_sync &callback_sync();
128129

130+
resp_stream &callback_stream();
131+
129132
request_data_t *request_data ();
133+
134+
http::request *req ();
130135
private:
131136
void check_type_ (int type);
132137

@@ -156,6 +161,8 @@ namespace manapi::net::http {
156161

157162
std::unique_ptr<std::map<std::string, std::string>> replacers_;
158163

164+
std::unique_ptr<http::request> req_;
165+
159166
#ifdef MANAPIHTTP_FETCH_SUPPORT
160167
std::unique_ptr<std::move_only_function<void(class manapi::net::fetch &)>> proxy_setup;
161168
#endif

‎include/components/FormData.hpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ namespace manapi::net {
2121

2222
typedef std::move_only_function<ondata_cb_t(std::string name)> onparam_cb_t;
2323

24-
typedef manapi::future<> (*onrecv_cb_t)(worker::base *worker, worker::shared_conn *conn, http::request_data_t *req, std::move_only_function<manapi::future<ssize_t>(const char *, ssize_t )> handler);
24+
typedef manapi::future<> (*onrecv_cb_t)(worker::base *worker, worker::shared_conn *conn, http::request_data_t *req, std::move_only_function<manapi::future<ssize_t>(const char *, ssize_t , bool fin)> handler);
2525

2626
formdata_recv (onrecv_cb_t onrecv_cb, manapi::net::worker::base *worker, worker::shared_conn *conn, http::request_data_t *req);
2727

‎include/http/HTTPv2.hpp

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -29,10 +29,10 @@ namespace manapi::net::http {
2929
HTTP2_STREAM_WANT_READ = manapi::ev::READ,
3030
HTTP2_STREAM_WANT_WRITE = manapi::ev::WRITE,
3131
HTTP2_STREAM_CLOSED = manapi::ev::DISCONNECT,
32-
HTTP2_STREAM_RECV_END = 8,
33-
HTTP2_STREAM_SEND_END = 16,
34-
HTTP2_STREAM_REMOVED = 32,
35-
HTTP2_STREAM_PRIORITY_INCR = 64,
32+
HTTP2_STREAM_PRIORITY_INCR = 8,
33+
HTTP2_STREAM_REMOVED = 16,
34+
HTTP2_STREAM_SEND_END = 32,
35+
HTTP2_STREAM_RECV_END = worker::base::CONN_RECV_END,
3636
HTTP2_STREAM_PRIORITY_LOCKED = 128,
3737
HTTP2_STREAM_IO_WAITING = 256
3838
};

‎include/worker/TCP.hpp

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,9 @@ namespace manapi::net::worker {
8080

8181
void flush_read_ (const shared_conn &conn, connection_interface *data);
8282

83-
void tcp_handle_read_data (const shared_conn &conn, connection_interface *data, int flags, const char *buffer, ssize_t size, ibuffpool_t *p);
83+
void tcp_handle_read_chunked (const shared_conn &conn, connection_interface *data, int flags, const char *buffer, ssize_t size, ibuffpool_t *p);
84+
85+
virtual void tcp_handle_read_data (const shared_conn &conn, connection_interface *data, int flags, const char *buffer, ssize_t size, ibuffpool_t *p);
8486

8587
void update_limit_rate ();
8688

‎include/worker/TLS.hpp

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ namespace manapi::net::worker {
3535

3636
ssize_t sync_write(const shared_conn &conn, const void *buff, ssize_t size, bool finish) override;
3737

38-
int event_flags(const shared_conn & conn, int flags) override;
38+
int event_flags(const shared_conn & conn, int flags) noexcept(true) override;
3939

4040
protected:
4141
int ssl_error_none_, ssl_error_want_read_,

‎main.cpp

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -204,7 +204,7 @@ int main () {
204204
manapi::net::hash::SHA256 hash{};
205205
hash.init();
206206
try {
207-
co_await req.callback_async([&result, &hash] (const char *buffer, ssize_t size)
207+
co_await req.callback_async([&result, &hash] (const char *buffer, ssize_t size, bool fin)
208208
-> manapi::future<ssize_t> {
209209
hash.update(reinterpret_cast<const uint8_t *>(buffer), size);
210210
result += size;
@@ -226,6 +226,16 @@ int main () {
226226
co_return resp.text(std::format("{} : {}", result, b));
227227
});
228228

229+
router.POST ("/upload2", [] (manapi::net::http::request &req, manapi::net::http::response &resp)
230+
-> manapi::future<> {
231+
co_return resp.callback_stream([&req] (manapi::net::http::response::resp_stream_cb cb) -> manapi::future<> {
232+
co_await req.callback_async([cb = std::move(cb)] (const char *buffer, ssize_t size, bool fin) mutable
233+
-> manapi::future<ssize_t> {
234+
co_return co_await cb (buffer, size, fin);
235+
});
236+
});
237+
});
238+
229239
router.GET("/download", [] (manapi::net::http::request &req, manapi::net::http::response &resp)
230240
-> manapi::future<> {
231241
co_return resp.file("/home/Timur/Desktop/WorkSpace/oneworld/test.ISO");

‎src/ManapiHttpRequest.cpp

Lines changed: 60 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -94,14 +94,14 @@ manapi::future<std::string> manapi::net::http::request::text() {
9494
size_t j = 0;
9595
//size_t socket_block_size = http_server->get_socket_block_size();
9696

97-
co_await this->read_body_(this->worker_.get(), this->conn_, this->request_data,[&body, &j] (const char *data, ssize_t size) -> ssize_t {
97+
co_await this->read_body_(this->worker_.get(), this->conn_, this->request_data,[&body, &j] (const char *data, ssize_t size, bool fin) -> ssize_t {
9898
memcpy (body.data() + j, data, size);
9999
j += size;
100100
return size;
101101
});
102102
}
103103
else {
104-
co_await this->read_body_(this->worker_.get(), this->conn_, this->request_data,[&body] (const char *data, ssize_t size)
104+
co_await this->read_body_(this->worker_.get(), this->conn_, this->request_data,[&body] (const char *data, ssize_t size, bool fin)
105105
-> ssize_t { body.append(data, size); return size; });
106106
}
107107

@@ -115,14 +115,14 @@ manapi::future<manapi::json> manapi::net::http::request::json()
115115

116116
if (post_mask) {
117117
json_builder builder = json_builder (*post_mask);
118-
co_await read_body_(this->worker_.get(), this->conn_, this->request_data,[&builder] (const char *data, ssize_t size)
118+
co_await read_body_(this->worker_.get(), this->conn_, this->request_data,[&builder] (const char *data, ssize_t size, bool fin)
119119
-> ssize_t { builder << std::string_view (data, size); return size; });
120120

121121
co_return std::move(builder.get());
122122
}
123123
else {
124124
json_builder builder = json_builder ();
125-
co_await read_body_(this->worker_.get(), this->conn_, this->request_data,[&builder] (const char *data, ssize_t size)
125+
co_await read_body_(this->worker_.get(), this->conn_, this->request_data,[&builder] (const char *data, ssize_t size, bool fin)
126126
-> ssize_t { builder << std::string_view (data, size); return size; });
127127

128128
co_return std::move(builder.get());
@@ -134,11 +134,11 @@ manapi::future<> manapi::net::http::request::form (formdata_recv::onparam_cb_t c
134134
co_await fdata.get(std::move(cb));
135135
}
136136

137-
manapi::future<> manapi::net::http::request::callback_sync( std::move_only_function<ssize_t(const char *buffer, ssize_t size)> callback) {
137+
manapi::future<> manapi::net::http::request::callback_sync(onrecv_sync_cb callback) {
138138
co_return co_await this->read_body_(this->worker_.get(), this->conn_, this->request_data,std::move(callback));
139139
}
140140

141-
manapi::future<> manapi::net::http::request::callback_async(std::move_only_function<manapi::future<ssize_t>(const char *buffer, ssize_t size)> callback) {
141+
manapi::future<> manapi::net::http::request::callback_async(onrecv_async_cb callback) {
142142
co_return co_await this->read_async_body_(this->worker_.get(), this->conn_, this->request_data,std::move(callback));
143143
}
144144

@@ -153,7 +153,7 @@ manapi::future<> manapi::net::http::request::file(std::string filepath) {
153153
std::exception_ptr err{nullptr};
154154

155155
try {
156-
co_await this->read_async_body_(this->worker_.get(), this->conn_, this->request_data, [&] (const char *data, ssize_t size)
156+
co_await this->read_async_body_(this->worker_.get(), this->conn_, this->request_data, [&] (const char *data, ssize_t size, bool fin)
157157
-> manapi::future<ssize_t> { return f.write(data, size); });
158158
}
159159
catch (...) {
@@ -240,7 +240,7 @@ bool manapi::net::http::request::propagation() const {
240240
return this->flags & internal::REQUEST_FLAG_IS_PROPAGATION;
241241
}
242242

243-
manapi::future<void> manapi::net::http::request::read_body_(worker::base *worker, worker::shared_conn *conn, request_data_t *req, std::move_only_function<ssize_t(const char *, ssize_t)> handler) {
243+
manapi::future<void> manapi::net::http::request::read_body_(worker::base *worker, worker::shared_conn *conn, request_data_t *req, onrecv_sync_cb handler) {
244244
using promise = manapi::async::promise<void, std::false_type>;
245245
std::unique_ptr<worker::worker_watcher_cb> prev{nullptr};
246246
int pflags;
@@ -262,16 +262,36 @@ manapi::future<void> manapi::net::http::request::read_body_(worker::base *worker
262262
if (flags & ev::READ) {
263263
try {
264264
ssize_t size;
265-
if (req->body_size >= 0)
265+
bool flg = false;
266+
267+
if (req->body_size >= 0) {
266268
size = std::min(req->body_size, static_cast<ssize_t> (nsize));
269+
if (req->body_size == size)
270+
flg = true;
271+
}
267272
else
268273
size = static_cast<ssize_t> (nsize);
269274

275+
if (flags & worker::base::CONN_RECV_END) {
276+
flags ^= worker::base::CONN_RECV_END;
277+
flg = true;
278+
279+
if (!nsize) {
280+
if (handler (nullptr, 0, flg) < 0) {
281+
reject (std::make_exception_ptr(RETHROW_MANAPIHTTP_EXCEPTION (
282+
manapi::ERR_INVALID_ARGUMENT, manapi::error::default_msgs[manapi::error::ERRMSG_CUSTOM_CALLBACK_ERR1], "invalid result")));
283+
goto finish;
284+
}
285+
resolve();
286+
goto finish;
287+
}
288+
}
289+
270290
ssize_t rhs = 0;
271291
while (rhs < size) {
272292
auto const copy = size - rhs;
273293

274-
auto const res = handler (buffer + rhs, copy);
294+
auto const res = handler (buffer + rhs, copy, flg);
275295
if (res >= 0) {
276296
if (copy > res) {
277297
reject(std::make_exception_ptr(RETHROW_MANAPIHTTP_EXCEPTION(
@@ -343,7 +363,7 @@ manapi::future<void> manapi::net::http::request::read_body_(worker::base *worker
343363
}
344364
}
345365

346-
manapi::future<> manapi::net::http::request::read_async_body_(worker::base *worker, worker::shared_conn *conn, request_data_t *req, std::move_only_function<manapi::future<ssize_t>(const char *, ssize_t)> handler) {
366+
manapi::future<> manapi::net::http::request::read_async_body_(worker::base *worker, worker::shared_conn *conn, request_data_t *req, onrecv_async_cb handler) {
347367
using promise = manapi::async::promise<void, std::false_type>;
348368
using handler_t = decltype(handler);
349369

@@ -402,21 +422,41 @@ manapi::future<> manapi::net::http::request::read_async_body_(worker::base *work
402422
ctx_cb.worker->event_flags(conn, 0);
403423
ctx_cb.cnt++;
404424
manapi::async::run (manapi::async::invoke(
405-
[] (const worker::shared_conn & conn, worker::ibuffpool_t p, const char * buffer, ssize_t nsize, ctx_cb_t_ *ctx_cb) -> manapi::future<> {
425+
[] (const worker::shared_conn & conn, worker::ibuffpool_t p, const char * buffer, ssize_t nsize, ctx_cb_t_ *ctx_cb, int flags) -> manapi::future<> {
406426
try {
407427

408428
ssize_t size;
409-
410-
if (ctx_cb->req->body_size >= 0)
429+
bool flg = false;
430+
if (ctx_cb->req->body_size >= 0) {
411431
size = std::min(ctx_cb->req->body_size, static_cast<ssize_t> (nsize));
412-
else
432+
433+
if (ctx_cb->req->body_size == size)
434+
flg = true;
435+
}
436+
else {
413437
size = static_cast<ssize_t> (nsize);
438+
}
439+
440+
if (flags & worker::base::CONN_RECV_END) {
441+
flags ^= worker::base::CONN_RECV_END;
442+
flg = true;
443+
444+
if (!nsize) {
445+
if (co_await ctx_cb->handler (nullptr, 0, flg) < 0) {
446+
ctx_cb->reject (std::make_exception_ptr(RETHROW_MANAPIHTTP_EXCEPTION (
447+
manapi::ERR_INVALID_ARGUMENT, manapi::error::default_msgs[manapi::error::ERRMSG_CUSTOM_CALLBACK_ERR1], "invalid result")));
448+
goto finish;
449+
}
450+
ctx_cb->resolve();
451+
goto finish;
452+
}
453+
}
414454

415455
ssize_t rhs = 0;
416456
while (rhs < size) {
417457
auto const copy = size - rhs;
418458

419-
auto const res = co_await ctx_cb->handler (buffer + rhs, copy);
459+
auto const res = co_await ctx_cb->handler (buffer + rhs, copy, flg);
420460
if (res >= 0) {
421461
if (copy > res) {
422462
ctx_cb->reject(std::make_exception_ptr(RETHROW_MANAPIHTTP_EXCEPTION(
@@ -448,26 +488,27 @@ manapi::future<> manapi::net::http::request::read_async_body_(worker::base *work
448488
ctx_cb->resolve();
449489
goto finish;
450490
}
491+
451492
}
452493
catch (...) {
453494
ctx_cb->reject (std::current_exception());
454495
goto finish;
455496
}
456497

457-
ctx_cb->worker->waiting(conn, true);
458498
ctx_cb->worker->event_flags(conn, ev::READ);
459499
ctx_cb->cnt--;
460500
ctx_cb->mx.unlock();
501+
461502
co_return;
462503

463504
finish: {
464-
auto &ctx_cb_ = ctx_cb;
505+
auto const ctx_cb_ = ctx_cb;
465506
ctx_cb_->worker->event_flags(conn, 0);
466507
ctx_cb_->worker->event_on(conn, nullptr);
467508
ctx_cb_->cnt--;
468509
ctx_cb_->mx.unlock();
469510
}
470-
}, ctx_cb.conn, std::move(*p), buffer, nsize, &ctx_cb));
511+
}, ctx_cb.conn, std::move(*p), buffer, nsize, &ctx_cb, flags));
471512
}
472513
else if (flags & worker::base::CONN_RECV_END) {
473514
ctx_cb.resolve();

0 commit comments

Comments
 (0)