diff --git a/src/docker/docker_client.erl b/src/docker/docker_client.erl index 6681faf..ff67ef4 100644 --- a/src/docker/docker_client.erl +++ b/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. -stream_worker(Owner, Ref, Method, Path, Body, Headers) -> +-spec stream_worker(pid(), reference(), string(), string(), binary(), list(), timeout(), timeout()) -> ok. +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()}. -receive_stream_response(Owner, OwnerRef, Ref, ConnPid, StreamRef) -> +-spec receive_stream_response(pid(), reference(), reference(), pid(), reference(), timeout(), timeout()) -> ok | {error, binary()}. +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()}. -receive_stream_body(Owner, OwnerRef, Ref, ConnPid, StreamRef) -> +-spec receive_stream_body(pid(), reference(), reference(), pid(), reference(), timeout()) -> ok | {error, binary()}. +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 diff --git a/src/docker/docker_commands.erl b/src/docker/docker_commands.erl index 5924a66..ec7d4eb 100644 --- a/src/docker/docker_commands.erl +++ b/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) ->