Handle chunked responses in the context of redirect requests.

The asynchronous was not handling situation, when the redirect
requests were sent chunked.

src/hackney_stream.erl

@@ -166,45 +166,71 @@ maybe_continue(Parent, Owner, Ref, #client{transport=Transport,
 %% - {redirect, To, Headers}
 %% - {see_other, To, Headers} for status 303 and POST requests.
 maybe_redirect(Parent, Owner, Ref, StatusInt, Reason,
-               #client{transport=Transport,
-                       socket=Socket,
-                       method=Method,
-                       follow_redirect=true}=Client) ->
+               #client{method=Method, follow_redirect=true}=Client) ->
   case lists:member(StatusInt, [301, 302, 307, 308]) of
     true ->
-      Transport:setopts(Socket, [{active, false}]),
-      case parse(Client) of
-        {loop, Client2} ->
-          maybe_redirect(Parent, Owner, Ref, StatusInt,
-            Reason, Client2);
-        {ok, {headers, Headers}, Client2} ->
-          Location = hackney:redirect_location(Headers),
-          case {Location, lists:member(Method, [get, head])} of
-            {undefined, _} ->
+      maybe_redirect_1(Parent, Owner, Ref, Reason, Client);
+   false when StatusInt =:= 303, Method =:= <<"POST">> ->
+      maybe_redirect_2(Parent, Owner, Ref, Reason, Client);
+    _ ->
+      Owner ! {hackney_response, Ref, {status, StatusInt, Reason}},
+      maybe_continue(Parent, Owner, Ref, Client)
+  end;
+maybe_redirect(Parent, Owner, Ref, StatusInt, Reason, Client) ->
+  Owner ! {hackney_response, Ref, {status, StatusInt, Reason}},
+  maybe_continue(Parent, Owner, Ref, Client).
+
+%% The first case for redirections are status codes 301, 302, 307 and 308.
+%% In this case {redirect, Location, Headers} is sent to the owner if
+%% the request is valid.
+maybe_redirect_1(Parent, Owner, Ref, Reason,
+               #client{transport=Transport, socket=Socket}=Client) ->
+  Transport:setopts(Socket, [{active, false}]),
+  case parse(Client) of
+    {loop, Client2} ->
+      maybe_redirect_1(Parent, Owner, Ref, Reason, Client2);
+    {more, Client2, Rest} ->
+      Continuation = fun(Client3) ->
+                         maybe_redirect_1(Parent, Owner, Ref, Reason, Client3)
+                     end,
+          async_recv(Parent, Owner, Ref, Client2, Rest, Continuation);
+    {ok, {headers, Headers}, Client2} ->
+      Location = hackney:redirect_location(Headers),
+      case Location of
+        undefined ->
+          Owner ! {hackney_response, Ref, {error,invalid_redirection}},
+          hackney_manager:handle_error(Client2);
+        _ ->
+          case hackney_response:skip_body(Client2) of
+            {skip, Client3} ->
+              hackney_manager:store_state(Client3),
               Owner ! {hackney_response, Ref,
-                {error,invalid_redirection}},
-              hackney_manager:handle_error(Client2);
-            {_, _} ->
-              case hackney_response:skip_body(Client2) of
-                {skip, Client3} ->
-                  hackney_manager:store_state(Client3),
-                  Owner ! {hackney_response, Ref,
-                    {redirect, Location, Headers}};
-                Error ->
-                  Owner ! {hackney_response, Ref, Error},
-                  hackney_manager:handle_error(Client2)
+                       {redirect, Location, Headers}};
+            Error ->
+              Owner ! {hackney_response, Ref, Error},
+              hackney_manager:handle_error(Client2)
               end
           end;
-        {error, Error} ->
-          Owner ! {hackney_response, Ref, {error, Error}},
-          hackney_manager:handle_error(Client)
-      end;
-    false when StatusInt =:= 303, Method =:= post ->
+    {error, Error} ->
+      Owner ! {hackney_response, Ref, {error, Error}},
+      hackney_manager:handle_error(Client)
+  end.
+
+%% The second case is for status code 303 and POST requests.
+%% This results in sending {see_other, Location, Headers} to the owner.
+maybe_redirect_2(Parent, Owner, Ref, Reason,
+               #client{transport=Transport,
+                       socket=Socket}=Client) ->
+      io:format("In post", []),
       Transport:setopts(Socket, [{active, false}]),
       case parse(Client) of
         {loop, Client2} ->
-          maybe_redirect(Parent, Owner, Ref, StatusInt,
-            Reason, Client2);
+          maybe_redirect_2(Parent, Owner, Ref, Reason, Client2);
+        {more, Client2, Rest} ->
+          Continuation = fun(Client3) ->
+                           maybe_redirect_2(Parent, Owner, Ref, Reason, Client3)
+                       end,
+          async_recv(Parent, Owner, Ref, Client2, Rest, Continuation);
         {ok, {headers, Headers}, Client2} ->
           case hackney:redirect_location(Headers) of
             undefined ->
@@ -225,21 +251,19 @@ maybe_redirect(Parent, Owner, Ref, StatusInt, Reason,
         {error, Error} ->
           Owner ! {hackney_response, Ref, {error, Error}},
           hackney_manager:handle_error(Client)
-      end;
-    _ ->
-      Owner ! {hackney_response, Ref, {status, StatusInt, Reason}},
-      maybe_continue(Parent, Owner, Ref, Client)
-  end;
-maybe_redirect(Parent, Owner, Ref, StatusInt, Reason, Client) ->
-  Owner ! {hackney_response, Ref, {status, StatusInt, Reason}},
-  maybe_continue(Parent, Owner, Ref, Client).
+      end.
 
 
+async_recv(Parent, Owner, Ref, Client, Buffer) ->
+  Continuation = fun(Client2) ->
+    stream_loop(Parent, Owner, Ref, Client2)
+  end,
+  async_recv(Parent, Owner, Ref, Client, Buffer, Continuation).
+
 async_recv(Parent, Owner, Ref,
            #client{transport=Transport,
                    socket=TSock,
-                   recv_timeout=Timeout}=Client, Buffer) ->
-
+                   recv_timeout=Timeout}=Client, Buffer, Continuation) ->
   {OK, Closed, Error} = Transport:messages(TSock),
   Sock = raw_sock(TSock),
   Transport:setopts(TSock, [{active, once}]),
@@ -263,7 +287,7 @@ async_recv(Parent, Owner, Ref,
       Transport:controlling_process(TSock, From),
       From ! {Ref, ok};
     {OK, Sock, Data} ->
-      stream_loop(Parent, Owner, Ref, Client#client{buffer=Data});
+      Continuation(Client#client{buffer=Data});
     {Closed, Sock} ->
       case Client#client.response_state of
         on_body when (Version =:= {1, 0} orelse Version =:= {1, 1})

test/hackney_integration_tests_async_long_headers.erl

@@ -0,0 +1,94 @@
+-module(hackney_integration_tests_async_long_headers).
+-include_lib("eunit/include/eunit.hrl").
+-include("hackney_lib.hrl").
+
+all_tests() ->
+  [fun absolute_redirect_request_follow/1].
+
+http_requests_test_() ->
+  TestMatrix = [
+                #{status_code => <<"301">>, method => get},
+                #{status_code => <<"303">>, method => post}
+           ],
+  lists:map(fun(Props) ->
+                {foreach,
+                 fun() -> start(Props) end,
+                 fun(StartResult) -> stop(StartResult, Props) end,
+                 all_tests()
+                }
+            end,
+            TestMatrix).
+
+start(#{status_code := StatusCode, method := Method}) ->
+  {ok, _} = application:ensure_all_started(hackney),
+  {ok, LSock} = gen_tcp:listen(0, [{active, false}]),
+  {ok, {_, Port}} = inet:sockname(LSock),
+  Pid = spawn_link(fun () -> dummy_server_loop(LSock, Port, StatusCode) end),
+  gen_tcp:controlling_process(LSock, Pid),
+  #{
+    dummy_http_pid => Pid,
+    method => Method,
+    port => Port
+   }.
+
+stop(#{dummy_http_pid := Pid}, _Props) ->
+  exit(Pid, normal),
+  application:stop(hackney),
+  error_logger:tty(true),
+  ok.
+
+
+absolute_redirect_request_follow(#{method := Method, port := Port}) ->
+  PortBin = list_to_binary(integer_to_list(Port)),
+  URL = <<"http://localhost:", PortBin/binary>>,
+  ExpectedRedirectUrl = <<"http://localhost:", PortBin/binary, "/redirected">>,
+  Options = [{follow_redirect, true}, {async, true}],
+  {ok, Ref} = hackney:request(Method, URL, [], <<>>, Options),
+  case Method of
+    get ->
+      [
+       receive
+         {hackney_response, Ref, {redirect, RedirectUrl, _headers}} ->
+           ?_assertEqual(ExpectedRedirectUrl, RedirectUrl);
+         Other ->
+           throw({error, Other})
+       after 1000 ->
+               throw(timeout)
+       end
+      ];
+    post ->
+      [
+       receive
+         {hackney_response, Ref, {see_other, RedirectUrl, _headers}} ->
+           ?_assertEqual(ExpectedRedirectUrl, RedirectUrl);
+         Other ->
+           throw({error, Other})
+       after 1000 ->
+               throw(timeout)
+       end
+      ]
+  end.
+
+dummy_server_loop(LSock, Port, StatusCode) ->
+  {ok, Sock} = gen_tcp:accept(LSock),
+  PortBin = list_to_binary(integer_to_list(Port)),
+  RedirectUrl = <<"http://localhost:", PortBin/binary, "/redirected">>,
+  Response = iolist_to_binary([
+                               "HTTP/1.1 ",
+                               StatusCode,
+                               " Moved Permanently\r\nLocation: ",
+                               RedirectUrl,
+                               "\r\nExtra-Header:",
+                               binary:copy(<<"a">>, 1024),
+                               "\r\n\r\n"
+                              ]),
+  send(Sock, Response),
+  ok = gen_tcp:shutdown(Sock, read_write),
+  dummy_server_loop(LSock, RedirectUrl, StatusCode).
+
+send(Sock, << Data :128/binary, Rest/binary>>) ->
+  ok = gen_tcp:send(Sock, Data),
+  send(Sock, Rest);
+
+send(Sock, Data) ->
+  ok = gen_tcp:send(Sock, Data).