Add request body reading options

The options were added to allow developers to fix timeout
issues when reading large bodies. It is also a cleaner and
easier to extend interface.

This commit deprecates the functions init_stream, stream_body
and skip_body which are no longer needed. They will be removed
in 1.0.

The body function can now take an additional argument that is a
list of options. The body_qs, part and part_body functions can
too and simply pass this argument down to the body call.

There are options for disabling the automatic continue reply,
setting a maximum length to be returned (soft limit), setting
the read length and read timeout, and setting the transfer and
content decode functions.

The return value of the body and body_qs have changed slightly.
The body function now works similarly to the part_body function,
in that it returns either an ok or a more tuple depending on
whether there is additional data to be read. The body_qs function
can return a badlength tuple if the body is too big. The default
size has been increased from 16KB to 64KB.

The default read length and timeout have been tweaked and vary
depending on the function called.

The body function will now adequately process chunked bodies,
which means that the body_qs function will too. But this means
that the behavior has changed slightly and your code should be
tested properly when updating your code.

The body and body_qs still accept a length as first argument
for compatibility purpose with older code. Note that this form
is deprecated and will be removed in 1.0. The part and part_body
function, being new and never having been in a release yet, have
this form completely removed in this commit.

Again, while most code should work as-is, you should make sure
that it actually does before pushing this to production.

guide/multipart_req.md

@@ -80,7 +80,9 @@ stream_file(Req) ->
 ```
 
 By default the body chunk Cowboy will return is limited
-to 8MB. This can of course be overriden.
+to 8MB. This can of course be overriden. Both functions
+can take a second argument, the same list of options that
+will be passed to `cowboy_req:body/2` function.
 
 Skipping unwanted parts
 -----------------------
@@ -107,6 +109,11 @@ Similarly, if you start reading the body and it ends up
 being too big, you can simply continue with the next part,
 Cowboy will automatically skip what remains.
 
+Note that the skipping rate may not be adequate for your
+application. If you observe poor performance when skipping,
+you might want to consider manually skipping by calling
+the `cowboy_req:part_body/1` function directly.
+
 And if you started reading the message but decide that you
 do not need the remaining parts, you can simply stop reading
 entirely and Cowboy will automatically figure out what to do.

guide/req.md

@@ -57,10 +57,9 @@ the socket or perform operations that may legitimately fail.
 They may return `{Result, Req}`, `{Result, Value, Req}`
 or `{error, atom()}`. This includes the following functions:
 `body/{1,2}`, `body_qs/{1,2}`, `chunked_reply/{2,3}`,
-`init_stream/4`, `parse_header/{2,3}`, `reply/{2,3,4}`,
-`skip_body/1`, `stream_body/{1,2}`. Finally, the group
-also includes the `chunk/2` function which always returns
-`ok`.
+`parse_header/{2,3}`, `part/{1,2}`, `part_body/{1,2}`
+and `reply/{2,3,4}`. Finally, the group also includes the
+`chunk/2` and `continue/1` functions which always return `ok`.
 
 The final group modifies the Req object state without
 performing any immediate operations. As these functions

guide/req_body.md

@@ -11,6 +11,12 @@ Cowboy will not attempt to read the body until you do.
 If handler execution ends without reading it, Cowboy
 will simply skip it.
 
+Cowboy provides different ways to read the request body.
+You can read it directly, stream it, but also read and
+parse in a single call for form urlencoded formats or
+multipart. All of these except multipart are covered in
+this chapter. Multipart is covered later on in the guide.
+
 Check for request body
 ----------------------
 
@@ -46,124 +52,115 @@ chunked bodies by using the stream functions.
 Reading the body
 ----------------
 
-If a content-length header was sent with the request,
-you can read the whole body directly.
+You can read the whole body directly in one call.
 
 ``` erlang
 {ok, Body, Req2} = cowboy_req:body(Req).
 ```
 
-If no content-length header is available, Cowboy will
-return the `{error, chunked}` tuple. You will need to
-stream the request body instead.
-
-By default, Cowboy will reject all body sizes above 8MB,
-to prevent an attacker from needlessly filling up memory.
-You can override this limit however.
+By default, Cowboy will attempt to read up to a
+size of 8MB. You can override this limit as needed.
 
 ``` erlang
-{ok, Body, Req2} = cowboy_req:body(100000000, Req).
+{ok, Body, Req2} = cowboy_req:body(Req, [{length, 100000000}]).
 ```
 
 You can also disable it.
 
 ``` erlang
-{ok, Body, Req2} = cowboy_req:body(infinity, Req).
+{ok, Body, Req2} = cowboy_req:body(Req, [{length, infinity}]).
 ```
 
 It is recommended that you do not disable it for public
 facing websites.
 
-Reading a body sent from an HTML form
--------------------------------------
+If the body is larger than the limit, then Cowboy will return
+a `more` tuple instead, allowing you to stream it if you
+would like to.
 
-You can directly obtain a list of key/value pairs if the
-body was sent using the application/x-www-form-urlencoded
-content-type.
+Streaming the body
+------------------
 
-``` erlang
-{ok, KeyValues, Req2} = cowboy_req:body_qs(Req).
-```
+You can stream the request body by chunks.
 
-You can then retrieve an individual value from that list.
+Cowboy returns a `more` tuple when there is more body to
+be read, and an `ok` tuple for the last chunk. This allows
+you to loop over all chunks.
 
 ``` erlang
-{_, Lang} = lists:keyfind(lang, 1, KeyValues).
+body_to_console(Req) ->
+    case cowboy_req:body(Req) of
+        {ok, Data, Req2} ->
+            io:format("~s", [Data]),
+            Req2;
+        {more, Data, Req2} ->
+            io:format("~s", [Data]),
+            body_to_console(Req2)
+    end.
 ```
 
-You should not attempt to match on the list as the order
-of the values is undefined.
+You can of course set the `length` option to configure the
+size of chunks.
 
-By default Cowboy will reject bodies with a size above
-16KB when using this function. You can override this limit.
+Rate of data transmission
+-------------------------
 
-``` erlang
-{ok, KeyValues, Req2} = cowboy_req:body_qs(500000, Req).
-```
+You can control the rate of data transmission by setting
+options when calling body functions. This applies not only
+to the functions described in this chapter, but also to
+the multipart functions.
 
-You can also disable it by passing the atom `infinity`,
-although it is not recommended.
+The `read_length` option defines the maximum amount of data
+to be received from the socket at once, in bytes.
 
-Streaming the body
-------------------
+The `read_timeout` option defines the time Cowboy waits
+before that amount is received, in milliseconds.
 
-You can stream the request body by chunks.
+Transfer and content decoding
+-----------------------------
 
-``` erlang
-{ok, Chunk, Req2} = cowboy_req:stream_body(Req).
-```
+Cowboy will by default decode the chunked transfer-encoding
+if any. It will not decode any content-encoding by default.
 
-By default, Cowboy will attempt to read chunks of up to
-1MB in size. The chunks returned by this function will
-often be smaller, however. You can also change this limit.
+The first time you call a body function you can set the
+`transfer_decode` and `content_decode` options. If the body
+was already started being read these options are simply
+ignored.
+
+The following example shows how to set both options.
 
 ``` erlang
-{ok, Chunk, Req2} = cowboy_req:stream_body(500000, Req).
+{ok, Req2} = cowboy_req:body(Req, [
+    {transfer_decode, fun transfer_decode/2, TransferState},
+    {content_decode, fun content_decode/1}
+]).
 ```
 
-When Cowboy finishes reading the body, any subsequent call
-will return `{done, Req2}`. You can thus write a recursive
-function to read the whole body and perform an action on
-all chunks, for example printing them to the console.
+Reading a form urlencoded body
+------------------------------
+
+You can directly obtain a list of key/value pairs if the
+body was sent using the application/x-www-form-urlencoded
+content-type.
 
 ``` erlang
-body_to_console(Req) ->
-    case cowboy_req:stream_body(Req) of
-        {ok, Chunk, Req2} ->
-            io:format("~s", [Chunk]),
-            body_to_console(Req2);
-        {done, Req2} ->
-            Req2
-    end.
+{ok, KeyValues, Req2} = cowboy_req:body_qs(Req).
 ```
 
-Advanced streaming
-------------------
-
-Cowboy will by default decode the chunked transfer-encoding
-if any. It will not decode any content-encoding by default.
-
-Before starting to stream, you can configure the functions
-that will be used for decoding both transfer-encoding and
-content-encoding.
+You can then retrieve an individual value from that list.
 
 ``` erlang
-{ok, Req2} = cowboy_req:init_stream(fun transfer_decode/2,
-    TransferStartState, fun content_decode/1, Req).
+{_, Lang} = lists:keyfind(lang, 1, KeyValues).
 ```
 
-Note that you do not need to call this function generally,
-as Cowboy will happily initialize the stream on its own.
-
-Skipping the body
------------------
+You should not attempt to match on the list as the order
+of the values is undefined.
 
-If you do not need the body, or if you started streaming
-the body but do not need the rest of it, you can skip it.
+By default Cowboy will reject bodies with a size above
+64KB when using this function. You can override this limit
+by setting the `length` option.
 
 ``` erlang
-{ok, Req2} = cowboy_req:skip_body(Req).
+{ok, KeyValues, Req2} = cowboy_req:body_qs(Req,
+    [{length, 2000000}]).
 ```
-
-You do not have to call this function though, as Cowboy will
-do it automatically when handler execution ends.

manual/cowboy_req.md

@@ -26,6 +26,15 @@ generally throw an error on the second call.
 Types
 -----
 
+### body_opts() = [{continue, boolean()}
+	| {length, non_neg_integer()}
+	| {read_length, non_neg_integer()}
+	| {read_timeout, timeout()}
+	| {transfer_decode, transfer_decode_fun(), any()}
+	| {content_decode, content_decode_fun()}]
+
+> Request body reading options.
+
 ### cookie_opts() = [{max_age, non_neg_integer()}
 	| {domain, binary()} | {path, binary()}
 	| {secure, boolean()} | {http_only, boolean()}]
@@ -327,20 +336,41 @@ Request related exports
 Request body related exports
 ----------------------------
 
-### body(Req) -> body(8000000, Req)
-### body(MaxLength, Req) -> {ok, Data, Req2} | {error, Reason}
+### body(Req) -> body(Req, [])
+### body(Req, Opts) -> {ok, Data, Req2} | {more, Data, Req2} | {error, Reason}
 
 > Types:
->  *  MaxLength = non_neg_integer() | infinity
+>  *  Opts = [body_opt()]
 >  *  Data = binary()
->  *  Reason = chunked | badlength | atom()
+>  *  Reason = atom()
+>
+> Read the request body.
+>
+> This function will read a chunk of the request body. If there is
+> more data to be read after this function call, then a `more` tuple
+> is returned. Otherwise an `ok` tuple is returned.
+>
+> Cowboy will automatically send a `100 Continue` reply if
+> required. If this behavior is not desirable, it can be disabled
+> by setting the `continue` option to `false`.
 >
-> Return the request body.
+> Cowboy will by default attempt to read up to 8MB of the body,
+> but in chunks of 1MB. It will use a timeout of 15s per chunk.
+> All these values can be changed using the `length`, `read_length`
+> and `read_timeout` options respectively. Note that the size
+> of the data may not be the same as requested as the decoding
+> functions may grow or shrink it, and Cowboy makes not attempt
+> at returning an exact amount.
 >
-> This function will return `{error, chunked}` if the request
-> body was sent using the chunked transfer-encoding. It will
-> also return `{error, badlength}` if the length of the body
-> exceeds the given `MaxLength`, which is 8MB by default.
+> Cowboy will properly handle chunked transfer-encoding by
+> default. If any other transfer-encoding or content-encoding
+> has been used for the request, custom decoding functions
+> can be used. The `content_decode` and `transfer_decode`
+> options allow setting the decode functions manually.
+>
+> After the body has been streamed fully, Cowboy will remove
+> the transfer-encoding header from the `Req` object, and add
+> the content-length header if it wasn't already there.
 >
 > This function can only be called once. Cowboy will not cache
 > the result of this call.
@@ -356,11 +386,13 @@ Request body related exports
 > use any transfer-encoding and if the content-length header
 > is present.
 
-### body_qs(Req) -> body_qs(16000, Req)
-### body_qs(MaxLength, Req) -> {ok, [{Name, Value}], Req2} | {error, Reason}
+### body_qs(Req) -> body_qs(Req,
+	[{length, 64000}, {read_length, 64000}, {read_timeout, 5000}])
+### body_qs(Req, Opts) -> {ok, [{Name, Value}], Req2}
+	| {badlength, Req2} | {error, Reason}
 
 > Types:
->  *  MaxLength = non_neg_integer() | infinity
+>  *  Opts = [body_opt()]
 >  *  Name = binary()
 >  *  Value = binary() | true
 >  *  Reason = chunked | badlength | atom()
@@ -371,10 +403,10 @@ Request body related exports
 > application/x-www-form-urlencoded, commonly used for the
 > query string.
 >
-> This function will return `{error, chunked}` if the request
-> body was sent using the chunked transfer-encoding. It will
-> also return `{error, badlength}` if the length of the body
-> exceeds the given `MaxLength`, which is 16KB by default.
+> This function calls `body/2` for reading the body, with the
+> same options it received. By default it will attempt to read
+> a body of 64KB in one chunk, with a timeout of 5s. If the
+> body is larger then a `badlength` tuple is returned.
 >
 > This function can only be called once. Cowboy will not cache
 > the result of this call.
@@ -383,34 +415,12 @@ Request body related exports
 
 > Return whether the request has a body.
 
-### init_stream(TransferDecode, TransferState, ContentDecode, Req) -> {ok, Req2}
-
-> Types:
->  *  TransferDecode = fun((Encoded, TransferState) -> OK | More | Done | {error, Reason})
->  *  Encoded = Decoded = Rest = binary()
->  *  TransferState = any()
->  *  OK = {ok, Decoded, Rest, TransferState}
->  *  More = more | {more, Length, Decoded, TransferState}
->  *  Done = {done, TotalLength, Rest} | {done, Decoded, TotalLength, Rest}
->  *  Length = TotalLength = non_neg_integer()
->  *  ContentDecode = fun((Encoded) -> {ok, Decoded} | {error, Reason})
->  *  Reason = atom()
->
-> Initialize streaming of the request body.
->
-> This function can be used to specify what function to use
-> for decoding the request body, generally specified in the
-> transfer-encoding and content-encoding request headers.
->
-> Cowboy will properly handle chunked transfer-encoding by
-> default. You do not need to call this function if you do
-> not need to decode other encodings, `stream_body/{1,2}`
-> will perform all the required initialization when it is
-> called the first time.
-
-### part(Req) -> {ok, Headers, Req2} | {done, Req2}
+### part(Req) -> part(Req,
+	[{length, 64000}, {read_length, 64000}, {read_timeout, 5000}])
+### part(Req, Opts) -> {ok, Headers, Req2} | {done, Req2}
 
 > Types:
+>  *  Opts = [body_opt()]
 >  *  Headers = cow_multipart:headers()
 >
 > Read the headers for the next part of the multipart message.
@@ -430,21 +440,28 @@ Request body related exports
 >
 > Note that once a part has been read, or skipped, it cannot
 > be read again.
+>
+> This function calls `body/2` for reading the body, with the
+> same options it received. By default it will only read chunks
+> of 64KB with a timeout of 5s. This is tailored for reading
+> part headers, not for skipping the previous part's body.
+> You might want to consider skipping large parts manually.
 
-### part_body(Req) -> part_body(8000000, Req)
-### part_body(MaxReadSize, Req) -> {ok, Data, Req2} | {more, Data, Req2}
+### part_body(Req) -> part_body(Req, [])
+### part_body(Req, Opts) -> {ok, Data, Req2} | {more, Data, Req2}
 
 > Types:
->  *  MaxReadSize = non_neg_integer()
+>  *  Opts = [body_opt()]
 >  *  Data = binary()
 >
 > Read the body of the current part of the multipart message.
 >
-> This function will read the body up to `MaxReadSize` bytes.
-> This is a soft limit. If there are more data to be read
-> from the socket for this part, the function will return
-> what it could read inside a `more` tuple. Otherwise, it
-> will return an `ok` tuple.
+> This function calls `body/2` for reading the body, with the
+> same options it received. It uses the same defaults.
+>
+> If there are more data to be read from the socket for this
+> part, the function will return what it could read inside a
+> `more` tuple. Otherwise, it will return an `ok` tuple.
 >
 > Calling this function again after receiving a `more` tuple
 > will return another chunk of body. The last chunk will be
@@ -453,46 +470,6 @@ Request body related exports
 > Note that once the body has been read, fully or partially,
 > it cannot be read again.
 
-### skip_body(Req) -> {ok, Req2} | {error, Reason}
-
-> Types:
->  *  Reason = atom()
->
-> Skip the request body.
->
-> This function will skip the body even if it was partially
-> read before.
-
-### stream_body(Req) -> stream_body(1000000, Req)
-### stream_body(MaxReadSize, Req) -> {ok, Data, Req2}
-	| {done, Req2} | {error, Reason}
-
-> Types:
->  *  MaxReadSize = non_neg_integer()
->  *  Data = binary()
->  *  Reason = atom()
->
-> Stream the request body.
->
-> This function will return the next segment of the body.
->
-> Cowboy will properly handle chunked transfer-encoding by
-> default. If any other transfer-encoding or content-encoding
-> has been used for the request, custom decoding functions
-> can be used. They must be specified using `init_stream/4`.
->
-> The amount of data returned by this function may vary
-> depending on the current state of the request. If data
-> is already available in the buffer then it is used fully,
-> otherwise Cowboy will read up to `MaxReadSize` bytes from
-> the socket. By default Cowboy will read up to 1MB of data.
-> It is then decoded, which may grow or shrink it, depending
-> on the encoding headers, before it is finally returned.
->
-> After the body has been streamed fully, Cowboy will remove
-> the transfer-encoding header from the `Req` object, and add
-> the content-length header if it wasn't already there.
-
 Response related exports
 ------------------------
 
@@ -536,6 +513,22 @@ Response related exports
 > This function can only be called once, with the exception
 > of overriding the response in the `onresponse` hook.
 
+### continue(Req) -> ok | {error, Reason}
+
+> Types:
+>  *  Reason = atom()
+>
+> Send a 100 Continue intermediate reply.
+>
+> This reply is required before the client starts sending the
+> body when the request contains the `expect` header with the
+> `100-continue` value.
+>
+> Cowboy will send this automatically when required. However
+> you may want to do it manually by disabling this behavior
+> with the `continue` body option and then calling this
+> function.
+
 ### delete_resp_header(Name, Req) -> Req2
 
 > Types:

src/cowboy_protocol.erl

@@ -469,8 +469,9 @@ next_request(Req, State=#state{req_keepalive=Keepalive, timeout=Timeout},
 		close ->
 			terminate(State);
 		_ ->
-			Buffer = case cowboy_req:skip_body(Req) of
-				{ok, Req2} -> cowboy_req:get(buffer, Req2);
+			%% Skip the body if it is reasonably sized. Close otherwise.
+			Buffer = case cowboy_req:body(Req) of
+				{ok, _, Req2} -> cowboy_req:get(buffer, Req2);
 				_ -> close
 			end,
 			%% Flush the resp_sent message before moving on.

src/cowboy_req.erl

@@ -49,10 +49,6 @@
 %% Request body API.
 -export([has_body/1]).
 -export([body_length/1]).
--export([init_stream/4]).
--export([stream_body/1]).
--export([stream_body/2]).
--export([skip_body/1]).
 -export([body/1]).
 -export([body/2]).
 -export([body_qs/1]).
@@ -79,6 +75,7 @@
 -export([chunked_reply/3]).
 -export([chunk/2]).
 -export([upgrade_reply/3]).
+-export([continue/1]).
 -export([maybe_reply/2]).
 -export([ensure_response/2]).
 
@@ -93,6 +90,16 @@
 -export([lock/1]).
 -export([to_list/1]).
 
+%% Deprecated API.
+-export([init_stream/4]).
+-deprecated({init_stream, 4}).
+-export([stream_body/1]).
+-deprecated({stream_body, 1}).
+-export([stream_body/2]).
+-deprecated({stream_body, 2}).
+-export([skip_body/1]).
+-deprecated({skip_body, 1}).
+
 -type cookie_opts() :: cow_cookie:cookie_opts().
 -export_type([cookie_opts/0]).
 
@@ -102,6 +109,14 @@
 -type transfer_decode_fun() :: fun((binary(), any())
 	-> cow_http_te:decode_ret()).
 
+-type body_opts() :: [{continue, boolean()}
+	| {length, non_neg_integer()}
+	| {read_length, non_neg_integer()}
+	| {read_timeout, timeout()}
+	| {transfer_decode, transfer_decode_fun(), any()}
+	| {content_decode, content_decode_fun()}].
+-export_type([body_opts/0]).
+
 -type resp_body_fun() :: fun((any(), module()) -> ok).
 -type send_chunk_fun() :: fun((iodata()) -> ok | {error, atom()}).
 -type resp_chunked_fun() :: fun((send_chunk_fun()) -> ok).
@@ -483,6 +498,107 @@ body_length(Req) ->
 			{undefined, Req2}
 	end.
 
+-spec body(Req)
+	-> {ok, binary(), Req} | {more, binary(), Req}
+	| {error, atom()} when Req::req().
+body(Req) ->
+	body(Req, []).
+
+-spec body(Req, body_opts())
+	-> {ok, binary(), Req} | {more, binary(), Req}
+	| {error, atom()} when Req::req().
+%% @todo This clause is kept for compatibility reasons, to be removed in 1.0.
+body(MaxBodyLength, Req) when is_integer(MaxBodyLength) ->
+	body(Req, [{length, MaxBodyLength}]);
+body(Req=#http_req{body_state=waiting}, Opts) ->
+	%% Send a 100 continue if needed (enabled by default).
+	Req1 = case lists:keyfind(continue, 1, Opts) of
+		{_, false} ->
+			Req;
+		_ ->
+			{ok, ExpectHeader, Req0} = parse_header(<<"expect">>, Req),
+			ok = case ExpectHeader of
+				[<<"100-continue">>] -> continue(Req0);
+				_ -> ok
+			end,
+			Req0
+	end,
+	%% Initialize body streaming state.
+	CFun = case lists:keyfind(content_decode, 1, Opts) of
+		false ->
+			fun cowboy_http:ce_identity/1;
+		{_, CFun0} ->
+			CFun0
+	end,
+	case lists:keyfind(transfer_decode, 1, Opts) of
+		false ->
+			case parse_header(<<"transfer-encoding">>, Req1) of
+				{ok, [<<"chunked">>], Req2} ->
+					body(Req2#http_req{body_state={stream, 0,
+						fun cow_http_te:stream_chunked/2, {0, 0}, CFun}}, Opts);
+				{ok, [<<"identity">>], Req2} ->
+					{Len, Req3} = body_length(Req2),
+					case Len of
+						0 ->
+							{ok, <<>>, Req3#http_req{body_state=done}};
+						_ ->
+							body(Req3#http_req{body_state={stream, Len,
+								fun cow_http_te:stream_identity/2, {0, Len},
+								CFun}}, Opts)
+					end
+			end;
+		{_, TFun, TState} ->
+			body(Req1#http_req{body_state={stream, 0,
+				TFun, TState, CFun}}, Opts)
+	end;
+body(Req=#http_req{body_state=done}, _) ->
+	{ok, <<>>, Req};
+body(Req, Opts) ->
+	ChunkLen = case lists:keyfind(length, 1, Opts) of
+		false -> 8000000;
+		{_, ChunkLen0} -> ChunkLen0
+	end,
+	ReadLen = case lists:keyfind(read_length, 1, Opts) of
+		false -> 1000000;
+		{_, ReadLen0} -> ReadLen0
+	end,
+	ReadTimeout = case lists:keyfind(read_timeout, 1, Opts) of
+		false -> 15000;
+		{_, ReadTimeout0} -> ReadTimeout0
+	end,
+	body_loop(Req, ReadTimeout, ReadLen, ChunkLen, <<>>).
+
+body_loop(Req=#http_req{buffer=Buffer, body_state={stream, Length, _, _, _}},
+		ReadTimeout, ReadLength, ChunkLength, Acc) ->
+	{Tag, Res, Req2} = case Buffer of
+		<<>> ->
+			body_recv(Req, ReadTimeout, min(Length, ReadLength));
+		_ ->
+			body_decode(Req, ReadTimeout)
+	end,
+	case {Tag, Res} of
+		{ok, {ok, Data}} ->
+			{ok, << Acc/binary, Data/binary >>, Req2};
+		{more, {ok, Data}} ->
+			Acc2 = << Acc/binary, Data/binary >>,
+			case byte_size(Acc2) >= ChunkLength of
+				true -> {more, Acc2, Req2};
+				false -> body_loop(Req2, ReadTimeout, ReadLength, ChunkLength, Acc2)
+			end;
+		_ -> %% Error.
+			Res
+	end.
+
+body_recv(Req=#http_req{transport=Transport, socket=Socket, buffer=Buffer},
+		ReadTimeout, ReadLength) ->
+	case Transport:recv(Socket, ReadLength, ReadTimeout) of
+		{ok, Data} ->
+			body_decode(Req#http_req{buffer= << Buffer/binary, Data/binary >>},
+				ReadTimeout);
+		Error = {error, _} ->
+			{error, Error, Req}
+	end.
+
 %% Two decodings happen. First a decoding function is applied to the
 %% transferred data, and then another is applied to the actual content.
 %%
@@ -491,149 +607,92 @@ body_length(Req) ->
 %% also initialized through this function.
 %%
 %% Content encoding is generally used for compression.
--spec init_stream(transfer_decode_fun(), any(), content_decode_fun(), Req)
-	-> {ok, Req} when Req::req().
-init_stream(TransferDecode, TransferState, ContentDecode, Req) ->
-	{ok, Req#http_req{body_state=
-		{stream, 0, TransferDecode, TransferState, ContentDecode}}}.
-
--spec stream_body(Req) -> {ok, binary(), Req}
-	| {done, Req} | {error, atom()} when Req::req().
-stream_body(Req) ->
-	stream_body(1000000, Req).
-
--spec stream_body(non_neg_integer(), Req) -> {ok, binary(), Req}
-	| {done, Req} | {error, atom()} when Req::req().
-stream_body(MaxLength, Req=#http_req{body_state=waiting, version=Version,
-		transport=Transport, socket=Socket}) ->
-	{ok, ExpectHeader, Req1} = parse_header(<<"expect">>, Req),
-	case ExpectHeader of
-		[<<"100-continue">>] ->
-			HTTPVer = atom_to_binary(Version, latin1),
-			Transport:send(Socket,
-				<< HTTPVer/binary, " ", (status(100))/binary, "\r\n\r\n" >>);
-		undefined ->
-			ok
-	end,
-	case parse_header(<<"transfer-encoding">>, Req1) of
-		{ok, [<<"chunked">>], Req2} ->
-			stream_body(MaxLength, Req2#http_req{body_state=
-				{stream, 0,
-					fun cow_http_te:stream_chunked/2, {0, 0},
-					fun cowboy_http:ce_identity/1}});
-		{ok, [<<"identity">>], Req2} ->
-			{Length, Req3} = body_length(Req2),
-			case Length of
-				0 ->
-					{done, Req3#http_req{body_state=done}};
-				Length ->
-					stream_body(MaxLength, Req3#http_req{body_state=
-						{stream, Length,
-							fun cow_http_te:stream_identity/2, {0, Length},
-							fun cowboy_http:ce_identity/1}})
-			end
-	end;
-stream_body(_, Req=#http_req{body_state=done}) ->
-	{done, Req};
-stream_body(_, Req=#http_req{buffer=Buffer})
-		when Buffer =/= <<>> ->
-	transfer_decode(Buffer, Req#http_req{buffer= <<>>});
-stream_body(MaxLength, Req) ->
-	stream_body_recv(MaxLength, Req).
-
--spec stream_body_recv(non_neg_integer(), Req)
-	-> {ok, binary(), Req} | {error, atom()} when Req::req().
-stream_body_recv(MaxLength, Req=#http_req{
-		transport=Transport, socket=Socket, buffer=Buffer,
-		body_state={stream, Length, _, _, _}}) ->
-	%% @todo Allow configuring the timeout.
-	case Transport:recv(Socket, min(Length, MaxLength), 5000) of
-		{ok, Data} -> transfer_decode(<< Buffer/binary, Data/binary >>,
-			Req#http_req{buffer= <<>>});
-		{error, Reason} -> {error, Reason}
-	end.
-
+%%
 %% @todo Handle chunked after-the-facts headers.
 %% @todo Depending on the length returned we might want to 0 or +5 it.
--spec transfer_decode(binary(), Req)
-	-> {ok, binary(), Req} | {error, atom()} when Req::req().
-transfer_decode(Data, Req=#http_req{body_state={stream, _,
-		TransferDecode, TransferState, ContentDecode}}) ->
-	case TransferDecode(Data, TransferState) of
+body_decode(Req=#http_req{buffer=Data, body_state={stream, _,
+		TDecode, TState, CDecode}}, ReadTimeout) ->
+	case TDecode(Data, TState) of
 		more ->
-			stream_body_recv(0, Req#http_req{buffer=Data, body_state={stream,
-				0, TransferDecode, TransferState, ContentDecode}});
-		{more, Data2, TransferState2} ->
-			content_decode(ContentDecode, Data2,
-				Req#http_req{body_state={stream, 0,
-				TransferDecode, TransferState2, ContentDecode}});
-		{more, Data2, Length, TransferState2} when is_integer(Length) ->
-			content_decode(ContentDecode, Data2,
-				Req#http_req{body_state={stream, Length,
-				TransferDecode, TransferState2, ContentDecode}});
-		{more, Data2, Rest, TransferState2} ->
-			content_decode(ContentDecode, Data2,
-				Req#http_req{buffer=Rest, body_state={stream, 0,
-				TransferDecode, TransferState2, ContentDecode}});
-		{done, Length, Rest} ->
-			Req2 = transfer_decode_done(Length, Rest, Req),
-			{done, Req2};
-		{done, Data2, Length, Rest} ->
-			Req2 = transfer_decode_done(Length, Rest, Req),
-			content_decode(ContentDecode, Data2, Req2)
+			body_recv(Req#http_req{body_state={stream, 0,
+				TDecode, TState, CDecode}}, ReadTimeout, 0);
+		{more, Data2, TState2} ->
+			{more, CDecode(Data2), Req#http_req{body_state={stream, 0,
+				TDecode, TState2, CDecode}, buffer= <<>>}};
+		{more, Data2, Length, TState2} when is_integer(Length) ->
+			{more, CDecode(Data2), Req#http_req{body_state={stream, Length,
+				TDecode, TState2, CDecode}, buffer= <<>>}};
+		{more, Data2, Rest, TState2} ->
+			{more, CDecode(Data2), Req#http_req{body_state={stream, 0,
+				TDecode, TState2, CDecode}, buffer=Rest}};
+		{done, TotalLength, Rest} ->
+			{ok, {ok, <<>>}, body_decode_end(Req, TotalLength, Rest)};
+		{done, Data2, TotalLength, Rest} ->
+			{ok, CDecode(Data2), body_decode_end(Req, TotalLength, Rest)}
 	end.
 
--spec transfer_decode_done(non_neg_integer(), binary(), Req)
-	-> Req when Req::req().
-transfer_decode_done(Length, Rest, Req=#http_req{
-		headers=Headers, p_headers=PHeaders}) ->
+body_decode_end(Req=#http_req{headers=Headers, p_headers=PHeaders},
+		TotalLength, Rest) ->
 	Headers2 = lists:keystore(<<"content-length">>, 1, Headers,
-		{<<"content-length">>, list_to_binary(integer_to_list(Length))}),
+		{<<"content-length">>, list_to_binary(integer_to_list(TotalLength))}),
 	%% At this point we just assume TEs were all decoded.
 	Headers3 = lists:keydelete(<<"transfer-encoding">>, 1, Headers2),
 	PHeaders2 = lists:keystore(<<"content-length">>, 1, PHeaders,
-		{<<"content-length">>, Length}),
+		{<<"content-length">>, TotalLength}),
 	PHeaders3 = lists:keydelete(<<"transfer-encoding">>, 1, PHeaders2),
 	Req#http_req{buffer=Rest, body_state=done,
 		headers=Headers3, p_headers=PHeaders3}.
 
--spec content_decode(content_decode_fun(), binary(), Req)
-	-> {ok, binary(), Req} | {error, atom()} when Req::req().
-content_decode(ContentDecode, Data, Req) ->
-	case ContentDecode(Data) of
-		{ok, Data2} -> {ok, Data2, Req};
-		{error, Reason} -> {error, Reason}
+-spec body_qs(Req)
+	-> {ok, [{binary(), binary() | true}], Req} | {error, atom()}
+	when Req::req().
+body_qs(Req) ->
+	body_qs(Req, [
+		{length, 64000},
+		{read_length, 64000},
+		{read_timeout, 5000}]).
+
+-spec body_qs(Req, body_opts()) -> {ok, [{binary(), binary() | true}], Req}
+	| {badlength, Req} | {error, atom()} when Req::req().
+%% @todo This clause is kept for compatibility reasons, to be removed in 1.0.
+body_qs(MaxBodyLength, Req) when is_integer(MaxBodyLength) ->
+	body_qs(Req, [{length, MaxBodyLength}]);
+body_qs(Req, Opts) ->
+	case body(Req, Opts) of
+		{ok, Body, Req2} ->
+			{ok, cow_qs:parse_qs(Body), Req2};
+		{more, _, Req2} ->
+			{badlength, Req2};
+		{error, Reason} ->
+			{error, Reason}
 	end.
 
--spec body(Req) -> {ok, binary(), Req} | {error, atom()} when Req::req().
-body(Req) ->
-	body(8000000, Req).
+%% Deprecated body API.
+%% @todo The following 4 functions will be removed in Cowboy 1.0.
 
--spec body(non_neg_integer() | infinity, Req)
-	-> {ok, binary(), Req} | {error, atom()} when Req::req().
-body(MaxBodyLength, Req) ->
-	case parse_header(<<"transfer-encoding">>, Req) of
-		{ok, [<<"identity">>], Req2} ->
-			{ok, Length, Req3} = parse_header(<<"content-length">>, Req2, 0),
-			if 	Length > MaxBodyLength ->
-					{error, badlength};
-				true ->
-					read_body(Req3, <<>>)
-			end;
-		{ok, _, _} ->
-			{error, chunked}
-	end.
+-spec init_stream(transfer_decode_fun(), any(), content_decode_fun(), Req)
+	-> {ok, Req} when Req::req().
+init_stream(TransferDecode, TransferState, ContentDecode, Req) ->
+	{ok, Req#http_req{body_state=
+		{stream, 0, TransferDecode, TransferState, ContentDecode}}}.
 
--spec read_body(Req, binary())
-	-> {ok, binary(), Req} | {error, atom()} when Req::req().
-read_body(Req, Acc) ->
-	case stream_body(Req) of
+-spec stream_body(Req) -> {ok, binary(), Req}
+	| {done, Req} | {error, atom()} when Req::req().
+stream_body(Req) ->
+	stream_body(1000000, Req).
+
+-spec stream_body(non_neg_integer(), Req) -> {ok, binary(), Req}
+	| {done, Req} | {error, atom()} when Req::req().
+stream_body(ChunkLength, Req) ->
+	case body(Req, [{length, ChunkLength}]) of
+		{ok, <<>>, Req2} ->
+			{done, Req2};
 		{ok, Data, Req2} ->
-			read_body(Req2, << Acc/binary, Data/binary >>);
-		{done, Req2} ->
-			{ok, Acc, Req2};
-		{error, Reason} ->
-			{error, Reason}
+			{ok, Data, Req2};
+		{more, Data, Req2} ->
+			{ok, Data, Req2};
+		Error = {error, _} ->
+			Error
 	end.
 
 -spec skip_body(Req) -> {ok, Req} | {error, atom()} when Req::req().
@@ -644,43 +703,34 @@ skip_body(Req) ->
 		{error, Reason} -> {error, Reason}
 	end.
 
--spec body_qs(Req)
-	-> {ok, [{binary(), binary() | true}], Req} | {error, atom()}
-	when Req::req().
-body_qs(Req) ->
-	body_qs(16000, Req).
-
-%% Essentially a POST query string.
--spec body_qs(non_neg_integer() | infinity, Req)
-	-> {ok, [{binary(), binary() | true}], Req} | {error, atom()}
-	when Req::req().
-body_qs(MaxBodyLength, Req) ->
-	case body(MaxBodyLength, Req) of
-		{ok, Body, Req2} ->
-			{ok, cow_qs:parse_qs(Body), Req2};
-		{error, Reason} ->
-			{error, Reason}
-	end.
-
 %% Multipart API.
 
 -spec part(Req)
 	-> {ok, cow_multipart:headers(), Req} | {done, Req}
 	when Req::req().
-part(Req=#http_req{multipart=undefined}) ->
-	part(init_multipart(Req));
 part(Req) ->
-	{ok, Data, Req2} = stream_multipart(Req),
-	part(Data, Req2).
+	part(Req, [
+		{length, 64000},
+		{read_length, 64000},
+		{read_timeout, 5000}]).
 
-part(Buffer, Req=#http_req{multipart={Boundary, _}}) ->
+-spec part(Req, body_opts())
+	-> {ok, cow_multipart:headers(), Req} | {done, Req}
+	when Req::req().
+part(Req=#http_req{multipart=undefined}, Opts) ->
+	part(init_multipart(Req), Opts);
+part(Req, Opts) ->
+	{Data, Req2} = stream_multipart(Req, Opts),
+	part(Data, Opts, Req2).
+
+part(Buffer, Opts, Req=#http_req{multipart={Boundary, _}}) ->
 	case cow_multipart:parse_headers(Buffer, Boundary) of
 		more ->
-			{ok, Data, Req2} = stream_multipart(Req),
-			part(<< Buffer/binary, Data/binary >>, Req2);
+			{Data, Req2} = stream_multipart(Req, Opts),
+			part(<< Buffer/binary, Data/binary >>, Opts, Req2);
 		{more, Buffer2} ->
-			{ok, Data, Req2} = stream_multipart(Req),
-			part(<< Buffer2/binary, Data/binary >>, Req2);
+			{Data, Req2} = stream_multipart(Req, Opts),
+			part(<< Buffer2/binary, Data/binary >>, Opts, Req2);
 		{ok, Headers, Rest} ->
 			{ok, Headers, Req#http_req{multipart={Boundary, Rest}}};
 		%% Ignore epilogue.
@@ -692,33 +742,39 @@ part(Buffer, Req=#http_req{multipart={Boundary, _}}) ->
 	-> {ok, binary(), Req} | {more, binary(), Req}
 	when Req::req().
 part_body(Req) ->
-	part_body(8000000, Req).
+	part_body(Req, []).
 
--spec part_body(non_neg_integer(), Req)
+-spec part_body(Req, body_opts())
 	-> {ok, binary(), Req} | {more, binary(), Req}
 	when Req::req().
-part_body(MaxLength, Req=#http_req{multipart=undefined}) ->
-	part_body(MaxLength, init_multipart(Req));
-part_body(MaxLength, Req) ->
-	part_body(<<>>, MaxLength, Req, <<>>).
-
-part_body(Buffer, MaxLength, Req=#http_req{multipart={Boundary, _}}, Acc)
-		when byte_size(Acc) > MaxLength ->
-	{more, Acc, Req#http_req{multipart={Boundary, Buffer}}};
-part_body(Buffer, MaxLength, Req=#http_req{multipart={Boundary, _}}, Acc) ->
-	{ok, Data, Req2} = stream_multipart(Req),
-	case cow_multipart:parse_body(<< Buffer/binary, Data/binary >>, Boundary) of
-		{ok, Body} ->
-			part_body(<<>>, MaxLength, Req2, << Acc/binary, Body/binary >>);
-		{ok, Body, Rest} ->
-			part_body(Rest, MaxLength, Req2, << Acc/binary, Body/binary >>);
-		done ->
-			{ok, Acc, Req2};
-		{done, Body} ->
-			{ok, << Acc/binary, Body/binary >>, Req2};
-		{done, Body, Rest} ->
-			{ok, << Acc/binary, Body/binary >>,
-				Req2#http_req{multipart={Boundary, Rest}}}
+part_body(Req=#http_req{multipart=undefined}, Opts) ->
+	part_body(init_multipart(Req), Opts);
+part_body(Req, Opts) ->
+	part_body(<<>>, Opts, Req, <<>>).
+
+part_body(Buffer, Opts, Req=#http_req{multipart={Boundary, _}}, Acc) ->
+	ChunkLen = case lists:keyfind(length, 1, Opts) of
+		false -> 8000000;
+		{_, ChunkLen0} -> ChunkLen0
+	end,
+	case byte_size(Acc) > ChunkLen of
+		true ->
+			{more, Acc, Req#http_req{multipart={Boundary, Buffer}}};
+		false ->
+			{Data, Req2} = stream_multipart(Req, Opts),
+			case cow_multipart:parse_body(<< Buffer/binary, Data/binary >>, Boundary) of
+				{ok, Body} ->
+					part_body(<<>>, Opts, Req2, << Acc/binary, Body/binary >>);
+				{ok, Body, Rest} ->
+					part_body(Rest, Opts, Req2, << Acc/binary, Body/binary >>);
+				done ->
+					{ok, Acc, Req2};
+				{done, Body} ->
+					{ok, << Acc/binary, Body/binary >>, Req2};
+				{done, Body, Rest} ->
+					{ok, << Acc/binary, Body/binary >>,
+						Req2#http_req{multipart={Boundary, Rest}}}
+			end
 	end.
 
 init_multipart(Req) ->
@@ -727,10 +783,12 @@ init_multipart(Req) ->
 	{_, Boundary} = lists:keyfind(<<"boundary">>, 1, Params),
 	Req2#http_req{multipart={Boundary, <<>>}}.
 
-stream_multipart(Req=#http_req{multipart={_, <<>>}}) ->
-	stream_body(Req);
-stream_multipart(Req=#http_req{multipart={Boundary, Buffer}}) ->
-	{ok, Buffer, Req#http_req{multipart={Boundary, <<>>}}}.
+stream_multipart(Req=#http_req{body_state=BodyState, multipart={_, <<>>}}, Opts) ->
+	true = BodyState =/= done,
+	{_, Data, Req2} = body(Req, Opts),
+	{Data, Req2};
+stream_multipart(Req=#http_req{multipart={Boundary, Buffer}}, _) ->
+	{Buffer, Req#http_req{multipart={Boundary, <<>>}}}.
 
 %% Response API.
 
@@ -970,6 +1028,13 @@ upgrade_reply(Status, Headers, Req=#http_req{transport=Transport,
 	], <<>>, Req),
 	{ok, Req2#http_req{resp_state=done, resp_headers=[], resp_body= <<>>}}.
 
+-spec continue(req()) -> ok | {error, atom()}.
+continue(#http_req{socket=Socket, transport=Transport,
+		version=Version}) ->
+	HTTPVer = atom_to_binary(Version, latin1),
+	Transport:send(Socket,
+		<< HTTPVer/binary, " ", (status(100))/binary, "\r\n\r\n" >>).
+
 %% Meant to be used internally for sending errors after crashes.
 -spec maybe_reply(cowboy:http_status(), req()) -> ok.
 maybe_reply(Status, Req) ->

test/http_SUITE.erl

@@ -336,7 +336,7 @@ echo_body(Config) ->
 %% Check if sending request whose size is bigger than 1000000 bytes causes 413
 echo_body_max_length(Config) ->
 	ConnPid = gun_open(Config),
-	Ref = gun:post(ConnPid, "/echo/body", [], << 0:8000008 >>),
+	Ref = gun:post(ConnPid, "/echo/body", [], << 0:2000000/unit:8 >>),
 	{response, nofin, 413, _} = gun:await(ConnPid, Ref),
 	ok.
 
@@ -350,7 +350,7 @@ echo_body_qs(Config) ->
 
 echo_body_qs_max_length(Config) ->
 	ConnPid = gun_open(Config),
-	Ref = gun:post(ConnPid, "/echo/body_qs", [], << "echo=", 0:15996/unit:8 >>),
+	Ref = gun:post(ConnPid, "/echo/body_qs", [], << "echo=", 0:2000000/unit:8 >>),
 	{response, nofin, 413, _} = gun:await(ConnPid, Ref),
 	ok.
 

test/http_SUITE_data/http_body_qs.erl

@@ -15,8 +15,8 @@ handle(Req, State) ->
 
 maybe_echo(<<"POST">>, true, Req) ->
 	case cowboy_req:body_qs(Req) of
-		{error,badlength} ->
-			echo(badlength, Req);
+		{badlength, Req2} ->
+			echo(badlength, Req2);
 		{ok, PostVals, Req2} ->
 			echo(proplists:get_value(<<"echo">>, PostVals), Req2)
 	end;

test/http_SUITE_data/http_echo_body.erl

@@ -10,17 +10,11 @@ init({_, http}, Req, _) ->
 handle(Req, State) ->
 	true = cowboy_req:has_body(Req),
 	{ok, Req3} = case cowboy_req:body(1000000, Req) of
-		{error, chunked} -> handle_chunked(Req);
-		{error, badlength} -> handle_badlength(Req);
-		{ok, Body, Req2} -> handle_body(Req2, Body)
+		{ok, Body, Req2} -> handle_body(Req2, Body);
+		{more, _, Req2} -> handle_badlength(Req2)
 	end,
 	{ok, Req3, State}.
 
-handle_chunked(Req) ->
-	{ok, Data, Req2} = read_body(Req, <<>>, 1000000),
-	{ok, Req3} = cowboy_req:reply(200, [], Data, Req2),
-	{ok, Req3}.
-
 handle_badlength(Req) ->
 	{ok, Req2} = cowboy_req:reply(413, [], <<"Request entity too large">>, Req),
 	{ok, Req2}.
@@ -33,13 +27,3 @@ handle_body(Req, Body) ->
 
 terminate(_, _, _) ->
 	ok.
-
-% Read chunked request content
-read_body(Req, Acc, BodyLengthRemaining) ->
-	case cowboy_req:stream_body(Req) of
-		{ok, Data, Req2} ->
-			BodyLengthRem = BodyLengthRemaining - byte_size(Data),
-			read_body(Req2, << Acc/binary, Data/binary >>, BodyLengthRem);
-		{done, Req2} ->
-			{ok, Acc, Req2}
-	end.

test/http_SUITE_data/http_loop_stream_recv.erl

@@ -14,12 +14,12 @@ info(stream, Req, undefined) ->
 	stream(Req, 1, <<>>).
 
 stream(Req, ID, Acc) ->
-	case cowboy_req:stream_body(Req) of
-		{ok, Data, Req2} ->
-			parse_id(Req2, ID, << Acc/binary, Data/binary >>);
-		{done, Req2} ->
+	case cowboy_req:body(Req) of
+		{ok, <<>>, Req2} ->
 			{ok, Req3} = cowboy_req:reply(200, Req2),
-			{ok, Req3, undefined}
+			{ok, Req3, undefined};
+		{_, Data, Req2} ->
+			parse_id(Req2, ID, << Acc/binary, Data/binary >>)
 	end.
 
 parse_id(Req, ID, Data) ->