net.http.parser: Allow specifying sink for large request bodies

This enables uses such as saving uploaded files directly to a file on
disk or streaming parsing of payloads.

See #726

3 files changed, 36 insertions, 7 deletions

diff --git a/net/http/parser.lua b/net/http/parser.lua
index 84b1e005..b8396518 100644
--- a/net/http/parser.lua
+++ b/net/http/parser.lua
@@ -88,8 +88,6 @@ function httpstream.new(success_cb, error_cb, parser_type, options_cb)
 					if not first_line then error = true; return error_cb("invalid-status-line"); end
 					chunked = have_body and headers["transfer-encoding"] == "chunked";
 					len = tonumber(headers["content-length"]); -- TODO check for invalid len
-					if len and len > bodylimit then error = true; return error_cb("content-length-limit-exceeded"); end
-					-- TODO ask a callback whether to proceed in case of large requests or Expect: 100-continue
 					if client then
 						-- FIXME handle '100 Continue' response (by skipping it)
 						if not have_body then len = 0; end
@@ -126,9 +124,17 @@ function httpstream.new(success_cb, error_cb, parser_type, options_cb)
 							body_sink = nil;
 						};
 					end
-					if chunked then
+					if len and len > bodylimit then
+						-- Early notification, for redirection
+						success_cb(packet);
+						if not packet.body_sink then error = true; return error_cb("content-length-limit-exceeded"); end
+					end
+					if chunked and not packet.body_sink then
+						success_cb(packet);
+						if not packet.body_sink then
 						packet.body_buffer = dbuffer.new(buflimit);
 					end
+					end
 					state = true;
 				end
 				if state then -- read body
@@ -154,11 +160,23 @@ function httpstream.new(success_cb, error_cb, parser_type, options_cb)
 							success_cb(packet);
 						elseif buffer:length() - chunk_start - 2 >= chunk_size then -- we have a chunk
 							buffer:discard(chunk_start - 1); -- TODO verify that it's not off-by-one
-							packet.body_buffer:write(buffer:read(chunk_size));
+							(packet.body_sink or packet.body_buffer):write(buffer:read(chunk_size));
 							buffer:discard(2); -- CRLF
 						else -- Partial chunk remaining
 							break;
 						end
+					elseif packet.body_sink then
+						local chunk = buffer:read_chunk(len);
+						while chunk and len > 0 do
+							if packet.body_sink:write(chunk) then
+								len = len - #chunk;
+								chunk = buffer:read_chunk(len);
+							else
+								error = true;
+								return error_cb("body-sink-write-failure");
+							end
+						end
+						if len == 0 then state = nil; success_cb(packet); end
 					elseif buffer:length() >= len then
 						assert(not chunked)
 						packet.body = buffer:read(len) or "";
diff --git a/plugins/mod_http.lua b/plugins/mod_http.lua
index cf63ecfb..54c6089b 100644
--- a/plugins/mod_http.lua
+++ b/plugins/mod_http.lua
@@ -160,6 +160,15 @@ function module.add_host(module)
 				elseif event_name:sub(-1, -1) == "/" then
 					module:hook_object_event(server, event_name:sub(1, -2), redir_handler, -1);
 				end
+				do
+					-- COMPAT Modules not compatible with streaming uploads behave as before.
+					local _handler = handler;
+					function handler(event) -- luacheck: ignore 432/event
+						if event.request.body ~= false then
+							return _handler(event);
+						end
+					end
+				end
 				if not app_handlers[event_name] then
 					app_handlers[event_name] = {
 						main = handler;
diff --git a/spec/net_http_parser_spec.lua b/spec/net_http_parser_spec.lua
index 8310a451..4ac3cab9 100644
--- a/spec/net_http_parser_spec.lua
+++ b/spec/net_http_parser_spec.lua
@@ -3,7 +3,9 @@ local http_parser = require "net.http.parser";
 local function test_stream(stream, expect)
 	local success_cb = spy.new(function (packet)
 		assert.is_table(packet);
-		assert.is_equal(expect.body, packet.body);
+		if packet.body ~= false then
+			assert.is_equal(expect.body, packet.body);
+		end
 	end);
 
 	stream = stream:gsub("\n", "\r\n");
@@ -79,7 +81,7 @@ o
 
 ]],
 				{
-					body = "Hello", count = 1;
+					body = "Hello", count = 2;
 				}
 			);
 		end);
@@ -108,7 +110,7 @@ o
 
 ]],
 				{
-					body = "Hello", count = 2;
+					body = "Hello", count = 3;
 				}
 			);
 		end);