fix docker_client

src/docker/docker_client.erl

@@ -9,7 +9,7 @@
 -module(docker_client).
 
 %% API
--export([request/4, start_stream/5]).
+-export([request/4, start_stream/5, start_stream_no_timeout/5]).
 
 -define(DOCKER_SOCKET, "/var/run/docker.sock").
 -define(RESPONSE_TIMEOUT, 5000).
@@ -44,7 +44,15 @@ request(Method, Path, Body, Headers) when is_list(Method), is_list(Path), is_bin
 start_stream(Owner, Method, Path, Body, Headers)
   when is_pid(Owner), is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) ->
     Ref = make_ref(),
-    {Pid, MRef} = spawn_monitor(fun() -> stream_worker(Owner, Ref, Method, Path, Body, Headers) end),
+    {Pid, MRef} = spawn_monitor(fun() -> stream_worker(Owner, Ref, Method, Path, Body, Headers, ?RESPONSE_TIMEOUT, ?STREAM_BODY_TIMEOUT) end),
+    {ok, Ref, Pid, MRef}.
+
+-spec start_stream_no_timeout(Owner :: pid(), Method :: string(), Path :: string(), Body :: binary(), Headers :: list()) ->
+    {ok, Ref :: reference(), Pid :: pid(), MRef :: reference()}.
+start_stream_no_timeout(Owner, Method, Path, Body, Headers)
+  when is_pid(Owner), is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) ->
+    Ref = make_ref(),
+    {Pid, MRef} = spawn_monitor(fun() -> stream_worker(Owner, Ref, Method, Path, Body, Headers, infinity, infinity) end),
     {ok, Ref, Pid, MRef}.
 
 %%%===================================================================
@@ -56,10 +64,10 @@ request_worker(Owner, Ref, Method, Path, Body, Headers) ->
     Result = do_request(Method, Path, Body, Headers),
     send_owner(Owner, Ref, {result, Result}).
 
--spec stream_worker(pid(), reference(), string(), string(), binary(), list()) -> ok.
+-spec stream_worker(pid(), reference(), string(), string(), binary(), list(), timeout(), timeout()) -> ok.
-stream_worker(Owner, Ref, Method, Path, Body, Headers) ->
+stream_worker(Owner, Ref, Method, Path, Body, Headers, ResponseTimeout, BodyTimeout) ->
     OwnerRef = erlang:monitor(process, Owner),
-    _Result = do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers),
+    _Result = do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers, ResponseTimeout, BodyTimeout),
     erlang:demonitor(OwnerRef, [flush]),
     ok.
 
@@ -82,14 +90,14 @@ do_request(Method, Path, Body, Headers) ->
             {error, Reason}
     end.
 
--spec do_stream(pid(), reference(), reference(), string(), string(), binary(), list()) ->
+-spec do_stream(pid(), reference(), reference(), string(), string(), binary(), list(), timeout(), timeout()) ->
     ok | {error, binary()}.
-do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers) ->
+do_stream(Owner, OwnerRef, Ref, Method, Path, Body, Headers, ResponseTimeout, BodyTimeout) ->
     case open_connection() of
         {ok, ConnPid} ->
             try
                 StreamRef = gun:request(ConnPid, Method, Path, Headers, Body),
-                receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef)
+                receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef, ResponseTimeout, BodyTimeout)
             after
                 close_connection(ConnPid)
             end;
@@ -136,18 +144,18 @@ receive_body(ConnPid, StreamRef, Status, Headers, Acc) ->
         {error, timeout}
     end.
 
--spec receive_stream_response(pid(), reference(), reference(), pid(), reference()) -> ok | {error, binary()}.
+-spec receive_stream_response(pid(), reference(), reference(), pid(), reference(), timeout(), timeout()) -> ok | {error, binary()}.
-receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
+receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef, ResponseTimeout, BodyTimeout) ->
     receive
         {gun_response, ConnPid, StreamRef, nofin, Status, Headers} when Status >= 200, Status < 300 ->
             send_owner(Owner, Ref, {response, Status, Headers}),
-            receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef);
+            receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, BodyTimeout);
         {gun_response, ConnPid, StreamRef, fin, Status, Headers} when Status >= 200, Status < 300 ->
             send_owner(Owner, Ref, {response, Status, Headers}),
             send_owner(Owner, Ref, done),
             ok;
         {gun_response, ConnPid, StreamRef, nofin, Status, _Headers} ->
-            receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, <<>>);
+            receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, <<>>, stream_error_body_timeout(BodyTimeout));
         {gun_response, ConnPid, StreamRef, fin, Status, _Headers} ->
             Reason = http_status_error(Status, <<>>),
             send_owner(Owner, Ref, {error, Reason}),
@@ -166,13 +174,13 @@ receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
             {error, Reason};
         {'DOWN', OwnerRef, process, _Pid, Reason0} ->
             {error, format_error({owner_down, Reason0})}
-    after ?RESPONSE_TIMEOUT ->
+    after ResponseTimeout ->
         send_owner(Owner, Ref, {error, <<"处理超时"/utf8>>}),
         {error, <<"timeout">>}
     end.
 
--spec receive_stream_body(pid(), reference(), reference(), pid(), reference()) -> ok | {error, binary()}.
+-spec receive_stream_body(pid(), reference(), reference(), pid(), reference(), timeout()) -> ok | {error, binary()}.
-receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
+receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, BodyTimeout) ->
     receive
         {gun_data, ConnPid, StreamRef, fin, Data} ->
             maybe_send_stream_data(Owner, Ref, Data),
@@ -180,7 +188,7 @@ receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
             ok;
         {gun_data, ConnPid, StreamRef, nofin, Data} ->
             maybe_send_stream_data(Owner, Ref, Data),
-            receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef);
+            receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, BodyTimeout);
         {gun_error, ConnPid, StreamRef, Reason0} ->
             Reason = format_error(Reason0),
             send_owner(Owner, Ref, {error, Reason}),
@@ -195,21 +203,21 @@ receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef) ->
             {error, Reason};
         {'DOWN', OwnerRef, process, _Pid, Reason0} ->
             {error, format_error({owner_down, Reason0})}
-    after ?STREAM_BODY_TIMEOUT ->
+    after BodyTimeout ->
         send_owner(Owner, Ref, {error, <<"timeout">>}),
         {error, <<"timeout">>}
     end.
 
--spec receive_stream_error_body(pid(), reference(), reference(), pid(), reference(), integer(), iodata()) ->
+-spec receive_stream_error_body(pid(), reference(), reference(), pid(), reference(), integer(), iodata(), timeout()) ->
     {error, binary()}.
-receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, Acc) ->
+receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, Acc, BodyTimeout) ->
     receive
         {gun_data, ConnPid, StreamRef, fin, Data} ->
             Reason = http_status_error(Status, iolist_to_binary([Acc, Data])),
             send_owner(Owner, Ref, {error, Reason}),
             {error, Reason};
         {gun_data, ConnPid, StreamRef, nofin, Data} ->
-            receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, [Acc, Data]);
+            receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, [Acc, Data], BodyTimeout);
         {gun_error, ConnPid, StreamRef, Reason0} ->
             Reason = format_error(Reason0),
             send_owner(Owner, Ref, {error, Reason}),
@@ -224,12 +232,18 @@ receive_stream_error_body(Owner, OwnerRef, Ref, ConnPid, StreamRef, Status, Acc)
             {error, Reason};
         {'DOWN', OwnerRef, process, _Pid, Reason0} ->
             {error, format_error({owner_down, Reason0})}
-    after ?BODY_TIMEOUT ->
+    after BodyTimeout ->
         Reason = http_status_error(Status, iolist_to_binary(Acc)),
         send_owner(Owner, Ref, {error, Reason}),
         {error, Reason}
     end.
 
+-spec stream_error_body_timeout(timeout()) -> timeout().
+stream_error_body_timeout(infinity) ->
+    infinity;
+stream_error_body_timeout(_BodyTimeout) ->
+    ?BODY_TIMEOUT.
+
 -spec open_connection() -> {ok, pid()} | {error, any()}.
 open_connection() ->
     case gun:open_unix(?DOCKER_SOCKET, #{}) of

src/docker/docker_commands.erl

@@ -18,7 +18,7 @@
 -spec pull_image(Image :: binary()) -> {ok, Ref :: reference(), Pid :: pid(), MRef :: reference()} | {error, Reason :: any()}.
 pull_image(Image) when is_binary(Image) ->
     Url = lists:flatten(io_lib:format("/images/create?fromImage=~s", [binary_to_list(Image)])),
-    docker_client:start_stream(self(), "POST", Url, <<>>, []).
+    docker_client:start_stream_no_timeout(self(), "POST", Url, <<>>, []).
 
 -spec check_image_exist(Image :: binary()) -> boolean().
 check_image_exist(Image) when is_binary(Image) ->