Add mg_http_stream_body() helper for generic streaming uploads

mongoose.c

@@ -4730,6 +4730,47 @@ void mg_http_start_ota(struct mg_connection *c, struct mg_http_message *hm,
   mg_call(c, MG_EV_READ, &c->recv.len);
 }
 
+void mg_http_stream_body(struct mg_connection *c, int ev, void *ev_data,
+                         struct mg_str uri_pattern,
+                         void (*cb)(struct mg_http_message *, struct mg_str *,
+                                    void **user_data)) {
+  struct mg_http_message *hm = (struct mg_http_message *) ev_data;
+
+  // Catch upload requests early, without buffering whole body
+  // When we receive MG_EV_HTTP_HDRS event, that means we've received all
+  // HTTP headers but not necessarily full HTTP body
+  if (ev == MG_EV_HTTP_HDRS &&
+      (mg_strcmp(hm->method, mg_str("POST")) == 0 ||
+       mg_strcmp(hm->method, mg_str("PUT")) == 0) &&
+      hm->body.len != (size_t) ~0 && mg_match(hm->uri, uri_pattern, NULL)) {
+    c->pfn = NULL;  // Silence HTTP protocol handler, we'll use MG_EV_READ
+    c->pfn_data = (void *) (ptrdiff_t) hm->body.len;  // Record expected len
+    cb(hm, NULL, &c->fn_data);  // Call handler before deleting headers
+    if (mg_http_get_header(hm, "Expect") != NULL) {
+      mg_http_reply(c, 100, NULL, "");  // If curl expects, we continue
+    }
+    mg_iobuf_del(&c->recv, 0, hm->head.len);  // Delete HTTP headers
+  }
+
+  if (c->pfn == NULL && ev != MG_EV_OPEN) {
+    if (c->pfn_data != NULL && c->recv.len > 0) {
+      // Got some uploaded data. Feed by 512-byte aligned chunks, for OTA
+      size_t left = (size_t) (ptrdiff_t) c->pfn_data, r = c->recv.len;
+      size_t aligned = r < left ? MG_ROUND_DOWN(r, 512) : left;
+      struct mg_str data = mg_str_n((char *) c->recv.buf, aligned);
+      if (data.len) cb(NULL, &data, &c->fn_data);
+      c->pfn_data = (void *) (ptrdiff_t) (left - aligned);
+      c->recv.len -= aligned;
+    }
+    if (c->pfn_data == NULL && c->is_draining == 0) {
+      // Finished upload
+      mg_http_reply(c, 200, NULL, "ok\n");
+      c->is_draining = 1;
+      cb(NULL, NULL, &c->fn_data);
+    }
+  }
+}
+
 static bool is_hex_digit(int c) {
   return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') ||
          (c >= 'A' && c <= 'F');

mongoose.h

@@ -2835,6 +2835,30 @@ size_t mg_http_next_multipart(struct mg_str, size_t, struct mg_http_part *);
 // Returns the HTTP status code from a parsed response message (e.g. 200, 404).
 int mg_http_status(const struct mg_http_message *hm);
 
+// Streams a POST/PUT body to a callback as it arrives, without buffering it.
+// Call it from your connection handler on every event.
+//
+// Callback cb(hm, data, &user_data) is called:
+//   - at start: hm is set, data is NULL. Open the file, set *user_data.
+//   - per chunk: hm is NULL, data is set. Chunks are multiples of 512 bytes,
+//     except the last one.
+//   - at end: both NULL. Close the file.
+// After the last chunk, "200 ok" is sent and the connection is drained.
+// Example:
+//   mg_http_stream_body(c, ev, ev_data, mg_str("/fs/*"), upload);
+// Full example:
+//   tutorials/http/file-upload-single-post
+// Related APIs:
+//   mg_http_start_upload(), mg_http_start_ota()
+// Notes:
+//   Requires Content-Length, chunked bodies are not streamed. user_data is
+//   c->fn_data. While streaming, c->pfn is NULL, so guard your own
+//   MG_EV_HTTP_MSG handling with c->pfn != NULL.
+void mg_http_stream_body(struct mg_connection *c, int ev, void *ev_data,
+                         struct mg_str uri_pattern,
+                         void (*upload_fn)(struct mg_http_message *,
+                                           struct mg_str *, void **user_data));
+
 
 void mg_http_serve_ssi(struct mg_connection *c, const char *root,
                        const char *fullpath);

src/http.c

@@ -1117,6 +1117,47 @@ void mg_http_start_ota(struct mg_connection *c, struct mg_http_message *hm,
   mg_call(c, MG_EV_READ, &c->recv.len);
 }
 
+void mg_http_stream_body(struct mg_connection *c, int ev, void *ev_data,
+                         struct mg_str uri_pattern,
+                         void (*cb)(struct mg_http_message *, struct mg_str *,
+                                    void **user_data)) {
+  struct mg_http_message *hm = (struct mg_http_message *) ev_data;
+
+  // Catch upload requests early, without buffering whole body
+  // When we receive MG_EV_HTTP_HDRS event, that means we've received all
+  // HTTP headers but not necessarily full HTTP body
+  if (ev == MG_EV_HTTP_HDRS &&
+      (mg_strcmp(hm->method, mg_str("POST")) == 0 ||
+       mg_strcmp(hm->method, mg_str("PUT")) == 0) &&
+      hm->body.len != (size_t) ~0 && mg_match(hm->uri, uri_pattern, NULL)) {
+    c->pfn = NULL;  // Silence HTTP protocol handler, we'll use MG_EV_READ
+    c->pfn_data = (void *) (ptrdiff_t) hm->body.len;  // Record expected len
+    cb(hm, NULL, &c->fn_data);  // Call handler before deleting headers
+    if (mg_http_get_header(hm, "Expect") != NULL) {
+      mg_http_reply(c, 100, NULL, "");  // If curl expects, we continue
+    }
+    mg_iobuf_del(&c->recv, 0, hm->head.len);  // Delete HTTP headers
+  }
+
+  if (c->pfn == NULL && ev != MG_EV_OPEN) {
+    if (c->pfn_data != NULL && c->recv.len > 0) {
+      // Got some uploaded data. Feed by 512-byte aligned chunks, for OTA
+      size_t left = (size_t) (ptrdiff_t) c->pfn_data, r = c->recv.len;
+      size_t aligned = r < left ? MG_ROUND_DOWN(r, 512) : left;
+      struct mg_str data = mg_str_n((char *) c->recv.buf, aligned);
+      if (data.len) cb(NULL, &data, &c->fn_data);
+      c->pfn_data = (void *) (ptrdiff_t) (left - aligned);
+      c->recv.len -= aligned;
+    }
+    if (c->pfn_data == NULL && c->is_draining == 0) {
+      // Finished upload
+      mg_http_reply(c, 200, NULL, "ok\n");
+      c->is_draining = 1;
+      cb(NULL, NULL, &c->fn_data);
+    }
+  }
+}
+
 static bool is_hex_digit(int c) {
   return (c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') ||
          (c >= 'A' && c <= 'F');

src/http.h

@@ -201,3 +201,26 @@ size_t mg_http_next_multipart(struct mg_str, size_t, struct mg_http_part *);
 
 // Returns the HTTP status code from a parsed response message (e.g. 200, 404).
 int mg_http_status(const struct mg_http_message *hm);
+
+// Streams a POST/PUT body to a callback as it arrives, without buffering it.
+// Call it from your connection handler on every event.
+//
+// Callback upload_fn(hm, data, user_data) is called:
+//   - at start: hm is set, data is NULL. Open the file, set user_data.
+//   - per chunk: hm is NULL, data is set. Chunks are 512 bytes aligned
+//     except the last one.
+//   - at end: both NULL. Close the file.
+// After the last chunk, "200 ok" is sent and the connection is drained.
+// Example:
+//   mg_http_stream_body(c, ev, ev_data, mg_str("/fs/*"), upload);
+// Full example:
+//   tutorials/http/file-upload-single-post
+// Related APIs:
+//   mg_http_start_upload(), mg_http_start_ota()
+// Notes:
+//   Requires Content-Length, chunked bodies are not streamed. user_data is
+//   &c->fn_data. While streaming, c->pfn is NULL
+void mg_http_stream_body(struct mg_connection *c, int ev, void *ev_data,
+                         struct mg_str uri_pattern,
+                         void (*upload_fn)(struct mg_http_message *,
+                                           struct mg_str *, void **user_data));

tutorials/http/file-upload-single-post/main.c

@@ -1,10 +1,11 @@
 // Copyright (c) 2020-2024 Cesanta Software Limited
 // All rights reserved
 //
-// Streaming upload example. Demonstrates how to use MG_EV_READ events
+// Streaming upload example. Demonstrates how to use mg_http_stream_body()
 // to save a large file without buffering it fully in memory.
+// The file is saved as /tmp/a.txt.
 //
-// curl http://localhost:8000/upload?name=a.txt --data-binary @large_file.txt
+// curl http://localhost:8000/fs/a.txt --data-binary @large_file.txt
 
 #include "mongoose.h"
 
@@ -28,48 +29,24 @@ static const char *s_tls_key =
     "4hf5Gx17YJkq5/z3k6ogPDPpoAYWIw1/sw==\n"
     "-----END EC PRIVATE KEY-----\n";
 
-struct upload_state {
-  size_t expected;  // POST data length, bytes
-  size_t received;  // Already received bytes
-  void *fp;         // Opened uploaded file
-};
-
-static void handle_uploads(struct mg_connection *c, int ev, void *ev_data) {
-  struct upload_state *us = (struct upload_state *) c->data;
+static void upload(struct mg_http_message *hm, struct mg_str *data, void **p) {
   struct mg_fs *fs = &mg_fs_posix;
-
-  // Catch /upload requests early, without buffering whole body
-  // When we receive MG_EV_HTTP_HDRS event, that means we've received all
-  // HTTP headers but not necessarily full HTTP body
-  if (ev == MG_EV_HTTP_HDRS) {
-    struct mg_http_message *hm = (struct mg_http_message *) ev_data;
-    if (mg_match(hm->uri, mg_str("/upload/*"), NULL)) {
-      char path[MG_PATH_MAX];
-      mg_snprintf(path, sizeof(path), "%s/%.*s", UPLOAD_DIR, hm->uri.len - 8,
-                  hm->uri.buf + 8);
-      us->expected = hm->body.len;  // Store number of bytes we expect
-      mg_iobuf_del(&c->recv, 0, hm->head.len);  // Delete HTTP headers
-      c->pfn = NULL;  // Silence HTTP protocol handler, we'll use MG_EV_READ
-      if (mg_path_is_sane(mg_str(path))) {
-        fs->rm(path);                        // Delete file if it exists
-        us->fp = fs->op(path, MG_FS_WRITE);  // Open file for writing
-      }
-    }
-  }
-
-  // Catch uploaded file data for both MG_EV_READ and MG_EV_HTTP_HDRS
-  if (us->expected > 0 && c->recv.len > 0) {
-    us->received += c->recv.len;
-    if (us->fp) fs->wr(us->fp, c->recv.buf, c->recv.len);  // Write to file
-    c->recv.len = 0;  // Delete received data
-    if (us->received >= us->expected) {
-      // Uploaded everything. Send response back
-      MG_INFO(("Uploaded %lu bytes", us->received));
-      mg_http_reply(c, 200, NULL, "%lu ok\n", us->received);
-      if (us->fp) fs->cl(us->fp);  // Close file
-      memset(us, 0, sizeof(*us));  // Cleanup upload state
-      c->is_draining = 1;          // Close connection when response gets sent
+  if (hm != NULL) {
+    char path[MG_PATH_MAX];
+    MG_INFO(("UPLOAD BEGIN, URI %.*s ", hm->uri.len, hm->uri.buf));
+    mg_snprintf(path, sizeof(path), "%s/%.*s", UPLOAD_DIR, hm->uri.len - 4,
+                hm->uri.buf + 4);
+    *p = NULL;
+    if (mg_path_is_sane(mg_str(path))) {
+      fs->rm(path);                     // Delete file if it exists
+      *p = fs->op(path, MG_FS_WRITE);  // Open file for writing
     }
+  } else if (data != NULL) {
+    MG_INFO(("UPLOAD DATA, %zu bytes", data->len));
+    if (*p) fs->wr(*p, data->buf, data->len);  // Write to file
+  } else {
+    MG_INFO(("UPLOAD END %p", p));
+    if (*p) fs->cl(*p), *p = NULL;
   }
 }
 
@@ -78,16 +55,13 @@ static void fn(struct mg_connection *c, int ev, void *ev_data) {
     struct mg_tls_opts opts = {.cert = mg_str(s_tls_cert),
                                .key = mg_str(s_tls_key)};
     mg_tls_init(c, &opts);
-  }
-
-  handle_uploads(c, ev, ev_data);
-
-  // Non-upload requests, we serve normally
-  // NOTE: handle_uploads() may delete request and reset c->pfn
-  if (ev == MG_EV_HTTP_MSG && c->pfn != NULL) {
+  } else if (ev == MG_EV_HTTP_MSG && c->pfn != NULL) {
     struct mg_http_serve_opts opts = {.root_dir = "web_root"};
     mg_http_serve_dir(c, ev_data, &opts);
   }
+
+  // Handle upload requests to /fs/*
+  mg_http_stream_body(c, ev, ev_data, mg_str("/fs/*"), upload);
 }
 
 int main(void) {

tutorials/http/file-upload-single-post/web_root/index.html

@@ -50,7 +50,7 @@
       r.onload = function() {
         ev.target.value = '';
         document.getElementById('el3').innerText = 'Uploading...';
-        fetch(`/upload/${encodeURIComponent(f.name)}`, {
+        fetch(`/fs/${encodeURIComponent(f.name)}`, {
           method: 'POST',
           body: r.result,
         }).then(function(res) {