fix deploy stream

apps/docker/src/docker_deploy_manager.erl

@@ -1,90 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @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,26 +11,17 @@
 -dialyzer([{nowarn_function, normalize_image/1}]).
 
 %% API
--export([start_monitor/3]).
+-export([deploy/4]).
--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",
@@ -41,34 +32,35 @@ start_monitor(TaskId, ContainerDir, Params)
 %   "command": ["nginx", "-g", "daemon off;"],
 %   "restart": "always"
 %}
--spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: map()) -> ok.
+-spec deploy(TaskId :: integer(), ContainerDir :: string(), Params :: map(), Reporter :: reporter()) -> ok.
-deploy(TaskId, ContainerDir, Params) when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
+deploy(TaskId, ContainerDir, Params, Reporter) when is_integer(TaskId), is_list(ContainerDir), is_map(Params) ->
     ContainerName = deploy_container_name(Params),
     Image0 = deploy_image(Params),
-    report_task_event(TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
+    report_stream_event(Reporter, TaskId, <<"info">>, <<"开始部署容器:"/utf8, ContainerName/binary>>),
     try
-        ok = ensure_container_absent(TaskId, ContainerName),
+        ok = ensure_container_absent(Reporter, TaskId, ContainerName),
-        {ok, Image} = ensure_image_ready(TaskId, Image0),
+        {ok, Image} = ensure_image_ready(Reporter, TaskId, Image0),
-        {ok, ContainerId} = create_container_and_config(TaskId, ContainerDir, Params),
+        {ok, ContainerId} = create_container_and_config(Reporter, TaskId, ContainerDir, Params),
         ShortContainerId = short_container_id(ContainerId),
-        report_task_event(TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>),
+        report_stream_event(Reporter, TaskId, <<"container_id">>, ContainerId),
-        report_task_event(TaskId, <<"info">>, <<"任务完成"/utf8>>),
+        report_stream_event(Reporter, TaskId, <<"info">>, <<"容器创建成功: "/utf8, ShortContainerId/binary>>),
+        report_stream_event(Reporter, TaskId, <<"info">>, <<"任务完成"/utf8>>),
         write_task_summary(TaskId, <<"success">>, ContainerName, Image, ContainerId),
-        docker_task_reporter:close(TaskId, ?TASK_SUCCESS)
+        close_task(Reporter, TaskId, ?TASK_SUCCESS)
     catch
         throw:{deploy_error, Reason} ->
-            report_task_event(TaskId, <<"error">>, Reason),
+            report_stream_event(Reporter, TaskId, <<"error">>, Reason),
-            report_task_event(TaskId, <<"error">>, <<"任务失败"/utf8>>),
+            report_stream_event(Reporter, TaskId, <<"error">>, <<"任务失败"/utf8>>),
             write_task_summary(TaskId, <<"fail">>, ContainerName, Image0, undefined),
             write_task_failure_reason(TaskId, Reason),
-            docker_task_reporter:close(TaskId, ?TASK_FAIL);
+            close_task(Reporter, TaskId, ?TASK_FAIL);
         Class:Reason:Stacktrace ->
             Error = iolist_to_binary(io_lib:format("deploy crashed: ~p:~p ~p", [Class, Reason, Stacktrace])),
-            report_task_event(TaskId, <<"error">>, Error),
+            report_stream_event(Reporter, TaskId, <<"error">>, Error),
-            report_task_event(TaskId, <<"error">>, <<"任务失败"/utf8>>),
+            report_stream_event(Reporter, TaskId, <<"error">>, <<"任务失败"/utf8>>),
             write_task_summary(TaskId, <<"fail">>, ContainerName, Image0, undefined),
             write_task_failure_reason(TaskId, Error),
-            docker_task_reporter:close(TaskId, ?TASK_FAIL)
+            close_task(Reporter, TaskId, ?TASK_FAIL)
     end.
 
 -spec normalize_image(binary()) -> binary().
@@ -81,9 +73,33 @@ normalize_image(Image) when is_binary(Image) ->
                      end,
     iolist_to_binary(lists:join(<<"/">>, PrefixParts ++ [NormalizedLast])).
 
--spec report_task_event(TaskId :: integer(), Level :: binary(), Msg :: binary()) -> ok.
+-spec report_stream_event(reporter(), TaskId :: integer(), Level :: binary(), Msg :: binary()) -> ok.
-report_task_event(TaskId, Level, Msg) when is_integer(TaskId), is_binary(Level), is_binary(Msg) ->
+report_stream_event({stream, StreamId}, _TaskId, Level, Msg) when is_integer(StreamId), is_binary(Level), is_binary(Msg) ->
-    docker_task_reporter:stream(TaskId, Level, 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 write_task_summary(integer(), binary(), binary(), binary(), undefined | binary()) -> ok.
 write_task_summary(TaskId, Status, ContainerName, Image, ContainerId)
@@ -113,55 +129,55 @@ write_task_failure_reason(TaskId, Reason) when is_integer(TaskId), is_binary(Rea
     ]),
     efka_logger:write(Info).
 
--spec ensure_container_absent(TaskId :: integer(), ContainerName :: binary()) -> ok.
+-spec ensure_container_absent(reporter(), TaskId :: integer(), ContainerName :: binary()) -> ok.
-ensure_container_absent(TaskId, ContainerName) when is_integer(TaskId), is_binary(ContainerName) ->
+ensure_container_absent(Reporter, TaskId, ContainerName) when is_integer(TaskId), is_binary(ContainerName) ->
-    report_task_event(TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>),
+    report_stream_event(Reporter, TaskId, <<"info">>, <<"开始创建容器: "/utf8, ContainerName/binary>>),
     ok.
 
--spec ensure_image_ready(TaskId :: integer(), Image0 :: binary()) -> {ok, binary()}.
+-spec ensure_image_ready(reporter(), TaskId :: integer(), Image0 :: binary()) -> {ok, binary()}.
-ensure_image_ready(TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
+ensure_image_ready(Reporter, TaskId, Image0) when is_integer(TaskId), is_binary(Image0) ->
     Image = normalize_image(Image0),
-    report_task_event(TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
+    report_stream_event(Reporter, TaskId, <<"info">>, <<"使用镜像:"/utf8, Image/binary>>),
     case docker_commands:check_image_exist(Image) of
         true ->
-            report_task_event(TaskId, <<"info">>, <<"本地镜像已存在，跳过拉取:"/utf8, Image/binary>>),
+            report_stream_event(Reporter, TaskId, <<"info">>, <<"本地镜像已存在，跳过拉取:"/utf8, Image/binary>>),
             {ok, Image};
         false ->
-            report_task_event(TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
+            report_stream_event(Reporter, TaskId, <<"info">>, <<"开始拉取镜像:"/utf8, Image/binary>>),
             case docker_commands:pull_image(Image) of
                 {ok, Ref, Pid, MRef} ->
-                    await_pull_image(TaskId, Ref, Pid, MRef),
+                    await_pull_image(Reporter, TaskId, Ref, Pid, MRef),
                     {ok, Image}
             end
     end.
 
--spec await_pull_image(TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok.
+-spec await_pull_image(reporter(), TaskId :: integer(), Ref :: reference(), Pid :: pid(), MRef :: reference()) -> ok.
-await_pull_image(TaskId, Ref, Pid, MRef) ->
+await_pull_image(Reporter, TaskId, Ref, Pid, MRef) ->
     receive
         {docker_client, Ref, {response, _Status, _Headers}} ->
-            await_pull_image(TaskId, Ref, Pid, MRef);
+            await_pull_image(Reporter, TaskId, Ref, Pid, MRef);
         {docker_client, Ref, {data, Data}} ->
-            report_task_event(TaskId, <<"info">>, Data),
+            report_stream_event(Reporter, TaskId, <<"info">>, Data),
-            await_pull_image(TaskId, Ref, Pid, MRef);
+            await_pull_image(Reporter, TaskId, Ref, Pid, MRef);
         {docker_client, Ref, done} ->
             erlang:demonitor(MRef, [flush]),
             ok;
         {docker_client, Ref, {error, Reason}} ->
             erlang:demonitor(MRef, [flush]),
-            report_task_event(TaskId, <<"error">>, Reason),
+            report_stream_event(Reporter, TaskId, <<"error">>, Reason),
             throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>});
         {'DOWN', MRef, process, Pid, Reason0} ->
             Reason = iolist_to_binary(io_lib:format("~p", [Reason0])),
             throw({deploy_error, <<"镜像拉取失败: "/utf8, Reason/binary>>})
     end.
 
--spec create_container_and_config(TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
+-spec create_container_and_config(reporter(), TaskId :: integer(), ContainerDir :: string(), Params :: map()) ->
     {ok, binary()}.
-create_container_and_config(TaskId, ContainerDir, Params)
+create_container_and_config(Reporter, 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(TaskId, ContainerDir),
+            ok = create_config_file(Reporter, TaskId, ContainerDir),
             {ok, ContainerId};
         {error, Reason} when is_binary(Reason) ->
             throw({deploy_error, format_create_container_error(Reason)});
@@ -170,8 +186,8 @@ create_container_and_config(TaskId, ContainerDir, Params)
             throw({deploy_error, Error})
     end.
 
--spec create_config_file(TaskId :: integer(), ContainerDir :: string()) -> ok.
+-spec create_config_file(reporter(), TaskId :: integer(), ContainerDir :: string()) -> ok.
-create_config_file(TaskId, ContainerDir) when is_integer(TaskId), is_list(ContainerDir) ->
+create_config_file(Reporter, 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} ->
@@ -180,7 +196,7 @@ create_config_file(TaskId, ContainerDir) when is_integer(TaskId), is_list(Contai
             ok;
         {error, Reason} ->
             ReasonBin = list_to_binary(io_lib:format("~p", [Reason])),
-            report_task_event(TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, ReasonBin/binary>>),
+            report_stream_event(Reporter, TaskId, <<"notice">>, <<"创建配置文件失败: "/utf8, ReasonBin/binary>>),
             ok
     end.
 

apps/docker/src/docker_sup.erl

@@ -19,24 +19,6 @@ 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

@@ -1,121 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @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/message.hrl

@@ -0,0 +1,25 @@
+%%%-------------------------------------------------------------------
+%%% @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

@@ -0,0 +1,225 @@
+%% -*- 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/include/service_pb.hrl

@@ -2,10 +2,10 @@
 %% Automatically generated, do not edit
 %% Generated by gpb_compile version 4.21.7
 
--ifndef(efka_service_pb).
+-ifndef(service_pb).
--define(efka_service_pb, true).
+-define(service_pb, true).
 
--define(efka_service_pb_gpb_version, "4.21.7").
+-define(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, 32 bits
+        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 64 bits
-         request                :: {register, efka_service_pb:'ServiceRequest.Register'()} | {subscribe, efka_service_pb:'ServiceRequest.Subscribe'()} | undefined % oneof
+         request                :: {register, service_pb:'ServiceRequest.Register'()} | {subscribe, 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, 32 bits
+        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 64 bits
-         reply                  :: {result, iodata()} | {error, efka_service_pb:'ServiceReply.Error'()} | undefined % oneof
+         reply                  :: {result, iodata()} | {error, 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, efka_service_pb:'ServiceCast.TopicEvent'()} | {metric_data, efka_service_pb:'ServiceCast.MetricData'()} | undefined % oneof
+        {body                   :: {topic_event, service_pb:'ServiceCast.TopicEvent'()} | {metric_data, service_pb:'ServiceCast.MetricData'()} | undefined % oneof
         }).
 -endif.
 

apps/efka/proto/message.proto

@@ -0,0 +1,179 @@
+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 {
-  uint32 packet_id = 1;
+  uint64 packet_id = 1;
 
   message Register {
     string service_id = 1;
@@ -21,7 +21,7 @@ message ServiceRequest {
 }
 
 message ServiceReply {
-  uint32 packet_id = 1;
+  uint64 packet_id = 1;
 
   message Error {
     int32 code = 1;

apps/efka/rebar.config

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

apps/efka/src/iot/efka_iot_client.erl

@@ -9,12 +9,15 @@
 -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, task_event_stream/3, close_task_event_stream/2]).
+-export([metric_data/2, ping/13]).
+-export([send_stream/2, stream_done/1]).
 -export([is_activated/0, dropped_message_count/0]).
 
 %% gen_statem callbacks
@@ -36,12 +39,21 @@
 -record(state, {
     socket :: undefined | ssl:sslsocket(),
     outbox :: efka_iot_outbox:outbox(),
-    %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
+    streams = #{},
-    auth_ref = undefined :: undefined | binary(),
+    %% 保存当前 auth 请求的 packet id，用来建立 auth 请求和响应的对应关系
+    auth_pkt_id = undefined :: undefined | pos_integer(),
+    next_pkt_id = 1 :: pos_integer(),
     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
 %%%===================================================================
@@ -51,13 +63,13 @@
 metric_data(RouteKey, Metric) when is_binary(RouteKey), is_binary(Metric) ->
     gen_statem:cast(?SERVER, {metric_data, RouteKey, Metric}).
 
--spec task_event_stream(TaskId :: integer(), Type :: binary(), Stream :: binary()) -> ok.
+-spec send_stream(StreamId :: stream_id(), Body :: term()) -> ok.
-task_event_stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type), is_binary(Stream) ->
+send_stream(StreamId, Body) when is_integer(StreamId), StreamId > 0 ->
-    gen_statem:cast(?SERVER, {task_event_stream, TaskId, Type, Stream}).
+    gen_statem:cast(?SERVER, {send_stream, StreamId, Body}).
 
--spec close_task_event_stream(TaskId :: integer(), Reason :: binary()) -> ok.
+-spec stream_done(StreamId :: stream_id()) -> ok.
-close_task_event_stream(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
+stream_done(StreamId) when is_integer(StreamId), StreamId > 0 ->
-    gen_statem:cast(?SERVER, {close_task_event_stream, TaskId, Reason}).
+    gen_statem:cast(?SERVER, {stream_done, StreamId}).
 
 -spec is_activated() -> boolean().
 is_activated() ->
@@ -106,7 +118,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 = term_to_binary({<<"message">>, {<<"data">>, #{<<"route_key">> => RouteKey, <<"metric">> => Metric}}}),
+    Packet = encode_message_frame({metric_data, RouteKey, Metric}),
     case StateName of
         ?STATE_ACTIVATED ->
             ok = ssl:send(Socket, Packet),
@@ -127,17 +139,23 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so
             end
     end;
 
-%% Task的stream流，只做实时的
+handle_event(cast, {send_stream, StreamId, Body}, ?STATE_ACTIVATED, State = #state{socket = Socket})
-handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    when is_integer(StreamId), StreamId > 0 ->
-    logger:debug("[efka_iot_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
+    Packet = encode_stream_frame(StreamId, Body),
-    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, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+handle_event(cast, {stream_done, StreamId}, _StateName, State = #state{streams = Streams})
-    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}),
+    when is_integer(StreamId), StreamId > 0 ->
-    ok = ssl:send(Socket, Packet),
+    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, _, _, State = #state{}) ->
@@ -151,14 +169,13 @@ 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) ->
+handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State = #state{next_pkt_id = PktId}) ->
     case connect_socket() of
         {ok, Socket} ->
-            Ref = request_ref(),
+            AuthPacket = auth_packet(PktId),
-            AuthPacket = auth_packet(Ref),
             ok = ssl:send(Socket, AuthPacket),
-            logger:debug("[efka_iot_client] send auth request, ref: ~p", [Ref]),
+            logger:debug("[efka_iot_client] send auth request, packet_id: ~p", [PktId]),
-            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
+            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_pkt_id = PktId, next_pkt_id = PktId + 1}, [{state_timeout, 5000, auth_timeout}]};
         {error, _Reason} ->
             schedule_reconnect(),
             {keep_state, State#state{socket = undefined}}
@@ -168,10 +185,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_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
 
 handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) ->
-    Packet = term_to_binary({<<"message">>, <<"ping">>}),
+    Packet = encode_message_frame(ping),
     case ssl:send(Socket, Packet) of
         ok ->
             {keep_state, schedule_ssl_ping(State)};
@@ -179,7 +196,8 @@ 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(),
-            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
+            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}}
     end;
 handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
     {keep_state, State};
@@ -207,110 +225,123 @@ handle_event(info, flush_cache, _, State) ->
 
 %% 处理收到的ssl消息
 handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
-    try binary_to_term(PacketBin, [safe]) of
+    case decode_frame(PacketBin) of
-        Packet ->
+        {ok, Packet} ->
-            {keep_state, State, [{next_event, internal, Packet}]}
+            {keep_state, State, [{next_event, internal, Packet}]};
-    catch
+        {error, Reason} ->
-        error:Error ->
+            logger:warning("[efka_iot_client] decode packet failed: ~p, packet_size: ~p", [Reason, byte_size(PacketBin)]),
-            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(),
-            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
+            NState = close_all_streams(bad_packet, State),
+            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = 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(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
+    NState = close_all_streams(ssl_error, State),
+    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = 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(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
+    NState = close_all_streams(ssl_closed, State),
+    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_pkt_id = undefined, ping_timer_ref = undefined}};
 
-%%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
+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 解码，再由这里按本地事件分发。
 
 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
-handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+handle_event(internal, #'Command'{packet_id = PktId, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    handle_container_command(Ref, Request, Socket),
+    handle_container_command(PktId, ContainerCommand, Socket),
     {keep_state, State};
-handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) ->
+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", [Request]),
+    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [ContainerCommand]),
-    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
+    send_container_response(Socket, PktId, {error, <<"agent invalid">>}),
     {keep_state, State};
 
 %% 处理response
-handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
+handle_event(internal, #'Response'{packet_id = AuthPktId, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_pkt_id = AuthPktId}) ->
     logger:debug("[efka_iot_client] auth success"),
-    State1 = schedule_ssl_ping(State#state{auth_ref = undefined}),
+    State1 = schedule_ssl_ping(State#state{auth_pkt_id = undefined}),
     {next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]};
-handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
+handle_event(internal, #'Response'{packet_id = AuthPktId, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_pkt_id = AuthPktId}) ->
     logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]),
     disconnect(Socket),
     schedule_reconnect(),
-    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_pkt_id = undefined}};
-handle_event(internal, {<<"response">>, _Ref, Reply}, StateName, State) ->
+handle_event(internal, #'Response'{} = Reply, StateName, State) ->
     logger:warning("[efka_iot_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
-handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) ->
+handle_event(internal, #'CommandResponse'{} = Reply, StateName, State) ->
     logger:warning("[efka_iot_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
 
-%% 处理Pub/Sub机制
+%% 透明 TCP stream 多路复用。
-handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) ->
+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};
-handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
+
+%% 处理Pub/Sub机制
+handle_event(internal, #'Message'{body = {pong, #'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) ->
     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) ->
+handle_event(internal, Packet, StateName, State) ->
-    logger:warning("[efka_iot_client] ignore unknown packet: ~p", [Packet]),
+    logger:warning("[efka_iot_client] ignore unknown packet: ~p, state_name: ~p", [Packet, StateName]),
     {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]),
@@ -324,18 +355,83 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
--spec auth_packet(binary()) -> binary().
+-spec auth_packet(pos_integer()) -> binary().
-auth_packet(Ref) when is_binary(Ref) ->
+auth_packet(PktId) when is_integer(PktId), PktId > 0 ->
     {ok, AuthInfo} = application:get_env(efka, auth),
     UUID = proplists:get_value(uuid, AuthInfo),
     Token = proplists:get_value(token, AuthInfo),
 
     Timestamp = efka_util:timestamp(),
-    term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{
+    encode_transport_frame(?CLASS_REQUEST, #'Request'{
-        <<"uuid">> => list_to_binary(UUID),
+        packet_id = PktId,
-        <<"token">> => list_to_binary(Token),
+        body = {auth_request, #'Request.AuthRequest'{
-        <<"timestamp">> => Timestamp
+            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.
 
 -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
 connect_socket() ->
@@ -377,52 +473,190 @@ cancel_timer(TimerRef) ->
     _ = erlang:cancel_timer(TimerRef),
     ok.
 
--spec request_ref() -> binary().
+-spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok.
-request_ref() ->
+send_container_response(Socket, PktId, Reply) ->
-    crypto:strong_rand_bytes(16).
+    Body = case Reply of
-
+               ok ->
--spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
+                   {result, <<"ok">>};
-send_container_response(Socket, Ref, Reply) ->
+               {ok, Result} ->
-    Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
+                   {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
+    }),
     ok = ssl:send(Socket, Packet).
 
--spec safe_reply(term()) -> term().
+-spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok.
-safe_reply(ok) ->
+handle_container_command(PktId, #'Command.Container'{action = {list, #'Command.Container.ContainerList'{}}}, Socket) ->
-    <<"ok">>;
+    Reply = docker_commands:get_containers(),
-safe_reply({ok, Result}) ->
+    send_container_response(Socket, PktId, Reply),
-    {<<"ok">>, safe_term(Result)};
+    ok;
-safe_reply({error, Reason}) ->
+handle_container_command(PktId, #'Command.Container'{action = {start, #'Command.Container.ContainerStart'{target = Target}}}, Socket) ->
-    {<<"error">>, safe_term(Reason)}.
+    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_term(term()) -> term().
+%% iot 发起 open 时携带 target；efka 根据本地 stream_targets 配置建连。
-safe_term(true) ->
+-spec handle_stream_frame(term(), term(), #state{}) -> gen_statem:event_handler_result(atom(), #state{}).
-    true;
+handle_stream_frame(StreamId, {open, Target}, State = #state{streams = Streams})
-safe_term(false) ->
+    when is_integer(StreamId), StreamId > 0, is_integer(Target), Target > 0 ->
-    false;
+    case valid_iot_stream_id(StreamId) andalso not maps:is_key(StreamId, Streams) of
-safe_term(undefined) ->
+        true ->
-    undefined;
+            case start_stream_worker(StreamId, Target) of
-safe_term(Value) when is_atom(Value) ->
+                {ok, {WorkerPid, MonitorRef}} ->
-    atom_to_binary(Value, utf8);
+                    StreamState = #stream_state{worker_pid = WorkerPid, monitor_ref = MonitorRef},
-safe_term(Value) when is_map(Value) ->
+                    {keep_state, State#state{streams = maps:put(StreamId, StreamState, Streams)}};
-    maps:from_list([{safe_term(K), safe_term(V)} || {K, V} <- maps:to_list(Value)]);
+                {error, Reason} ->
-safe_term(Value) when is_list(Value) ->
+                    send_stream(StreamId, {open_error, Reason}),
-    [safe_term(Item) || Item <- Value];
+                    {keep_state, State}
-safe_term(Value) when is_tuple(Value) ->
+            end;
-    list_to_tuple([safe_term(Item) || Item <- tuple_to_list(Value)]);
+        false ->
-safe_term(Value) ->
+            send_stream(StreamId, {reset, <<"invalid stream open">>}),
-    Value.
+            {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 container_target(map()) -> binary().
+-spec start_stream_worker(stream_id(), pos_integer()) -> {ok, {pid(), reference()}} | {error, term()}.
-container_target(Target) when is_map(Target) ->
+start_stream_worker(StreamId, ?STREAM_TARGET_MANAGER) ->
-    NameBin = to_binary(maps:get(<<"name">>, Target, <<>>)),
+    efka_iot_stream:start_stream(StreamId, ?STREAM_TARGET_MANAGER);
-    IdBin = to_binary(maps:get(<<"id">>, Target, <<>>)),
+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),
     case NameBin of
         <<>> ->
             true = IdBin =/= <<>>,
             IdBin;
         _ ->
             NameBin
-    end.
+    end;
+container_target(undefined) ->
+    error(bad_container_target).
 
 -spec to_binary(binary() | list()) -> binary().
 to_binary(Value) when is_binary(Value) ->

apps/efka/src/iot/efka_iot_deploy_stream.erl

@@ -0,0 +1,119 @@
+%%%-------------------------------------------------------------------
+%%% @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

@@ -0,0 +1,156 @@
+%%%-------------------------------------------------------------------
+%%% @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/service_pb.erl

@@ -3,7 +3,7 @@
 %% Automatically @generated, do not edit
 %% Generated by gpb_compile version 4.21.7
 %% Version source: file
--module(efka_service_pb).
+-module(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("efka_service_pb.hrl").
+-include("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 4294967295, TrUserData), Rest},
+    {NewFValue, RestF} = {id((X bsl N + Acc) band 18446744073709551615, 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 4294967295, TrUserData), Rest},
+    {NewFValue, RestF} = {id((X bsl N + Acc) band 18446744073709551615, 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_uint32(F1, [packet_id | Path], TrUserData)
+       true -> v_type_uint64(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_uint32(F1, [packet_id | Path], TrUserData)
+       true -> v_type_uint64(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_uint32/3}).
+-compile({nowarn_unused_function,v_type_uint64/3}).
--dialyzer({nowarn_function,v_type_uint32/3}).
+-dialyzer({nowarn_function,v_type_uint64/3}).
-v_type_uint32(N, _Path, _TrUserData) when is_integer(N), 0 =< N, N =< 4294967295 -> ok;
+v_type_uint64(N, _Path, _TrUserData) when is_integer(N), 0 =< N, N =< 18446744073709551615 -> 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_uint64(N, Path, _TrUserData) when is_integer(N) -> mk_type_error({value_out_of_range, uint64, unsigned, 64}, N, Path);
-v_type_uint32(X, Path, _TrUserData) -> mk_type_error({bad_integer, uint32, unsigned, 32}, X, Path).
+v_type_uint64(X, Path, _TrUserData) -> mk_type_error({bad_integer, uint64, unsigned, 64}, 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 = uint32, occurrence = optional, opts = []},
+      [#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, 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 = uint32, occurrence = optional, opts = []},
+      [#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, 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 = uint32, occurrence = optional, opts = []},
+    [#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, 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 = uint32, occurrence = optional, opts = []},
+    [#field{name = packet_id, fnum = 1, rnum = 2, type = uint64, 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("efka_service_pb.hrl").
+-include("service_pb.hrl").
 
 %% 一级帧类型
 %% REQUEST: 需要响应的请求帧
@@ -50,11 +50,11 @@ websocket_handle(ping, State) ->
     {reply, pong, State};
 
 websocket_handle({binary, <<?FRAME_REQUEST, PacketBin/binary>>}, State) ->
-    Request = efka_service_pb:decode_msg(PacketBin, 'ServiceRequest'),
+    Request = 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 = efka_service_pb:decode_msg(PacketBin, 'ServiceCast'),
+    Cast = 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 = efka_service_pb:encode_msg(#'ServiceCast'{
+    Packet = 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(efka_service_pb:'ServiceRequest'(), #state{}) -> {reply, {binary, binary()}, #state{}}.
+-spec handle_request(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(efka_service_pb:'ServiceCast'(), #state{}) -> {ok, #state{}}.
+-spec handle_cast(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 = efka_service_pb:encode_msg(#'ServiceReply'{
+    Reply = 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 = efka_service_pb:encode_msg(#'ServiceReply'{
+    Reply = service_pb:encode_msg(#'ServiceReply'{
         packet_id = PacketId,
         reply = {error, #'ServiceReply.Error'{code = Code, message = Message}}
     }),

config/sys.config.src

@@ -15,6 +15,16 @@
       {udp_port, 24000}
     ]},
 
+    {stream_targets, [
+      {1, [
+        {name, manager},
+        {host, "127.0.0.1"},
+        {port, 81},
+        {connect_timeout, 3000},
+        {idle_timeout, 120000}
+      ]}
+    ]},
+
     {heartbeat, [
       {interval, 5000}
     ]},

docs/efka_iot_protocol.md

@@ -21,6 +21,7 @@
 {<<"command">>, Ref, {Domain, Payload}}
 {<<"command_response">>, Ref, {Domain, Reply}}
 {<<"message">>, Body}
+{<<"stream">>, StreamId, Body}
 ```
 
 语义说明：
@@ -32,11 +33,56 @@
 | `{<<"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` 会直接关闭连接。

rebar.config

@@ -16,8 +16,8 @@
          [efka,
           sasl]},
 
-        {mode, prod},
+        {mode, dev},
-        {include_erts, true},
+        {include_erts, false},
 
         {sys_config_src, "./config/sys.config.src"},
         {vm_args_src, "./config/vm.args.src"}