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).
+-ifndef(service_pb).
--define(efka_service_pb, true).
+-define(service_pb, true).
 
--define(efka_service_pb_gpb_version, "4.21.7").
+-define(service_pb_gpb_version, "4.21.7").
 
 
 -ifndef('SERVICEREQUEST.REGISTER_PB_H').
@@ -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,6 +9,8 @@
 -module(efka_iot_client).
 -author("anlicheng").
 -include("efka_tables.hrl").
+-include("message.hrl").
+-include("message_pb.hrl").
 
 -behaviour(gen_statem).
 
@@ -39,7 +41,8 @@
     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()
 }).
@@ -123,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),
@@ -147,18 +150,18 @@ 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 = term_to_binary({<<"stream">>, StreamId, Body}),
+    Packet = encode_stream_frame(StreamId, Body),
     ok = ssl:send(Socket, Packet),
     {keep_state, State};
 handle_event(cast, {send_stream, _StreamId, _Body}, _StateName, State) ->
@@ -189,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}}
@@ -206,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)};
@@ -243,12 +246,11 @@ handle_event(info, flush_cache, _, State) ->
 
 %% 处理收到的ssl消息
 handle_event(info, {ssl, Socket, PacketBin}, _, State = #state{socket = Socket}) when is_binary(PacketBin) ->
-    try binary_to_term(PacketBin, [safe]) of
+    case decode_frame(PacketBin) of
-        Packet ->
+        {ok, Packet} ->
-            {keep_state, State, [{next_event, internal, Packet}]}
+            {keep_state, State, [{next_event, internal, Packet}]};
-    catch
+        {error, Reason} ->
-        error:Error ->
+            logger:warning("[efka_iot_client] decode packet failed: ~p, packet_size: ~p", [Reason, byte_size(PacketBin)]),
-            logger:warning("[efka_iot_client] binary_to_term get error: ~p, packet_size: ~p", [Error, byte_size(PacketBin)]),
             disconnect(Socket),
             cancel_ssl_ping(State),
             schedule_reconnect(),
@@ -281,57 +283,72 @@ handle_event(info, {'DOWN', MonitorRef, process, WorkerPid, Reason}, _StateName,
                 undefined ->
                     {keep_state, State#state{streams = NStreams}};
                 Socket ->
-                    Packet = term_to_binary({<<"stream">>, StreamId, {<<"reset">>, safe_term({worker_down, Reason})}}),
+                    Packet = encode_stream_frame(StreamId, {reset, safe_term({worker_down, Reason})}),
                     ok = ssl:send(Socket, Packet),
                     {keep_state, State#state{streams = NStreams}}
             end
     end;
 
-%%% 处理内部消息，ssl收到的消息会先 binary_to_term，再由这里按协议结构模式匹配
+%%% 处理内部消息。TLS 收到的数据先经过 protobuf 解码，再由这里按本地事件分发。
 
 %% 容器管理命令由 iot 发起，使用 command/command_response 语义。
-handle_event(internal, {<<"command">>, Ref, {<<"container">>, Request}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
+handle_event(internal, #'Command'{packet_id = Ref, body = {container, ContainerCommand}}, ?STATE_ACTIVATED, State = #state{socket = Socket}) ->
-    handle_container_command(Ref, Request, 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};
 
-%% 透明 TCP stream 多路复用，StreamId 只出现在新顶层 <<"stream">> 帧里。
+%% 透明 TCP stream 多路复用。
-handle_event(internal, {<<"stream">>, StreamId, Body}, ?STATE_ACTIVATED, State) ->
+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">>, StreamId, Body}, StateName, 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, Body]),
+        [StateName, StreamId, Payload]),
     {keep_state, State};
 
 %% 处理Pub/Sub机制
-handle_event(internal, {<<"message">>, <<"pong">>}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, #'Message'{body = {pong, #'Message.Pong'{}}}, ?STATE_ACTIVATED, State) ->
     {keep_state, State};
-handle_event(internal, {<<"message">>, {<<"pub">>, #{<<"topic">> := Topic, <<"qos">> := Qos, <<"content">> := Content}}}, ?STATE_ACTIVATED, State) ->
+handle_event(internal, #'Message'{body = {pub, #'Message.Pub'{topic = Topic, qos = Qos, content = Content}}}, ?STATE_ACTIVATED, State) ->
     logger:debug("[efka_iot_client] get pub topic: ~p, qos: ~p, content: ~p", [Topic, Qos, Content]),
     efka_subscription:publish(Topic, Qos, Content),
     {keep_state, State};
-handle_event(internal, Packet, _StateName, State) ->
+handle_event(internal, Packet, StateName, State) ->
-    logger:warning("[efka_iot_client] ignore unknown packet: ~p", [Packet]),
+    logger:warning("[efka_iot_client] ignore unknown packet: ~p, state_name: ~p", [Packet, StateName]),
     {keep_state, State};
 
 handle_event(info, Info, _, State = #state{}) ->
@@ -355,18 +372,98 @@ code_change(_OldVsn, StateName, State = #state{}, _Extra) ->
 %%% Internal functions
 %%%===================================================================
 
--spec auth_packet(binary()) -> binary().
+-spec auth_packet(pos_integer()) -> binary().
-auth_packet(Ref) when is_binary(Ref) ->
+auth_packet(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">>, #{
+    encode_transport_frame(?CLASS_REQUEST, #'Request'{
-        <<"uuid">> => list_to_binary(UUID),
+        packet_id = Ref,
-        <<"token">> => list_to_binary(Token),
+        body = {auth_request, #'Request.AuthRequest'{
-        <<"timestamp">> => Timestamp
+            uuid = list_to_binary(UUID),
-    }}}).
+            token = list_to_binary(Token),
+            timestamp = Timestamp
+        }}
+    }).
+
+-spec encode_message_frame(term()) -> binary().
+encode_message_frame(ping) ->
+    encode_transport_frame(?CLASS_MESSAGE, #'Message'{body = {ping, #'Message.Ping'{}}});
+encode_message_frame({metric_data, RouteKey, Metric}) ->
+    encode_transport_frame(?CLASS_MESSAGE, #'Message'{body = {metric_data, #'Message.MetricData'{
+        route_key = RouteKey,
+        metric = Metric
+    }}});
+encode_message_frame({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() ->
@@ -408,52 +505,100 @@ cancel_timer(TimerRef) ->
     _ = erlang:cancel_timer(TimerRef),
     ok.
 
--spec request_ref() -> binary().
+-spec send_container_response(ssl:sslsocket(), pos_integer(), term()) -> ok.
-request_ref() ->
-    crypto:strong_rand_bytes(16).
-
--spec send_container_response(ssl:sslsocket(), binary(), term()) -> ok.
 send_container_response(Socket, Ref, Reply) ->
-    Packet = term_to_binary({<<"command_response">>, Ref, {<<"container">>, safe_reply(Reply)}}),
+    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 handle_container_command(binary(), term(), ssl:sslsocket()) -> ok.
+-spec handle_container_command(pos_integer(), message_pb:'Command.Container'(), ssl:sslsocket()) -> ok.
-handle_container_command(Ref, #{<<"action">> := <<"list">>}, Socket) ->
+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_command(Ref, #{<<"action">> := <<"deploy">>, <<"task_id">> := TaskId, <<"params">> := Params}, Socket) ->
+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_command(Ref, #{<<"action">> := <<"start">>, <<"target">> := Target}, Socket) ->
+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_command(Ref, #{<<"action">> := <<"stop">>, <<"target">> := Target, <<"timeout_seconds">> := TimeoutSeconds}, Socket) ->
+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_command(Ref, #{<<"action">> := <<"kill">>, <<"target">> := Target, <<"signal">> := Signal}, Socket) ->
+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_command(Ref, #{<<"action">> := <<"remove">>, <<"target">> := Target, <<"force">> := Force, <<"remove_volumes">> := RemoveVolumes}, Socket) ->
+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_command(Ref, #{<<"action">> := <<"config">>, <<"target">> := Target, <<"config">> := Config}, Socket) ->
+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_command(Ref, Request, Socket) ->
+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})
+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 ->
@@ -461,11 +606,9 @@ handle_stream_frame(StreamId, <<"open">>, State = #state{streams = Streams})
             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">>}),
+            send_stream(StreamId, {reset, <<"invalid stream open">>}),
             {keep_state, State}
     end;
-handle_stream_frame(StreamId, {<<"open">>, _Params}, State) ->
-    handle_stream_frame(StreamId, <<"open">>, State);
 handle_stream_frame(StreamId, Body, State = #state{streams = Streams})
     when is_integer(StreamId), StreamId > 0 ->
     case maps:get(StreamId, Streams, undefined) of
@@ -475,7 +618,7 @@ handle_stream_frame(StreamId, Body, State = #state{streams = Streams})
         StreamState = #stream_state{worker_pid = WorkerPid} ->
             WorkerPid ! {stream, StreamId, Body},
             case Body of
-                {<<"reset">>, _Reason} ->
+                {reset, _Reason} ->
                     demonitor_stream(StreamState),
                     {keep_state, State#state{streams = maps:remove(StreamId, Streams)}};
                 _ ->
@@ -487,14 +630,14 @@ handle_stream_frame(StreamId, Body, State) ->
     {keep_state, State}.
 
 -spec maybe_reset_unknown_stream(stream_id(), term()) -> ok.
-maybe_reset_unknown_stream(_StreamId, {<<"reset">>, _Reason}) ->
+maybe_reset_unknown_stream(_StreamId, {reset, _Reason}) ->
     ok;
-maybe_reset_unknown_stream(_StreamId, <<"fin">>) ->
+maybe_reset_unknown_stream(_StreamId, fin) ->
     ok;
-maybe_reset_unknown_stream(_StreamId, {<<"open_error">>, _Reason}) ->
+maybe_reset_unknown_stream(_StreamId, {open_error, _Reason}) ->
     ok;
 maybe_reset_unknown_stream(StreamId, _Body) ->
-    send_stream(StreamId, {<<"reset">>, <<"unknown stream">>}).
+    send_stream(StreamId, {reset, <<"unknown stream">>}).
 
 -spec valid_iot_stream_id(stream_id()) -> boolean().
 valid_iot_stream_id(StreamId) ->
@@ -519,7 +662,7 @@ take_stream_by_monitor(MonitorRef, WorkerPid, Iter, Streams) ->
 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">>, safe_term(Reason)}}}
+        WorkerPid ! {stream, StreamId, {reset, {channel_closed, Reason}}}
     end, Streams),
     State#state{streams = #{}}.
 
@@ -528,13 +671,32 @@ demonitor_stream(#stream_state{monitor_ref = MonitorRef}) ->
     erlang:demonitor(MonitorRef, [flush]),
     ok.
 
--spec safe_reply(term()) -> term().
+-spec reply_to_binary(term()) -> binary().
-safe_reply(ok) ->
+reply_to_binary(Value) when is_binary(Value) ->
-    <<"ok">>;
+    Value;
-safe_reply({ok, Result}) ->
+reply_to_binary(Value) ->
-    {<<"ok">>, safe_term(Result)};
+    SafeValue = safe_term(Value),
-safe_reply({error, Reason}) ->
+    try iolist_to_binary(Value) of
-    {<<"error">>, safe_term(Reason)}.
+        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

@@ -1,7 +1,7 @@
 %%%-------------------------------------------------------------------
 %%% @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.
+%%% in data frames.
 %%% @end
 %%%-------------------------------------------------------------------
 -module(efka_iot_stream).
@@ -27,7 +27,7 @@ run(StreamId) when is_integer(StreamId), StreamId > 0 ->
         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})}),
+            efka_iot_client:send_stream(StreamId, {reset, safe_term({Class, Reason})}),
             ok
     after
         efka_iot_client:stream_done(StreamId)
@@ -37,11 +37,11 @@ run(StreamId) when is_integer(StreamId), StreamId > 0 ->
 run0(StreamId) ->
     case open_target_socket() of
         {ok, Socket, IdleTimeout} ->
-            efka_iot_client:send_stream(StreamId, <<"opened">>),
+            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)}),
+            efka_iot_client:send_stream(StreamId, {open_error, safe_term(Reason)}),
             ok
     end.
 
@@ -69,34 +69,34 @@ open_target_socket() ->
 -spec loop(stream_id(), gen_tcp:socket(), timeout()) -> ok.
 loop(StreamId, Socket, IdleTimeout) ->
     receive
-        {stream, StreamId, {<<"data">>, Data}} when is_binary(Data) ->
+        {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)}),
+                    efka_iot_client:send_stream(StreamId, {reset, safe_term(Reason)}),
                     close_socket(Socket)
             end;
-        {stream, StreamId, <<"fin">>} ->
+        {stream, StreamId, fin} ->
             _ = gen_tcp:shutdown(Socket, write),
             loop(StreamId, Socket, IdleTimeout);
-        {stream, StreamId, {<<"reset">>, _Reason}} ->
+        {stream, StreamId, {reset, _Reason}} ->
             close_socket(Socket);
         {tcp, Socket, Data} ->
-            efka_iot_client:send_stream(StreamId, {<<"data">>, 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">>),
+            efka_iot_client:send_stream(StreamId, fin),
             close_socket(Socket);
         {tcp_error, Socket, Reason} ->
-            efka_iot_client:send_stream(StreamId, {<<"reset">>, safe_term(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">>}),
+        efka_iot_client:send_stream(StreamId, {reset, <<"idle_timeout">>}),
         close_socket(Socket)
     end.
 

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}}
     }),

rebar.config

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