Merge branch 'bmk/inets/httpc/support_upload_body_streaming/OTP-9094' into bmk/inets/inet56_integration

4 files changed, 181 insertions, 49 deletions

diff --git a/lib/inets/src/http_client/httpc.erl b/lib/inets/src/http_client/httpc.erl
index 04fae13b20..b70b16f57f 100644
--- a/lib/inets/src/http_client/httpc.erl
+++ b/lib/inets/src/http_client/httpc.erl
@@ -126,7 +126,10 @@ request(Url, Profile) ->
 %%      Header = {Field, Value}
 %%	Field = string()
 %%	Value = string()
-%%	Body = string() | binary() - HTLM-code
+%%	Body = string() | binary() | {fun(SendAcc) -> SendFunResult, SendAcc} |
+%%              {chunkify, fun(SendAcc) -> SendFunResult, SendAcc} - HTLM-code
+%%      SendFunResult = eof | {ok, iolist(), NewSendAcc}
+%%      SendAcc = NewSendAcc = term()
 %%
 %% Description: Sends a HTTP-request. The function can be both
 %% syncronus and asynchronous in the later case the function will
@@ -426,26 +429,44 @@ service_info(Pid) ->
 
 handle_request(Method, Url, 
 	       {Scheme, UserInfo, Host, Port, Path, Query}, 
-	       Headers, ContentType, Body, 
+	       Headers0, ContentType, Body0,
 	       HTTPOptions0, Options0, Profile) ->
 
-    Started    = http_util:timestamp(), 
-    NewHeaders = [{http_util:to_lower(Key), Val} || {Key, Val} <- Headers],
+    Started     = http_util:timestamp(), 
+    NewHeaders0 = [{http_util:to_lower(Key), Val} || {Key, Val} <- Headers0],
 
     try
 	begin
+	    ?hcrt("begin processing", [{started,     Started}, 
+				       {new_headers, NewHeaders0}]),
+
+	    {NewHeaders, Body} = 
+		case Body0 of
+		    {chunkify, ProcessBody, Acc} 
+		      when is_function(ProcessBody, 1) ->
+			NewHeaders1 = ensure_chunked_encoding(NewHeaders0), 
+			Body1       = {mk_chunkify_fun(ProcessBody), Acc}, 
+			{NewHeaders1, Body1};
+		    {ProcessBody, _} 
+		      when is_function(ProcessBody, 1) ->
+			{NewHeaders0, Body0};
+		    _ when is_list(Body0) orelse is_binary(Body0) ->
+			{NewHeaders0, Body0};
+		    _ ->
+			throw({error, {bad_body, Body0}})
+		end,
+
 	    HTTPOptions   = http_options(HTTPOptions0),
 	    Options       = request_options(Options0), 
 	    Sync          = proplists:get_value(sync,   Options),
 	    Stream        = proplists:get_value(stream, Options),
 	    Host2         = header_host(Scheme, Host, Port), 
 	    HeadersRecord = header_record(NewHeaders, Host2, HTTPOptions),
-	    Receiver   = proplists:get_value(receiver, Options),
-	    SocketOpts = proplists:get_value(socket_opts, Options),
-	    UrlEncodeBool =  HTTPOptions#http_options.url_encode,
-	    MaybeEscPath = url_encode(Path, UrlEncodeBool),
-	    MaybeEscQuery = url_encode(Query, UrlEncodeBool),
-	    AbsUri  = url_encode(Url, UrlEncodeBool),
+	    Receiver      = proplists:get_value(receiver, Options),
+	    SocketOpts    = proplists:get_value(socket_opts, Options),
+	    MaybeEscPath  = maybe_encode_uri(HTTPOptions, Path),
+	    MaybeEscQuery = maybe_encode_uri(HTTPOptions, Query),
+	    AbsUri        = maybe_encode_uri(HTTPOptions, Url),
 
 	    Request = #request{from          = Receiver,
 			       scheme        = Scheme, 
@@ -458,38 +479,71 @@ handle_request(Method, Url,
 			       settings      = HTTPOptions, 
 			       abs_uri       = AbsUri,
 			       userinfo      = UserInfo, 
-			       stream        = Stream,
-			       headers_as_is = headers_as_is(Headers, Options),
+			       stream        = Stream, 
+			       headers_as_is = headers_as_is(Headers0, Options),
 			       socket_opts   = SocketOpts, 
 			       started       = Started},
+
 	    case httpc_manager:request(Request, profile_name(Profile)) of
 		{ok, RequestId} ->
 		    handle_answer(RequestId, Sync, Options);
 		{error, Reason} ->
+		    ?hcrd("request failed", [{reason, Reason}]),
 		    {error, Reason}
 	    end
 	end
     catch
 	error:{noproc, _} ->
+	    ?hcrv("noproc", [{profile, Profile}]),
 	    {error, {not_started, Profile}};
 	throw:Error ->
+	    ?hcrv("throw", [{error, Error}]),
 	    Error
     end.
 
-url_encode(URI, true) ->
+ensure_chunked_encoding(Hdrs) ->
+    Key = "transfer-encoding",
+    lists:keystore(Key, 1, Hdrs, {Key, "chunked"}).
+
+maybe_encode_uri(#http_options{url_encode = true}, URI) ->
     http_uri:encode(URI);
-url_encode(URI, false) ->
+maybe_encode_uri(_, URI) ->
     URI.
 
+mk_chunkify_fun(ProcessBody) ->
+    fun(eof_body) ->
+	    eof;
+       (Acc) ->
+	    case ProcessBody(Acc) of
+		eof ->
+		    {ok, <<"0\r\n\r\n">>, eof_body};
+		{ok, Data, NewAcc} ->
+		    {ok, mk_chunk_bin(Data), NewAcc}
+	    end
+    end.
+
+mk_chunk_bin(Data) ->
+    Bin = iolist_to_binary(Data),
+    iolist_to_binary([hex_size(Bin), "\r\n", Bin, "\r\n"]).
+
+hex_size(Bin) ->
+    hd(io_lib:format("~.16B", [size(Bin)])).
+
+
 handle_answer(RequestId, false, _) ->
     {ok, RequestId};
 handle_answer(RequestId, true, Options) ->
     receive
 	{http, {RequestId, saved_to_file}} ->
+	    ?hcrt("received saved-to-file", [{request_id, RequestId}]),
 	    {ok, saved_to_file};
 	{http, {RequestId, {_,_,_} = Result}} ->
+	    ?hcrt("received answer", [{request_id, RequestId}, 
+				      {result,     Result}]),
 	    return_answer(Options, Result);
 	{http, {RequestId, {error, Reason}}} ->
+	    ?hcrt("received error", [{request_id, RequestId}, 
+				     {reason,     Reason}]),
 	    {error, Reason}
     end.
 
diff --git a/lib/inets/src/http_client/httpc_handler.erl b/lib/inets/src/http_client/httpc_handler.erl
index cb6f3e2841..5e22400fe0 100644
--- a/lib/inets/src/http_client/httpc_handler.erl
+++ b/lib/inets/src/http_client/httpc_handler.erl
@@ -1724,9 +1724,27 @@ handle_verbose(_) ->
 %%     ok.
 
 
+send_raw(#session{socket = Socket, socket_type = SocketType}, 
+	 {ProcessBody, Acc}) when is_function(ProcessBody, 1) ->
+    ?hcrt("send raw", [{acc, Acc}]),
+    send_raw(SocketType, Socket, ProcessBody, Acc);
 send_raw(#session{socket = Socket, socket_type = SocketType}, Body) ->
     http_transport:send(SocketType, Socket, Body).
 
+send_raw(SocketType, Socket, ProcessBody, Acc) ->
+    case ProcessBody(Acc) of
+        eof ->
+            ok;
+        {ok, Data, NewAcc} ->
+            DataBin = iolist_to_binary(Data),
+            ?hcrd("send", [{data, DataBin}]),
+            case http_transport:send(SocketType, Socket, DataBin) of
+                ok ->
+                    send_raw(SocketType, Socket, ProcessBody, NewAcc);
+                Error ->
+                    Error
+            end
+    end.
 
 
 call(Msg, Pid) ->
diff --git a/lib/inets/src/http_client/httpc_request.erl b/lib/inets/src/http_client/httpc_request.erl
index d4df97ad40..0d602adb11 100644
--- a/lib/inets/src/http_client/httpc_request.erl
+++ b/lib/inets/src/http_client/httpc_request.erl
@@ -79,37 +79,65 @@ send(SendAddr, Socket, SocketType,
 	   {settings,      HttpOptions}, 
 	   {userinfo,      UserInfo}]),
 
-    TmpHeaders = handle_user_info(UserInfo, Headers),
+    TmpHdrs = handle_user_info(UserInfo, Headers),
 
-    {TmpHeaders2, Body} = 
-	post_data(Method, TmpHeaders, Content, HeadersAsIs),
+    {TmpHdrs2, Body} = post_data(Method, TmpHdrs, Content, HeadersAsIs),
     
-    {NewHeaders, Uri} = case Address of
-			    SendAddr ->
-				{TmpHeaders2, Path ++ Query};
-			    _Proxy ->
-				TmpHeaders3 =
-				    handle_proxy(HttpOptions, TmpHeaders2),
-				{TmpHeaders3, AbsUri}
-			end,
-
-    FinalHeaders = case NewHeaders of
-		       HeaderList when is_list(HeaderList) ->
-			   http_headers(HeaderList, []);
-		       _  ->
-			   http_request:http_headers(NewHeaders)
-		   end,
+    {NewHeaders, Uri} = 
+	case Address of
+	    SendAddr ->
+		{TmpHdrs2, Path ++ Query};
+	    _Proxy ->
+		TmpHdrs3 = handle_proxy(HttpOptions, TmpHdrs2), 
+		{TmpHdrs3, AbsUri}
+	end,
+    
+    FinalHeaders = 
+	case NewHeaders of
+	    HeaderList when is_list(HeaderList) ->
+		http_headers(HeaderList, []);
+	    _  ->
+		http_request:http_headers(NewHeaders)
+	end,
     Version = HttpOptions#http_options.version,
 
-    Message = [method(Method), " ", Uri, " ", 
-	       version(Version), ?CRLF, 
-	       headers(FinalHeaders, Version), ?CRLF, Body],
+    do_send_body(SocketType, Socket, Method, Uri, Version, FinalHeaders, Body).
+
+
+do_send_body(SocketType, Socket, Method, Uri, Version, Headers, 
+	     {ProcessBody, Acc}) when is_function(ProcessBody, 1) ->
+    ?hcrt("send", [{acc, Acc}]),
+    case do_send_body(SocketType, Socket, Method, Uri, Version, Headers, []) of
+        ok ->
+            do_send_body(SocketType, Socket, ProcessBody, Acc);
+        Error ->
+            Error
+    end;
 
+do_send_body(SocketType, Socket, Method, Uri, Version, Headers, Body) ->
+    ?hcrt("create message", [{body, Body}]),
+    Message = [method(Method), " ", Uri, " ",
+	       version(Version), ?CRLF,
+	       headers(Headers, Version), ?CRLF, Body],
     ?hcrd("send", [{message, Message}]),
-    
     http_transport:send(SocketType, Socket, lists:append(Message)).
 
 
+do_send_body(SocketType, Socket, ProcessBody, Acc) ->
+    case ProcessBody(Acc) of
+        eof ->
+            ok;
+        {ok, Data, NewAcc} ->
+            DataBin = iolist_to_binary(Data),
+            ?hcrd("send", [{data, DataBin}]),
+            case http_transport:send(SocketType, Socket, DataBin) of
+                ok ->
+                    do_send_body(SocketType, Socket, ProcessBody, NewAcc);
+                Error ->
+                    Error
+            end
+    end.
+
 
 %%-------------------------------------------------------------------------
 %% is_idempotent(Method) ->
@@ -161,7 +189,6 @@ is_client_closing(Headers) ->
 %%%========================================================================
 post_data(Method, Headers, {ContentType, Body}, HeadersAsIs) 
   when (Method =:= post) orelse (Method =:= put) ->
-    ContentLength = body_length(Body),	      
     NewBody = case Headers#http_request_h.expect of
 		  "100-continue" ->
 		      "";
@@ -170,14 +197,22 @@ post_data(Method, Headers, {ContentType, Body}, HeadersAsIs)
 	      end,
     
     NewHeaders = case HeadersAsIs of
-		     [] ->
-			 Headers#http_request_h{'content-type' = 
-						ContentType, 
-						'content-length' = 
-						ContentLength};
-		     _ ->
-			 HeadersAsIs
-		 end,
+        [] ->
+            Headers#http_request_h{
+                'content-type' = ContentType,
+                'content-length' = case body_length(Body) of
+                    undefined ->
+                        % on upload streaming the caller must give a
+                        % value to the Content-Length header
+                        % (or use chunked Transfer-Encoding)
+                        Headers#http_request_h.'content-length';
+                    Len when is_list(Len) ->
+                        Len
+                    end
+            };
+        _ ->
+            HeadersAsIs
+    end,
     
     {NewHeaders, NewBody};
 
@@ -190,7 +225,10 @@ body_length(Body) when is_binary(Body) ->
    integer_to_list(size(Body));
 
 body_length(Body) when is_list(Body) ->
-  integer_to_list(length(Body)).
+  integer_to_list(length(Body));
+
+body_length({DataFun, _Acc}) when is_function(DataFun, 1) ->
+  undefined.
 
 method(Method) ->
     http_util:to_upper(atom_to_list(Method)).
diff --git a/lib/inets/src/inets_app/inets.appup.src b/lib/inets/src/inets_app/inets.appup.src
index 07da8ca961..65ce0cf0eb 100644
--- a/lib/inets/src/inets_app/inets.appup.src
+++ b/lib/inets/src/inets_app/inets.appup.src
@@ -18,9 +18,20 @@
 
 {"%VSN%",
  [
+  {"5.5.2", 
+   [
+    {load_module, httpc, soft_purge, soft_purge, [httpc_handler]},
+    {load_module, httpc_request, soft_purge, soft_purge, []},
+    {update, httpc_handler, soft, soft_purge, soft_purge, [httpc_request]}
+   ]
+  }, 
   {"5.5.1", 
    [
-    {load_module, http_chunk, soft_purge, soft_purge, []}
+    {load_module, http_chunk, soft_purge, soft_purge, []},
+    {load_module, httpc, soft_purge, soft_purge, [httpc_handler]},
+    {load_module, httpc_request, soft_purge, soft_purge, []},
+    {update, httpc_handler, soft, soft_purge, soft_purge, 
+     [httpc_request,http_chunk]}
    ]
   }, 
   {"5.5",
@@ -34,10 +45,21 @@
    ]
   } 
  ],
- [	
+ [
+  {"5.5.2",
+   [
+    {load_module, httpc, soft_purge, soft_purge, [httpc_handler]},
+    {load_module, httpc_request, soft_purge, soft_purge, []},
+    {update, httpc_handler, soft, soft_purge, soft_purge, [httpc_request]}
+   ]
+  },
   {"5.5.1", 
    [
-    {load_module, http_chunk, soft_purge, soft_purge, []}
+    {load_module, http_chunk, soft_purge, soft_purge, []},
+    {load_module, httpc, soft_purge, soft_purge, [httpc_handler]},
+    {load_module, httpc_request, soft_purge, soft_purge, []},
+    {update, httpc_handler, soft, soft_purge, soft_purge, 
+     [httpc_request,http_chunk]}
    ]
   }, 
   {"5.5",