fix config

apps/docker/src/docker.app.src

@@ -9,6 +9,9 @@
     kernel,
     stdlib
    ]},
+  {env, [
+    {root_dir, "/usr/local/code/efka"}
+  ]},
   {modules, []},
   {licenses, ["Apache-2.0"]},
   {links, []}

apps/docker/src/docker_deploy_manager.erl

@@ -0,0 +1,90 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2026, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 20. 4月 2026
+%%%-------------------------------------------------------------------
+-module(docker_deploy_manager).
+-author("anlicheng").
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0, deploy/2]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(SERVER, ?MODULE).
+
+-record(state, {
+    root_dir :: string(),
+    task_map = #{}
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+-spec deploy(integer(), map()) -> ok | {error, binary()}.
+deploy(TaskId, Params) when is_integer(TaskId), is_map(Params) ->
+    gen_server:call(?SERVER, {deploy, TaskId, Params}).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+-spec init(list()) -> {ok, #state{}}.
+init([]) ->
+    erlang:process_flag(trap_exit, true),
+    {ok, RootDir} = docker_helper:root_dir(),
+    {ok, #state{root_dir = RootDir}}.
+
+-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, term(), #state{}}.
+handle_call({deploy, TaskId, Params}, _From, State = #state{root_dir = RootDir, task_map = TaskMap})
+  when is_map(Params) ->
+    ContainerName = maps:get(<<"container_name">>, Params),
+    {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName),
+    {ok, {TaskPid, _Ref}} = docker_deployer:start_monitor(TaskId, ContainerDir, Params),
+    logger:debug("[docker_deploy_manager] start deploy task_id: ~p, params: ~p", [TaskId, Params]),
+    {reply, ok, State#state{task_map = maps:put(TaskPid, TaskId, TaskMap)}};
+handle_call(_Request, _From, State) ->
+    {reply, ok, State}.
+
+-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
+handle_info({'DOWN', _Ref, process, TaskPid, Reason}, State = #state{task_map = TaskMap}) ->
+    case maps:take(TaskPid, TaskMap) of
+        error ->
+            {noreply, State};
+        {TaskId, NTaskMap} ->
+            case Reason of
+                normal ->
+                    logger:debug("[docker_deploy_manager] task_id: ~p, exit normal", [TaskId]);
+                Error0 ->
+                    Error = iolist_to_binary(io_lib:format("~p", [Error0])),
+                    docker_task_reporter:stream(TaskId, <<"error">>, <<"任务失败: "/utf8, Error/binary>>),
+                    docker_task_reporter:close(TaskId, <<"task exited">>),
+                    logger:notice("[docker_deploy_manager] task_id: ~p, exit with error: ~p", [TaskId, Error])
+            end,
+            {noreply, State#state{task_map = NTaskMap}}
+    end;
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+-spec terminate(term(), #state{}) -> ok.
+terminate(_Reason, _State) ->
+    ok.
+
+-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.

apps/docker/src/docker_deployer.erl

@@ -11,17 +11,26 @@
 -dialyzer([{nowarn_function, normalize_image/1}]).
 
 %% API
--export([deploy/4]).
+-export([start_monitor/3]).
+-export([deploy/3]).
 
 -define(TASK_SUCCESS, <<"success">>).
 -define(TASK_FAIL, <<"fail">>).
 
--type reporter() :: {stream, pos_integer()}.
-
 %%%===================================================================
 %%% API
 %%%===================================================================
 
+-spec(start_monitor(TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
+    {ok, {pid(), reference()}}).
+start_monitor(TaskId, ContainerDir, Params)
+  when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
+    {ok, spawn_monitor(?MODULE, deploy, [TaskId, ContainerDir, Params])}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
 %{
 %   "image": "nginx:latest",
 %   "container_name": "my_nginx",
@@ -32,35 +41,34 @@
 %   "command": ["nginx", "-g", "daemon off;"],
 %   "restart": "always"
 %}
--spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: map(), Reporter :: reporter()) -> ok.
-deploy(TaskId, ContainerDir, Params, Reporter) when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
+-spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: map()) -> ok.
+deploy(TaskId, ContainerDir, Params) when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
     ContainerName = deploy_container_name(Params),
     Image0 = deploy_image(Params),
-    report_stream_event(Reporter, TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
+    report_task_event(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
     try
-        ok = ensure_container_absent(Reporter, TaskId, ContainerName),
-        {ok, Image} = ensure_image_ready(Reporter, TaskId, Image0),
-        {ok, ContainerId} = create_container_and_config(Reporter, TaskId, ContainerDir, Params),
+        ok = ensure_container_absent(TaskId, ContainerName),
+        {ok, Image} = ensure_image_ready(TaskId, Image0),
+        {ok, ContainerId} = create_container_and_config(TaskId, ContainerDir, Params),
         ShortContainerId = short_container_id(ContainerId),
-        report_stream_event(Reporter, TaskId, <<"container_id">>, ContainerId),
-        report_stream_event(Reporter, TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>),
-        report_stream_event(Reporter, TaskId, <<"info">>, <<"任务完成"/utf8>>),
+        report_task_event(TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>),
+        report_task_event(TaskId, <<"info">>, <<"任务完成"/utf8>>),
         write_task_summary(TaskId, <<"success">>, ContainerName, Image, ContainerId),
-        close_task(Reporter, TaskId, ?TASK_SUCCESS)
+        docker_task_reporter:close(TaskId, ?TASK_SUCCESS)
     catch
         throw:{deploy_error, Reason} ->
-            report_stream_event(Reporter, TaskId, <<"error">>, Reason),
-            report_stream_event(Reporter, TaskId, <<"error">>, <<"任务失败"/utf8>>),
+            report_task_event(TaskId, <<"error">>, Reason),
+            report_task_event(TaskId, <<"error">>, <<"任务失败"/utf8>>),
             write_task_summary(TaskId, <<"fail">>, ContainerName, Image0, undefined),
             write_task_failure_reason(TaskId, Reason),
-            close_task(Reporter, TaskId, ?TASK_FAIL);
+            docker_task_reporter:close(TaskId, ?TASK_FAIL);
         Class:Reason:Stacktrace ->
             Error = iolist_to_binary(io_lib:format("deploy crashed: ~p:~p ~p", [Class, Reason, Stacktrace])),
-            report_stream_event(Reporter, TaskId, <<"error">>, Error),
-            report_stream_event(Reporter, TaskId, <<"error">>, <<"任务失败"/utf8>>),
+            report_task_event(TaskId, <<"error">>, Error),
+            report_task_event(TaskId, <<"error">>, <<"任务失败"/utf8>>),
             write_task_summary(TaskId, <<"fail">>, ContainerName, Image0, undefined),
             write_task_failure_reason(TaskId, Error),
-            close_task(Reporter, TaskId, ?TASK_FAIL)
+            docker_task_reporter:close(TaskId, ?TASK_FAIL)
     end.
 
 -spec normalize_image(binary()) -> binary().
@@ -73,33 +81,9 @@ normalize_image(Image) when is_binary(Image) ->
                      end,
     iolist_to_binary(lists:join(<<"/">>, PrefixParts ++ [NormalizedLast])).
 
--spec report_stream_event(reporter(), TaskId :: integer(), Level :: binary(), Msg :: binary()) -> ok.
-report_stream_event({stream, StreamId}, _TaskId, Level, Msg) when is_integer(StreamId), is_binary(Level), is_binary(Msg) ->
-    efka_iot_client:send_stream(StreamId, {data, encode_stream_event(Level, Msg)}).
-
--spec close_task(reporter(), integer(), binary()) -> ok.
-close_task({stream, StreamId}, _TaskId, Reason) when is_integer(StreamId), is_binary(Reason) ->
-    ok = efka_iot_client:send_stream(StreamId, {data, encode_close_event(Reason)}),
-    efka_iot_client:send_stream(StreamId, fin).
-
--spec encode_stream_event(binary(), binary()) -> binary().
-encode_stream_event(<<"container_id">>, ContainerId) ->
-    iolist_to_binary(json:encode(#{
-        <<"type">> => <<"container_id">>,
-        <<"container_id">> => ContainerId
-    }));
-encode_stream_event(Type, Msg) ->
-    iolist_to_binary(json:encode(#{
-        <<"type">> => Type,
-        <<"message">> => Msg
-    })).
-
--spec encode_close_event(binary()) -> binary().
-encode_close_event(Reason) ->
-    iolist_to_binary(json:encode(#{
-        <<"type">> => <<"close">>,
-        <<"reason">> => Reason
-    })).
+-spec report_task_event(TaskId :: integer(), Level :: binary(), Msg :: binary()) -> ok.
+report_task_event(TaskId, Level, Msg) when is_integer(TaskId), is_binary(Level), is_binary(Msg) ->
+    docker_task_reporter:stream(TaskId, Level, Msg).
 
 -spec write_task_summary(integer(), binary(), binary(), binary(), undefined | binary()) -> ok.
 write_task_summary(TaskId, Status, ContainerName, Image, ContainerId)
@@ -129,55 +113,49 @@ write_task_failure_reason(TaskId, Reason) when is_integer(TaskId), is_binary(Rea
     ]),
     efka_logger:write(Info).
 
--spec ensure_container_absent(reporter(), TaskId :: integer(), ContainerName :: binary()) -> ok.
-ensure_container_absent(Reporter, TaskId, ContainerName) when is_integer(TaskId), is_binary(ContainerName) ->
-    report_stream_event(Reporter, TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>),
+-spec ensure_container_absent(TaskId :: integer(), ContainerName :: binary()) -> ok.
+ensure_container_absent(TaskId, ContainerName) when is_integer(TaskId), is_binary(ContainerName) ->
+    report_task_event(TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>),
     ok.
 
--spec ensure_image_ready(reporter(), TaskId :: integer(), Image0 :: binary()) -> {ok, binary()}.
-ensure_image_ready(Reporter, TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
+-spec ensure_image_ready(TaskId :: integer(), Image0 :: binary()) -> {ok, binary()}.
+ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
     Image = normalize_image(Image0),
-    report_stream_event(Reporter, TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
-    case docker_commands:check_image_exist(Image) of
-        true ->
-            report_stream_event(Reporter, TaskId, <<"info">>, <<"本地镜像已存在，跳过拉取:"/utf8, Image/binary>>),
-            {ok, Image};
-        false ->
-            report_stream_event(Reporter, TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
-            case docker_commands:pull_image(Image) of
-                {ok, Ref, Pid, MRef} ->
-                    await_pull_image(Reporter, TaskId, Ref, Pid, MRef),
-                    {ok, Image}
-            end
+    report_task_event(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
+    report_task_event(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
+    case docker_commands:pull_image(Image) of
+        {ok, Ref, Pid, MRef} ->
+            await_pull_image(TaskId, Ref, Pid, MRef),
+            {ok, Image}
     end.
 
--spec await_pull_image(reporter(), TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok.
-await_pull_image(Reporter, 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(Reporter, TaskId, Ref, Pid, MRef);
+            await_pull_image(TaskId, Ref, Pid, MRef);
         {docker_client, Ref, {data, Data}} ->
-            report_stream_event(Reporter, TaskId, <<"info">>, Data),
-            await_pull_image(Reporter, TaskId, Ref, Pid, MRef);
+            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_stream_event(Reporter, TaskId, <<"error">>, Reason),
+            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(reporter(), TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
+-spec create_container_and_config(TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
     {ok, binary()}.
-create_container_and_config(Reporter, TaskId, ContainerDir, Params)
+create_container_and_config(TaskId, ContainerDir, Params)
   when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
     case docker_commands:create_container(ContainerDir, Params) of
         {ok, ContainerId} ->
-            ok = create_config_file(Reporter, TaskId, ContainerDir),
+            ok = create_config_file(TaskId, ContainerDir),
             {ok, ContainerId};
         {error, Reason} when is_binary(Reason) ->
             throw({deploy_error, format_create_container_error(Reason)});
@@ -186,8 +164,8 @@ create_container_and_config(Reporter, TaskId, ContainerDir, Params)
             throw({deploy_error, Error})
     end.
 
--spec create_config_file(reporter(), TaskId :: integer(), ContainerDir :: string()) -> ok.
-create_config_file(Reporter, TaskId, ContainerDir) when is_integer(TaskId), is_list(ContainerDir) ->
+-spec create_config_file(TaskId :: integer(), ContainerDir :: string()) -> ok.
+create_config_file(TaskId, ContainerDir) when is_integer(TaskId), is_list(ContainerDir) ->
     ConfigFile = docker_helper:get_config_file(ContainerDir),
     case file:open(ConfigFile, [write, exclusive]) of
         {ok, FD} ->
@@ -196,7 +174,7 @@ create_config_file(Reporter, TaskId, ContainerDir) when is_integer(TaskId), is_l
             ok;
         {error, Reason} ->
             ReasonBin = list_to_binary(io_lib:format("~p", [Reason])),
-            report_stream_event(Reporter, TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, ReasonBin/binary>>),
+            report_task_event(TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, ReasonBin/binary>>),
             ok
     end.
 

apps/docker/src/docker_sup.erl

@@ -19,6 +19,24 @@ start_link() ->
 -spec init(list()) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
 init([]) ->
     SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
-    ChildSpecs = [],
+    ChildSpecs = [
+        #{
+            id => docker_task_reporter,
+            start => {docker_task_reporter, start_link, []},
+            restart => permanent,
+            shutdown => 2000,
+            type => worker,
+            modules => [docker_task_reporter]
+        },
+
+        #{
+            id => docker_deploy_manager,
+            start => {docker_deploy_manager, start_link, []},
+            restart => permanent,
+            shutdown => 2000,
+            type => worker,
+            modules => [docker_deploy_manager]
+        }
+    ],
 
     {ok, {SupFlags, ChildSpecs}}.

apps/docker/src/docker_task_reporter.erl

@@ -0,0 +1,121 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2026, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 20. 4月 2026
+%%%-------------------------------------------------------------------
+-module(docker_task_reporter).
+-author("anlicheng").
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0]).
+-export([stream/3, close/2]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(SERVER, ?MODULE).
+-define(FLUSH_INTERVAL, 1000).
+
+-record(state, {
+    pending = queue:new(),
+    flush_ref = undefined
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> ok.
+stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
+    gen_server:cast(?SERVER, {stream, TaskId, Type, Stream}).
+
+-spec close(TaskId :: integer(), Reason :: binary()) -> ok.
+close(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
+    gen_server:cast(?SERVER, {close, TaskId, Reason}).
+
+-spec start_link() -> {ok, pid()} | ignore | {error, term()}.
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+-spec init(list()) -> {ok, #state{}}.
+init([]) ->
+    {ok, #state{}}.
+
+-spec handle_call(term(), {pid(), term()}, #state{}) -> {reply, ok, #state{}}.
+handle_call(_Request, _From, State) ->
+    {reply, ok, State}.
+
+-spec handle_cast(term(), #state{}) -> {noreply, #state{}}.
+handle_cast({stream, TaskId, Type, Stream}, State0 = #state{pending = Pending0}) ->
+    Pending = queue:in({stream, TaskId, Type, Stream}, Pending0),
+    {noreply, ensure_flush_timer(State0#state{pending = Pending}, 0)};
+handle_cast({close, TaskId, Reason}, State0 = #state{pending = Pending0}) ->
+    Pending = queue:in({close, TaskId, Reason}, Pending0),
+    {noreply, ensure_flush_timer(State0#state{pending = Pending}, 0)};
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+-spec handle_info(term(), #state{}) -> {noreply, #state{}}.
+handle_info(flush, State0 = #state{}) ->
+    State1 = State0#state{flush_ref = undefined},
+    {noreply, flush_pending(State1)};
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+-spec terminate(term(), #state{}) -> ok.
+terminate(_Reason, _State) ->
+    ok.
+
+-spec code_change(term(), #state{}, term()) -> {ok, #state{}}.
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+-spec ensure_flush_timer(#state{}, non_neg_integer()) -> #state{}.
+ensure_flush_timer(State = #state{flush_ref = undefined}, Delay) when is_integer(Delay), Delay >= 0 ->
+    Ref = erlang:send_after(Delay, self(), flush),
+    State#state{flush_ref = Ref};
+ensure_flush_timer(State = #state{}, _Delay) ->
+    State.
+
+-spec flush_pending(#state{}) -> #state{}.
+flush_pending(State = #state{pending = Pending0}) ->
+    case queue:out(Pending0) of
+        {empty, _} ->
+            State;
+        {{value, Event}, Pending1} ->
+            case send_event(Event) of
+                ok ->
+                    flush_pending(State#state{pending = Pending1});
+                not_ready ->
+                    ensure_flush_timer(State#state{pending = Pending0}, ?FLUSH_INTERVAL)
+            end
+    end.
+
+-spec send_event({stream, integer(), binary(), binary()} | {close, integer(), binary()}) -> ok | not_ready.
+send_event({stream, TaskId, Type, Stream}) ->
+    maybe_send(fun() -> efka_iot_client:task_event_stream(TaskId, Type, Stream) end);
+send_event({close, TaskId, Reason}) ->
+    maybe_send(fun() -> efka_iot_client:close_task_event_stream(TaskId, Reason) end).
+
+-spec maybe_send(fun(() -> any())) -> ok | not_ready.
+maybe_send(SendFun) ->
+    case catch efka_iot_client:is_activated() of
+        true ->
+            _ = catch SendFun(),
+            ok;
+        _ ->
+            not_ready
+    end.

apps/efka/include/efka_service_pb.hrl

@@ -2,10 +2,10 @@
 %% Automatically generated, do not edit
 %% Generated by gpb_compile version 4.21.7
 
--ifndef(service_pb).
--define(service_pb, true).
+-ifndef(efka_service_pb).
+-define(efka_service_pb, true).
 
--define(service_pb_gpb_version, "4.21.7").
+-define(efka_service_pb_gpb_version, "4.21.7").
 
 
 -ifndef('SERVICEREQUEST.REGISTER_PB_H').
@@ -25,8 +25,8 @@
 -ifndef('SERVICEREQUEST_PB_H').
 -define('SERVICEREQUEST_PB_H', true).
 -record('ServiceRequest',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 64 bits
-         request                :: {register, service_pb:'ServiceRequest.Register'()} | {subscribe, service_pb:'ServiceRequest.Subscribe'()} | undefined % oneof
+        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
+         request                :: {register, efka_service_pb:'ServiceRequest.Register'()} | {subscribe, efka_service_pb:'ServiceRequest.Subscribe'()} | undefined % oneof
         }).
 -endif.
 
@@ -41,8 +41,8 @@
 -ifndef('SERVICEREPLY_PB_H').
 -define('SERVICEREPLY_PB_H', true).
 -record('ServiceReply',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 64 bits
-         reply                  :: {result, iodata()} | {error, service_pb:'ServiceReply.Error'()} | undefined % oneof
+        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
+         reply                  :: {result, iodata()} | {error, efka_service_pb:'ServiceReply.Error'()} | undefined % oneof
         }).
 -endif.
 
@@ -65,7 +65,7 @@
 -ifndef('SERVICECAST_PB_H').
 -define('SERVICECAST_PB_H', true).
 -record('ServiceCast',
-        {body                   :: {topic_event, service_pb:'ServiceCast.TopicEvent'()} | {metric_data, service_pb:'ServiceCast.MetricData'()} | undefined % oneof
+        {body                   :: {topic_event, efka_service_pb:'ServiceCast.TopicEvent'()} | {metric_data, efka_service_pb:'ServiceCast.MetricData'()} | undefined % oneof
         }).
 -endif.
 

apps/efka/include/message.hrl

@@ -1,25 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @author anlicheng
-%%% @copyright (C) 2026, <COMPANY>
-%%% @doc
-%%%
-%%% @end
-%%% Created : 07. 7月 2026 17:09
-%%%-------------------------------------------------------------------
--author("anlicheng").
-
-%%--------------------------------------------------------------------
-%% Wire classes
-%%--------------------------------------------------------------------
--define(CLASS_REQUEST, 1).
--define(CLASS_RESPONSE, 2).
--define(CLASS_COMMAND, 3).
--define(CLASS_COMMAND_RESPONSE, 4).
--define(CLASS_MESSAGE, 5).
--define(CLASS_STREAM, 6).
-
-%%--------------------------------------------------------------------
-%% Stream targets
-%%--------------------------------------------------------------------
--define(STREAM_TARGET_MANAGER, 1).
--define(STREAM_TARGET_CONTAINER_DEPLOY, 2).

apps/efka/include/message_pb.hrl

@@ -1,225 +0,0 @@
-%% -*- coding: utf-8 -*-
-%% Automatically generated, do not edit
-%% Generated by gpb_compile version 4.21.7
-
--ifndef(message_pb).
--define(message_pb, true).
-
--define(message_pb_gpb_version, "4.21.7").
-
-
--ifndef('REQUEST.AUTHREQUEST_PB_H').
--define('REQUEST.AUTHREQUEST_PB_H', true).
--record('Request.AuthRequest',
-        {uuid = <<>>            :: iodata() | undefined, % = 1, optional
-         token = <<>>           :: iodata() | undefined, % = 2, optional
-         timestamp = 0          :: non_neg_integer() | undefined % = 3, optional, 64 bits
-        }).
--endif.
-
--ifndef('REQUEST_PB_H').
--define('REQUEST_PB_H', true).
--record('Request',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 64 bits
-         body                   :: {auth_request, message_pb:'Request.AuthRequest'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('RESPONSE.ERROR_PB_H').
--define('RESPONSE.ERROR_PB_H', true).
--record('Response.Error',
-        {code = 0               :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         reason = <<>>          :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('RESPONSE.AUTHRESPONSE_PB_H').
--define('RESPONSE.AUTHRESPONSE_PB_H', true).
--record('Response.AuthResponse',
-        {
-        }).
--endif.
-
--ifndef('RESPONSE_PB_H').
--define('RESPONSE_PB_H', true).
--record('Response',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 64 bits
-         body                   :: {auth_response, message_pb:'Response.AuthResponse'()} | {error, message_pb:'Response.Error'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('COMMAND.CONTAINER_PB_H').
--define('COMMAND.CONTAINER_PB_H', true).
--record('Command.Container',
-        {action                 :: {list, message_pb:'Command.Container.ContainerList'()} | {start, message_pb:'Command.Container.ContainerStart'()} | {stop, message_pb:'Command.Container.ContainerStop'()} | {kill, message_pb:'Command.Container.ContainerKill'()} | {remove, message_pb:'Command.Container.ContainerRemove'()} | {config, message_pb:'Command.Container.ContainerConfig'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('COMMAND.CONTAINER.CONTAINERCONFIG_PB_H').
--define('COMMAND.CONTAINER.CONTAINERCONFIG_PB_H', true).
--record('Command.Container.ContainerConfig',
-        {target = undefined     :: message_pb:'Command.Container.ContainerTarget'() | undefined, % = 1, optional
-         config = <<>>          :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('COMMAND.CONTAINER.CONTAINERREMOVE_PB_H').
--define('COMMAND.CONTAINER.CONTAINERREMOVE_PB_H', true).
--record('Command.Container.ContainerRemove',
-        {target = undefined     :: message_pb:'Command.Container.ContainerTarget'() | undefined, % = 1, optional
-         force = false          :: boolean() | 0 | 1 | undefined, % = 2, optional
-         remove_volumes = false :: boolean() | 0 | 1 | undefined % = 3, optional
-        }).
--endif.
-
--ifndef('COMMAND.CONTAINER.CONTAINERKILL_PB_H').
--define('COMMAND.CONTAINER.CONTAINERKILL_PB_H', true).
--record('Command.Container.ContainerKill',
-        {target = undefined     :: message_pb:'Command.Container.ContainerTarget'() | undefined, % = 1, optional
-         signal = <<>>          :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('COMMAND.CONTAINER.CONTAINERSTOP_PB_H').
--define('COMMAND.CONTAINER.CONTAINERSTOP_PB_H', true).
--record('Command.Container.ContainerStop',
-        {target = undefined     :: message_pb:'Command.Container.ContainerTarget'() | undefined, % = 1, optional
-         timeout_seconds = 0    :: non_neg_integer() | undefined % = 2, optional, 32 bits
-        }).
--endif.
-
--ifndef('COMMAND.CONTAINER.CONTAINERSTART_PB_H').
--define('COMMAND.CONTAINER.CONTAINERSTART_PB_H', true).
--record('Command.Container.ContainerStart',
-        {target = undefined     :: message_pb:'Command.Container.ContainerTarget'() | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('COMMAND.CONTAINER.CONTAINERLIST_PB_H').
--define('COMMAND.CONTAINER.CONTAINERLIST_PB_H', true).
--record('Command.Container.ContainerList',
-        {
-        }).
--endif.
-
--ifndef('COMMAND.CONTAINER.CONTAINERTARGET_PB_H').
--define('COMMAND.CONTAINER.CONTAINERTARGET_PB_H', true).
--record('Command.Container.ContainerTarget',
-        {name = <<>>            :: iodata() | undefined, % = 1, optional
-         id = <<>>              :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('COMMAND_PB_H').
--define('COMMAND_PB_H', true).
--record('Command',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 64 bits
-         body                   :: {container, message_pb:'Command.Container'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('COMMANDRESPONSE.ERROR_PB_H').
--define('COMMANDRESPONSE.ERROR_PB_H', true).
--record('CommandResponse.Error',
-        {code = 0               :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         reason = <<>>          :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('COMMANDRESPONSE_PB_H').
--define('COMMANDRESPONSE_PB_H', true).
--record('CommandResponse',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 64 bits
-         body                   :: {result, iodata()} | {error, message_pb:'CommandResponse.Error'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('MESSAGE.PING_PB_H').
--define('MESSAGE.PING_PB_H', true).
--record('Message.Ping',
-        {
-        }).
--endif.
-
--ifndef('MESSAGE.PONG_PB_H').
--define('MESSAGE.PONG_PB_H', true).
--record('Message.Pong',
-        {
-        }).
--endif.
-
--ifndef('MESSAGE.PUB_PB_H').
--define('MESSAGE.PUB_PB_H', true).
--record('Message.Pub',
-        {topic = <<>>           :: iodata() | undefined, % = 1, optional
-         qos = 0                :: non_neg_integer() | undefined, % = 2, optional, 32 bits
-         content = <<>>         :: iodata() | undefined % = 3, optional
-        }).
--endif.
-
--ifndef('MESSAGE.METRICDATA_PB_H').
--define('MESSAGE.METRICDATA_PB_H', true).
--record('Message.MetricData',
-        {route_key = <<>>       :: iodata() | undefined, % = 1, optional
-         metric = <<>>          :: iodata() | undefined % = 2, optional
-        }).
--endif.
-
--ifndef('MESSAGE_PB_H').
--define('MESSAGE_PB_H', true).
--record('Message',
-        {body                   :: {ping, message_pb:'Message.Ping'()} | {pong, message_pb:'Message.Pong'()} | {pub, message_pb:'Message.Pub'()} | {metric_data, message_pb:'Message.MetricData'()} | undefined % oneof
-        }).
--endif.
-
--ifndef('STREAM.OPEN_PB_H').
--define('STREAM.OPEN_PB_H', true).
--record('Stream.Open',
-        {target = 0             :: non_neg_integer() | undefined % = 1, optional, 32 bits
-        }).
--endif.
-
--ifndef('STREAM.OPENED_PB_H').
--define('STREAM.OPENED_PB_H', true).
--record('Stream.Opened',
-        {
-        }).
--endif.
-
--ifndef('STREAM.OPENERROR_PB_H').
--define('STREAM.OPENERROR_PB_H', true).
--record('Stream.OpenError',
-        {reason = <<>>          :: iodata() | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('STREAM.DATA_PB_H').
--define('STREAM.DATA_PB_H', true).
--record('Stream.Data',
-        {bytes = <<>>           :: iodata() | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('STREAM.FIN_PB_H').
--define('STREAM.FIN_PB_H', true).
--record('Stream.Fin',
-        {
-        }).
--endif.
-
--ifndef('STREAM.RESET_PB_H').
--define('STREAM.RESET_PB_H', true).
--record('Stream.Reset',
-        {reason = <<>>          :: iodata() | undefined % = 1, optional
-        }).
--endif.
-
--ifndef('STREAM_PB_H').
--define('STREAM_PB_H', true).
--record('Stream',
-        {stream_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         payload                :: {open, message_pb:'Stream.Open'()} | {opened, message_pb:'Stream.Opened'()} | {open_error, message_pb:'Stream.OpenError'()} | {data, message_pb:'Stream.Data'()} | {fin, message_pb:'Stream.Fin'()} | {reset, message_pb:'Stream.Reset'()} | undefined % oneof
-        }).
--endif.
-
--endif.

apps/efka/proto/message.proto

@@ -1,179 +0,0 @@
-syntax = "proto3";
-
-// iot <-> efka protocol payload definitions.
-//
-// The transport frame keeps the first byte as the coarse message class:
-//
-//   [CLASS_REQUEST][protobuf(Request)]
-//   [CLASS_RESPONSE][protobuf(Response)]
-//   [CLASS_COMMAND][protobuf(Command)]
-//   [CLASS_COMMAND_RESPONSE][protobuf(CommandResponse)]
-//   [CLASS_MESSAGE][protobuf(Message)]
-//   [CLASS_STREAM][protobuf(Stream)]
-//
-// HTTP proxy bytes carried by Stream.Data are transparent payload bytes.
-
-message Request {
-  uint64 packet_id = 1;
-
-  message AuthRequest {
-    bytes uuid = 1;
-    bytes token = 2;
-    uint64 timestamp = 3;
-  }
-
-  oneof body {
-    AuthRequest auth_request = 10;
-  }
-
-}
-
-message Response {
-  uint64 packet_id = 1;
-
-  message Error {
-    uint32 code = 1;
-    bytes reason = 2;
-  }
-
-  message AuthResponse {
-
-  }
-
-  oneof body {
-    AuthResponse auth_response = 10;
-    Error error = 11;
-  }
-
-}
-
-message Command {
-  uint64 packet_id = 1;
-
-  message Container {
-    message ContainerTarget {
-      bytes name = 1;
-      bytes id = 2;
-    }
-
-    message ContainerList {
-
-    }
-
-    message ContainerStart {
-      ContainerTarget target = 1;
-    }
-
-    message ContainerStop {
-      ContainerTarget target = 1;
-      uint32 timeout_seconds = 2;
-    }
-
-    message ContainerKill {
-      ContainerTarget target = 1;
-      bytes signal = 2;
-    }
-
-    message ContainerRemove {
-      ContainerTarget target = 1;
-      bool force = 2;
-      bool remove_volumes = 3;
-    }
-
-    message ContainerConfig {
-      ContainerTarget target = 1;
-      bytes config = 2;
-    }
-
-    oneof action {
-      ContainerList list = 1;
-      ContainerStart start = 3;
-      ContainerStop stop = 4;
-      ContainerKill kill = 5;
-      ContainerRemove remove = 6;
-      ContainerConfig config = 7;
-    }
-
-  }
-
-  oneof body {
-    Container container = 10;
-  }
-
-}
-
-message CommandResponse {
-  uint64 packet_id = 1;
-
-  message Error {
-    uint32 code = 1;
-    bytes reason = 2;
-  }
-
-  oneof body {
-    bytes result = 10;
-    Error error = 11;
-  }
-
-}
-
-message Message {
-  message Ping {
-  }
-
-  message Pong {
-  }
-
-  message Pub {
-    bytes topic = 1;
-    uint32 qos = 2;
-    bytes content = 3;
-  }
-
-  message MetricData {
-    bytes route_key = 1;
-    bytes metric = 2;
-  }
-
-  oneof body {
-    Ping ping = 10;
-    Pong pong = 11;
-    Pub pub = 12;
-    MetricData metric_data = 13;
-  }
-}
-
-message Stream {
-  uint32 stream_id = 1;
-
-  message Open {
-    uint32 target = 1;
-  }
-
-  message Opened {
-  }
-
-  message OpenError {
-    bytes reason = 1;
-  }
-
-  message Data {
-    bytes bytes = 1;
-  }
-
-  message Fin {
-  }
-
-  message Reset {
-    bytes reason = 1;
-  }
-
-  oneof payload {
-    Open open = 10;
-    Opened opened = 11;
-    OpenError open_error = 12;
-    Data data = 13;
-    Fin fin = 14;
-    Reset reset = 15;
-  }
-}

apps/efka/proto/service.proto

@@ -4,7 +4,7 @@ syntax = "proto3";
 // 该协议用于替代当前基于 JSON 的请求/响应/推送格式。
 
 message ServiceRequest {
-  uint64 packet_id = 1;
+  uint32 packet_id = 1;
 
   message Register {
     string service_id = 1;
@@ -21,7 +21,7 @@ message ServiceRequest {
 }
 
 message ServiceReply {
-  uint64 packet_id = 1;
+  uint32 packet_id = 1;
 
   message Error {
     int32 code = 1;

apps/efka/rebar.config

@@ -2,9 +2,9 @@
 
 {gpb_opts, [
     {i, "proto"},
-    {f, ["service.proto", "message.proto"]},
+    {f, ["service.proto"]},
     recursive,
-    {module_name_prefix, ""},
+    {module_name_prefix, "efka_"},
     {module_name_suffix, "_pb"},
     {o_erl, "src/protobuf"},
     {o_hrl, "include"},

apps/efka/src/efka.app.src

@@ -5,7 +5,7 @@
   {mod, {efka_app, []}},
   {applications,
    [
-    %sync,
+    sync,
     docker,
     cowboy,
     ranch,

apps/efka/src/iot/efka_iot_client.erl

@@ -9,15 +9,12 @@
 -module(efka_iot_client).
 -author("anlicheng").
 -include("efka_tables.hrl").
--include("message.hrl").
--include("message_pb.hrl").
 
 -behaviour(gen_statem).
 
 %% API
 -export([start_link/0]).
--export([metric_data/2, ping/13]).
--export([send_stream/2, stream_done/1]).
+-export([metric_data/2, ping/13, task_event_stream/3, close_task_event_stream/2]).
 -export([is_activated/0, dropped_message_count/0]).
 
 %% gen_statem callbacks
@@ -39,21 +36,12 @@
 -record(state, {
     socket :: undefined | ssl:sslsocket(),
     outbox :: efka_iot_outbox:outbox(),
-    streams = #{},
-    %% 保存当前 auth 请求的 packet id，用来建立 auth 请求和响应的对应关系
-    auth_pkt_id = undefined :: undefined | pos_integer(),
-    next_pkt_id = 1 :: pos_integer(),
+    %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
+    auth_ref = undefined :: undefined | binary(),
     ping_timer_ref = undefined :: undefined | reference(),
     dropped_message_count = 0 :: non_neg_integer()
 }).
 
--record(stream_state, {
-    worker_pid :: pid(),
-    monitor_ref :: reference()
-}).
-
--type stream_id() :: pos_integer().
-
 %%%===================================================================
 %%% API
 %%%===================================================================
@@ -63,13 +51,13 @@
 metric_data(RouteKey, Metric) when is_binary(RouteKey), is_binary(Metric) ->
     gen_statem:cast(?SERVER, {metric_data, RouteKey, Metric}).
 
--spec send_stream(StreamId :: stream_id(), Body :: term()) -> ok.
-send_stream(StreamId, Body) when is_integer(StreamId), StreamId > 0 ->
-    gen_statem:cast(?SERVER, {send_stream, StreamId, Body}).
+-spec task_event_stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> ok.
+task_event_stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
+    gen_statem:cast(?SERVER, {task_event_stream, TaskId, Type, Stream}).
 
--spec stream_done(StreamId :: stream_id()) -> ok.
-stream_done(StreamId) when is_integer(StreamId), StreamId > 0 ->
-    gen_statem:cast(?SERVER, {stream_done, StreamId}).
+-spec close_task_event_stream(TaskId :: integer(), Reason :: binary()) -> ok.
+close_task_event_stream(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
+    gen_statem:cast(?SERVER, {close_task_event_stream, TaskId, Reason}).
 
 -spec is_activated() -> boolean().
 is_activated() ->
@@ -118,7 +106,7 @@ outbox_options() ->
 %% 异步发送数据，连接存在时直接发送；否则写入持久化 outbox。
 -spec handle_event(term(), term(), atom(), #state{}) -> term().
 handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
-    Packet = encode_message_frame({metric_data, RouteKey, Metric}),
+    Packet = term_to_binary({<<"message">>, {<<"data">>, #{<<"route_key">> => RouteKey, <<"metric">> => Metric}}}),
     case StateName of
         ?STATE_ACTIVATED ->
             ok = ssl:send(Socket, Packet),
@@ -139,23 +127,17 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so
             end
     end;
 
-handle_event(cast, {send_stream, StreamId, Body}, ?STATE_ACTIVATED, State = #state{socket = Socket})
-    when is_integer(StreamId), StreamId > 0 ->
-    Packet = encode_stream_frame(StreamId, Body),
+%% Task的stream流，只做实时的
+handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    logger:debug("[efka_iot_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
+    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => Type, <<"stream">> => Stream}}}),
     ok = ssl:send(Socket, Packet),
     {keep_state, State};
-handle_event(cast, {send_stream, _StreamId, _Body}, _StateName, State) ->
-    {keep_state, State};
 
-handle_event(cast, {stream_done, StreamId}, _StateName, State = #state{streams = Streams})
-    when is_integer(StreamId), StreamId > 0 ->
-    case maps:take(StreamId, Streams) of
-        error ->
-            {keep_state, State};
-        {StreamState, NStreams} ->
-            demonitor_stream(StreamState),
-            {keep_state, State#state{streams = NStreams}}
-    end;
+handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}),
+    ok = ssl:send(Socket, Packet),
+    {keep_state, State};
 
 %% 其他情况下直接忽略
 handle_event(cast, _, _, State = #state{}) ->
@@ -169,13 +151,14 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro
     {keep_state, State, [{reply, From, DroppedCount}]};
 
 %% 异步建立到服务器的连接
-handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = #state{next_pkt_id = PktId}) ->
+handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
     case connect_socket() of
         {ok, Socket} ->
-            AuthPacket = auth_packet(PktId),
+            Ref = request_ref(),
+            AuthPacket = auth_packet(Ref),
             ok = ssl:send(Socket, AuthPacket),
-            logger:debug("[efka_iot_client] send auth request, packet_id: ~p", [PktId]),
-            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_pkt_id = PktId, next_pkt_id = PktId + 1}, [{state_timeout, 5000, auth_timeout}]};
+            logger:debug("[efka_iot_client] send auth request, ref: ~p", [Ref]),
+            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
         {error, _Reason} ->
             schedule_reconnect(),
             {keep_state, State#state{socket = undefined}}
@@ -185,10 +168,10 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S
     logger:debug("[efka_iot_client] auth request timeout"),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
 
 handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) ->
-    Packet = encode_message_frame(ping),
+    Packet = term_to_binary({<<"message">>, <<"ping">>}),
     case ssl:send(Socket, Packet) of
         ok ->
             {keep_state, schedule_ssl_ping(State)};
@@ -196,8 +179,7 @@ handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #sta
             logger:warning("[efka_iot_client] send ssl ping failed, reason: ~p", [Reason]),
             disconnect(Socket),
             schedule_reconnect(),
-            NState = close_all_streams({send_ping_failed, Reason}, State),
-            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}}
+            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
     {keep_state, State};
@@ -225,123 +207,110 @@ handle_event(info, flush_cache, _, State) ->
 
 %% 处理收到的ssl消息
 handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
-    case decode_frame(PacketBin) of
-        {ok, Packet} ->
-            {keep_state, State, [{next_event, internal, Packet}]};
-        {error, Reason} ->
-            logger:warning("[efka_iot_client] decode packet failed: ~p, packet_size: ~p", [Reason, byte_size(PacketBin)]),
+    try binary_to_term(PacketBin, [safe]) of
+        Packet ->
+            {keep_state, State, [{next_event, internal, Packet}]}
+    catch
+        error:Error ->
+            logger:warning("[efka_iot_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
             disconnect(Socket),
             cancel_ssl_ping(State),
             schedule_reconnect(),
-            NState = close_all_streams(bad_packet, State),
-            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}}
+            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
     logger:debug("[efka_iot_client] ssl error: ~p", [Reason]),
     disconnect(Socket),
     cancel_ssl_ping(State),
     schedule_reconnect(),
-    NState = close_all_streams(ssl_error, State),
-    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
 handle_event(info, {ssl_closed, Socket}, _, State = #state{}) ->
     logger:debug("[efka_iot_client] ssl closed"),
     disconnect(Socket),
     cancel_ssl_ping(State),
     schedule_reconnect(),
-    NState = close_all_streams(ssl_closed, State),
-    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
 
-handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName, State = #state{streams = Streams}) ->
-    case take_stream_by_monitor(MonitorRef, WorkerPid, Streams) of
-        error ->
-            {keep_state, State};
-        {StreamId, StreamState, NStreams} ->
-            demonitor_stream(StreamState),
-            case Reason of
-                normal ->
-                    {keep_state, State#state{streams = NStreams}};
-                _ ->
-                    logger:warning("[efka_iot_client] stream worker down, stream_id: ~p, reason: ~p", [StreamId, Reason]),
-                    case State#state.socket of
-                        undefined ->
-                            {keep_state, State#state{streams = NStreams}};
-                        Socket ->
-                            Packet = encode_stream_frame(StreamId, {reset, {worker_down, Reason}}),
-                            ok = ssl:send(Socket, Packet),
-                            {keep_state, State#state{streams = NStreams}}
-                    end
-            end
-    end;
-
-%%% 处理内部消息。TLS 收到的数据先经过 protobuf 解码，再由这里按本地事件分发。
+%%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
 
 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
-handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    handle_container_command(PktId, ContainerCommand, Socket),
+handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    handle_container_command(Ref, Request, Socket),
     {keep_state, State};
-handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, _StateName, State = #state{socket = Socket}) ->
-    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
-    send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
+handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) ->
+    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
+    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
     {keep_state, State};
 
 %% 处理response
-handle_event(internal, #'Response'{packet_id = AuthPktId, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_pkt_id = AuthPktId}) ->
+handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
     logger:debug("[efka_iot_client] auth success"),
-    State1 = schedule_ssl_ping(State#state{auth_pkt_id = undefined}),
+    State1 = schedule_ssl_ping(State#state{auth_ref = undefined}),
     {next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]};
-handle_event(internal, #'Response'{packet_id = AuthPktId, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_pkt_id = AuthPktId}) ->
+handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
     logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
-handle_event(internal, #'Response'{} = Reply, StateName, State) ->
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+handle_event(internal, {<<"response">>, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_iot_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
-handle_event(internal, #'CommandResponse'{} = Reply, StateName, State) ->
+handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_iot_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
 
-%% 透明 TCP stream 多路复用。
-handle_event(internal, #'Stream'{stream_id = StreamId, payload = Payload}, ?STATE_ACTIVATED, State) ->
-    Body = case Payload of
-               {open, #'Stream.Open'{target = Target}} ->
-                   {open, Target};
-               {opened, #'Stream.Opened'{}} ->
-                   opened;
-               {open_error, #'Stream.OpenError'{reason = Reason}} ->
-                   {open_error, Reason};
-               {data, #'Stream.Data'{bytes = Data}} ->
-                   {data, Data};
-               {fin, #'Stream.Fin'{}} ->
-                   fin;
-               {reset, #'Stream.Reset'{reason = Reason}} ->
-                   {reset, Reason}
-           end,
-    handle_stream_frame(StreamId, Body, State);
-handle_event(internal, #'Stream'{stream_id = StreamId, payload = Payload}, StateName, State) ->
-    logger:warning("[efka_iot_client] ignore stream frame in state ~p, stream_id: ~p, body: ~p",
-        [StateName, StreamId, Payload]),
-    {keep_state, State};
-
 %% 处理Pub/Sub机制
-handle_event(internal, #'Message'{body = {pong, #'Message.Pong'{}}}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) ->
     {keep_state, State};
-handle_event(internal, #'Message'{body = {pub, #'Message.Pub'{topic = Topic, qos = Qos, content = Content}}}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
     logger:debug("[efka_iot_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
     efka_subscription:publish(Topic, Qos, Content),
     {keep_state, State};
-handle_event(internal, Packet, StateName, State) ->
-    logger:warning("[efka_iot_client] ignore unknown packet: ~p, state_name: ~p", [Packet, StateName]),
+handle_event(internal, Packet, _StateName, State) ->
+    logger:warning("[efka_iot_client] ignore unknown packet: ~p", [Packet]),
     {keep_state, State};
 
 handle_event(info, Info, _, State = #state{}) ->
     logger:notice("[efka_iot_client] get unknown info: ~p", [Info]),
     {keep_state, State}.
 
+-spec handle_container_command(binary(), term(), ssl:sslsocket()) -> ok.
+handle_container_command(Ref, #{<<"action">> := <<"list">>}, Socket) ->
+    Reply = docker_commands:get_containers(),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) ->
+    Reply = docker_deploy_manager:deploy(TaskId, Params),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
+    Reply = docker_commands:start_container(container_target(Target)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) ->
+    Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) ->
+    Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) ->
+    Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) ->
+    Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, Request, Socket) ->
+    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
+    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
+    ok.
+
 -spec terminate(term(), atom(), #state{}) -> ok.
 terminate(Reason, _StateName, State = #state{socket = Socket, outbox = Outbox}) ->
     cancel_ssl_ping(State),
-    _ = close_all_streams(Reason, State),
     disconnect(Socket),
     efka_iot_outbox:close(Outbox),
     logger:notice("[efka_iot_client] terminate with reason: ~p", [Reason]),
@@ -355,83 +324,18 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
--spec auth_packet(pos_integer()) -> binary().
-auth_packet(PktId) when is_integer(PktId), PktId > 0 ->
+-spec auth_packet(binary()) -> binary().
+auth_packet(Ref) when is_binary(Ref) ->
     {ok, AuthInfo} = application:get_env(efka, auth),
     UUID = proplists:get_value(uuid, AuthInfo),
     Token = proplists:get_value(token, AuthInfo),
 
     Timestamp = efka_util:timestamp(),
-    encode_transport_frame(?CLASS_REQUEST, #'Request'{
-        packet_id = PktId,
-        body = {auth_request, #'Request.AuthRequest'{
-            uuid = list_to_binary(UUID),
-            token = list_to_binary(Token),
-            timestamp = Timestamp
-        }}
-    }).
-
--spec encode_message_frame(term()) -> binary().
-encode_message_frame(ping) ->
-    encode_transport_frame(?CLASS_MESSAGE, #'Message'{body = {ping, #'Message.Ping'{}}});
-encode_message_frame({metric_data, RouteKey, Metric}) ->
-    encode_transport_frame(?CLASS_MESSAGE, #'Message'{
-        body = {metric_data, #'Message.MetricData'{route_key = RouteKey, metric = Metric}}
-    });
-encode_message_frame(Body) ->
-    error({unsupported_message_body, Body}).
-
--spec encode_stream_frame(stream_id(), term()) -> binary().
-encode_stream_frame(StreamId, Body) ->
-    Payload = case Body of
-                  open ->
-                      {open, #'Stream.Open'{target = ?STREAM_TARGET_MANAGER}};
-                  {open, Target} when is_integer(Target), Target > 0 ->
-                      {open, #'Stream.Open'{target = Target}};
-                  opened ->
-                      {opened, #'Stream.Opened'{}};
-                  {open_error, Reason} ->
-                      {open_error, #'Stream.OpenError'{reason = reason_to_binary(Reason)}};
-                  {data, Data} when is_binary(Data) ->
-                      {data, #'Stream.Data'{bytes = Data}};
-                  fin ->
-                      {fin, #'Stream.Fin'{}};
-                  {reset, Reason} ->
-                      {reset, #'Stream.Reset'{reason = reason_to_binary(Reason)}}
-              end,
-    encode_transport_frame(?CLASS_STREAM, #'Stream'{stream_id = StreamId, payload = Payload}).
-
--spec encode_transport_frame(byte(), message_pb:'$msg'()) -> binary().
-encode_transport_frame(Class, Msg) ->
-    Payload = message_pb:encode_msg(Msg),
-    <<Class, Payload/binary>>.
-
--spec decode_frame(binary()) -> {ok, tuple()} | {error, term()}.
-decode_frame(<<?CLASS_RESPONSE, Payload/binary>>) ->
-    decode_pb_frame(Payload, 'Response');
-decode_frame(<<?CLASS_COMMAND, Payload/binary>>) ->
-    decode_pb_frame(Payload, 'Command');
-decode_frame(<<?CLASS_COMMAND_RESPONSE, Payload/binary>>) ->
-    decode_pb_frame(Payload, 'CommandResponse');
-decode_frame(<<?CLASS_MESSAGE, Payload/binary>>) ->
-    decode_pb_frame(Payload, 'Message');
-decode_frame(<<?CLASS_STREAM, Payload/binary>>) ->
-    decode_pb_frame(Payload, 'Stream');
-decode_frame(<<?CLASS_REQUEST, _Payload/binary>>) ->
-    {error, unsupported_request_frame};
-decode_frame(Other) ->
-    {error, {invalid_frame, Other}}.
-
--spec decode_pb_frame(binary(), message_pb:'$msg_name'()) ->
-    {ok, tuple()} | {error, term()}.
-decode_pb_frame(Payload, MsgName) ->
-    try message_pb:decode_msg(Payload, MsgName) of
-        Msg ->
-            {ok, Msg}
-    catch
-        Class:Reason ->
-            {error, {bad_protobuf, MsgName, Class, Reason}}
-    end.
+    term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{
+        <<"uuid">> => list_to_binary(UUID),
+        <<"token">> => list_to_binary(Token),
+        <<"timestamp">> => Timestamp
+    }}}).
 
 -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
 connect_socket() ->
@@ -473,190 +377,52 @@ cancel_timer(TimerRef) ->
     _ = erlang:cancel_timer(TimerRef),
     ok.
 
--spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok.
-send_container_response(Socket, PktId, Reply) ->
-    Body = case Reply of
-               ok ->
-                   {result, <<"ok">>};
-               {ok, Result} ->
-                   {result, reply_to_binary(Result)};
-               {error, Reason} ->
-                   {error, #'CommandResponse.Error'{code = 1, reason = reason_to_binary(Reason)}};
-               Other ->
-                   {result, reply_to_binary(Other)}
-           end,
-    Packet = encode_transport_frame(?CLASS_COMMAND_RESPONSE, #'CommandResponse'{
-        packet_id = PktId,
-        body = Body
-    }),
+-spec request_ref() -> binary().
+request_ref() ->
+    crypto:strong_rand_bytes(16).
+
+-spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
+send_container_response(Socket, Ref, Reply) ->
+    Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
     ok = ssl:send(Socket, Packet).
 
--spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok.
-handle_container_command(PktId, #'Command.Container'{action = {list, #'Command.Container.ContainerList'{}}}, Socket) ->
-    Reply = docker_commands:get_containers(),
-    send_container_response(Socket, PktId, Reply),
-    ok;
-handle_container_command(PktId, #'Command.Container'{action = {start, #'Command.Container.ContainerStart'{target = Target}}}, Socket) ->
-    Reply = docker_commands:start_container(container_target(Target)),
-    send_container_response(Socket, PktId, Reply),
-    ok;
-handle_container_command(PktId, #'Command.Container'{action = {stop, #'Command.Container.ContainerStop'{target = Target, timeout_seconds = TimeoutSeconds}}}, Socket) ->
-    Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
-    send_container_response(Socket, PktId, Reply),
-    ok;
-handle_container_command(PktId, #'Command.Container'{action = {kill, #'Command.Container.ContainerKill'{target = Target, signal = Signal}}}, Socket) ->
-    Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
-    send_container_response(Socket, PktId, Reply),
-    ok;
-handle_container_command(PktId, #'Command.Container'{action = {remove, #'Command.Container.ContainerRemove'{target = Target, force = Force, remove_volumes = RemoveVolumes}}}, Socket) ->
-    Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
-    send_container_response(Socket, PktId, Reply),
-    ok;
-handle_container_command(PktId, #'Command.Container'{action = {config, #'Command.Container.ContainerConfig'{target = Target, config = Config}}}, Socket) ->
-    Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
-    send_container_response(Socket, PktId, Reply),
-    ok;
-handle_container_command(PktId, ContainerCommand, Socket) ->
-    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
-    send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
-    ok.
+-spec safe_reply(term()) -> term().
+safe_reply(ok) ->
+    <<"ok">>;
+safe_reply({ok, Result}) ->
+    {<<"ok">>, safe_term(Result)};
+safe_reply({error, Reason}) ->
+    {<<"error">>, safe_term(Reason)}.
 
-%% iot 发起 open 时携带 target；efka 根据本地 stream_targets 配置建连。
--spec handle_stream_frame(term(), term(), #state{}) -> gen_statem:event_handler_result(atom(), #state{}).
-handle_stream_frame(StreamId, {open, Target}, State = #state{streams = Streams})
-    when is_integer(StreamId), StreamId > 0, is_integer(Target), Target > 0 ->
-    case valid_iot_stream_id(StreamId) andalso not maps:is_key(StreamId, Streams) of
-        true ->
-            case start_stream_worker(StreamId, Target) of
-                {ok, {WorkerPid, MonitorRef}} ->
-                    StreamState = #stream_state{worker_pid = WorkerPid, monitor_ref = MonitorRef},
-                    {keep_state, State#state{streams = maps:put(StreamId, StreamState, Streams)}};
-                {error, Reason} ->
-                    send_stream(StreamId, {open_error, Reason}),
-                    {keep_state, State}
-            end;
-        false ->
-            send_stream(StreamId, {reset, <<"invalid stream open">>}),
-            {keep_state, State}
-    end;
-handle_stream_frame(StreamId, {open, Target}, State)
-    when is_integer(StreamId), StreamId > 0 ->
-    logger:warning("[efka_iot_client] invalid stream target, stream_id: ~p, target: ~p", [StreamId, Target]),
-    send_stream(StreamId, {open_error, <<"invalid stream target">>}),
-    {keep_state, State};
-handle_stream_frame(StreamId, Body, State = #state{streams = Streams})
-    when is_integer(StreamId), StreamId > 0 ->
-    case maps:get(StreamId, Streams, undefined) of
-        undefined ->
-            maybe_reset_unknown_stream(StreamId, Body),
-            {keep_state, State};
-        StreamState = #stream_state{worker_pid = WorkerPid} ->
-            WorkerPid ! {stream, StreamId, Body},
-            case Body of
-                {reset, _Reason} ->
-                    demonitor_stream(StreamState),
-                    {keep_state, State#state{streams = maps:remove(StreamId, Streams)}};
-                _ ->
-                    {keep_state, State}
-            end
-    end;
-handle_stream_frame(StreamId, Body, State) ->
-    logger:warning("[efka_iot_client] invalid stream frame, stream_id: ~p, body: ~p", [StreamId, Body]),
-    {keep_state, State}.
+-spec safe_term(term()) -> term().
+safe_term(true) ->
+    true;
+safe_term(false) ->
+    false;
+safe_term(undefined) ->
+    undefined;
+safe_term(Value) when is_atom(Value) ->
+    atom_to_binary(Value, utf8);
+safe_term(Value) when is_map(Value) ->
+    maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]);
+safe_term(Value) when is_list(Value) ->
+    [safe_term(Item) || Item <- Value];
+safe_term(Value) when is_tuple(Value) ->
+    list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]);
+safe_term(Value) ->
+    Value.
 
--spec start_stream_worker(stream_id(), pos_integer()) -> {ok, {pid(), reference()}} | {error, term()}.
-start_stream_worker(StreamId, ?STREAM_TARGET_MANAGER) ->
-    efka_iot_stream:start_stream(StreamId, ?STREAM_TARGET_MANAGER);
-start_stream_worker(StreamId, ?STREAM_TARGET_CONTAINER_DEPLOY) ->
-    efka_iot_deploy_stream:start_stream(StreamId);
-start_stream_worker(_StreamId, Target) ->
-    {error, {unknown_stream_target, Target}}.
-
--spec maybe_reset_unknown_stream(stream_id(), term()) -> ok.
-maybe_reset_unknown_stream(_StreamId, {reset, _Reason}) ->
-    ok;
-maybe_reset_unknown_stream(_StreamId, fin) ->
-    ok;
-maybe_reset_unknown_stream(_StreamId, {open_error, _Reason}) ->
-    ok;
-maybe_reset_unknown_stream(StreamId, _Body) ->
-    send_stream(StreamId, {reset, <<"unknown stream">>}).
-
--spec valid_iot_stream_id(stream_id()) -> boolean().
-valid_iot_stream_id(StreamId) ->
-    StreamId rem 2 =:= 1.
-
--spec take_stream_by_monitor(reference(), pid(), map()) ->
-    {stream_id(), #stream_state{}, map()} | error.
-take_stream_by_monitor(MonitorRef, WorkerPid, Streams) ->
-    take_stream_by_monitor(MonitorRef, WorkerPid, maps:iterator(Streams), Streams).
-
-take_stream_by_monitor(MonitorRef, WorkerPid, Iter, Streams) ->
-    case maps:next(Iter) of
-        none ->
-            error;
-        {StreamId, StreamState = #stream_state{worker_pid = WorkerPid, monitor_ref = MonitorRef}, _NextIter} ->
-            {StreamId, StreamState, maps:remove(StreamId, Streams)};
-        {_StreamId, _StreamState, NextIter} ->
-            take_stream_by_monitor(MonitorRef, WorkerPid, NextIter, Streams)
-    end.
-
--spec close_all_streams(term(), #state{}) -> #state{}.
-close_all_streams(Reason, State = #state{streams = Streams}) ->
-    maps:foreach(fun(StreamId, StreamState = #stream_state{worker_pid = WorkerPid}) ->
-        demonitor_stream(StreamState),
-        WorkerPid ! {stream, StreamId, {reset, {channel_closed, Reason}}}
-    end, Streams),
-    State#state{streams = #{}}.
-
--spec demonitor_stream(#stream_state{}) -> ok.
-demonitor_stream(#stream_state{monitor_ref = MonitorRef}) ->
-    erlang:demonitor(MonitorRef, [flush]),
-    ok.
-
--spec reply_to_binary(term()) -> binary().
-reply_to_binary(Value) when is_binary(Value) ->
-    Value;
-reply_to_binary(Value) ->
-    try iolist_to_binary(Value) of
-        Bin ->
-            Bin
-    catch
-        _:_ ->
-            try iolist_to_binary(json:encode(Value)) of
-                JsonBin ->
-                    JsonBin
-            catch
-                _:_ ->
-                    unicode:characters_to_binary(io_lib:format("~p", [Value]))
-            end
-    end.
-
--spec reason_to_binary(term()) -> binary().
-reason_to_binary(Reason) when is_binary(Reason) ->
-    Reason;
-reason_to_binary(Reason) ->
-    try iolist_to_binary(Reason) of
-        Bin ->
-            Bin
-    catch
-        _:_ ->
-            unicode:characters_to_binary(io_lib:format("~p", [Reason]))
-    end.
-
--spec container_target(message_pb:'Command.Container.ContainerTarget'() | undefined) -> binary().
-container_target(#'Command.Container.ContainerTarget'{name = Name, id = Id}) ->
-    NameBin = iolist_to_binary(Name),
-    IdBin = iolist_to_binary(Id),
+-spec container_target(map()) -> binary().
+container_target(Target) when is_map(Target) ->
+    NameBin = to_binary(maps:get(<<"name">>, Target, <<>>)),
+    IdBin = to_binary(maps:get(<<"id">>, Target, <<>>)),
     case NameBin of
         <<>> ->
             true = IdBin =/= <<>>,
             IdBin;
         _ ->
             NameBin
-    end;
-container_target(undefined) ->
-    error(bad_container_target).
+    end.
 
 -spec to_binary(binary() | list()) -> binary().
 to_binary(Value) when is_binary(Value) ->

apps/efka/src/iot/efka_iot_deploy_stream.erl

@@ -1,119 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @doc One container deploy feedback stream from iot.
-%%% @end
-%%%-------------------------------------------------------------------
--module(efka_iot_deploy_stream).
-
--export([start_stream/1]).
--export([run/1]).
-
--define(REQUEST_TIMEOUT, 10000).
--define(MAX_REQUEST_BYTES, 16 * 1024 * 1024).
-
--type stream_id() :: pos_integer().
-
--spec start_stream(StreamId :: stream_id()) -> {ok, {pid(), reference()}}.
-start_stream(StreamId) when is_integer(StreamId), StreamId > 0 ->
-    {ok, spawn_monitor(?MODULE, run, [StreamId])}.
-
--spec run(StreamId :: stream_id()) -> ok.
-run(StreamId) when is_integer(StreamId), StreamId > 0 ->
-    try run0(StreamId) of
-        ok ->
-            ok
-    catch
-        Class:Reason:Stack ->
-            logger:warning("[efka_iot_deploy_stream] stream_id: ~p crashed, class: ~p, reason: ~p, stack: ~p",
-                [StreamId, Class, Reason, Stack]),
-            send_error_and_close(StreamId, iolist_to_binary(io_lib:format("~p:~p", [Class, Reason]))),
-            ok
-    after
-        efka_iot_client:stream_done(StreamId)
-    end.
-
--spec run0(stream_id()) -> ok.
-run0(StreamId) ->
-    efka_iot_client:send_stream(StreamId, opened),
-    case receive_request(StreamId, <<>>) of
-        {ok, Request} ->
-            handle_request(StreamId, Request);
-        {error, reset} ->
-            ok;
-        {error, Reason} ->
-            send_error_and_close(StreamId, reason_to_binary(Reason))
-    end.
-
--spec receive_request(stream_id(), binary()) -> {ok, binary()} | {error, term()}.
-receive_request(StreamId, Acc) ->
-    receive
-        {stream, StreamId, {data, Data}} when is_binary(Data) ->
-            NAcc = <<Acc/binary, Data/binary>>,
-            case byte_size(NAcc) =< ?MAX_REQUEST_BYTES of
-                true ->
-                    receive_request(StreamId, NAcc);
-                false ->
-                    {error, request_too_large}
-            end;
-        {stream, StreamId, fin} ->
-            {ok, Acc};
-        {stream, StreamId, {reset, _Reason}} ->
-            {error, reset};
-        Info ->
-            logger:debug("[efka_iot_deploy_stream] stream_id: ~p ignore unknown info: ~p", [StreamId, Info]),
-            receive_request(StreamId, Acc)
-    after ?REQUEST_TIMEOUT ->
-        {error, request_timeout}
-    end.
-
--spec handle_request(stream_id(), binary()) -> ok.
-handle_request(StreamId, RequestBin) ->
-    case decode_request(RequestBin) of
-        {ok, TaskId, Params} ->
-            deploy(StreamId, TaskId, Params);
-        {error, Reason} ->
-            send_error_and_close(StreamId, reason_to_binary(Reason))
-    end.
-
--spec decode_request(binary()) -> {ok, non_neg_integer(), map()} | {error, term()}.
-decode_request(RequestBin) ->
-    try json:decode(RequestBin) of
-        #{<<"task_id">> := TaskId, <<"params">> := Params}
-          when is_integer(TaskId), TaskId >= 0, is_map(Params) ->
-            {ok, TaskId, Params};
-        _ ->
-            {error, invalid_deploy_request}
-    catch
-        Class:Reason ->
-            {error, {bad_json, Class, Reason}}
-    end.
-
--spec deploy(stream_id(), non_neg_integer(), map()) -> ok.
-deploy(StreamId, TaskId, Params) ->
-    try
-        ContainerName = maps:get(<<"container_name">>, Params),
-        {ok, RootDir} = docker_helper:root_dir(),
-        {ok, ContainerDir} = docker_helper:ensure_container_dir(RootDir, ContainerName),
-        docker_deployer:deploy(TaskId, ContainerDir, Params, {stream, StreamId})
-    catch
-        Class:Reason ->
-            Error = iolist_to_binary(io_lib:format("deploy stream failed: ~p:~p", [Class, Reason])),
-            send_error_and_close(StreamId, Error)
-    end.
-
--spec send_error_and_close(stream_id(), binary()) -> ok.
-send_error_and_close(StreamId, Reason) ->
-    ok = efka_iot_client:send_stream(StreamId, {data, iolist_to_binary(json:encode(#{
-        <<"type">> => <<"error">>,
-        <<"message">> => Reason
-    }))}),
-    ok = efka_iot_client:send_stream(StreamId, {data, iolist_to_binary(json:encode(#{
-        <<"type">> => <<"close">>,
-        <<"reason">> => <<"fail">>
-    }))}),
-    efka_iot_client:send_stream(StreamId, fin).
-
--spec reason_to_binary(term()) -> binary().
-reason_to_binary(Reason) when is_binary(Reason) ->
-    Reason;
-reason_to_binary(Reason) ->
-    unicode:characters_to_binary(io_lib:format("~p", [Reason])).

apps/efka/src/iot/efka_iot_stream.erl

@@ -1,156 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @doc One transparent TCP stream from iot to the local manager service.
-%%% The iot/efka protocol does not inspect HTTP; all HTTP bytes are carried
-%%% in data frames.
-%%% @end
-%%%-------------------------------------------------------------------
--module(efka_iot_stream).
--include("message.hrl").
-
--export([start_stream/2]).
--export([run/2]).
-
--define(DEFAULT_CONNECT_TIMEOUT, 3000).
--define(DEFAULT_IDLE_TIMEOUT, 120000).
-
--type stream_id() :: pos_integer().
--type stream_target() :: pos_integer().
-
--spec start_stream(StreamId :: integer(), Target :: stream_target()) -> {ok, {pid(), reference()}}.
-start_stream(StreamId, Target) when is_integer(StreamId), StreamId > 0, is_integer(Target), Target > 0 ->
-    {ok, spawn_monitor(?MODULE, run, [StreamId, Target])}.
-
--spec run(StreamId :: stream_id(), Target :: stream_target()) -> ok.
-run(StreamId, Target) when is_integer(StreamId), StreamId > 0, is_integer(Target), Target > 0 ->
-    try run0(StreamId, Target) of
-        ok ->
-            ok
-    catch
-        Class:Reason:Stack ->
-            logger:warning("[efka_iot_stream] stream_id: ~p crashed, class: ~p, reason: ~p, stack: ~p",
-                [StreamId, Class, Reason, Stack]),
-            efka_iot_client:send_stream(StreamId, {reset, safe_term({Class, Reason})}),
-            ok
-    after
-        efka_iot_client:stream_done(StreamId)
-    end.
-
--spec run0(stream_id(), stream_target()) -> ok.
-run0(StreamId, Target) ->
-    case open_target_socket(Target) of
-        {ok, Socket, IdleTimeout} ->
-            efka_iot_client:send_stream(StreamId, opened),
-            ok = inet:setopts(Socket, [{active, once}]),
-            loop(StreamId, Socket, IdleTimeout);
-        {error, Reason} ->
-            efka_iot_client:send_stream(StreamId, {open_error, safe_term(Reason)}),
-            ok
-    end.
-
--spec open_target_socket(stream_target()) -> {ok, gen_tcp:socket(), timeout()} | {error, term()}.
-open_target_socket(Target) ->
-    case stream_target_props(Target) of
-        {ok, Props} ->
-            open_target_socket(Target, Props);
-        {error, Reason} ->
-            {error, Reason}
-    end.
-
--spec open_target_socket(stream_target(), proplists:proplist()) ->
-    {ok, gen_tcp:socket(), timeout()} | {error, term()}.
-open_target_socket(Target, Props) ->
-    Host = proplists:get_value(host, Props),
-    Port = proplists:get_value(port, Props),
-    ConnectTimeout = proplists:get_value(connect_timeout, Props, ?DEFAULT_CONNECT_TIMEOUT),
-    IdleTimeout = proplists:get_value(idle_timeout, Props, ?DEFAULT_IDLE_TIMEOUT),
-
-    SocketOpts = [
-        binary,
-        {packet, raw},
-        {active, false},
-        {nodelay, true}
-    ],
-    case gen_tcp:connect(Host, Port, SocketOpts, ConnectTimeout) of
-        {ok, Socket} ->
-            {ok, Socket, IdleTimeout};
-        {error, Reason} ->
-            {error, {connect_failed, Target, Reason}}
-    end.
-
--spec stream_target_props(stream_target()) -> {ok, proplists:proplist()} | {error, term()}.
-stream_target_props(Target) ->
-    case application:get_env(efka, stream_targets) of
-        {ok, Targets} ->
-            case proplists:get_value(Target, Targets) of
-                undefined ->
-                    {error, {unknown_stream_target, Target}};
-                Props ->
-                    {ok, Props}
-            end;
-        undefined when Target =:= ?STREAM_TARGET_MANAGER ->
-            case application:get_env(efka, stream_target) of
-                {ok, Props} ->
-                    {ok, Props};
-                undefined ->
-                    {error, {unknown_stream_target, Target}}
-            end;
-        undefined ->
-            {error, {unknown_stream_target, Target}}
-    end.
-
--spec loop(stream_id(), gen_tcp:socket(), timeout()) -> ok.
-loop(StreamId, Socket, IdleTimeout) ->
-    receive
-        {stream, StreamId, {data, Data}} when is_binary(Data) ->
-            case gen_tcp:send(Socket, Data) of
-                ok ->
-                    loop(StreamId, Socket, IdleTimeout);
-                {error, Reason} ->
-                    efka_iot_client:send_stream(StreamId, {reset, safe_term(Reason)}),
-                    close_socket(Socket)
-            end;
-        {stream, StreamId, fin} ->
-            _ = gen_tcp:shutdown(Socket, write),
-            loop(StreamId, Socket, IdleTimeout);
-        {stream, StreamId, {reset, _Reason}} ->
-            close_socket(Socket);
-        {tcp, Socket, Data} ->
-            efka_iot_client:send_stream(StreamId, {data, Data}),
-            ok = inet:setopts(Socket, [{active, once}]),
-            loop(StreamId, Socket, IdleTimeout);
-        {tcp_closed, Socket} ->
-            efka_iot_client:send_stream(StreamId, fin),
-            close_socket(Socket);
-        {tcp_error, Socket, Reason} ->
-            efka_iot_client:send_stream(StreamId, {reset, safe_term(Reason)}),
-            close_socket(Socket);
-        Info ->
-            logger:debug("[efka_iot_stream] stream_id: ~p ignore unknown info: ~p", [StreamId, Info]),
-            loop(StreamId, Socket, IdleTimeout)
-    after IdleTimeout ->
-        efka_iot_client:send_stream(StreamId, {reset, <<"idle_timeout">>}),
-        close_socket(Socket)
-    end.
-
--spec close_socket(gen_tcp:socket()) -> ok.
-close_socket(Socket) ->
-    catch gen_tcp:close(Socket),
-    ok.
-
--spec safe_term(term()) -> term().
-safe_term(true) ->
-    true;
-safe_term(false) ->
-    false;
-safe_term(undefined) ->
-    undefined;
-safe_term(Value) when is_atom(Value) ->
-    atom_to_binary(Value, utf8);
-safe_term(Value) when is_map(Value) ->
-    maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]);
-safe_term(Value) when is_list(Value) ->
-    [safe_term(Item) || Item <- Value];
-safe_term(Value) when is_tuple(Value) ->
-    list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]);
-safe_term(Value) ->
-    Value.

apps/efka/src/protobuf/efka_service_pb.erl

@@ -3,7 +3,7 @@
 %% Automatically @generated, do not edit
 %% Generated by gpb_compile version 4.21.7
 %% Version source: file
--module(service_pb).
+-module(efka_service_pb).
 
 -export([encode_msg/1, encode_msg/2, encode_msg/3]).
 -export([decode_msg/2, decode_msg/3]).
@@ -46,7 +46,7 @@
 -export([gpb_version_as_string/0, gpb_version_as_list/0]).
 -export([gpb_version_source/0]).
 
--include("service_pb.hrl").
+-include("efka_service_pb.hrl").
 -include_lib("gpb/include/gpb.hrl").
 
 %% enumerated types
@@ -558,7 +558,7 @@ dg_read_field_def_ServiceRequest(<<>>, 0, 0, _, F@_1, F@_2, _) -> #'ServiceReque
 
 d_field_ServiceRequest_packet_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_ServiceRequest_packet_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
 d_field_ServiceRequest_packet_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) ->
-    {NewFValue, RestF} = {id((X bsl N + Acc) band 18446744073709551615, TrUserData), Rest},
+    {NewFValue, RestF} = {id((X bsl N + Acc) band 4294967295, TrUserData), Rest},
     dfp_read_field_def_ServiceRequest(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
 
 d_field_ServiceRequest_register(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_ServiceRequest_register(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
@@ -687,7 +687,7 @@ dg_read_field_def_ServiceReply(<<>>, 0, 0, _, F@_1, F@_2, _) -> #'ServiceReply'{
 
 d_field_ServiceReply_packet_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_ServiceReply_packet_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
 d_field_ServiceReply_packet_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) ->
-    {NewFValue, RestF} = {id((X bsl N + Acc) band 18446744073709551615, TrUserData), Rest},
+    {NewFValue, RestF} = {id((X bsl N + Acc) band 4294967295, TrUserData), Rest},
     dfp_read_field_def_ServiceReply(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
 
 d_field_ServiceReply_result(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_ServiceReply_result(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
@@ -1111,7 +1111,7 @@ verify_msg(Msg, MsgName, Opts) ->
 -dialyzer({nowarn_function,v_msg_ServiceRequest/3}).
 v_msg_ServiceRequest(#'ServiceRequest'{packet_id = F1, request = F2}, Path, TrUserData) ->
     if F1 == undefined -> ok;
-       true -> v_type_uint64(F1, [packet_id | Path], TrUserData)
+       true -> v_type_uint32(F1, [packet_id | Path], TrUserData)
     end,
     case F2 of
         undefined -> ok;
@@ -1142,7 +1142,7 @@ v_msg_ServiceRequest(X, Path, _TrUserData) -> mk_type_error({expected_msg, 'Serv
 -dialyzer({nowarn_function,v_msg_ServiceReply/3}).
 v_msg_ServiceReply(#'ServiceReply'{packet_id = F1, reply = F2}, Path, TrUserData) ->
     if F1 == undefined -> ok;
-       true -> v_type_uint64(F1, [packet_id | Path], TrUserData)
+       true -> v_type_uint32(F1, [packet_id | Path], TrUserData)
     end,
     case F2 of
         undefined -> ok;
@@ -1203,11 +1203,11 @@ v_type_int32(N, _Path, _TrUserData) when is_integer(N), -2147483648 =< N, N =< 2
 v_type_int32(N, Path, _TrUserData) when is_integer(N) -> mk_type_error({value_out_of_range, int32, signed, 32}, N, Path);
 v_type_int32(X, Path, _TrUserData) -> mk_type_error({bad_integer, int32, signed, 32}, X, Path).
 
--compile({nowarn_unused_function,v_type_uint64/3}).
--dialyzer({nowarn_function,v_type_uint64/3}).
-v_type_uint64(N, _Path, _TrUserData) when is_integer(N), 0 =< N, N =< 18446744073709551615 -> ok;
-v_type_uint64(N, Path, _TrUserData) when is_integer(N) -> mk_type_error({value_out_of_range, uint64, unsigned, 64}, N, Path);
-v_type_uint64(X, Path, _TrUserData) -> mk_type_error({bad_integer, uint64, unsigned, 64}, X, Path).
+-compile({nowarn_unused_function,v_type_uint32/3}).
+-dialyzer({nowarn_function,v_type_uint32/3}).
+v_type_uint32(N, _Path, _TrUserData) when is_integer(N), 0 =< N, N =< 4294967295 -> ok;
+v_type_uint32(N, Path, _TrUserData) when is_integer(N) -> mk_type_error({value_out_of_range, uint32, unsigned, 32}, N, Path);
+v_type_uint32(X, Path, _TrUserData) -> mk_type_error({bad_integer, uint32, unsigned, 32}, X, Path).
 
 -compile({nowarn_unused_function,v_type_string/3}).
 -dialyzer({nowarn_function,v_type_string/3}).
@@ -1267,14 +1267,14 @@ get_msg_defs() ->
     [{{msg, 'ServiceRequest.Register'}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}]},
      {{msg, 'ServiceRequest.Subscribe'}, [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}]},
      {{msg, 'ServiceRequest'},
-      [#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, occurrence = optional, opts = []},
+      [#field{name = packet_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
        #gpb_oneof{name = request, rnum = 3,
                   fields =
                       [#field{name = register, fnum = 10, rnum = 3, type = {msg, 'ServiceRequest.Register'}, occurrence = optional, opts = []}, #field{name = subscribe, fnum = 11, rnum = 3, type = {msg, 'ServiceRequest.Subscribe'}, occurrence = optional, opts = []}],
                   opts = []}]},
      {{msg, 'ServiceReply.Error'}, [#field{name = code, fnum = 1, rnum = 2, type = int32, occurrence = optional, opts = []}, #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
      {{msg, 'ServiceReply'},
-      [#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, occurrence = optional, opts = []},
+      [#field{name = packet_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
        #gpb_oneof{name = reply, rnum = 3, fields = [#field{name = result, fnum = 10, rnum = 3, type = bytes, occurrence = optional, opts = []}, #field{name = error, fnum = 11, rnum = 3, type = {msg, 'ServiceReply.Error'}, occurrence = optional, opts = []}],
                   opts = []}]},
      {{msg, 'ServiceCast.MetricData'}, [#field{name = route_key, fnum = 1, rnum = 2, type = bytes, occurrence = optional, opts = []}, #field{name = metric, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}]},
@@ -1312,14 +1312,14 @@ fetch_enum_def(EnumName) -> erlang:error({no_such_enum, EnumName}).
 find_msg_def('ServiceRequest.Register') -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}];
 find_msg_def('ServiceRequest.Subscribe') -> [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}];
 find_msg_def('ServiceRequest') ->
-    [#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, occurrence = optional, opts = []},
+    [#field{name = packet_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
      #gpb_oneof{name = request, rnum = 3,
                 fields =
                     [#field{name = register, fnum = 10, rnum = 3, type = {msg, 'ServiceRequest.Register'}, occurrence = optional, opts = []}, #field{name = subscribe, fnum = 11, rnum = 3, type = {msg, 'ServiceRequest.Subscribe'}, occurrence = optional, opts = []}],
                 opts = []}];
 find_msg_def('ServiceReply.Error') -> [#field{name = code, fnum = 1, rnum = 2, type = int32, occurrence = optional, opts = []}, #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
 find_msg_def('ServiceReply') ->
-    [#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, occurrence = optional, opts = []},
+    [#field{name = packet_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
      #gpb_oneof{name = reply, rnum = 3, fields = [#field{name = result, fnum = 10, rnum = 3, type = bytes, occurrence = optional, opts = []}, #field{name = error, fnum = 11, rnum = 3, type = {msg, 'ServiceReply.Error'}, occurrence = optional, opts = []}],
                 opts = []}];
 find_msg_def('ServiceCast.MetricData') -> [#field{name = route_key, fnum = 1, rnum = 2, type = bytes, occurrence = optional, opts = []}, #field{name = metric, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}];

apps/efka/src/service/efka_service_channel.erl

@@ -9,7 +9,7 @@
 -module(efka_service_channel).
 -author("licheng5").
 -include("efka_tables.hrl").
--include("service_pb.hrl").
+-include("efka_service_pb.hrl").
 
 %% 一级帧类型
 %% REQUEST: 需要响应的请求帧
@@ -50,11 +50,11 @@ websocket_handle(ping, State) ->
     {reply, pong, State};
 
 websocket_handle({binary, <<?FRAME_REQUEST, PacketBin/binary>>}, State) ->
-    Request = service_pb:decode_msg(PacketBin, 'ServiceRequest'),
+    Request = efka_service_pb:decode_msg(PacketBin, 'ServiceRequest'),
     logger:debug("[efka_service_channel] get request: ~p", [Request]),
     handle_request(Request, State);
 websocket_handle({binary, <<?FRAME_CAST, PacketBin/binary>>}, State) ->
-    Cast = service_pb:decode_msg(PacketBin, 'ServiceCast'),
+    Cast = efka_service_pb:decode_msg(PacketBin, 'ServiceCast'),
     logger:debug("[efka_service_channel] get cast: ~p", [Cast]),
     handle_cast(Cast, State);
 
@@ -66,7 +66,7 @@ websocket_handle(Info, State) ->
 -spec websocket_info(term(), #state{}) ->
     {reply, term(), #state{}} | {stop, #state{}} | {ok, #state{}}.
 websocket_info({topic_broadcast, Topic, Content}, State = #state{}) ->
-    Packet = service_pb:encode_msg(#'ServiceCast'{
+    Packet = efka_service_pb:encode_msg(#'ServiceCast'{
         body = {topic_event, #'ServiceCast.TopicEvent'{topic = Topic, content = Content}}
     }),
     logger:debug("[efka_service_channel] will publish topic: ~p", [Topic]),
@@ -105,7 +105,7 @@ terminate(Reason, _Req, State = #state{service_id = ServiceId, is_registered = I
 %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
 
 %% 注册, 要建立程序和容器之间的关系
--spec handle_request(service_pb:'ServiceRequest'(), #state{}) -> {reply, {binary, binary()}, #state{}}.
+-spec handle_request(efka_service_pb:'ServiceRequest'(), #state{}) -> {reply, {binary, binary()}, #state{}}.
 handle_request(#'ServiceRequest'{packet_id = PacketId, request = {register, #'ServiceRequest.Register'{service_id = ServiceId}}}, State) ->
     {ok, ServicePid} = efka_service_sup:start_service(ServiceId),
     case efka_service:attach_channel(ServicePid, self()) of
@@ -144,7 +144,7 @@ handle_request(#'ServiceRequest'{packet_id = PacketId, request = {subscribe, #'S
 handle_request(#'ServiceRequest'{packet_id = PacketId}, State) ->
     {reply, {binary, error_reply_packet(PacketId, -1, <<"invalid request">>)}, State}.
 
--spec handle_cast(service_pb:'ServiceCast'(), #state{}) -> {ok, #state{}}.
+-spec handle_cast(efka_service_pb:'ServiceCast'(), #state{}) -> {ok, #state{}}.
 handle_cast(#'ServiceCast'{body = {metric_data, #'ServiceCast.MetricData'{route_key = RouteKey, metric = Metric}}},
     State = #state{service_pid = ServicePid, is_registered = true}) ->
     efka_service:metric_data(ServicePid, RouteKey, Metric),
@@ -154,7 +154,7 @@ handle_cast(#'ServiceCast'{body = _Body}, State) ->
 
 -spec result_reply_packet(integer(), binary()) -> binary().
 result_reply_packet(PacketId, Result) when is_integer(PacketId), is_binary(Result) ->
-    Reply = service_pb:encode_msg(#'ServiceReply'{
+    Reply = efka_service_pb:encode_msg(#'ServiceReply'{
         packet_id = PacketId,
         reply = {result, Result}
     }),
@@ -162,7 +162,7 @@ result_reply_packet(PacketId, Result) when is_integer(PacketId), is_binary(Resul
 
 -spec error_reply_packet(integer(), integer(), binary()) -> binary().
 error_reply_packet(PacketId, Code, Message) when is_integer(PacketId), is_integer(Code), is_binary(Message) ->
-    Reply = service_pb:encode_msg(#'ServiceReply'{
+    Reply = efka_service_pb:encode_msg(#'ServiceReply'{
         packet_id = PacketId,
         reply = {error, #'ServiceReply.Error'{code = Code, message = Message}}
     }),

config/sys.config

@@ -1,6 +1,6 @@
 [
   {efka, [
-    {dets_dir, "${EFKA_DETS_DIR}"},
+    {dets_dir, "/var/lib/efka/dets/"},
 
     {websocket_server, [
       {port, 18080},
@@ -10,19 +10,9 @@
     ]},
 
     {iot_server, [
-      {host, "${EFKA_IOT_HOST}"},
+      {host, "localhost"},
       {tls_port, 1443},
-      {udp_port, 24000}
-    ]},
-
-    {stream_targets, [
-      {1, [
-        {name, manager},
-        {host, "127.0.0.1"},
-        {port, 81},
-        {connect_timeout, 3000},
-        {idle_timeout, 120000}
-      ]}
+      {udp_port, 18080}
     ]},
 
     {heartbeat, [
@@ -30,14 +20,14 @@
     ]},
 
     {auth, [
-      {uuid, "${EFKA_AUTH_UUID}"},
-      {token, "${EFKA_AUTH_TOKEN}"}
+      {uuid, "qbxmjyzrkpntfgswaevodhluicqzxplkm"},
+      {token, "zpxlkvmqwnbghytrujsdieofazxcvbnm"}
     ]}
 
   ]},
 
   {docker, [
-    {root_dir, "${EFKA_DOCKER_ROOT_DIR}"}
+    {root_dir, "/var/lib/efka/docker/"}
   ]},
 
   %% 系统日志配置，使用 OTP logger
@@ -56,10 +46,10 @@
         #{
           level => debug,
           config => #{
-            file => "log/debug.log",
-            max_no_files => 10,
-            max_no_bytes => 524288000
-          },
+          file => "log/debug.log",
+          max_no_files => 10,
+          max_no_bytes => 524288000
+        },
           formatter => {logger_formatter, #{template => [time, " [", level, "] ", msg, "\n"]}}
         }
       }

config/vm.args

@@ -1,9 +1,9 @@
--sname efka
+-name efka
 
 -setcookie efka_cookie
 -kernel start_group false
 
--mnesia dir '"${EFKA_MNESIA_DIR}"'
+-mnesia dir '"/var/lib/efka/mnesia"'
 -mnesia dump_log_write_threshold 5000
 -mnesia dc_dump_limit 40
 

docs/efka_iot_protocol.md

@@ -21,7 +21,6 @@
 {<<"command">>, Ref, {Domain, Payload}}
 {<<"command_response">>, Ref, {Domain, Reply}}
 {<<"message">>, Body}
-{<<"stream">>, StreamId, Body}
 ```
 
 语义说明：
@@ -33,56 +32,11 @@
 | `{<<"command">>, Ref, {Domain, Payload}}` | iot -> efka | iot 下发命令，需要 efka 回复 |
 | `{<<"command_response">>, Ref, {Domain, Reply}}` | efka -> iot | efka 对 iot command 的回复 |
 | `{<<"message">>, Body}` | 双向 | 异步消息，不要求回复 |
-| `{<<"stream">>, StreamId, Body}` | 双向 | 透明 TCP 字节流多路复用 |
 
 `command` 和 `command_response` 的 `Domain` 表示业务域，目前支持：
 
 - `<<"container">>`
 
-## 透明 TCP Stream
-
-`stream` 用于在同一条 TLS 长连接上复用多条透明 TCP 字节流。旧的 `request`、`response`、`command`、`command_response`、`message` 帧格式保持不变；`StreamId = 0` 只作为保留概念，不出现在新的 `stream` 帧里。
-
-`StreamId` 规则：
-
-- `0`：保留给现有控制流。
-- 奇数：`iot` 发起。
-- 偶数：`efka` 发起，当前预留。
-- 同一条 TLS 连接内 `StreamId` 不复用。
-
-`Body` 取值：
-
-```erlang
-<<"open">>
-<<"opened">>
-{<<"open_error">>, Reason}
-{<<"data">>, Chunk}
-<<"fin">>
-{<<"reset">>, Reason}
-```
-
-语义：
-
-- `open`：发起方请求打开一个透明 TCP stream。`open` 不携带参数；`efka` 端固定连接本机 `stream_target`，通常指向本机 manager/nginx 入口。
-- `opened`：本地 TCP 连接建立成功。
-- `open_error`：本地 TCP 连接建立失败，stream 结束。
-- `data`：透明字节数据，`Chunk` 是 binary。HTTP 请求头、请求体、响应头、响应体都只是该 binary 的内容，iot/efka 中间协议不解析 HTTP。
-- `fin`：半关闭，发送方后续不再发送 `data`，但仍可继续接收对端 `data`。
-- `reset`：异常关闭，双方应立即释放该 `StreamId` 的资源。
-
-典型 iot 发起访问 efka 本机监控服务的流程：
-
-```erlang
-iot  -> efka: {<<"stream">>, 1, <<"open">>}
-efka -> iot:  {<<"stream">>, 1, <<"opened">>}
-
-iot  -> efka: {<<"stream">>, 1, {<<"data">>, RawHttpRequestBytes}}
-iot  -> efka: {<<"stream">>, 1, <<"fin">>}
-
-efka -> iot:  {<<"stream">>, 1, {<<"data">>, RawHttpResponseBytes}}
-efka -> iot:  {<<"stream">>, 1, <<"fin">>}
-```
-
 ## 鉴权请求
 
 初始连接由 `efka` 发起鉴权 request。每条 TLS 连接只允许一次鉴权；`iot` 侧鉴权成功后会在 `ssl_channel` 标记该连接已鉴权，如果同一连接再次发送 `auth_request`，`iot` 会直接关闭连接。

env.file

@@ -1,15 +0,0 @@
-[dev]
-EFKA_DETS_DIR="/usr/local/var/lib/efka/dets/"
-EFKA_IOT_HOST="localhost"
-EFKA_AUTH_UUID="qbxmjyzrkpntfgswaevodhluicqzxplkm"
-EFKA_AUTH_TOKEN="zpxlkvmqwnbghytrujsdieofazxcvbnm"
-EFKA_DOCKER_ROOT_DIR="/usr/local/var/lib/efka/docker/"
-EFKA_MNESIA_DIR="/usr/local/var/lib/efka/mnesia"
-
-[prod]
-EFKA_DETS_DIR="/var/lib/efka/dets/"
-EFKA_IOT_HOST="localhost"
-EFKA_AUTH_UUID="qbxmjyzrkpntfgswaevodhluicqzxplkm"
-EFKA_AUTH_TOKEN="zpxlkvmqwnbghytrujsdieofazxcvbnm"
-EFKA_DOCKER_ROOT_DIR="/var/lib/efka/docker/"
-EFKA_MNESIA_DIR="/var/lib/efka/mnesia"

rebar.config

@@ -17,10 +17,18 @@
           sasl]},
 
         {mode, dev},
-        {include_erts, false},
 
-        {sys_config_src, "./config/sys.config.src"},
-        {vm_args_src, "./config/vm.args.src"}
+        %% automatically picked up if the files
+        %% exist but can be set manually, which
+        %% is required if the names aren't exactly
+        %% sys.config and vm.args
+        {sys_config, "./config/sys.config"},
+        {vm_args, "./config/vm.args"}
+
+        %% the .src form of the configuration files do
+        %% not require setting RELX_REPLACE_OS_VARS
+        %% {sys_config_src, "./config/sys.config.src"},
+        %% {vm_args_src, "./config/vm.args.src"}
 ]}.
 
 {profiles, [{prod, [{relx,

run

@@ -1,10 +1,6 @@
 #! /bin/sh
 
-export EFKA_DETS_DIR="/usr/local/var/lib/efka/dets/"
-export EFKA_IOT_HOST="localhost"
-export EFKA_AUTH_UUID="qbxmjyzrkpntfgswaevodhluicqzxplkm"
-export EFKA_AUTH_TOKEN="zpxlkvmqwnbghytrujsdieofazxcvbnm"
-export EFKA_DOCKER_ROOT_DIR="/usr/local/var/lib/efka/docker/"
-export EFKA_MNESIA_DIR="/usr/local/var/lib/efka/mnesia"
+rebar3 compile
+rebar3 release
 
 _build/default/rel/efka/bin/efka console