fix docker client

src/docker/docker_client.erl

@@ -12,7 +12,7 @@
 -behaviour(gen_server).
 
 %% API
--export([request/4, start_stream/5, stream_request/5]).
+-export([request/4, start_stream/5]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
@@ -73,21 +73,6 @@ start_stream(Owner, Method, Path, Body, Headers)
             {error, Reason}
     end.
 
-%% Backward-compatible wrapper for existing callers. New stream users should prefer start_stream/5.
--spec stream_request(Callback :: fun((term()) -> any()), Method :: string(), Path :: string(), Body :: binary(), Headers :: list()) ->
-    ok | {error, Reason :: any()}.
-stream_request(Callback, Method, Path, Body, Headers)
-  when is_function(Callback, 1), is_list(Method), is_list(Path), is_binary(Body), is_list(Headers) ->
-    case start_stream(self(), Method, Path, Body, Headers) of
-        {ok, Ref, Pid} ->
-            MRef = erlang:monitor(process, Pid),
-            await_stream(Ref, Callback, Pid, MRef);
-        {error, Reason0} ->
-            Reason = format_error(Reason0),
-            Callback({error, Reason}),
-            {error, Reason}
-    end.
-
 %%%===================================================================
 %%% gen_server callbacks
 %%%===================================================================
@@ -307,27 +292,6 @@ receive_stream_error_body(State, ConnPid, StreamRef, Status, Acc) ->
         {error, Reason}
     end.
 
--spec await_stream(reference(), fun((term()) -> any()), pid(), reference()) -> ok | {error, binary()}.
-await_stream(Ref, Callback, Pid, MRef) ->
-    receive
-        {docker_client, Ref, {response, _Status, _Headers}} ->
-            await_stream(Ref, Callback, Pid, MRef);
-        {docker_client, Ref, {data, Data}} ->
-            Callback({message, Data}),
-            await_stream(Ref, Callback, Pid, MRef);
-        {docker_client, Ref, done} ->
-            erlang:demonitor(MRef, [flush]),
-            ok;
-        {docker_client, Ref, {error, Reason}} ->
-            erlang:demonitor(MRef, [flush]),
-            Callback({error, Reason}),
-            {error, Reason};
-        {'DOWN', MRef, process, Pid, Reason0} ->
-            Reason = format_error({client_down, Reason0}),
-            Callback({error, Reason}),
-            {error, Reason}
-    end.
-
 -spec open_connection() -> {ok, pid()} | {error, any()}.
 open_connection() ->
     case gun:open_unix(?DOCKER_SOCKET, #{}) of

src/docker/docker_commands.erl

@@ -11,15 +11,15 @@
 -include("message_pb.hrl").
 
 %% API
--export([pull_image/2, check_image_exist/1]).
+-export([pull_image/1, check_image_exist/1]).
 -export([create_container/2, check_container_exist/1, is_container_running/1,
     start_container/1, stop_container/1, stop_container/2, remove_container/1, remove_container/3, kill_container/1, kill_container/2,
     get_containers/0]).
 
--spec pull_image(Image :: binary(), Callback :: fun((Msg :: any()) -> no_return())) -> ok | {error, Reason :: any()}.
+-spec pull_image(Image :: binary()) -> {ok, Ref :: reference(), Pid :: pid()} | {error, Reason :: any()}.
-pull_image(Image, Callback) when is_binary(Image), is_function(Callback, 1) ->
+pull_image(Image) when is_binary(Image) ->
     Url = lists:flatten(io_lib:format("/images/create?fromImage=~s", [binary_to_list(Image)])),
-    docker_client:stream_request(Callback, "POST", Url, <<>>, []).
+    docker_client:start_stream(self(), "POST", Url, <<>>, []).
 
 -spec check_image_exist(Image :: binary()) -> boolean().
 check_image_exist(Image) when is_binary(Image) ->

src/docker/docker_deployer.erl

@@ -125,14 +125,9 @@ ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
     Image = normalize_image(Image0),
     report_task_event(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
     report_task_event(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
-    CB = fun
+    case docker_commands:pull_image(Image) of
-             ({message, M}) ->
+        {ok, Ref, Pid} ->
-                 report_task_event(TaskId, <<"info">>, M);
+            await_pull_image(TaskId, Ref, Pid),
-             ({error, Error}) ->
-                 report_task_event(TaskId, <<"error">>, Error)
-         end,
-    case docker_commands:pull_image(Image, CB) of
-        ok ->
             {ok, Image};
         {error, Reason} when is_binary(Reason) ->
             throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>});
@@ -141,6 +136,31 @@ ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
             throw({deploy_error, Error})
     end.
 
+-spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid()) -> ok.
+await_pull_image(TaskId, Ref, Pid) when is_integer(TaskId), is_reference(Ref), is_pid(Pid) ->
+    MRef = erlang:monitor(process, Pid),
+    await_pull_image(TaskId, Ref, Pid, MRef).
+
+-spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok.
+await_pull_image(TaskId, Ref, Pid, MRef) ->
+    receive
+        {docker_client, Ref, {response, _Status, _Headers}} ->
+            await_pull_image(TaskId, Ref, Pid, MRef);
+        {docker_client, Ref, {data, Data}} ->
+            report_task_event(TaskId, <<"info">>, Data),
+            await_pull_image(TaskId, Ref, Pid, MRef);
+        {docker_client, Ref, done} ->
+            erlang:demonitor(MRef, [flush]),
+            ok;
+        {docker_client, Ref, {error, Reason}} ->
+            erlang:demonitor(MRef, [flush]),
+            report_task_event(TaskId, <<"error">>, Reason),
+            throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>});
+        {'DOWN', MRef, process, Pid, Reason0} ->
+            Reason = iolist_to_binary(io_lib:format("~p", [Reason0])),
+            throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>})
+    end.
+
 -spec create_container_and_config(TaskId :: integer(), ContainerDir :: string(), Params :: message_pb:'ContainerDeployParams'()) ->
     {ok, binary()}.
 create_container_and_config(TaskId, ContainerDir, Params)

src/tests/docker_commands_tests.erl

@@ -67,7 +67,31 @@ test_all() ->
 
 -spec test_pull() -> ok.
 test_pull() ->
-    ok = docker_commands:pull_image(?TEST_IMAGE, fun(Msg) -> logger:debug("msg is: ~p", [Msg]) end).
+    {ok, Ref, Pid} = docker_commands:pull_image(?TEST_IMAGE),
+    await_pull(Ref, Pid).
+
+-spec await_pull(reference(), pid()) -> ok.
+await_pull(Ref, Pid) when is_reference(Ref), is_pid(Pid) ->
+    MRef = erlang:monitor(process, Pid),
+    await_pull(Ref, Pid, MRef).
+
+-spec await_pull(reference(), pid(), reference()) -> ok.
+await_pull(Ref, Pid, MRef) ->
+    receive
+        {docker_client, Ref, {response, _Status, _Headers}} ->
+            await_pull(Ref, Pid, MRef);
+        {docker_client, Ref, {data, Data}} ->
+            logger:debug("msg is: ~p", [Data]),
+            await_pull(Ref, Pid, MRef);
+        {docker_client, Ref, done} ->
+            erlang:demonitor(MRef, [flush]),
+            ok;
+        {docker_client, Ref, {error, Reason}} ->
+            erlang:demonitor(MRef, [flush]),
+            error({pull_failed, Reason});
+        {'DOWN', MRef, process, Pid, Reason} ->
+            error({pull_client_down, Reason})
+    end.
 
 -spec test_check_image_exist() -> ok.
 test_check_image_exist() ->