add protobuf

apps/efka/include/message.hrl

@@ -0,0 +1,19 @@
+%%%-------------------------------------------------------------------
+%%% @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).

apps/efka/include/message_pb.hrl

@@ -0,0 +1,242 @@
+%% -*- 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, 32 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, 32 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'()} | {deploy, message_pb:'Command.Container.ContainerDeploy'()} | {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.CONTAINERDEPLOY_PB_H').
+-define('COMMAND.CONTAINER.CONTAINERDEPLOY_PB_H', true).
+-record('Command.Container.ContainerDeploy',
+        {task_id = 0            :: non_neg_integer() | undefined, % = 1, optional, 32 bits
+         params = <<>>          :: iodata() | undefined % = 2, 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, 32 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, 32 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.TASKEVENT_PB_H').
+-define('MESSAGE.TASKEVENT_PB_H', true).
+-record('Message.TaskEvent',
+        {task_id = 0            :: non_neg_integer() | undefined, % = 1, optional, 32 bits
+         type = <<>>            :: iodata() | undefined, % = 2, optional
+         stream = <<>>          :: iodata() | undefined % = 3, 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'()} | {task_event, message_pb:'Message.TaskEvent'()} | undefined % oneof
+        }).
+-endif.
+
+-ifndef('STREAM.OPEN_PB_H').
+-define('STREAM.OPEN_PB_H', true).
+-record('Stream.Open',
+        {
+        }).
+-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).
--define(efka_service_pb, true).
+-ifndef(service_pb).
+-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').
@@ -26,7 +26,7 @@
 -define('SERVICEREQUEST_PB_H', true).
 -record('ServiceRequest',
         {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 bits
-         request                :: {register, efka_service_pb:'ServiceRequest.Register'()} | {subscribe, efka_service_pb:'ServiceRequest.Subscribe'()} | undefined % oneof
+         request                :: {register, service_pb:'ServiceRequest.Register'()} | {subscribe, service_pb:'ServiceRequest.Subscribe'()} | undefined % oneof
         }).
 -endif.
 
@@ -42,7 +42,7 @@
 -define('SERVICEREPLY_PB_H', true).
 -record('ServiceReply',
         {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 32 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,193 @@
+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 {
+  uint32 packet_id = 1;
+
+  message AuthRequest {
+    bytes uuid = 1;
+    bytes token = 2;
+    uint64 timestamp = 3;
+  }
+
+  oneof body {
+    AuthRequest auth_request = 10;
+  }
+
+}
+
+message Response {
+  uint32 packet_id = 1;
+
+  message Error {
+    uint32 code = 1;
+    bytes reason = 2;
+  }
+
+  message AuthResponse {
+
+  }
+
+  oneof body {
+    AuthResponse auth_response = 10;
+    Error error = 11;
+  }
+
+}
+
+message Command {
+  uint32 packet_id = 1;
+
+  message Container {
+    message ContainerTarget {
+      bytes name = 1;
+      bytes id = 2;
+    }
+
+    message ContainerList {
+
+    }
+
+    message ContainerDeploy {
+      uint32 task_id = 1;
+      // The current deploy config is still an application-level structured
+      // payload. Keep it opaque until the Docker create schema is finalized.
+      bytes params = 2;
+    }
+
+    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;
+      ContainerDeploy deploy = 2;
+      ContainerStart start = 3;
+      ContainerStop stop = 4;
+      ContainerKill kill = 5;
+      ContainerRemove remove = 6;
+      ContainerConfig config = 7;
+    }
+
+  }
+
+  oneof body {
+    Container container = 10;
+  }
+
+}
+
+message CommandResponse {
+  uint32 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;
+  }
+
+  message TaskEvent {
+    uint32 task_id = 1;
+    bytes type = 2;
+    bytes stream = 3;
+  }
+
+  oneof body {
+    Ping ping = 10;
+    Pong pong = 11;
+    Pub pub = 12;
+    MetricData metric_data = 13;
+    TaskEvent task_event = 14;
+  }
+}
+
+message Stream {
+  uint32 stream_id = 1;
+
+  message Open {
+  }
+
+  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/rebar.config

@@ -4,7 +4,7 @@
     {i, "proto"},
     {f, ["service.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([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(),
+    streams = #{},
     %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
-    auth_ref = undefined :: undefined | binary(),
+    auth_ref = 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
 %%%===================================================================
@@ -59,6 +71,14 @@ task_event_stream(TaskId, Type, Stream) when is_integer(TaskId), is_binary(Type)
 close_task_event_stream(TaskId, Reason) when is_integer(TaskId), is_binary(Reason) ->
     gen_statem:cast(?SERVER, {close_task_event_stream, TaskId, Reason}).
 
+-spec send_stream(StreamId :: stream_id(), Body :: term()) -> ok.
+send_stream(StreamId, Body) when is_integer(StreamId), StreamId > 0 ->
+    gen_statem:cast(?SERVER, {send_stream, StreamId, Body}).
+
+-spec stream_done(StreamId :: stream_id()) -> ok.
+stream_done(StreamId) when is_integer(StreamId), StreamId > 0 ->
+    gen_statem:cast(?SERVER, {stream_done, StreamId}).
+
 -spec is_activated() -> boolean().
 is_activated() ->
     gen_statem:call(?SERVER, is_activated).
@@ -106,7 +126,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),
@@ -130,15 +150,33 @@ handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{so
 %% Task的stream流，只做实时的
 handle_event(cast, {task_event_stream, TaskId, Type, Stream}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
     logger:debug("[efka_iot_client] event_stream task_id: ~p, stream: ~ts", [TaskId, Stream]),
-    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => Type, <<"stream">> => Stream}}}),
+    Packet = encode_message_frame({task_event, TaskId, Type, Stream}),
     ok = ssl:send(Socket, Packet),
     {keep_state, State};
 
 handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}),
+    Packet = encode_message_frame({task_event, TaskId, <<"close">>, Reason}),
     ok = ssl:send(Socket, Packet),
     {keep_state, State};
 
+handle_event(cast, {send_stream, StreamId, Body}, ?STATE_ACTIVATED, State = #state{socket = Socket})
+    when is_integer(StreamId), StreamId > 0 ->
+    Packet = encode_stream_frame(StreamId, Body),
+    ok = ssl:send(Socket, Packet),
+    {keep_state, State};
+handle_event(cast, {send_stream, _StreamId, _Body}, _StateName, State) ->
+    {keep_state, State};
+
+handle_event(cast, {stream_done, StreamId}, _StateName, State = #state{streams = Streams})
+    when is_integer(StreamId), StreamId > 0 ->
+    case maps:take(StreamId, Streams) of
+        error ->
+            {keep_state, State};
+        {StreamState, NStreams} ->
+            demonitor_stream(StreamState),
+            {keep_state, State#state{streams = NStreams}}
+    end;
+
 %% 其他情况下直接忽略
 handle_event(cast, _, _, State = #state{}) ->
     {keep_state, State};
@@ -154,11 +192,11 @@ handle_event({call, From}, dropped_message_count, _StateName, State = #state{dro
 handle_event(info, {timeout, _, create_transport}, ?STATE_DISCONNECTED, State) ->
     case connect_socket() of
         {ok, Socket} ->
-            Ref = request_ref(),
+            {Ref, State1} = next_packet_id(State),
             AuthPacket = auth_packet(Ref),
             ok = ssl:send(Socket, AuthPacket),
             logger:debug("[efka_iot_client] send auth request, ref: ~p", [Ref]),
-            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
+            {next_state, ?STATE_AUTH, State1#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
         {error, _Reason} ->
             schedule_reconnect(),
             {keep_state, State#state{socket = undefined}}
@@ -171,7 +209,7 @@ handle_event(state_timeout, auth_timeout, ?STATE_AUTH, State = #state{socket = S
     {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
 
 handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #state{socket = Socket, ping_timer_ref = TimerRef}) ->
-    Packet = 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 +217,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_ref = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
     {keep_state, State};
@@ -207,110 +246,119 @@ 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
-        Packet ->
-            {keep_state, State, [{next_event, internal, Packet}]}
-    catch
-        error:Error ->
-            logger:warning("[efka_iot_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
+    case decode_frame(PacketBin) of
+        {ok, Packet} ->
+            {keep_state, State, [{next_event, internal, Packet}]};
+        {error, Reason} ->
+            logger:warning("[efka_iot_client] decode packet failed: ~p, packet_size: ~p", [Reason, byte_size(PacketBin)]),
             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_ref = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
     logger:debug("[efka_iot_client] ssl error: ~p", [Reason]),
     disconnect(Socket),
     cancel_ssl_ping(State),
     schedule_reconnect(),
-    {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_ref = undefined, ping_timer_ref = undefined}};
 handle_event(info, {ssl_closed, Socket}, _, State = #state{}) ->
     logger:debug("[efka_iot_client] ssl closed"),
     disconnect(Socket),
     cancel_ssl_ping(State),
     schedule_reconnect(),
-    {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_ref = 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),
+            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, safe_term({worker_down, Reason})}),
+                    ok = ssl:send(Socket, Packet),
+                    {keep_state, State#state{streams = NStreams}}
+            end
+    end;
+
+%%% 处理内部消息。TLS 收到的数据先经过 protobuf 解码，再由这里按本地事件分发。
 
 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
-handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    handle_container_command(Ref, Request, Socket),
+handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    handle_container_command(Ref, ContainerCommand, Socket),
     {keep_state, State};
-handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) ->
+handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, _StateName, State = #state{socket = Socket}) ->
+    Request = decode_container_command(ContainerCommand),
     logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
     send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
     {keep_state, State};
 
 %% 处理response
-handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
+handle_event(internal, #'Response'{packet_id = AuthRef, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
     logger:debug("[efka_iot_client] auth success"),
     State1 = schedule_ssl_ping(State#state{auth_ref = 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 = AuthRef, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
     logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]),
     disconnect(Socket),
     schedule_reconnect(),
     {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = 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机制
-handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) ->
+%% 透明 TCP stream 多路复用。
+handle_event(internal, #'Stream'{stream_id = StreamId, payload = Payload}, ?STATE_ACTIVATED, State) ->
+    Body = case Payload of
+               {open, #'Stream.Open'{}} ->
+                   open;
+               {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) ->
-    logger:warning("[efka_iot_client] ignore unknown packet: ~p", [Packet]),
+handle_event(internal, Packet, StateName, State) ->
+    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 +372,98 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
--spec auth_packet(binary()) -> binary().
-auth_packet(Ref) when is_binary(Ref) ->
+-spec auth_packet(pos_integer()) -> binary().
+auth_packet(Ref) when is_integer(Ref), Ref > 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">>, #{
-        <<"uuid">> => list_to_binary(UUID),
-        <<"token">> => list_to_binary(Token),
-        <<"timestamp">> => Timestamp
-    }}}).
+    encode_transport_frame(?CLASS_REQUEST, #'Request'{
+        packet_id = Ref,
+        body = {auth_request, #'Request.AuthRequest'{
+            uuid = list_to_binary(UUID),
+            token = list_to_binary(Token),
+            timestamp = Timestamp
+        }}
+    }).
+
+-spec encode_message_frame(term()) -> binary().
+encode_message_frame(ping) ->
+    encode_transport_frame(?CLASS_MESSAGE, #'Message'{body = {ping, #'Message.Ping'{}}});
+encode_message_frame({metric_data, RouteKey, Metric}) ->
+    encode_transport_frame(?CLASS_MESSAGE, #'Message'{body = {metric_data, #'Message.MetricData'{
+        route_key = RouteKey,
+        metric = Metric
+    }}});
+encode_message_frame({task_event, TaskId, Type, Stream}) ->
+    encode_transport_frame(?CLASS_MESSAGE, #'Message'{body = {task_event, #'Message.TaskEvent'{
+        task_id = TaskId,
+        type = Type,
+        stream = Stream
+    }}});
+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'{}};
+                  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 next_packet_id(#state{}) -> {pos_integer(), #state{}}.
+next_packet_id(State = #state{next_pkt_id = PacketId}) ->
+    {PacketId, State#state{next_pkt_id = next_packet_id_value(PacketId)}}.
+
+-spec next_packet_id_value(pos_integer()) -> pos_integer().
+next_packet_id_value(16#ffffffff) ->
+    1;
+next_packet_id_value(PacketId) when is_integer(PacketId), PacketId > 0 ->
+    PacketId + 1.
 
 -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
 connect_socket() ->
@@ -377,22 +505,198 @@ cancel_timer(TimerRef) ->
     _ = erlang:cancel_timer(TimerRef),
     ok.
 
--spec request_ref() -> binary().
-request_ref() ->
-    crypto:strong_rand_bytes(16).
-
--spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
+-spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok.
 send_container_response(Socket, Ref, Reply) ->
-    Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
+    Body = case Reply of
+               ok ->
+                   {result, <<"ok">>};
+               {ok, Result} ->
+                   {result, reply_to_binary(Result)};
+               {error, Reason} ->
+                   {error, #'CommandResponse.Error'{code = 1, reason = reason_to_binary(Reason)}};
+               Other ->
+                   {result, reply_to_binary(Other)}
+           end,
+    Packet = encode_transport_frame(?CLASS_COMMAND_RESPONSE, #'CommandResponse'{
+        packet_id = Ref,
+        body = Body
+    }),
     ok = ssl:send(Socket, Packet).
 
--spec safe_reply(term()) -> term().
-safe_reply(ok) ->
-    <<"ok">>;
-safe_reply({ok, Result}) ->
-    {<<"ok">>, safe_term(Result)};
-safe_reply({error, Reason}) ->
-    {<<"error">>, safe_term(Reason)}.
+-spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok.
+handle_container_command(Ref, ContainerCommand, Socket) ->
+    handle_container_request(Ref, decode_container_command(ContainerCommand), Socket).
+
+-spec handle_container_request(pos_integer(), map(), ssl:sslsocket()) -> ok.
+handle_container_request(Ref, #{<<"action">> := <<"list">>}, Socket) ->
+    Reply = docker_commands:get_containers(),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_request(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_request(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
+    Reply = docker_commands:start_container(container_target(Target)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_request(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_request(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_request(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_request(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_request(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 decode_container_command(message_pb:'Command.Container'()) -> map().
+decode_container_command(#'Command.Container'{action = {list, #'Command.Container.ContainerList'{}}}) ->
+    #{<<"action">> => <<"list">>};
+decode_container_command(#'Command.Container'{action = {deploy, #'Command.Container.ContainerDeploy'{task_id = TaskId, params = Params}}}) ->
+    #{<<"action">> => <<"deploy">>, <<"task_id">> => TaskId, <<"params">> => decode_term_payload(Params)};
+decode_container_command(#'Command.Container'{action = {start, #'Command.Container.ContainerStart'{target = Target}}}) ->
+    #{<<"action">> => <<"start">>, <<"target">> => decode_container_target(Target)};
+decode_container_command(#'Command.Container'{action = {stop, #'Command.Container.ContainerStop'{target = Target, timeout_seconds = TimeoutSeconds}}}) ->
+    #{<<"action">> => <<"stop">>, <<"target">> => decode_container_target(Target), <<"timeout_seconds">> => TimeoutSeconds};
+decode_container_command(#'Command.Container'{action = {kill, #'Command.Container.ContainerKill'{target = Target, signal = Signal}}}) ->
+    #{<<"action">> => <<"kill">>, <<"target">> => decode_container_target(Target), <<"signal">> => Signal};
+decode_container_command(#'Command.Container'{action = {remove, #'Command.Container.ContainerRemove'{target = Target, force = Force, remove_volumes = RemoveVolumes}}}) ->
+    #{<<"action">> => <<"remove">>, <<"target">> => decode_container_target(Target), <<"force">> => Force, <<"remove_volumes">> => RemoveVolumes};
+decode_container_command(#'Command.Container'{action = {config, #'Command.Container.ContainerConfig'{target = Target, config = Config}}}) ->
+    #{<<"action">> => <<"config">>, <<"target">> => decode_container_target(Target), <<"config">> => Config};
+decode_container_command(Other) ->
+    #{<<"action">> => <<"invalid">>, <<"payload">> => safe_term(Other)}.
+
+-spec decode_container_target(message_pb:'Command.Container.ContainerTarget'() | undefined) -> map().
+decode_container_target(#'Command.Container.ContainerTarget'{name = Name, id = Id}) ->
+    #{<<"name">> => iolist_to_binary(Name), <<"id">> => iolist_to_binary(Id)};
+decode_container_target(undefined) ->
+    #{<<"name">> => <<>>, <<"id">> => <<>>}.
+
+-spec decode_term_payload(iodata()) -> term().
+decode_term_payload(Payload) ->
+    Bin = iolist_to_binary(Payload),
+    try binary_to_term(Bin, [safe]) of
+        Term ->
+            Term
+    catch
+        error:_ ->
+            Bin
+    end.
+
+%% iot 发起 open 时不携带参数；efka 固定按本地 stream_target 建连。
+-spec handle_stream_frame(term(), term(), #state{}) -> gen_statem:event_handler_result(atom(), #state{}).
+handle_stream_frame(StreamId, open, State = #state{streams = Streams})
+    when is_integer(StreamId), StreamId > 0 ->
+    case valid_iot_stream_id(StreamId) andalso not maps:is_key(StreamId, Streams) of
+        true ->
+            {ok, {WorkerPid, MonitorRef}} = efka_iot_stream:start_stream(StreamId),
+            StreamState = #stream_state{worker_pid = WorkerPid, monitor_ref = MonitorRef},
+            {keep_state, State#state{streams = maps:put(StreamId, StreamState, Streams)}};
+        false ->
+            send_stream(StreamId, {reset, <<"invalid stream open">>}),
+            {keep_state, State}
+    end;
+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 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) ->
+    SafeValue = safe_term(Value),
+    try iolist_to_binary(Value) of
+        Bin ->
+            Bin
+    catch
+        _:_ ->
+            try iolist_to_binary(json:encode(SafeValue)) of
+                JsonBin ->
+                    JsonBin
+            catch
+                _:_ ->
+                    term_to_binary(SafeValue)
+            end
+    end.
+
+-spec reason_to_binary(term()) -> binary().
+reason_to_binary(Reason) when is_binary(Reason) ->
+    Reason;
+reason_to_binary(Reason) when is_list(Reason) ->
+    iolist_to_binary(Reason);
+reason_to_binary(Reason) ->
+    term_to_binary(safe_term(Reason)).
 
 -spec safe_term(term()) -> term().
 safe_term(true) ->

apps/efka/src/iot/efka_iot_stream.erl

@@ -0,0 +1,124 @@
+%%%-------------------------------------------------------------------
+%%% @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).
+
+-export([start_stream/1]).
+-export([run/1]).
+
+-define(DEFAULT_CONNECT_TIMEOUT, 3000).
+-define(DEFAULT_IDLE_TIMEOUT, 120000).
+
+-type stream_id() :: pos_integer().
+
+-spec start_stream(StreamId :: integer()) -> {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_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()) -> ok.
+run0(StreamId) ->
+    case open_target_socket() 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() -> {ok, gen_tcp:socket(), timeout()} | {error, term()}.
+open_target_socket() ->
+    {ok, Props} = application:get_env(efka, stream_target),
+    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, Reason}
+    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

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,13 @@
       {udp_port, 24000}
     ]},
 
+    {stream_target, [
+        {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},
-        {include_erts, true},
+        {mode, dev},
+        {include_erts, false},
 
         {sys_config_src, "./config/sys.config.src"},
         {vm_args_src, "./config/vm.args.src"}