apps/efka/include/efka_service_pb.hrl

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

apps/efka/include/message.hrl

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

apps/efka/include/message_pb.hrl

@@ -1,242 +0,0 @@
-%% -*- coding: utf-8 -*-
-%% Automatically generated, do not edit
-%% Generated by gpb_compile version 4.21.7
-
--ifndef(message_pb).
--define(message_pb, true).
-
--define(message_pb_gpb_version, "4.21.7").
-
-
--ifndef('REQUEST.AUTHREQUEST_PB_H').
--define('REQUEST.AUTHREQUEST_PB_H', true).
--record('Request.AuthRequest',
-        {uuid = <<>>            :: iodata() | undefined, % = 1, optional
-         token = <<>>           :: iodata() | undefined, % = 2, optional
-         timestamp = 0          :: non_neg_integer() | undefined % = 3, optional, 64 bits
-        }).
--endif.
-
--ifndef('REQUEST_PB_H').
--define('REQUEST_PB_H', true).
--record('Request',
-        {packet_id = 0          :: non_neg_integer() | undefined, % = 1, optional, 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/proto/message.proto

@@ -1,193 +0,0 @@
-syntax = "proto3";
-
-// iot <-> efka protocol payload definitions.
-//
-// The transport frame keeps the first byte as the coarse message class:
-//
-//   [CLASS_REQUEST][protobuf(Request)]
-//   [CLASS_RESPONSE][protobuf(Response)]
-//   [CLASS_COMMAND][protobuf(Command)]
-//   [CLASS_COMMAND_RESPONSE][protobuf(CommandResponse)]
-//   [CLASS_MESSAGE][protobuf(Message)]
-//   [CLASS_STREAM][protobuf(Stream)]
-//
-// HTTP proxy bytes carried by Stream.Data are transparent payload bytes.
-
-message Request {
-  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, ""},
+    {module_name_prefix, "efka_"},
     {module_name_suffix, "_pb"},
     {o_erl, "src/protobuf"},
     {o_hrl, "include"},

apps/efka/src/iot/efka_iot_client.erl

@@ -9,15 +9,12 @@
 -module(efka_iot_client).
 -author("anlicheng").
 -include("efka_tables.hrl").
--include("message.hrl").
--include("message_pb.hrl").
 
 -behaviour(gen_statem).
 
 %% API
 -export([start_link/0]).
 -export([metric_data/2, ping/13, 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
@@ -39,21 +36,12 @@
 -record(state, {
     socket :: undefined | ssl:sslsocket(),
     outbox :: efka_iot_outbox:outbox(),
-    streams = #{},
     %% 保存当前auth请求的ref，用来建立auth请求和响应的对应关系
-    auth_ref = undefined :: undefined | pos_integer(),
-    next_pkt_id = 1 :: pos_integer(),
+    auth_ref = undefined :: undefined | binary(),
     ping_timer_ref = undefined :: undefined | reference(),
     dropped_message_count = 0 :: non_neg_integer()
 }).
 
--record(stream_state, {
-    worker_pid :: pid(),
-    monitor_ref :: reference()
-}).
-
--type stream_id() :: pos_integer().
-
 %%%===================================================================
 %%% API
 %%%===================================================================
@@ -71,14 +59,6 @@ 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).
@@ -126,7 +106,7 @@ outbox_options() ->
 %% 异步发送数据，连接存在时直接发送；否则写入持久化 outbox。
 -spec handle_event(term(), term(), atom(), #state{}) -> term().
 handle_event(cast, {metric_data, RouteKey, Metric}, StateName, State = #state{socket = Socket}) ->
-    Packet = encode_message_frame({metric_data, RouteKey, Metric}),
+    Packet = term_to_binary({<<"message">>, {<<"data">>, #{<<"route_key">> => RouteKey, <<"metric">> => Metric}}}),
     case StateName of
         ?STATE_ACTIVATED ->
             ok = ssl:send(Socket, Packet),
@@ -150,33 +130,15 @@ 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 = encode_message_frame({task_event, TaskId, Type, Stream}),
+    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => Type, <<"stream">> => Stream}}}),
     ok = ssl:send(Socket, Packet),
     {keep_state, State};
 
 handle_event(cast, {close_task_event_stream, TaskId, Reason}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    Packet = encode_message_frame({task_event, TaskId, <<"close">>, Reason}),
+    Packet = term_to_binary({<<"message">>, {<<"task_event">>, #{<<"task_id">> => TaskId, <<"type">> => <<"close">>, <<"stream">> => Reason}}}),
     ok = ssl:send(Socket, Packet),
     {keep_state, State};
 
-handle_event(cast, {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};
@@ -192,11 +154,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, State1} = next_packet_id(State),
+            Ref = request_ref(),
             AuthPacket = auth_packet(Ref),
             ok = ssl:send(Socket, AuthPacket),
             logger:debug("[efka_iot_client] send auth request, ref: ~p", [Ref]),
-            {next_state, ?STATE_AUTH, State1#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
+            {next_state, ?STATE_AUTH, State#state{socket = Socket, auth_ref = Ref}, [{state_timeout, 5000, auth_timeout}]};
         {error, _Reason} ->
             schedule_reconnect(),
             {keep_state, State#state{socket = undefined}}
@@ -209,7 +171,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 = encode_message_frame(ping),
+    Packet = term_to_binary({<<"message">>, <<"ping">>}),
     case ssl:send(Socket, Packet) of
         ok ->
             {keep_state, schedule_ssl_ping(State)};
@@ -217,8 +179,7 @@ handle_event(info, {timeout, TimerRef, ssl_ping}, ?STATE_ACTIVATED, State = #sta
             logger:warning("[efka_iot_client] send ssl ping failed, reason: ~p", [Reason]),
             disconnect(Socket),
             schedule_reconnect(),
-            NState = close_all_streams({send_ping_failed, Reason}, State),
-            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
+            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {timeout, _TimerRef, ssl_ping}, _StateName, State) ->
     {keep_state, State};
@@ -246,119 +207,110 @@ handle_event(info, flush_cache, _, State) ->
 
 %% 处理收到的ssl消息
 handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
-    case decode_frame(PacketBin) of
-        {ok, Packet} ->
-            {keep_state, State, [{next_event, internal, Packet}]};
-        {error, Reason} ->
-            logger:warning("[efka_iot_client] decode packet failed: ~p, packet_size: ~p", [Reason, byte_size(PacketBin)]),
+    try binary_to_term(PacketBin, [safe]) of
+        Packet ->
+            {keep_state, State, [{next_event, internal, Packet}]}
+    catch
+        error:Error ->
+            logger:warning("[efka_iot_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
             disconnect(Socket),
             cancel_ssl_ping(State),
             schedule_reconnect(),
-            NState = close_all_streams(bad_packet, State),
-            {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
+            {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}}
     end;
 handle_event(info, {ssl_error, Socket, Reason}, _, State = #state{}) ->
     logger:debug("[efka_iot_client] ssl error: ~p", [Reason]),
     disconnect(Socket),
     cancel_ssl_ping(State),
     schedule_reconnect(),
-    NState = close_all_streams(ssl_error, State),
-    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
 handle_event(info, {ssl_closed, Socket}, _, State = #state{}) ->
     logger:debug("[efka_iot_client] ssl closed"),
     disconnect(Socket),
     cancel_ssl_ping(State),
     schedule_reconnect(),
-    NState = close_all_streams(ssl_closed, State),
-    {next_state, ?STATE_DISCONNECTED, NState#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
+    {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined, ping_timer_ref = undefined}};
 
-handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName, State = #state{streams = Streams}) ->
-    case take_stream_by_monitor(MonitorRef, WorkerPid, Streams) of
-        error ->
-            {keep_state, State};
-        {StreamId, StreamState, NStreams} ->
-            demonitor_stream(StreamState),
-            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 解码，再由这里按本地事件分发。
+%%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
 
 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
-handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    handle_container_command(Ref, ContainerCommand, Socket),
+handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+    handle_container_command(Ref, Request, Socket),
     {keep_state, State};
-handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, _StateName, State = #state{socket = Socket}) ->
-    Request = decode_container_command(ContainerCommand),
+handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, _StateName, State = #state{socket = Socket}) ->
     logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
     send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
     {keep_state, State};
 
 %% 处理response
-handle_event(internal, #'Response'{packet_id = AuthRef, body = {auth_response, #'Response.AuthResponse'{}}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
+handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, <<"ok">>}}, ?STATE_AUTH, State = #state{auth_ref = AuthRef}) ->
     logger:debug("[efka_iot_client] auth success"),
     State1 = schedule_ssl_ping(State#state{auth_ref = undefined}),
     {next_state, ?STATE_ACTIVATED, State1, [{next_event, info, flush_cache}]};
-handle_event(internal, #'Response'{packet_id = AuthRef, body = {error, #'Response.Error'{reason = Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
+handle_event(internal, {<<"response">>, AuthRef, {<<"auth_response">>, {<<"error">>, Reason}}}, ?STATE_AUTH, State = #state{socket = Socket, auth_ref = AuthRef}) ->
     logger:debug("[efka_iot_client] auth failed, reason: ~p", [Reason]),
     disconnect(Socket),
     schedule_reconnect(),
     {next_state, ?STATE_DISCONNECTED, State#state{socket = undefined, auth_ref = undefined}};
-handle_event(internal, #'Response'{} = Reply, StateName, State) ->
+handle_event(internal, {<<"response">>, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_iot_client] ignore unexpected response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
-handle_event(internal, #'CommandResponse'{} = Reply, StateName, State) ->
+handle_event(internal, {<<"command_response">>, _Ref, Reply}, StateName, State) ->
     logger:warning("[efka_iot_client] ignore unexpected command_response in state ~p: ~p", [StateName, Reply]),
     {keep_state, State};
 
-%% 透明 TCP stream 多路复用。
-handle_event(internal, #'Stream'{stream_id = StreamId, payload = Payload}, ?STATE_ACTIVATED, State) ->
-    Body = case Payload of
-               {open, #'Stream.Open'{}} ->
-                   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};
-
 %% 处理Pub/Sub机制
-handle_event(internal, #'Message'{body = {pong, #'Message.Pong'{}}}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) ->
     {keep_state, State};
-handle_event(internal, #'Message'{body = {pub, #'Message.Pub'{topic = Topic, qos = Qos, content = Content}}}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
     logger:debug("[efka_iot_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
     efka_subscription:publish(Topic, Qos, Content),
     {keep_state, State};
-handle_event(internal, Packet, StateName, State) ->
-    logger:warning("[efka_iot_client] ignore unknown packet: ~p, state_name: ~p", [Packet, StateName]),
+handle_event(internal, Packet, _StateName, State) ->
+    logger:warning("[efka_iot_client] ignore unknown packet: ~p", [Packet]),
     {keep_state, State};
 
 handle_event(info, Info, _, State = #state{}) ->
     logger:notice("[efka_iot_client] get unknown info: ~p", [Info]),
     {keep_state, State}.
 
+-spec handle_container_command(binary(), term(), ssl:sslsocket()) -> ok.
+handle_container_command(Ref, #{<<"action">> := <<"list">>}, Socket) ->
+    Reply = docker_commands:get_containers(),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) ->
+    Reply = docker_deploy_manager:deploy(TaskId, Params),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
+    Reply = docker_commands:start_container(container_target(Target)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) ->
+    Reply = docker_commands:stop_container(container_target(Target), TimeoutSeconds),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) ->
+    Reply = docker_commands:kill_container(container_target(Target), to_binary(Signal)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) ->
+    Reply = docker_commands:remove_container(container_target(Target), to_bool(Force), to_bool(RemoveVolumes)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) ->
+    Reply = docker_helper:update_container_config(container_target(Target), iolist_to_binary(Config)),
+    send_container_response(Socket, Ref, Reply),
+    ok;
+handle_container_command(Ref, Request, Socket) ->
+    logger:notice("[efka_iot_client] get an invalid command: ~p, agent invalid", [Request]),
+    send_container_response(Socket, Ref, {error, <<"agent invalid">>}),
+    ok.
+
 -spec terminate(term(), atom(), #state{}) -> ok.
 terminate(Reason, _StateName, State = #state{socket = Socket, outbox = Outbox}) ->
     cancel_ssl_ping(State),
-    _ = close_all_streams(Reason, State),
     disconnect(Socket),
     efka_iot_outbox:close(Outbox),
     logger:notice("[efka_iot_client] terminate with reason: ~p", [Reason]),
@@ -372,98 +324,18 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
--spec auth_packet(pos_integer()) -> binary().
-auth_packet(Ref) when is_integer(Ref), Ref > 0 ->
+-spec auth_packet(binary()) -> binary().
+auth_packet(Ref) when is_binary(Ref) ->
     {ok, AuthInfo} = application:get_env(efka, auth),
     UUID = proplists:get_value(uuid, AuthInfo),
     Token = proplists:get_value(token, AuthInfo),
 
     Timestamp = efka_util:timestamp(),
-    encode_transport_frame(?CLASS_REQUEST, #'Request'{
-        packet_id = 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.
+    term_to_binary({<<"request">>, Ref, {<<"auth_request">>, #{
+        <<"uuid">> => list_to_binary(UUID),
+        <<"token">> => list_to_binary(Token),
+        <<"timestamp">> => Timestamp
+    }}}).
 
 -spec connect_socket() -> {ok, ssl:sslsocket()} | {error, term()}.
 connect_socket() ->
@@ -505,198 +377,22 @@ cancel_timer(TimerRef) ->
     _ = erlang:cancel_timer(TimerRef),
     ok.
 
--spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok.
+-spec request_ref() -> binary().
+request_ref() ->
+    crypto:strong_rand_bytes(16).
+
+-spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
 send_container_response(Socket, Ref, Reply) ->
-    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
-    }),
+    Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
     ok = ssl:send(Socket, Packet).
 
--spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok.
-handle_container_command(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_reply(term()) -> term().
+safe_reply(ok) ->
+    <<"ok">>;
+safe_reply({ok, Result}) ->
+    {<<"ok">>, safe_term(Result)};
+safe_reply({error, Reason}) ->
+    {<<"error">>, safe_term(Reason)}.
 
 -spec safe_term(term()) -> term().
 safe_term(true) ->

apps/efka/src/iot/efka_iot_stream.erl

@@ -1,124 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @doc One transparent TCP stream from iot to the local manager service.
-%%% The iot/efka protocol does not inspect HTTP; all HTTP bytes are carried
-%%% in data frames.
-%%% @end
-%%%-------------------------------------------------------------------
--module(efka_iot_stream).
-
--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/efka_service_pb.erl

@@ -3,7 +3,7 @@
 %% Automatically @generated, do not edit
 %% Generated by gpb_compile version 4.21.7
 %% Version source: file
--module(service_pb).
+-module(efka_service_pb).
 
 -export([encode_msg/1, encode_msg/2, encode_msg/3]).
 -export([decode_msg/2, decode_msg/3]).
@@ -46,7 +46,7 @@
 -export([gpb_version_as_string/0, gpb_version_as_list/0]).
 -export([gpb_version_source/0]).
 
--include("service_pb.hrl").
+-include("efka_service_pb.hrl").
 -include_lib("gpb/include/gpb.hrl").
 
 %% enumerated types

apps/efka/src/service/efka_service_channel.erl

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

config/sys.config.src

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

rebar.config

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