fix channel

apps/efka/include/efka_service.hrl

@@ -0,0 +1,18 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 26. 8月 2025 14:42
+%%%-------------------------------------------------------------------
+-author("anlicheng").
+
+%% 服务注册
+-define(PACKET_REQUEST, 16#01).
+%% 消息响应
+-define(PACKET_RESPONSE, 16#02).
+%% 上传数据
+-define(PACKET_PUSH, 16#03).
+
+-define(PACKET_PUB, 16#04).

apps/efka/include/message_pb.hrl

@@ -31,7 +31,7 @@
 -define('PUB_PB_H', true).
 -record(pub,
         {topic = <<>>           :: unicode:chardata() | undefined, % = 1, optional
-         content = <<>>         :: unicode:chardata() | undefined % = 2, optional
+         content = <<>>         :: iodata() | undefined % = 2, optional
         }).
 -endif.
 
@@ -64,7 +64,7 @@
 -define('INVOKE_PB_H', true).
 -record(invoke,
         {service_id = <<>>      :: unicode:chardata() | undefined, % = 1, optional
-         payload = <<>>         :: unicode:chardata() | undefined, % = 2, optional
+         payload = <<>>         :: iodata() | undefined, % = 2, optional
          timeout = 0            :: non_neg_integer() | undefined % = 3, optional, 32 bits
         }).
 -endif.
@@ -83,7 +83,8 @@
 -record(data,
         {service_id = <<>>      :: unicode:chardata() | undefined, % = 1, optional
          device_uuid = <<>>     :: unicode:chardata() | undefined, % = 2, optional
-         metric = <<>>          :: unicode:chardata() | undefined % = 3, optional
+         route_key = <<>>       :: unicode:chardata() | undefined, % = 3, optional
+         metric = <<>>          :: iodata() | undefined % = 4, optional
         }).
 -endif.
 
@@ -125,4 +126,12 @@
         }).
 -endif.
 
+-ifndef('ALARM_PB_H').
+-define('ALARM_PB_H', true).
+-record(alarm,
+        {service_id = <<>>      :: unicode:chardata() | undefined, % = 1, optional
+         params = <<>>          :: unicode:chardata() | undefined % = 2, optional
+        }).
+-endif.
+
 -endif.

apps/efka/src/efka.app.src

@@ -8,9 +8,10 @@
     sync,
     jiffy,
     %gpb,
-    mnesia,
     parse_trans,
     lager,
+    cowboy,
+    ranch,
     crypto,
     inets,
     ssl,

apps/efka/src/efka_app.erl

@@ -11,44 +11,12 @@
 
 start(_StartType, _StartArgs) ->
     io:setopts([{encoding, unicode}]),
-    %% 启动mnesia数据库
-    start_mnesia(),
     %% 加速内存的回收
     erlang:system_flag(fullsweep_after, 16),
+
+    ws_server:start_server(),
+
     efka_sup:start_link().
 
 stop(_State) ->
     ok.
-
-%% internal functions
-
-%% 启动内存数据库
-start_mnesia() ->
-    %% 启动数据库
-    ensure_mnesia_schema(),
-    ok = mnesia:start(),
-    Tables = mnesia:system_info(tables),
-    lager:debug("[efka_app] tables: ~p", [Tables]),
-    %% 创建数据库表
-    not lists:member(id_generator, Tables) andalso id_generator_model:create_table(),
-    not lists:member(service, Tables) andalso service_model:create_table(),
-    not lists:member(cache, Tables) andalso cache_model:create_table(),
-    not lists:member(task_log, Tables) andalso task_log_model:create_table(),
-    ok.
-
--spec ensure_mnesia_schema() -> any().
-ensure_mnesia_schema() ->
-    case mnesia:system_info(use_dir) of
-        true ->
-            lager:debug("[efka_app] mnesia schema exists"),
-            ok;
-        false ->
-            mnesia:stop(),
-            case mnesia:create_schema([node()]) of
-                ok -> ok;
-                {error, {_, {already_exists, _}}} -> ok;
-                Error ->
-                    lager:debug("[iot_app] create mnesia schema failed with error: ~p", [Error]),
-                    throw({init_schema, Error})
-            end
-    end.

apps/efka/src/efka_remote_agent.erl

@@ -6,7 +6,7 @@
 %%% @end
 %%% Created : 21. 5月 2025 18:38
 %%%-------------------------------------------------------------------
--module(efka_agent).
+-module(efka_remote_agent).
 -author("anlicheng").
 -include("message_pb.hrl").
 -include("efka.hrl").
@@ -16,7 +16,7 @@
 
 %% API
 -export([start_link/0]).
--export([metric_data/3, event/3, ping/13, request_service_config/2, await_reply/2]).
+-export([metric_data/4, event/3, ping/13, request_service_config/2, await_reply/2]).
 
 %% gen_statem callbacks
 -export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]).
@@ -46,9 +46,9 @@
 %%%===================================================================
 
 %% 发送数据
--spec metric_data(ServiceId :: binary(), DeviceUUID::binary(), LineProtocolData :: binary()) -> no_return().
+-spec metric_data(ServiceId :: binary(), DeviceUUID::binary(), RouteKey :: binary(), Metric :: binary()) -> no_return().
-metric_data(ServiceId, DeviceUUID, LineProtocolData) when is_binary(ServiceId), is_binary(DeviceUUID), is_binary(LineProtocolData) ->
+metric_data(ServiceId, DeviceUUID, RouteKey, Metric) when is_binary(ServiceId), is_binary(DeviceUUID), is_binary(RouteKey), is_binary(Metric) ->
-    gen_statem:cast(?SERVER, {metric_data, ServiceId, DeviceUUID, LineProtocolData}).
+    gen_statem:cast(?SERVER, {metric_data, ServiceId, DeviceUUID, RouteKey, Metric}).
 
 -spec event(ServiceId :: binary(), EventType :: integer(), Params :: binary()) -> no_return().
 event(ServiceId, EventType, Params) when is_binary(ServiceId), is_integer(EventType), is_binary(Params) ->
@@ -108,11 +108,12 @@ handle_event({call, From}, {request_service_config, _ReceiverPid, _ServiceId}, _
     {keep_state, State, [{reply, From, {error, <<"transport is not alive">>}}]};
 
 %% 异步发送数据, 连接存在时候直接发送；否则缓存到mnesia
-handle_event(cast, {metric_data, ServiceId, DeviceUUID, LineProtocolData}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
+handle_event(cast, {metric_data, ServiceId, DeviceUUID, RouteKey, Metric}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
     Packet = message_pb:encode_msg(#data{
         service_id = ServiceId,
         device_uuid = DeviceUUID,
-        metric = LineProtocolData
+        route_key = RouteKey,
+        metric = Metric
     }),
     efka_transport:send(TransportPid, ?METHOD_DATA, Packet),
     {keep_state, State};
@@ -182,7 +183,7 @@ handle_event(info, {connect_reply, Reply}, ?STATE_CONNECTING, State = #state{tra
             efka_transport:auth_request(TransportPid, AuthBin),
             {next_state, ?STATE_AUTH, State};
         {error, Reason} ->
-            lager:debug("[efka_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
+            lager:debug("[efka_remote_agent] connect failed, error: ~p, pid: ~p", [Reason, TransportPid]),
             efka_transport:stop(TransportPid),
             {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
     end;
@@ -193,26 +194,26 @@ handle_event(info, {auth_reply, Reply}, ?STATE_AUTH, State = #state{transport_pi
             #auth_reply{code = Code, message = Message} = message_pb:decode_msg(ReplyBin, auth_reply),
             case Code of
                 0 ->
-                    lager:debug("[efka_agent] auth success, message: ~p", [Message]),
+                    lager:debug("[efka_remote_agent] auth success, message: ~p", [Message]),
                     {next_state, ?STATE_ACTIVATED, State, [{next_event, info, flush_cache}]};
                 1 ->
                     %% 主机在后台的授权未通过；此时agent不能推送数据给云端服务器，但是云端服务器可以推送命令给agent
                     %% socket的连接状态需要维持
-                    lager:debug("[efka_agent] auth denied, message: ~p", [Message]),
+                    lager:debug("[efka_remote_agent] auth denied, message: ~p", [Message]),
                     {next_state, ?STATE_RESTRICTED, State};
                 2 ->
                     % 其他类型的错误，需要间隔时间重试
-                    lager:debug("[efka_agent] auth failed, message: ~p", [Message]),
+                    lager:debug("[efka_remote_agent] auth failed, message: ~p", [Message]),
                     efka_transport:stop(TransportPid),
                     {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}};
                 _ ->
                     % 其他类型的错误，需要间隔时间重试
-                    lager:debug("[efka_agent] auth failed, invalid message"),
+                    lager:debug("[efka_remote_agent] auth failed, invalid message"),
                     efka_transport:stop(TransportPid),
                     {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
             end;
         {error, Reason} ->
-            lager:debug("[efka_agent] auth_request failed, error: ~p", [Reason]),
+            lager:debug("[efka_remote_agent] auth_request failed, error: ~p", [Reason]),
             efka_transport:stop(TransportPid),
             {next_state, ?STATE_DENIED, State#state{transport_pid = undefined}}
     end;
@@ -315,7 +316,7 @@ handle_event(info, {server_async_call, PacketId, <<?PUSH_INVOKE:8, InvokeBin/bin
 %% 处理task_log
 handle_event(info, {server_async_call, PacketId, <<?PUSH_TASK_LOG:8, TaskLogBin/binary>>}, ?STATE_ACTIVATED, State = #state{transport_pid = TransportPid}) ->
     #fetch_task_log{task_id = TaskId} = message_pb:decode_msg(TaskLogBin, fetch_task_log),
-    lager:debug("[efka_agent] get task_log request: ~p", [TaskId]),
+    lager:debug("[efka_remote_agent] get task_log request: ~p", [TaskId]),
     {ok, Logs} = efka_inetd_task_log:get_logs(TaskId),
     Reply = case length(Logs) > 0 of
                 true ->
@@ -345,7 +346,7 @@ handle_event(info, {server_command, ?COMMAND_AUTH, <<Auth:8>>}, StateName, State
 
 %% 处理Pub/Sub机制
 handle_event(info, {server_pub, Topic, Content}, ?STATE_ACTIVATED, State) ->
-    lager:debug("[efka_agent] get pub topic: ~p, content: ~p", [Topic, Content]),
+    lager:debug("[efka_remote_agent] get pub topic: ~p, content: ~p", [Topic, Content]),
     %% 消息发送到订阅系统
     efka_subscription:publish(Topic, Content),
     {keep_state, State};
@@ -391,7 +392,7 @@ handle_event(info, {timeout, _, {request_timeout, Ref}}, ?STATE_ACTIVATED, State
 
 %% transport进程退出
 handle_event(info, {'DOWN', MRef, process, TransportPid, Reason}, _, State = #state{transport_ref = MRef}) ->
-    lager:debug("[efka_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]),
+    lager:debug("[efka_remote_agent] transport pid: ~p, exit with reason: ~p", [TransportPid, Reason]),
     erlang:start_timer(5000, self(), create_transport),
     {next_state, ?STATE_DENIED, State#state{transport_pid = undefined, transport_ref = undefined}}.
 

apps/efka/src/efka_service.erl

@@ -18,7 +18,7 @@
 -export([start_link/2]).
 -export([get_name/1, get_pid/1, attach_channel/2]).
 -export([push_config/3, request_config/1, invoke/3]).
--export([metric_data/3, send_event/3]).
+-export([metric_data/4, send_event/3]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
@@ -63,9 +63,9 @@ invoke(Pid, Ref, Payload) when is_pid(Pid), is_reference(Ref), is_binary(Payload
 request_config(Pid) when is_pid(Pid) ->
     gen_server:call(Pid, request_config).
 
--spec metric_data(Pid :: pid(), DeviceUUID :: binary(), Data :: binary()) -> no_return().
+-spec metric_data(Pid :: pid(), DeviceUUID :: binary(), RouteKey :: binary(), Metric :: binary()) -> no_return().
-metric_data(Pid, DeviceUUID, Data) when is_pid(Pid), is_binary(DeviceUUID), is_binary(Data) ->
+metric_data(Pid, DeviceUUID, RouteKey, Metric) when is_pid(Pid), is_binary(DeviceUUID), is_binary(RouteKey),  is_binary(Metric) ->
-    gen_server:cast(Pid, {metric_data, DeviceUUID, Data}).
+    gen_server:cast(Pid, {metric_data, DeviceUUID, RouteKey, Metric}).
 
 -spec send_event(Pid :: pid(), EventType :: integer(), Params :: binary()) -> no_return().
 send_event(Pid, EventType, Params) when is_pid(Pid), is_integer(EventType), is_binary(Params) ->
@@ -155,13 +155,13 @@ handle_call(_Request, _From, State = #state{}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-handle_cast({metric_data, DeviceUUID, LineProtocolData}, State = #state{service_id = ServiceId}) ->
+handle_cast({metric_data, DeviceUUID, RouteKey, Metric}, State = #state{service_id = ServiceId}) ->
-    lager:debug("[efka_service] metric_data service_id: ~p, device_uuid: ~p, metric data: ~p", [ServiceId, DeviceUUID, LineProtocolData]),
+    lager:debug("[efka_service] metric_data service_id: ~p, device_uuid: ~p, route_key: ~p, metric data: ~p", [ServiceId, DeviceUUID, RouteKey, Metric]),
-    efka_agent:metric_data(ServiceId, DeviceUUID, LineProtocolData),
+    efka_remote_agent:metric_data(ServiceId, DeviceUUID, RouteKey, Metric),
     {noreply, State};
 
 handle_cast({send_event, EventType, Params}, State = #state{service_id = ServiceId}) ->
-    efka_agent:event(ServiceId, EventType, Params),
+    efka_remote_agent:event(ServiceId, EventType, Params),
     lager:debug("[efka_service] send_event, service_id: ~p, event_type: ~p, params: ~p", [ServiceId, EventType, Params]),
     {noreply, State};
 
@@ -169,7 +169,7 @@ handle_cast({send_event, EventType, Params}, State = #state{service_id = Service
 handle_cast({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{channel_pid = ChannelPid, service_id = ServiceId, inflight = Inflight, callbacks = Callbacks}) ->
     case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of
         true ->
-            efka_tcp_channel:push_config(ChannelPid, Ref, self(), ConfigJson),
+            gen_channel:push_config(ChannelPid, Ref, self(), ConfigJson),
             %% 设置成功，需要更新微服务的配置
             CB = fun() -> service_model:set_config(ServiceId, ConfigJson) end,
             {noreply, State#state{inflight = maps:put(Ref, ReceiverPid, Inflight), callbacks = maps:put(Ref, CB, Callbacks)}};
@@ -182,7 +182,7 @@ handle_cast({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{channel_
 handle_cast({invoke, Ref, ReceiverPid, Payload}, State = #state{channel_pid = ChannelPid, inflight = Inflight}) ->
     case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of
         true ->
-            efka_tcp_channel:invoke(ChannelPid, Ref, self(), Payload),
+            gen_channel:invoke(ChannelPid, Ref, self(), Payload),
             {noreply, State#state{inflight = maps:put(Ref, ReceiverPid, Inflight)}};
         false ->
             ReceiverPid ! {service_reply, Ref, {error, <<"channel is not alive">>}},

apps/efka/src/efka_std_modbus_service.erl

@@ -157,11 +157,11 @@ handle_call(_Request, _From, State = #state{}) ->
     {stop, Reason :: term(), NewState :: #state{}}).
 handle_cast({metric_data, DeviceUUID, LineProtocolData}, State = #state{service_id = ServiceId}) ->
     lager:debug("[efka_service] metric_data service_id: ~p, device_uuid: ~p, metric data: ~p", [ServiceId, DeviceUUID, LineProtocolData]),
-    efka_agent:metric_data(ServiceId, DeviceUUID, LineProtocolData),
+    efka_remote_agent:metric_data(ServiceId, DeviceUUID, LineProtocolData),
     {noreply, State};
 
 handle_cast({send_event, EventType, Params}, State = #state{service_id = ServiceId}) ->
-    efka_agent:event(ServiceId, EventType, Params),
+    efka_remote_agent:event(ServiceId, EventType, Params),
     lager:debug("[efka_service] send_event, service_id: ~p, event_type: ~p, params: ~p", [ServiceId, EventType, Params]),
     {noreply, State};
 

apps/efka/src/efka_sup.erl

@@ -28,6 +28,15 @@ start_link() ->
 init([]) ->
     SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
     ChildSpecs = [
+        #{
+            id => 'efka_model_sup',
+            start => {'efka_model_sup', start_link, []},
+            restart => permanent,
+            shutdown => 2000,
+            type => supervisor,
+            modules => ['efka_model_sup']
+        },
+
         #{
             id => 'efka_inetd_task_log',
             start => {'efka_inetd_task_log', start_link, []},
@@ -56,30 +65,21 @@ init([]) ->
         },
 
         #{
-            id => 'efka_agent',
+            id => 'efka_remote_agent',
-            start => {'efka_agent', start_link, []},
+            start => {'efka_remote_agent', start_link, []},
             restart => permanent,
             shutdown => 2000,
             type => worker,
-            modules => ['efka_agent']
+            modules => ['efka_remote_agent']
         },
 
         #{
-            id => 'efka_tcp_sup',
+            id => 'tcp_sup',
-            start => {'efka_tcp_sup', start_link, []},
+            start => {'tcp_sup', start_link, []},
             restart => permanent,
             shutdown => 2000,
             type => supervisor,
-            modules => ['efka_tcp_sup']
+            modules => ['tcp_sup']
-        },
-
-        #{
-            id => 'efka_tcp_server',
-            start => {'efka_tcp_server', start_link, []},
-            restart => permanent,
-            shutdown => 2000,
-            type => worker,
-            modules => ['efka_tcp_server']
         },
 
         #{

apps/efka/src/gen_channel.erl

@@ -0,0 +1,90 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 27. 8月 2025 15:22
+%%%-------------------------------------------------------------------
+-module(gen_channel).
+-author("anlicheng").
+-include("efka_service.hrl").
+
+-export([register/2]).
+-export([push_config/4, invoke/4, channel_reply/3]).
+-export([next_packet_id/1]).
+-export([json_result/2, json_error/3]).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+-spec push_config(ChannelPid :: pid(), Ref :: reference(), ReceiverPid :: pid(), ConfigJson :: binary()) -> no_return().
+push_config(ChannelPid, Ref, ReceiverPid, ConfigJson) when is_pid(ChannelPid), is_pid(ReceiverPid), is_binary(ConfigJson), is_reference(Ref) ->
+    ChannelPid ! {push_config, Ref, ReceiverPid, ConfigJson}.
+
+-spec invoke(ChannelPid :: pid(), Ref :: reference(), ReceiverPid :: pid(), Payload :: binary()) -> no_return().
+invoke(ChannelPid, Ref, ReceiverPid, Payload) when is_pid(ChannelPid), is_pid(ReceiverPid), is_binary(Payload), is_reference(Ref) ->
+    ChannelPid ! {invoke, Ref, ReceiverPid, Payload}.
+
+%% 超时逻辑处理
+channel_reply(Id, Reply, Inflight) ->
+    case maps:take(Id, Inflight) of
+        error ->
+            Inflight;
+        {{ReceiverPid, Ref}, NInflight} ->
+            case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
+                true ->
+                    ReceiverPid ! {channel_reply, Ref, Reply};
+                false ->
+                    ok
+            end,
+            NInflight
+    end.
+
+%% 注册
+-spec register(Id :: integer(), ServiceId :: binary()) -> {error, Reply :: binary()} | {ok, Reply :: binary(), ServicePid :: pid()}.
+register(Id, ServiceId) when is_integer(Id), is_binary(ServiceId) ->
+    case efka_service:get_pid(ServiceId) of
+        undefined ->
+            lager:warning("[gen_channel] service_id: ~p, not running", [ServiceId]),
+            Reply = json_error(Id, -1, <<"service not running">>),
+            {error, Reply};
+        ServicePid when is_pid(ServicePid) ->
+            case efka_service:attach_channel(ServicePid, self()) of
+                ok ->
+                    Reply = json_result(Id, <<"ok">>),
+                    erlang:monitor(process, ServicePid),
+                    {ok, Reply, ServicePid};
+                {error, Error} ->
+                    lager:warning("[gen_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]),
+                    Reply = json_error(Id, -1, Error),
+                    {error, Reply}
+            end
+    end.
+
+%% 采用32位编码
+-spec next_packet_id(PacketId :: integer()) -> NextPacketId :: integer().
+next_packet_id(PacketId) when PacketId >= 4294967295 ->
+    1;
+next_packet_id(PacketId) ->
+    PacketId + 1.
+
+-spec json_result(Id :: integer(), Result :: term()) -> binary().
+json_result(Id, Result) when is_integer(Id) ->
+    Response = #{
+        <<"id">> => Id,
+        <<"result">> => Result
+    },
+    jiffy:encode(Response, [force_utf8]).
+
+-spec json_error(Id :: integer(), Code :: integer(), Message :: binary()) -> binary().
+json_error(Id, Code, Message) when is_integer(Id), is_integer(Code), is_binary(Message) ->
+    Response = #{
+        <<"id">> => Id,
+        <<"error">> => #{
+            <<"code">> => Code,
+            <<"message">> => Message
+        }
+    },
+    jiffy:encode(Response, [force_utf8]).

apps/efka/src/mnesia/cache_model.erl

@@ -1,74 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @author aresei
-%%% @copyright (C) 2023, <COMPANY>
-%%% @doc
-%%%
-%%% @end
-%%% Created : 04. 7月 2023 12:31
-%%%-------------------------------------------------------------------
--module(cache_model).
--author("aresei").
--include("efka_tables.hrl").
--include_lib("stdlib/include/qlc.hrl").
-
--define(TAB, cache).
-
-%% API
--export([create_table/0]).
--export([insert/2, get_all_cache/0, fetch_next/0, delete/1, next_id/0]).
--export([first_key/0]).
-
-create_table() ->
-    %% id生成器
-    {atomic, ok} = mnesia:create_table(cache, [
-        {attributes, record_info(fields, cache)},
-        {record_name, cache},
-        {disc_copies, [node()]},
-        {type, ordered_set}
-    ]).
-
-next_id() ->
-    id_generator_model:next_id(?TAB).
-
--spec insert(Method :: integer(), Data :: binary()) -> ok | {error, Reason :: any()}.
-insert(Method, Data) when is_integer(Method), is_binary(Data) ->
-    Cache = #cache{id = next_id(), method = Method, data = Data},
-    case mnesia:transaction(fun() -> mnesia:write(?TAB, Cache, write) end) of
-        {'atomic', ok} ->
-            ok;
-        {'aborted', Reason} ->
-            {error, Reason}
-    end.
-
-fetch_next() ->
-    case mnesia:dirty_first(?TAB) of
-        '$end_of_table' ->
-            error;
-        Id ->
-            [Entry] = mnesia:dirty_read(?TAB, Id),
-            {ok, Entry}
-    end.
-
-delete(Id) when is_integer(Id) ->
-    case mnesia:transaction(fun() -> mnesia:delete(?TAB, Id, write) end) of
-        {'atomic', ok} ->
-            ok;
-        {'aborted', Reason} ->
-            {error, Reason}
-    end.
-
--spec get_all_cache() -> [#cache{}].
-get_all_cache() ->
-    Fun = fun() ->
-        Q = qlc:q([E || E <- mnesia:table(?TAB)]),
-        qlc:e(Q)
-          end,
-    case mnesia:transaction(Fun) of
-        {'atomic', Res} ->
-            Res;
-        {'aborted', _} ->
-            []
-    end.
-
-first_key() ->
-    mnesia:dirty_first(?TAB).

apps/efka/src/mnesia/id_generator_model.erl

@@ -1,26 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @author anlicheng
-%%% @copyright (C) 2025, <COMPANY>
-%%% @doc
-%%%
-%%% @end
-%%% Created : 06. 5月 2025 10:32
-%%%-------------------------------------------------------------------
--module(id_generator_model).
--author("anlicheng").
--include("efka_tables.hrl").
-
-%% API
--export([create_table/0, next_id/1]).
-
-create_table() ->
-    %% id生成器
-    {atomic, ok} = mnesia:create_table(id_generator, [
-        {attributes, record_info(fields, id_generator)},
-        {record_name, id_generator},
-        {disc_copies, [node()]},
-        {type, ordered_set}
-    ]).
-
-next_id(Tab) when is_atom(Tab) ->
-    mnesia:dirty_update_counter(id_generator, Tab, 1).

apps/efka/src/mnesia/service_model.erl

@@ -1,140 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @author aresei
-%%% @copyright (C) 2023, <COMPANY>
-%%% @doc
-%%%
-%%% @end
-%%% Created : 04. 7月 2023 12:31
-%%%-------------------------------------------------------------------
--module(service_model).
--author("aresei").
--include("efka_tables.hrl").
--include_lib("stdlib/include/qlc.hrl").
-
--define(TAB, service).
-
-%% API
--export([create_table/0]).
--export([insert/1, get_all_services/0, get_all_service_ids/0, get_running_services/0]).
--export([get_config_json/1, set_config/2, get_service/1, get_status/1, change_status/2]).
--export([display_services/0]).
-
-create_table() ->
-    %% id生成器
-    {atomic, ok} = mnesia:create_table(service, [
-        {attributes, record_info(fields, service)},
-        {record_name, service},
-        {disc_copies, [node()]},
-        {type, ordered_set}
-    ]).
-
-insert(Service = #service{}) ->
-    case mnesia:transaction(fun() -> mnesia:write(?TAB, Service, write) end) of
-        {'atomic', Res} ->
-            Res;
-        {'aborted', Reason} ->
-            {error, Reason}
-    end.
-
-change_status(ServiceId, NewStatus) when is_binary(ServiceId), is_integer(NewStatus) ->
-    Fun = fun() ->
-        case mnesia:read(?TAB, ServiceId, write) of
-            [] ->
-                mnesia:abort(<<"service not found">>);
-            [Service] ->
-                mnesia:write(?TAB, Service#service{status = NewStatus}, write)
-        end
-          end,
-    case mnesia:transaction(Fun) of
-        {'atomic', ok} ->
-            ok;
-        {'aborted', Reason} ->
-            {error, Reason}
-    end.
-
--spec set_config(ServiceId :: binary(), ConfigJson :: binary()) -> ok | {error, Reason :: any()}.
-set_config(ServiceId, ConfigJson) when is_binary(ServiceId), is_binary(ConfigJson) ->
-    Fun = fun() ->
-        case mnesia:read(?TAB, ServiceId, write) of
-            [] ->
-                mnesia:abort(<<"service not found">>);
-            [S] ->
-                mnesia:write(?TAB, S#service{config_json = ConfigJson}, write)
-        end
-          end,
-    case mnesia:transaction(Fun) of
-        {'atomic', ok} ->
-            ok;
-        {'aborted', Reason} ->
-            {error, Reason}
-    end.
-
--spec get_config_json(ServiceId :: binary()) -> error | {ok, ConfigJson :: binary()}.
-get_config_json(ServiceId) when is_binary(ServiceId) ->
-    case mnesia:dirty_read(?TAB, ServiceId) of
-        [] ->
-            error;
-        [#service{config_json = ConfigJson}] ->
-            {ok, ConfigJson}
-    end.
-
--spec get_status(ServiceId :: binary()) -> Status :: integer().
-get_status(ServiceId) when is_binary(ServiceId) ->
-    case mnesia:dirty_read(?TAB, ServiceId) of
-        [] ->
-            0;
-        [#service{status = Status}] ->
-            Status
-    end.
-
--spec get_service(ServiceId :: binary()) -> error | {ok, Service :: #service{}}.
-get_service(ServiceId) when is_binary(ServiceId) ->
-    case mnesia:dirty_read(?TAB, ServiceId) of
-        [] ->
-            error;
-        [Service] ->
-            {ok, Service}
-    end.
-
--spec get_all_services() -> [#service{}].
-get_all_services() ->
-    Fun = fun() ->
-        Q = qlc:q([E || E <- mnesia:table(?TAB)]),
-        qlc:e(Q)
-          end,
-
-    case mnesia:transaction(Fun) of
-        {'atomic', Res} ->
-            Res;
-        {'aborted', _} ->
-            []
-    end.
-
--spec get_all_service_ids() -> [ServiceId :: binary()].
-get_all_service_ids() ->
-    mnesia:dirty_all_keys(?TAB).
-
--spec get_running_services() -> {ok, [#service{}]} | {error, Reason :: term()}.
-get_running_services() ->
-    F = fun() ->
-        Q = qlc:q([E || E <- mnesia:table(?TAB), E#service.status == 1]),
-        qlc:e(Q)
-        end,
-    case mnesia:transaction(F) of
-        {atomic, Services} ->
-            {ok, Services};
-        {aborted, Error} ->
-            {error, Error}
-    end.
-
-display_services() ->
-    F = fun() ->
-        Q = qlc:q([E || E <- mnesia:table(?TAB)]),
-        qlc:e(Q)
-        end,
-    case mnesia:transaction(F) of
-        {atomic, Services} ->
-            {ok, Services};
-        {aborted, Error} ->
-            {error, Error}
-    end.

apps/efka/src/mnesia/task_log_model.erl

@@ -1,46 +0,0 @@
-%%%-------------------------------------------------------------------
-%%% @author aresei
-%%% @copyright (C) 2023, <COMPANY>
-%%% @doc
-%%%
-%%% @end
-%%% Created : 04. 7月 2023 12:31
-%%%-------------------------------------------------------------------
--module(task_log_model).
--author("aresei").
--include("efka_tables.hrl").
--include_lib("stdlib/include/qlc.hrl").
-
--define(TAB, task_log).
-
-%% API
--export([create_table/0]).
--export([insert/2, get_logs/1]).
-
-create_table() ->
-    %% id生成器
-    {atomic, ok} = mnesia:create_table(task_log, [
-        {attributes, record_info(fields, task_log)},
-        {record_name, task_log},
-        {disc_copies, [node()]},
-        {type, ordered_set}
-    ]).
-
--spec insert(TaskId :: integer(), Logs :: [binary()]) -> ok | {error, Reason :: term()}.
-insert(TaskId, Logs) when is_integer(TaskId), is_list(Logs) ->
-    TaskLog = #task_log{task_id = TaskId, logs = Logs},
-    case mnesia:transaction(fun() -> mnesia:write(?TAB, TaskLog, write) end) of
-        {'atomic', Res} ->
-            Res;
-        {'aborted', Reason} ->
-            {error, Reason}
-    end.
-
--spec get_logs(TaskId :: integer()) -> Logs :: [binary()].
-get_logs(TaskId) when is_integer(TaskId) ->
-    case mnesia:dirty_read(?TAB, TaskId) of
-        [] ->
-            [];
-        [#task_log{logs = Logs}] ->
-            Logs
-    end.

apps/efka/src/models/efka_model_sup.erl

@@ -0,0 +1,74 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 18. 4月 2025 16:42
+%%%-------------------------------------------------------------------
+-module(efka_model_sup).
+-author("anlicheng").
+-include("efka_tables.hrl").
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/0]).
+
+%% Supervisor callbacks
+-export([init/1]).
+
+-define(SERVER, ?MODULE).
+
+%%%===================================================================
+%%% API functions
+%%%===================================================================
+
+%% @doc Starts the supervisor
+-spec(start_link() -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
+start_link() ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+
+%%%===================================================================
+%%% Supervisor callbacks
+%%%===================================================================
+
+%% @private
+%% @doc Whenever a supervisor is started using supervisor:start_link/[2,3],
+%% this function is called by the new process to find out about
+%% restart strategy, maximum restart frequency and child
+%% specifications.
+init([]) ->
+    SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
+    Specs = [
+        #{
+            id => cache_model,
+            start => {cache_model, start_link, []},
+            restart => permanent,
+            shutdown => 5000,
+            type => worker,
+            modules => ['cache_model']
+        },
+        #{
+            id => service_model,
+            start => {service_model, start_link, []},
+            restart => permanent,
+            shutdown => 5000,
+            type => worker,
+            modules => ['service_model']
+        },
+        #{
+            id => task_log_model,
+            start => {task_log_model, start_link, []},
+            restart => permanent,
+            shutdown => 5000,
+            type => worker,
+            modules => ['task_log_model']
+        }
+    ],
+
+    {ok, {SupFlags, Specs}}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================

apps/efka/src/models/service_model.erl

@@ -14,6 +14,7 @@
 
 %% API
 -export([start_link/0]).
+-export([insert/1, change_status/2, set_config/2, get_config_json/1, get_status/1, get_service/1, get_all_services/0, get_running_services/0]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
@@ -55,11 +56,7 @@ get_service(ServiceId) when is_binary(ServiceId) ->
 get_all_services() ->
     gen_server:call(?SERVER, get_all_services).
 
--spec get_all_service_ids() -> [ServiceId :: binary()].
+-spec get_running_services() -> {ok, [#service{}]}.
-get_all_service_ids() ->
-    gen_server:call(?SERVER, get_all_service_ids).
-
--spec get_running_services() -> {ok, [#service{}]} | {error, Reason :: term()}.
 get_running_services() ->
     gen_server:call(?SERVER, get_running_services).
 
@@ -141,7 +138,7 @@ handle_call(get_all_services, _From, State = #state{}) ->
 handle_call(get_running_services, _From, State = #state{}) ->
     Items = dets:foldl(fun(Record, Acc) -> [Record|Acc] end, [], ?TAB),
     RunningItems = lists:filter(fun(#service{status = Status}) -> Status =:= 1 end, lists:reverse(Items)),
-    {reply, RunningItems, State};
+    {reply, {ok, RunningItems}, State};
 
 handle_call(_Request, _From, State = #state{}) ->
     {reply, ok, State}.

apps/efka/src/models/task_log_model.erl

@@ -0,0 +1,122 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 13. 8月 2025 17:01
+%%%-------------------------------------------------------------------
+-module(task_log_model).
+-author("anlicheng").
+-include("efka_tables.hrl").
+
+-behaviour(gen_server).
+
+%% API
+-export([start_link/0]).
+-export([insert/2, get_logs/1]).
+
+%% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
+
+-define(SERVER, ?MODULE).
+-define(TAB, task_log).
+
+-record(state, {
+
+}).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+-spec insert(TaskId :: integer(), Logs :: [binary()]) -> ok | {error, Reason :: term()}.
+insert(TaskId, Logs) when is_integer(TaskId), is_list(Logs) ->
+    TaskLog = #task_log{task_id = TaskId, logs = Logs},
+    gen_server:call(?SERVER, {insert, TaskLog}).
+
+-spec get_logs(TaskId :: integer()) -> Logs :: [binary()].
+get_logs(TaskId) when is_integer(TaskId) ->
+    gen_server:call(?SERVER, {get_logs, TaskId}).
+
+%% @doc Spawns the server and registers the local name (unique)
+-spec(start_link() ->
+    {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+%%%===================================================================
+%%% gen_server callbacks
+%%%===================================================================
+
+%% @private
+%% @doc Initializes the server
+-spec(init(Args :: term()) ->
+    {ok, State :: #state{}} | {ok, State :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term()} | ignore).
+init([]) ->
+    {ok, DetsDir} = application:get_env(efka, dets_dir),
+    File = DetsDir ++ "task_log.dets",
+    {ok, ?TAB} = dets:open_file(?TAB, [{file, File}, {type, set}, {keypos, 2}]),
+    {ok, #state{}}.
+
+%% @private
+%% @doc Handling call messages
+-spec(handle_call(Request :: term(), From :: {pid(), Tag :: term()},
+    State :: #state{}) ->
+    {reply, Reply :: term(), NewState :: #state{}} |
+    {reply, Reply :: term(), NewState :: #state{}, timeout() | hibernate} |
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), Reply :: term(), NewState :: #state{}} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_call({insert, TaskLog}, _From, State = #state{}) ->
+    ok = dets:insert(?TAB, TaskLog),
+    {reply, ok, State};
+handle_call({get_logs, TaskId}, _From, State = #state{}) ->
+    Reply = case dets:lookup(?TAB, TaskId) of
+                [] ->
+                    [];
+                [#task_log{logs = Logs}|_] ->
+                    Logs
+            end,
+    {reply, Reply, State}.
+
+%% @private
+%% @doc Handling cast messages
+-spec(handle_cast(Request :: term(), State :: #state{}) ->
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_cast(_Request, State = #state{}) ->
+    {noreply, State}.
+
+%% @private
+%% @doc Handling all non call/cast messages
+-spec(handle_info(Info :: timeout() | term(), State :: #state{}) ->
+    {noreply, NewState :: #state{}} |
+    {noreply, NewState :: #state{}, timeout() | hibernate} |
+    {stop, Reason :: term(), NewState :: #state{}}).
+handle_info(_Info, State = #state{}) ->
+    {noreply, State}.
+
+%% @private
+%% @doc This function is called by a gen_server when it is about to
+%% terminate. It should be the opposite of Module:init/1 and do any
+%% necessary cleaning up. When it returns, the gen_server terminates
+%% with Reason. The return value is ignored.
+-spec(terminate(Reason :: (normal | shutdown | {shutdown, term()} | term()),
+    State :: #state{}) -> term()).
+terminate(_Reason, _State = #state{}) ->
+    ok.
+
+%% @private
+%% @doc Convert process state when code is changed
+-spec(code_change(OldVsn :: term() | {down, term()}, State :: #state{},
+    Extra :: term()) ->
+    {ok, NewState :: #state{}} | {error, Reason :: term()}).
+code_change(_OldVsn, State = #state{}, _Extra) ->
+    {ok, State}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================

apps/efka/src/proto/message_pb.erl

@@ -79,9 +79,11 @@
 
 -type event() :: #event{}.
 
--export_type(['auth_request'/0, 'auth_reply'/0, 'pub'/0, 'async_call_reply'/0, 'deploy'/0, 'fetch_task_log'/0, 'invoke'/0, 'push_service_config'/0, 'data'/0, 'ping'/0, 'service_inform'/0, 'event'/0]).
+-type alarm() :: #alarm{}.
--type '$msg_name'() :: auth_request | auth_reply | pub | async_call_reply | deploy | fetch_task_log | invoke | push_service_config | data | ping | service_inform | event.
+
--type '$msg'() :: auth_request() | auth_reply() | pub() | async_call_reply() | deploy() | fetch_task_log() | invoke() | push_service_config() | data() | ping() | service_inform() | event().
+-export_type(['auth_request'/0, 'auth_reply'/0, 'pub'/0, 'async_call_reply'/0, 'deploy'/0, 'fetch_task_log'/0, 'invoke'/0, 'push_service_config'/0, 'data'/0, 'ping'/0, 'service_inform'/0, 'event'/0, 'alarm'/0]).
+-type '$msg_name'() :: auth_request | auth_reply | pub | async_call_reply | deploy | fetch_task_log | invoke | push_service_config | data | ping | service_inform | event | alarm.
+-type '$msg'() :: auth_request() | auth_reply() | pub() | async_call_reply() | deploy() | fetch_task_log() | invoke() | push_service_config() | data() | ping() | service_inform() | event() | alarm().
 -export_type(['$msg_name'/0, '$msg'/0]).
 
 -if(?OTP_RELEASE >= 24).
@@ -119,7 +121,8 @@ encode_msg(Msg, MsgName, Opts) ->
         data -> encode_msg_data(id(Msg, TrUserData), TrUserData);
         ping -> encode_msg_ping(id(Msg, TrUserData), TrUserData);
         service_inform -> encode_msg_service_inform(id(Msg, TrUserData), TrUserData);
-        event -> encode_msg_event(id(Msg, TrUserData), TrUserData)
+        event -> encode_msg_event(id(Msg, TrUserData), TrUserData);
+        alarm -> encode_msg_alarm(id(Msg, TrUserData), TrUserData)
     end.
 
 
@@ -219,9 +222,9 @@ encode_msg_pub(#pub{topic = F1, content = F2}, Bin, TrUserData) ->
        true ->
            begin
                TrF2 = id(F2, TrUserData),
-               case is_empty_string(TrF2) of
+               case iolist_size(TrF2) of
-                   true -> B1;
+                   0 -> B1;
-                   false -> e_type_string(TrF2, <<B1/binary, 18>>, TrUserData)
+                   _ -> e_type_bytes(TrF2, <<B1/binary, 18>>, TrUserData)
                end
            end
     end.
@@ -326,9 +329,9 @@ encode_msg_invoke(#invoke{service_id = F1, payload = F2, timeout = F3}, Bin, TrU
             true ->
                 begin
                     TrF2 = id(F2, TrUserData),
-                    case is_empty_string(TrF2) of
+                    case iolist_size(TrF2) of
-                        true -> B1;
+                        0 -> B1;
-                        false -> e_type_string(TrF2, <<B1/binary, 18>>, TrUserData)
+                        _ -> e_type_bytes(TrF2, <<B1/binary, 18>>, TrUserData)
                     end
                 end
          end,
@@ -379,7 +382,7 @@ encode_msg_push_service_config(#push_service_config{service_id = F1, config_json
 encode_msg_data(Msg, TrUserData) -> encode_msg_data(Msg, <<>>, TrUserData).
 
 
-encode_msg_data(#data{service_id = F1, device_uuid = F2, metric = F3}, Bin, TrUserData) ->
+encode_msg_data(#data{service_id = F1, device_uuid = F2, route_key = F3, metric = F4}, Bin, TrUserData) ->
     B1 = if F1 == undefined -> Bin;
             true ->
                 begin
@@ -400,7 +403,7 @@ encode_msg_data(#data{service_id = F1, device_uuid = F2, metric = F3}, Bin, TrUs
                     end
                 end
          end,
-    if F3 == undefined -> B2;
+    B3 = if F3 == undefined -> B2;
             true ->
                 begin
                     TrF3 = id(F3, TrUserData),
@@ -409,6 +412,16 @@ encode_msg_data(#data{service_id = F1, device_uuid = F2, metric = F3}, Bin, TrUs
                         false -> e_type_string(TrF3, <<B2/binary, 26>>, TrUserData)
                     end
                 end
+         end,
+    if F4 == undefined -> B3;
+       true ->
+           begin
+               TrF4 = id(F4, TrUserData),
+               case iolist_size(TrF4) of
+                   0 -> B3;
+                   _ -> e_type_bytes(TrF4, <<B3/binary, 34>>, TrUserData)
+               end
+           end
     end.
 
 encode_msg_ping(Msg, TrUserData) -> encode_msg_ping(Msg, <<>>, TrUserData).
@@ -607,6 +620,31 @@ encode_msg_event(#event{service_id = F1, event_type = F2, params = F3}, Bin, TrU
            end
     end.
 
+encode_msg_alarm(Msg, TrUserData) -> encode_msg_alarm(Msg, <<>>, TrUserData).
+
+
+encode_msg_alarm(#alarm{service_id = F1, params = F2}, Bin, TrUserData) ->
+    B1 = if F1 == undefined -> Bin;
+            true ->
+                begin
+                    TrF1 = id(F1, TrUserData),
+                    case is_empty_string(TrF1) of
+                        true -> Bin;
+                        false -> e_type_string(TrF1, <<Bin/binary, 10>>, TrUserData)
+                    end
+                end
+         end,
+    if F2 == undefined -> B1;
+       true ->
+           begin
+               TrF2 = id(F2, TrUserData),
+               case is_empty_string(TrF2) of
+                   true -> B1;
+                   false -> e_type_string(TrF2, <<B1/binary, 18>>, TrUserData)
+               end
+           end
+    end.
+
 e_field_ping_ips([Elem | Rest], Bin, TrUserData) ->
     Bin2 = <<Bin/binary, 58>>,
     Bin3 = e_type_string(id(Elem, TrUserData), Bin2, TrUserData),
@@ -786,7 +824,8 @@ decode_msg_2_doit(push_service_config, Bin, TrUserData) -> id(decode_msg_push_se
 decode_msg_2_doit(data, Bin, TrUserData) -> id(decode_msg_data(Bin, TrUserData), TrUserData);
 decode_msg_2_doit(ping, Bin, TrUserData) -> id(decode_msg_ping(Bin, TrUserData), TrUserData);
 decode_msg_2_doit(service_inform, Bin, TrUserData) -> id(decode_msg_service_inform(Bin, TrUserData), TrUserData);
-decode_msg_2_doit(event, Bin, TrUserData) -> id(decode_msg_event(Bin, TrUserData), TrUserData).
+decode_msg_2_doit(event, Bin, TrUserData) -> id(decode_msg_event(Bin, TrUserData), TrUserData);
+decode_msg_2_doit(alarm, Bin, TrUserData) -> id(decode_msg_alarm(Bin, TrUserData), TrUserData).
 
 
 
@@ -1240,63 +1279,70 @@ skip_32_push_service_config(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3,
 
 skip_64_push_service_config(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_push_service_config(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData).
 
-decode_msg_data(Bin, TrUserData) -> dfp_read_field_def_data(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData).
+decode_msg_data(Bin, TrUserData) -> dfp_read_field_def_data(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData).
 
-dfp_read_field_def_data(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_data_service_id(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData);
+dfp_read_field_def_data(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> d_field_data_service_id(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-dfp_read_field_def_data(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_data_device_uuid(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData);
+dfp_read_field_def_data(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> d_field_data_device_uuid(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-dfp_read_field_def_data(<<26, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> d_field_data_metric(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData);
+dfp_read_field_def_data(<<26, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> d_field_data_route_key(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-dfp_read_field_def_data(<<>>, 0, 0, _, F@_1, F@_2, F@_3, _) -> #data{service_id = F@_1, device_uuid = F@_2, metric = F@_3};
+dfp_read_field_def_data(<<34, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> d_field_data_metric(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-dfp_read_field_def_data(Other, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dg_read_field_def_data(Other, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData).
+dfp_read_field_def_data(<<>>, 0, 0, _, F@_1, F@_2, F@_3, F@_4, _) -> #data{service_id = F@_1, device_uuid = F@_2, route_key = F@_3, metric = F@_4};
+dfp_read_field_def_data(Other, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> dg_read_field_def_data(Other, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData).
 
-dg_read_field_def_data(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 32 - 7 -> dg_read_field_def_data(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData);
+dg_read_field_def_data(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData) when N < 32 - 7 -> dg_read_field_def_data(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-dg_read_field_def_data(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, F@_3, TrUserData) ->
+dg_read_field_def_data(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, F@_3, F@_4, TrUserData) ->
     Key = X bsl N + Acc,
     case Key of
-        10 -> d_field_data_service_id(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData);
+        10 -> d_field_data_service_id(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, TrUserData);
-        18 -> d_field_data_device_uuid(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData);
+        18 -> d_field_data_device_uuid(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, TrUserData);
-        26 -> d_field_data_metric(Rest, 0, 0, 0, F@_1, F@_2, F@_3, TrUserData);
+        26 -> d_field_data_route_key(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, TrUserData);
+        34 -> d_field_data_metric(Rest, 0, 0, 0, F@_1, F@_2, F@_3, F@_4, TrUserData);
         _ ->
             case Key band 7 of
-                0 -> skip_varint_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData);
+                0 -> skip_varint_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, TrUserData);
-                1 -> skip_64_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData);
+                1 -> skip_64_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, TrUserData);
-                2 -> skip_length_delimited_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData);
+                2 -> skip_length_delimited_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, TrUserData);
-                3 -> skip_group_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData);
+                3 -> skip_group_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, TrUserData);
-                5 -> skip_32_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, TrUserData)
+                5 -> skip_32_data(Rest, 0, 0, Key bsr 3, F@_1, F@_2, F@_3, F@_4, TrUserData)
             end
     end;
-dg_read_field_def_data(<<>>, 0, 0, _, F@_1, F@_2, F@_3, _) -> #data{service_id = F@_1, device_uuid = F@_2, metric = F@_3}.
+dg_read_field_def_data(<<>>, 0, 0, _, F@_1, F@_2, F@_3, F@_4, _) -> #data{service_id = F@_1, device_uuid = F@_2, route_key = F@_3, metric = F@_4}.
 
-d_field_data_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_data_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData);
+d_field_data_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData) when N < 57 -> d_field_data_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-d_field_data_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, F@_3, TrUserData) ->
+d_field_data_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, F@_3, F@_4, TrUserData) ->
     {NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
-    dfp_read_field_def_data(RestF, 0, 0, F, NewFValue, F@_2, F@_3, TrUserData).
+    dfp_read_field_def_data(RestF, 0, 0, F, NewFValue, F@_2, F@_3, F@_4, TrUserData).
 
-d_field_data_device_uuid(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_data_device_uuid(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData);
+d_field_data_device_uuid(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData) when N < 57 -> d_field_data_device_uuid(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-d_field_data_device_uuid(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, F@_3, TrUserData) ->
+d_field_data_device_uuid(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, F@_3, F@_4, TrUserData) ->
     {NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
-    dfp_read_field_def_data(RestF, 0, 0, F, F@_1, NewFValue, F@_3, TrUserData).
+    dfp_read_field_def_data(RestF, 0, 0, F, F@_1, NewFValue, F@_3, F@_4, TrUserData).
 
-d_field_data_metric(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> d_field_data_metric(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData);
+d_field_data_route_key(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData) when N < 57 -> d_field_data_route_key(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-d_field_data_metric(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, _, TrUserData) ->
+d_field_data_route_key(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, _, F@_4, TrUserData) ->
     {NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
-    dfp_read_field_def_data(RestF, 0, 0, F, F@_1, F@_2, NewFValue, TrUserData).
+    dfp_read_field_def_data(RestF, 0, 0, F, F@_1, F@_2, NewFValue, F@_4, TrUserData).
 
-skip_varint_data(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> skip_varint_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData);
+d_field_data_metric(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData) when N < 57 -> d_field_data_metric(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-skip_varint_data(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData).
+d_field_data_metric(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, _, TrUserData) ->
+    {NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
+    dfp_read_field_def_data(RestF, 0, 0, F, F@_1, F@_2, F@_3, NewFValue, TrUserData).
 
-skip_length_delimited_data(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) when N < 57 -> skip_length_delimited_data(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, TrUserData);
+skip_varint_data(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> skip_varint_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
-skip_length_delimited_data(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, TrUserData) ->
+skip_varint_data(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData).
+
+skip_length_delimited_data(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData) when N < 57 -> skip_length_delimited_data(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData);
+skip_length_delimited_data(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, F@_3, F@_4, TrUserData) ->
     Length = X bsl N + Acc,
     <<_:Length/binary, Rest2/binary>> = Rest,
-    dfp_read_field_def_data(Rest2, 0, 0, F, F@_1, F@_2, F@_3, TrUserData).
+    dfp_read_field_def_data(Rest2, 0, 0, F, F@_1, F@_2, F@_3, F@_4, TrUserData).
 
-skip_group_data(Bin, _, Z2, FNum, F@_1, F@_2, F@_3, TrUserData) ->
+skip_group_data(Bin, _, Z2, FNum, F@_1, F@_2, F@_3, F@_4, TrUserData) ->
     {_, Rest} = read_group(Bin, FNum),
-    dfp_read_field_def_data(Rest, 0, Z2, FNum, F@_1, F@_2, F@_3, TrUserData).
+    dfp_read_field_def_data(Rest, 0, Z2, FNum, F@_1, F@_2, F@_3, F@_4, TrUserData).
 
-skip_32_data(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData).
+skip_32_data(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData).
 
-skip_64_data(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData).
+skip_64_data(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData) -> dfp_read_field_def_data(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, F@_4, TrUserData).
 
 decode_msg_ping(Bin, TrUserData) ->
     dfp_read_field_def_ping(Bin,
@@ -1641,6 +1687,57 @@ skip_32_event(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) ->
 
 skip_64_event(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData) -> dfp_read_field_def_event(Rest, Z1, Z2, F, F@_1, F@_2, F@_3, TrUserData).
 
+decode_msg_alarm(Bin, TrUserData) -> dfp_read_field_def_alarm(Bin, 0, 0, 0, id(<<>>, TrUserData), id(<<>>, TrUserData), TrUserData).
+
+dfp_read_field_def_alarm(<<10, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_alarm_service_id(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+dfp_read_field_def_alarm(<<18, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> d_field_alarm_params(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+dfp_read_field_def_alarm(<<>>, 0, 0, _, F@_1, F@_2, _) -> #alarm{service_id = F@_1, params = F@_2};
+dfp_read_field_def_alarm(Other, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dg_read_field_def_alarm(Other, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+dg_read_field_def_alarm(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 32 - 7 -> dg_read_field_def_alarm(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+dg_read_field_def_alarm(<<0:1, X:7, Rest/binary>>, N, Acc, _, F@_1, F@_2, TrUserData) ->
+    Key = X bsl N + Acc,
+    case Key of
+        10 -> d_field_alarm_service_id(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
+        18 -> d_field_alarm_params(Rest, 0, 0, 0, F@_1, F@_2, TrUserData);
+        _ ->
+            case Key band 7 of
+                0 -> skip_varint_alarm(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                1 -> skip_64_alarm(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                2 -> skip_length_delimited_alarm(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                3 -> skip_group_alarm(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData);
+                5 -> skip_32_alarm(Rest, 0, 0, Key bsr 3, F@_1, F@_2, TrUserData)
+            end
+    end;
+dg_read_field_def_alarm(<<>>, 0, 0, _, F@_1, F@_2, _) -> #alarm{service_id = F@_1, params = F@_2}.
+
+d_field_alarm_service_id(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_alarm_service_id(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+d_field_alarm_service_id(<<0:1, X:7, Rest/binary>>, N, Acc, F, _, F@_2, TrUserData) ->
+    {NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
+    dfp_read_field_def_alarm(RestF, 0, 0, F, NewFValue, F@_2, TrUserData).
+
+d_field_alarm_params(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> d_field_alarm_params(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+d_field_alarm_params(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, _, TrUserData) ->
+    {NewFValue, RestF} = begin Len = X bsl N + Acc, <<Bytes:Len/binary, Rest2/binary>> = Rest, Bytes2 = binary:copy(Bytes), {id(Bytes2, TrUserData), Rest2} end,
+    dfp_read_field_def_alarm(RestF, 0, 0, F, F@_1, NewFValue, TrUserData).
+
+skip_varint_alarm(<<1:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> skip_varint_alarm(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData);
+skip_varint_alarm(<<0:1, _:7, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_alarm(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+skip_length_delimited_alarm(<<1:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) when N < 57 -> skip_length_delimited_alarm(Rest, N + 7, X bsl N + Acc, F, F@_1, F@_2, TrUserData);
+skip_length_delimited_alarm(<<0:1, X:7, Rest/binary>>, N, Acc, F, F@_1, F@_2, TrUserData) ->
+    Length = X bsl N + Acc,
+    <<_:Length/binary, Rest2/binary>> = Rest,
+    dfp_read_field_def_alarm(Rest2, 0, 0, F, F@_1, F@_2, TrUserData).
+
+skip_group_alarm(Bin, _, Z2, FNum, F@_1, F@_2, TrUserData) ->
+    {_, Rest} = read_group(Bin, FNum),
+    dfp_read_field_def_alarm(Rest, 0, Z2, FNum, F@_1, F@_2, TrUserData).
+
+skip_32_alarm(<<_:32, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_alarm(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
+skip_64_alarm(<<_:64, Rest/binary>>, Z1, Z2, F, F@_1, F@_2, TrUserData) -> dfp_read_field_def_alarm(Rest, Z1, Z2, F, F@_1, F@_2, TrUserData).
+
 read_group(Bin, FieldNum) ->
     {NumBytes, EndTagLen} = read_gr_b(Bin, 0, 0, 0, 0, FieldNum),
     <<Group:NumBytes/binary, _:EndTagLen/binary, Rest/binary>> = Bin,
@@ -1718,7 +1815,8 @@ merge_msgs(Prev, New, MsgName, Opts) ->
         data -> merge_msg_data(Prev, New, TrUserData);
         ping -> merge_msg_ping(Prev, New, TrUserData);
         service_inform -> merge_msg_service_inform(Prev, New, TrUserData);
-        event -> merge_msg_event(Prev, New, TrUserData)
+        event -> merge_msg_event(Prev, New, TrUserData);
+        alarm -> merge_msg_alarm(Prev, New, TrUserData)
     end.
 
 -compile({nowarn_unused_function,merge_msg_auth_request/3}).
@@ -1834,7 +1932,7 @@ merge_msg_push_service_config(#push_service_config{service_id = PFservice_id, co
                              end}.
 
 -compile({nowarn_unused_function,merge_msg_data/3}).
-merge_msg_data(#data{service_id = PFservice_id, device_uuid = PFdevice_uuid, metric = PFmetric}, #data{service_id = NFservice_id, device_uuid = NFdevice_uuid, metric = NFmetric}, _) ->
+merge_msg_data(#data{service_id = PFservice_id, device_uuid = PFdevice_uuid, route_key = PFroute_key, metric = PFmetric}, #data{service_id = NFservice_id, device_uuid = NFdevice_uuid, route_key = NFroute_key, metric = NFmetric}, _) ->
     #data{service_id =
               if NFservice_id =:= undefined -> PFservice_id;
                  true -> NFservice_id
@@ -1843,6 +1941,10 @@ merge_msg_data(#data{service_id = PFservice_id, device_uuid = PFdevice_uuid, met
               if NFdevice_uuid =:= undefined -> PFdevice_uuid;
                  true -> NFdevice_uuid
               end,
+          route_key =
+              if NFroute_key =:= undefined -> PFroute_key;
+                 true -> NFroute_key
+              end,
           metric =
               if NFmetric =:= undefined -> PFmetric;
                  true -> NFmetric
@@ -1944,6 +2046,17 @@ merge_msg_event(#event{service_id = PFservice_id, event_type = PFevent_type, par
                   true -> NFparams
                end}.
 
+-compile({nowarn_unused_function,merge_msg_alarm/3}).
+merge_msg_alarm(#alarm{service_id = PFservice_id, params = PFparams}, #alarm{service_id = NFservice_id, params = NFparams}, _) ->
+    #alarm{service_id =
+               if NFservice_id =:= undefined -> PFservice_id;
+                  true -> NFservice_id
+               end,
+           params =
+               if NFparams =:= undefined -> PFparams;
+                  true -> NFparams
+               end}.
+
 
 verify_msg(Msg) when tuple_size(Msg) >= 1 -> verify_msg(Msg, element(1, Msg), []);
 verify_msg(X) -> mk_type_error(not_a_known_message, X, []).
@@ -1967,6 +2080,7 @@ verify_msg(Msg, MsgName, Opts) ->
         ping -> v_msg_ping(Msg, [MsgName], TrUserData);
         service_inform -> v_msg_service_inform(Msg, [MsgName], TrUserData);
         event -> v_msg_event(Msg, [MsgName], TrUserData);
+        alarm -> v_msg_alarm(Msg, [MsgName], TrUserData);
         _ -> mk_type_error(not_a_known_message, Msg, [])
     end.
 
@@ -2011,7 +2125,7 @@ v_msg_pub(#pub{topic = F1, content = F2}, Path, TrUserData) ->
        true -> v_type_string(F1, [topic | Path], TrUserData)
     end,
     if F2 == undefined -> ok;
-       true -> v_type_string(F2, [content | Path], TrUserData)
+       true -> v_type_bytes(F2, [content | Path], TrUserData)
     end,
     ok;
 v_msg_pub(X, Path, _TrUserData) -> mk_type_error({expected_msg, pub}, X, Path).
@@ -2062,7 +2176,7 @@ v_msg_invoke(#invoke{service_id = F1, payload = F2, timeout = F3}, Path, TrUserD
        true -> v_type_string(F1, [service_id | Path], TrUserData)
     end,
     if F2 == undefined -> ok;
-       true -> v_type_string(F2, [payload | Path], TrUserData)
+       true -> v_type_bytes(F2, [payload | Path], TrUserData)
     end,
     if F3 == undefined -> ok;
        true -> v_type_uint32(F3, [timeout | Path], TrUserData)
@@ -2087,7 +2201,7 @@ v_msg_push_service_config(X, Path, _TrUserData) -> mk_type_error({expected_msg,
 
 -compile({nowarn_unused_function,v_msg_data/3}).
 -dialyzer({nowarn_function,v_msg_data/3}).
-v_msg_data(#data{service_id = F1, device_uuid = F2, metric = F3}, Path, TrUserData) ->
+v_msg_data(#data{service_id = F1, device_uuid = F2, route_key = F3, metric = F4}, Path, TrUserData) ->
     if F1 == undefined -> ok;
        true -> v_type_string(F1, [service_id | Path], TrUserData)
     end,
@@ -2095,7 +2209,10 @@ v_msg_data(#data{service_id = F1, device_uuid = F2, metric = F3}, Path, TrUserDa
        true -> v_type_string(F2, [device_uuid | Path], TrUserData)
     end,
     if F3 == undefined -> ok;
-       true -> v_type_string(F3, [metric | Path], TrUserData)
+       true -> v_type_string(F3, [route_key | Path], TrUserData)
+    end,
+    if F4 == undefined -> ok;
+       true -> v_type_bytes(F4, [metric | Path], TrUserData)
     end,
     ok;
 v_msg_data(X, Path, _TrUserData) -> mk_type_error({expected_msg, data}, X, Path).
@@ -2184,6 +2301,18 @@ v_msg_event(#event{service_id = F1, event_type = F2, params = F3}, Path, TrUserD
     ok;
 v_msg_event(X, Path, _TrUserData) -> mk_type_error({expected_msg, event}, X, Path).
 
+-compile({nowarn_unused_function,v_msg_alarm/3}).
+-dialyzer({nowarn_function,v_msg_alarm/3}).
+v_msg_alarm(#alarm{service_id = F1, params = F2}, Path, TrUserData) ->
+    if F1 == undefined -> ok;
+       true -> v_type_string(F1, [service_id | Path], TrUserData)
+    end,
+    if F2 == undefined -> ok;
+       true -> v_type_string(F2, [params | Path], TrUserData)
+    end,
+    ok;
+v_msg_alarm(X, Path, _TrUserData) -> mk_type_error({expected_msg, alarm}, X, Path).
+
 -compile({nowarn_unused_function,v_type_int32/3}).
 -dialyzer({nowarn_function,v_type_int32/3}).
 v_type_int32(N, _Path, _TrUserData) when is_integer(N), -2147483648 =< N, N =< 2147483647 -> ok;
@@ -2216,6 +2345,12 @@ v_type_string(S, Path, _TrUserData) when is_list(S); is_binary(S) ->
     end;
 v_type_string(X, Path, _TrUserData) -> mk_type_error(bad_unicode_string, X, Path).
 
+-compile({nowarn_unused_function,v_type_bytes/3}).
+-dialyzer({nowarn_function,v_type_bytes/3}).
+v_type_bytes(B, _Path, _TrUserData) when is_binary(B) -> ok;
+v_type_bytes(B, _Path, _TrUserData) when is_list(B) -> ok;
+v_type_bytes(X, Path, _TrUserData) -> mk_type_error(bad_binary_value, X, Path).
+
 -compile({nowarn_unused_function,mk_type_error/3}).
 -spec mk_type_error(_, _, list()) -> no_return().
 mk_type_error(Error, ValueSeen, Path) ->
@@ -2261,7 +2396,7 @@ get_msg_defs() ->
        #field{name = token, fnum = 5, rnum = 5, type = string, occurrence = optional, opts = []},
        #field{name = timestamp, fnum = 6, rnum = 6, type = uint32, occurrence = optional, opts = []}]},
      {{msg, auth_reply}, [#field{name = code, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}, #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
-     {{msg, pub}, [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]},
+     {{msg, pub}, [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}]},
      {{msg, async_call_reply},
       [#field{name = code, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
        #field{name = result, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
@@ -2273,7 +2408,7 @@ get_msg_defs() ->
      {{msg, fetch_task_log}, [#field{name = task_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}]},
      {{msg, invoke},
       [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
-       #field{name = payload, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
+       #field{name = payload, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []},
        #field{name = timeout, fnum = 3, rnum = 4, type = uint32, occurrence = optional, opts = []}]},
      {{msg, push_service_config},
       [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
@@ -2282,7 +2417,8 @@ get_msg_defs() ->
      {{msg, data},
       [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
        #field{name = device_uuid, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
-       #field{name = metric, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}]},
+       #field{name = route_key, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []},
+       #field{name = metric, fnum = 4, rnum = 5, type = bytes, occurrence = optional, opts = []}]},
      {{msg, ping},
       [#field{name = adcode, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
        #field{name = boot_time, fnum = 2, rnum = 3, type = uint32, occurrence = optional, opts = []},
@@ -2305,16 +2441,17 @@ get_msg_defs() ->
      {{msg, event},
       [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
        #field{name = event_type, fnum = 2, rnum = 3, type = uint32, occurrence = optional, opts = []},
-       #field{name = params, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}]}].
+       #field{name = params, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}]},
+     {{msg, alarm}, [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = params, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}]}].
 
 
-get_msg_names() -> [auth_request, auth_reply, pub, async_call_reply, deploy, fetch_task_log, invoke, push_service_config, data, ping, service_inform, event].
+get_msg_names() -> [auth_request, auth_reply, pub, async_call_reply, deploy, fetch_task_log, invoke, push_service_config, data, ping, service_inform, event, alarm].
 
 
 get_group_names() -> [].
 
 
-get_msg_or_group_names() -> [auth_request, auth_reply, pub, async_call_reply, deploy, fetch_task_log, invoke, push_service_config, data, ping, service_inform, event].
+get_msg_or_group_names() -> [auth_request, auth_reply, pub, async_call_reply, deploy, fetch_task_log, invoke, push_service_config, data, ping, service_inform, event, alarm].
 
 
 get_enum_names() -> [].
@@ -2338,7 +2475,7 @@ find_msg_def(auth_request) ->
      #field{name = token, fnum = 5, rnum = 5, type = string, occurrence = optional, opts = []},
      #field{name = timestamp, fnum = 6, rnum = 6, type = uint32, occurrence = optional, opts = []}];
 find_msg_def(auth_reply) -> [#field{name = code, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}, #field{name = message, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
-find_msg_def(pub) -> [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
+find_msg_def(pub) -> [#field{name = topic, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = content, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []}];
 find_msg_def(async_call_reply) ->
     [#field{name = code, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []},
      #field{name = result, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
@@ -2350,7 +2487,7 @@ find_msg_def(deploy) ->
 find_msg_def(fetch_task_log) -> [#field{name = task_id, fnum = 1, rnum = 2, type = uint32, occurrence = optional, opts = []}];
 find_msg_def(invoke) ->
     [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
-     #field{name = payload, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
+     #field{name = payload, fnum = 2, rnum = 3, type = bytes, occurrence = optional, opts = []},
      #field{name = timeout, fnum = 3, rnum = 4, type = uint32, occurrence = optional, opts = []}];
 find_msg_def(push_service_config) ->
     [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
@@ -2359,7 +2496,8 @@ find_msg_def(push_service_config) ->
 find_msg_def(data) ->
     [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
      #field{name = device_uuid, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []},
-     #field{name = metric, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}];
+     #field{name = route_key, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []},
+     #field{name = metric, fnum = 4, rnum = 5, type = bytes, occurrence = optional, opts = []}];
 find_msg_def(ping) ->
     [#field{name = adcode, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
      #field{name = boot_time, fnum = 2, rnum = 3, type = uint32, occurrence = optional, opts = []},
@@ -2383,6 +2521,7 @@ find_msg_def(event) ->
     [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []},
      #field{name = event_type, fnum = 2, rnum = 3, type = uint32, occurrence = optional, opts = []},
      #field{name = params, fnum = 3, rnum = 4, type = string, occurrence = optional, opts = []}];
+find_msg_def(alarm) -> [#field{name = service_id, fnum = 1, rnum = 2, type = string, occurrence = optional, opts = []}, #field{name = params, fnum = 2, rnum = 3, type = string, occurrence = optional, opts = []}];
 find_msg_def(_) -> error.
 
 
@@ -2453,6 +2592,7 @@ fqbin_to_msg_name(<<"Data">>) -> data;
 fqbin_to_msg_name(<<"Ping">>) -> ping;
 fqbin_to_msg_name(<<"ServiceInform">>) -> service_inform;
 fqbin_to_msg_name(<<"Event">>) -> event;
+fqbin_to_msg_name(<<"Alarm">>) -> alarm;
 fqbin_to_msg_name(E) -> error({gpb_error, {badmsg, E}}).
 
 
@@ -2468,6 +2608,7 @@ msg_name_to_fqbin(data) -> <<"Data">>;
 msg_name_to_fqbin(ping) -> <<"Ping">>;
 msg_name_to_fqbin(service_inform) -> <<"ServiceInform">>;
 msg_name_to_fqbin(event) -> <<"Event">>;
+msg_name_to_fqbin(alarm) -> <<"Alarm">>;
 msg_name_to_fqbin(E) -> error({gpb_error, {badmsg, E}}).
 
 
@@ -2506,7 +2647,7 @@ get_all_source_basenames() -> ["message_pb.proto"].
 get_all_proto_names() -> ["message_pb"].
 
 
-get_msg_containment("message_pb") -> [async_call_reply, auth_reply, auth_request, data, deploy, event, fetch_task_log, invoke, ping, pub, push_service_config, service_inform];
+get_msg_containment("message_pb") -> [alarm, async_call_reply, auth_reply, auth_request, data, deploy, event, fetch_task_log, invoke, ping, pub, push_service_config, service_inform];
 get_msg_containment(P) -> error({gpb_error, {badproto, P}}).
 
 
@@ -2538,6 +2679,7 @@ get_proto_by_msg_name_as_fqbin(<<"Deploy">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(<<"AuthReply">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(<<"AsyncCallReply">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(<<"ServiceInform">>) -> "message_pb";
+get_proto_by_msg_name_as_fqbin(<<"Alarm">>) -> "message_pb";
 get_proto_by_msg_name_as_fqbin(E) -> error({gpb_error, {badmsg, E}}).
 
 

apps/efka/src/tcp_server/tcp_channel.erl

@@ -6,14 +6,14 @@
 %%% @end
 %%% Created : 30. 4月 2025 09:22
 %%%-------------------------------------------------------------------
--module(efka_tcp_channel).
+-module(tcp_channel).
 -author("anlicheng").
+-include("efka_service.hrl").
 
 -behaviour(gen_server).
 
 %% API
 -export([start_link/1]).
--export([push_config/4, invoke/4]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]).
@@ -24,15 +24,6 @@
 -define(PENDING_TIMEOUT, 10 * 1000).
 %% 消息类型
 
-%% 服务注册
--define(PACKET_REQUEST, 16#01).
-%% 消息响应
--define(PACKET_RESPONSE, 16#02).
-%% 上传数据
--define(PACKET_PUSH, 16#03).
-
--define(PACKET_PUB, 16#04).
-
 -record(state, {
     packet_id = 1,
     socket :: gen_tcp:socket(),
@@ -48,14 +39,6 @@
 %%% API
 %%%===================================================================
 
--spec push_config(ChannelPid :: pid(), Ref :: reference(), ReceiverPid :: pid(), ConfigJson :: binary()) -> no_return().
-push_config(ChannelPid, Ref, ReceiverPid, ConfigJson) when is_pid(ChannelPid), is_pid(ReceiverPid), is_binary(ConfigJson), is_reference(Ref) ->
-    gen_server:cast(ChannelPid, {push_config, Ref, ReceiverPid, ConfigJson}).
-
--spec invoke(ChannelPid :: pid(), Ref :: reference(), ReceiverPid :: pid(), Payload :: binary()) -> no_return().
-invoke(ChannelPid, Ref, ReceiverPid, Payload) when is_pid(ChannelPid), is_pid(ReceiverPid), is_binary(Payload), is_reference(Ref) ->
-    gen_server:cast(ChannelPid, {invoke, Ref, ReceiverPid, Payload}).
-
 %% @doc Spawns the server and registers the local name (unique)
 -spec(start_link(Socket :: gen_tcp:socket()) ->
     {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
@@ -95,24 +78,6 @@ handle_call(_Request, _From, State = #state{}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
-%% 推送配置项目
-handle_cast({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
-    PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"push_config">>, <<"params">> => #{<<"config">> => ConfigJson}},
-    Packet = jiffy:encode(PushConfig, [force_utf8]),
-    ok = gen_tcp:send(Socket, <<?PACKET_PUSH:8, Packet/binary>>),
-
-    erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
-    {noreply, State#state{packet_id = next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
-
-%% 远程调用
-handle_cast({invoke, Ref, ReceiverPid, Payload}, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
-    PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"invoke">>, <<"params">> => #{<<"payload">> => Payload}},
-    Packet = jiffy:encode(PushConfig, [force_utf8]),
-    ok = gen_tcp:send(Socket, <<?PACKET_PUSH:8, Packet/binary>>),
-
-    erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
-    {noreply, State#state{packet_id = next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
-
 handle_cast(_Request, State = #state{}) ->
     {noreply, State}.
 
@@ -122,64 +87,47 @@ handle_cast(_Request, State = #state{}) ->
     {noreply, NewState :: #state{}} |
     {noreply, NewState :: #state{}, timeout() | hibernate} |
     {stop, Reason :: term(), NewState :: #state{}}).
+
+%% 推送配置项目
+handle_info({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
+    PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"push_config">>, <<"params">> => #{<<"config">> => ConfigJson}},
+    Packet = jiffy:encode(PushConfig, [force_utf8]),
+    ok = gen_tcp:send(Socket, <<?PACKET_PUSH:8, Packet/binary>>),
+
+    erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
+    {noreply, State#state{packet_id = gen_channel:next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
+
+%% 远程调用
+handle_info({invoke, Ref, ReceiverPid, Payload}, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
+    PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"invoke">>, <<"params">> => #{<<"payload">> => Payload}},
+    Packet = jiffy:encode(PushConfig, [force_utf8]),
+    ok = gen_tcp:send(Socket, <<?PACKET_PUSH:8, Packet/binary>>),
+
+    erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
+    {noreply, State#state{packet_id = gen_channel:next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
+
 %% 处理micro-client:request => efka 主动的请求
 handle_info({tcp, Socket, <<?PACKET_REQUEST:8, Data/binary>>}, State = #state{socket = Socket}) ->
     Request = jiffy:decode(Data, [return_maps]),
-    case handle_request(Request, State) of
+    handle_request(Request, State);
-        {ok, NewState} ->
-            {noreply, NewState};
-        {stop, Reason, NewState} ->
-            {stop, Reason, NewState}
-    end;
 
 %% 处理micro-client:response => efka 的响应
 handle_info({tcp, Socket, <<?PACKET_RESPONSE:8, Data/binary>>}, State = #state{socket = Socket, inflight = Inflight}) ->
     Resp = jiffy:decode(Data, [return_maps]),
-    case Resp of
+    {PacketId, Reply} = case Resp of
                             #{<<"id">> := Id, <<"result">> := Result} ->
-            case maps:take(Id, Inflight) of
+                                {Id, {ok, Result}};
-                error ->
-                    lager:warning("[tcp_channel] get unknown publish response message: ~p, packet_id: ~p", [Resp, Id]),
-                    {noreply, State};
-                {{ReceiverPid, Ref}, NInflight} ->
-                    case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
-                        true ->
-                            ReceiverPid ! {channel_reply, Ref, {ok, Result}};
-                        false ->
-                            lager:warning("[tcp_channel] get publish response message: ~p, packet_id: ~p, but receiver_pid is deaded", [Resp, Id])
-                    end,
-                    {noreply, State#state{inflight = NInflight}}
-            end;
                             #{<<"id">> := Id, <<"error">> := #{<<"code">> := _Code, <<"message">> := Error}} ->
-            case maps:take(Id, Inflight) of
+                                {Id, {error, Error}}
-                error ->
-                    lager:warning("[tcp_channel] get unknown publish response message: ~p, packet_id: ~p", [Resp, Id]),
-                    {noreply, State};
-                {{ReceiverPid, Ref}, NInflight} ->
-                    case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
-                        true ->
-                            ReceiverPid ! {channel_reply, Ref, {error, Error}};
-                        false ->
-                            lager:warning("[tcp_channel] get publish response message: ~p, packet_id: ~p, but receiver_pid is deaded", [Resp, Id])
                         end,
-                    {noreply, State#state{inflight = NInflight}}
+
-            end
+    NInflight = gen_channel:channel_reply(PacketId, Reply, Inflight),
-    end;
+    {noreply, State#state{inflight = NInflight}};
 
 %% 超时逻辑处理
 handle_info({timeout, _, {pending_timeout, Id}}, State = #state{inflight = Inflight}) ->
-    case maps:take(Id, Inflight) of
+    NInflight = gen_channel:channel_reply(Id, {error, <<"timeout">>}, Inflight),
-        error ->
+    {noreply, State#state{inflight = NInflight}};
-            {noreply, State};
-        {{ReceiverPid, Ref}, NInflight} ->
-            case is_pid(ReceiverPid) andalso is_process_alive(ReceiverPid) of
-                true ->
-                    ReceiverPid ! {channel_reply, Ref, {error, <<"timeout">>}};
-                false ->
-                    ok
-            end,
-            {noreply, State#state{inflight = NInflight}}
-    end;
 
 %% 订阅的消息
 handle_info({topic_broadcast, Topic, Content}, State = #state{socket = Socket}) ->
@@ -224,72 +172,30 @@ code_change(_OldVsn, State = #state{}, _Extra) ->
 
 %% 注册
 handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"service_id">> := ServiceId}}, State = #state{socket = Socket}) ->
-    case efka_service:get_pid(ServiceId) of
+    case gen_channel:register(Id, ServiceId) of
-        undefined ->
+        {error, Reply} ->
-            lager:warning("[efka_tcp_channel] service_id: ~p, not running", [ServiceId]),
+            ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Reply/binary>>),
-            Packet = json_error(Id, -1, <<"service not running">>),
-            ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Packet/binary>>),
             {stop, normal, State};
-        ServicePid when is_pid(ServicePid) ->
+        {ok, Reply, ServicePid} ->
-            case efka_service:attach_channel(ServicePid, self()) of
+            ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Reply/binary>>),
-                ok ->
+            {noreply, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}}
-                    Packet = json_result(Id, <<"ok">>),
-                    erlang:monitor(process, ServicePid),
-
-                    ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Packet/binary>>),
-                    {ok, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}};
-                {error, Error} ->
-                    lager:warning("[efka_tcp_channel] service_id: ~p, attach_channel get error: ~p", [ServiceId, Error]),
-                    Packet = json_error(Id, -1, Error),
-                    ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Packet/binary>>),
-                    {stop, normal, State}
-            end
     end;
-
 %% 请求参数
 handle_request(#{<<"id">> := Id, <<"method">> := <<"request_config">>}, State = #state{socket = Socket, service_pid = ServicePid, is_registered = true}) ->
     {ok, ConfigJson} = efka_service:request_config(ServicePid),
-    Packet = json_result(Id, ConfigJson),
+    Packet = gen_channel:json_result(Id, ConfigJson),
     ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Packet/binary>>),
-    {ok, State};
+    {noreply, State};
-
 %% 数据项
-handle_request(#{<<"id">> := 0, <<"method">> := <<"metric_data">>, <<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"metric">> := Metric}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
+handle_request(#{<<"id">> := 0, <<"method">> := <<"metric_data">>,
-    efka_service:metric_data(ServicePid, DeviceUUID, Metric),
+    <<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
-    {ok, State};
+    efka_service:metric_data(ServicePid, DeviceUUID, RouteKey, Metric),
-
+    {noreply, State};
 %% Event事件
 handle_request(#{<<"id">> := 0, <<"method">> := <<"event">>, <<"params">> := #{<<"event_type">> := EventType, <<"body">> := Body}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
     efka_service:send_event(ServicePid, EventType, Body),
-    {ok, State};
+    {noreply, State};
-
 %% 订阅事件
 handle_request(#{<<"id">> := 0, <<"method">> := <<"subscribe">>, <<"params">> := #{<<"topic">> := Topic}}, State = #state{is_registered = true}) ->
     efka_subscription:subscribe(Topic, self()),
-    {ok, State}.
+    {noreply, State}.
-
-%% 采用32位编码
--spec next_packet_id(PacketId :: integer()) -> NextPacketId :: integer().
-next_packet_id(PacketId) when PacketId >= 4294967295 ->
-    1;
-next_packet_id(PacketId) ->
-    PacketId + 1.
-
--spec json_result(Id :: integer(), Result :: term()) -> binary().
-json_result(Id, Result) when is_integer(Id) ->
-    Response = #{
-        <<"id">> => Id,
-        <<"result">> => Result
-    },
-    jiffy:encode(Response, [force_utf8]).
-
--spec json_error(Id :: integer(), Code :: integer(), Message :: binary()) -> binary().
-json_error(Id, Code, Message) when is_integer(Id), is_integer(Code), is_binary(Message) ->
-    Response = #{
-        <<"id">> => Id,
-        <<"error">> => #{
-            <<"code">> => Code,
-            <<"message">> => Message
-        }
-    },
-    jiffy:encode(Response, [force_utf8]).

apps/efka/src/tcp_server/tcp_connection_sup.erl

@@ -3,7 +3,7 @@
 %% @end
 %%%-------------------------------------------------------------------
 
--module(efka_tcp_sup).
+-module(tcp_connection_sup).
 
 -behaviour(supervisor).
 
@@ -28,8 +28,8 @@ start_link() ->
 init([]) ->
     SupFlags = #{strategy => simple_one_for_one, intensity => 0, period => 1},
     ChildSpec = #{
-        id => efka_tcp_channel,
+        id => tcp_channel,
-        start => {efka_tcp_channel, start_link, []},
+        start => {tcp_channel, start_link, []},
         restart => temporary,
         type => worker
     },

apps/efka/src/tcp_server/tcp_server.erl

@@ -6,7 +6,7 @@
 %%% @end
 %%% Created : 29. 4月 2025 23:24
 %%%-------------------------------------------------------------------
--module(efka_tcp_server).
+-module(tcp_server).
 -author("anlicheng").
 
 %% API
@@ -33,7 +33,7 @@ main_loop(ListenSocket) ->
     case gen_tcp:accept(ListenSocket) of
         {ok, Socket} ->
             % 为每个新连接生成一个处理进程
-            {ok, ChannelPid} = efka_tcp_sup:start_child(Socket),
+            {ok, ChannelPid} = tcp_connection_sup:start_child(Socket),
             ok = gen_tcp:controlling_process(Socket, ChannelPid),
             % 继续监听下一个连接
             main_loop(ListenSocket);

apps/efka/src/tcp_server/tcp_sup.erl

@@ -0,0 +1,72 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 26. 8月 2025 14:36
+%%%-------------------------------------------------------------------
+-module(tcp_sup).
+-author("anlicheng").
+
+-behaviour(supervisor).
+
+%% API
+-export([start_link/0]).
+
+%% Supervisor callbacks
+-export([init/1]).
+
+-define(SERVER, ?MODULE).
+
+%%%===================================================================
+%%% API functions
+%%%===================================================================
+
+%% @doc Starts the supervisor
+-spec(start_link() -> {ok, Pid :: pid()} | ignore | {error, Reason :: term()}).
+start_link() ->
+    supervisor:start_link({local, ?SERVER}, ?MODULE, []).
+
+%%%===================================================================
+%%% Supervisor callbacks
+%%%===================================================================
+
+%% @private
+%% @doc Whenever a supervisor is started using supervisor:start_link/[2,3],
+%% this function is called by the new process to find out about
+%% restart strategy, maximum restart frequency and child
+%% specifications.
+-spec(init(Args :: term()) ->
+    {ok, {SupFlags :: {RestartStrategy :: supervisor:strategy(),
+        MaxR :: non_neg_integer(), MaxT :: non_neg_integer()},
+        [ChildSpec :: supervisor:child_spec()]}}
+    | ignore | {error, Reason :: term()}).
+init([]) ->
+    SupFlags = #{strategy => one_for_one, intensity => 1000, period => 3600},
+
+    Specs = [
+        #{
+            id => 'tcp_connection_sup',
+            start => {'tcp_connection_sup', start_link, []},
+            restart => permanent,
+            shutdown => 2000,
+            type => supervisor,
+            modules => ['tcp_connection_sup']
+        },
+
+        #{
+            id => 'tcp_server',
+            start => {'tcp_server', start_link, []},
+            restart => permanent,
+            shutdown => 2000,
+            type => worker,
+            modules => ['tcp_server']
+        }
+    ],
+
+    {ok, {SupFlags, Specs}}.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================

apps/efka/src/websocket_server/ws_channel.erl

@@ -0,0 +1,150 @@
+%%%-------------------------------------------------------------------
+%%% @author licheng5
+%%% @copyright (C) 2021, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 11. 1月 2021 上午12:17
+%%%-------------------------------------------------------------------
+-module(ws_channel).
+-author("licheng5").
+-include("efka_service.hrl").
+
+%% API
+-export([init/2]).
+-export([websocket_init/1, websocket_handle/2, websocket_info/2, terminate/3]).
+
+%% 最大的等待时间
+-define(PENDING_TIMEOUT, 10 * 1000).
+
+-record(state, {
+    packet_id = 1,
+    service_id :: undefined | binary(),
+    service_pid :: undefined | pid(),
+    is_registered = false :: boolean(),
+
+    %% 请求响应的对应关系, #{packet_id => {ReceiverPid, Ref}}; 自身的inflight需要超时逻辑处理
+    inflight = #{}
+}).
+
+%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+%% 逻辑处理方法
+%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+
+init(Req, Opts) ->
+    {cowboy_websocket, Req, Opts}.
+
+websocket_init(_State) ->
+    lager:debug("[ws_channel] get a new connection"),
+    %% 初始状态为true
+    {ok, #state{packet_id = 1}}.
+
+websocket_handle({binary, <<?PACKET_REQUEST, Data/binary>>}, State) ->
+    Request = jiffy:decode(Data, [return_maps]),
+    handle_request(Request, State);
+
+%% 处理micro-client:response => efka 的响应
+websocket_handle({binary, <<?PACKET_RESPONSE:8, Data/binary>>}, State = #state{inflight = Inflight}) ->
+    {PacketId, Reply} = case jiffy:decode(Data, [return_maps]) of
+                            #{<<"id">> := Id, <<"result">> := Result} ->
+                                {Id, {ok, Result}};
+                            #{<<"id">> := Id, <<"error">> := #{<<"code">> := _Code, <<"message">> := Error}} ->
+                                {Id, {error, Error}}
+                        end,
+    NInflight = gen_channel:channel_reply(PacketId, Reply, Inflight),
+    {ok, State#state{inflight = NInflight}};
+
+websocket_handle(Info, State) ->
+    lager:error("[ws_channel] get a unknown message: ~p, channel will closed", [Info]),
+    {stop, State}.
+
+websocket_info({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{packet_id = PacketId, inflight = Inflight}) ->
+    PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"push_config">>, <<"params">> => #{<<"config">> => ConfigJson}},
+    Packet = jiffy:encode(PushConfig, [force_utf8]),
+    Reply = <<?PACKET_PUSH:8, Packet/binary>>,
+    erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
+
+    {reply, {binary, Reply}, State#state{packet_id = gen_channel:next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
+
+%% 远程调用
+websocket_info({invoke, Ref, ReceiverPid, Payload}, State = #state{packet_id = PacketId, inflight = Inflight}) ->
+    PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"invoke">>, <<"params">> => #{<<"payload">> => Payload}},
+    Packet = jiffy:encode(PushConfig, [force_utf8]),
+    Reply = <<?PACKET_PUSH:8, Packet/binary>>,
+    erlang:start_timer(?PENDING_TIMEOUT, self(), {pending_timeout, PacketId}),
+
+    {reply, {binary, Reply}, State#state{packet_id = gen_channel:next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
+
+%% 订阅的消息
+websocket_info({topic_broadcast, Topic, Content}, State = #state{}) ->
+    Packet = jiffy:encode(#{<<"topic">> => Topic, <<"content">> => Content}, [force_utf8]),
+    Reply = <<?PACKET_PUB:8, Packet/binary>>,
+    {reply, {binary, Reply}, State};
+
+%% 超时逻辑处理
+websocket_info({timeout, _, {pending_timeout, Id}}, State = #state{inflight = Inflight}) ->
+    NInflight = gen_channel:channel_reply(Id, {error, <<"timeout">>}, Inflight),
+    {ok, State#state{inflight = NInflight}};
+
+%% service进程关闭
+websocket_info({'DOWN', _Ref, process, ServicePid, Reason}, State = #state{service_pid = ServicePid}) ->
+    lager:debug("[tcp_channel] service_pid: ~p, exited: ~p", [ServicePid, Reason]),
+    {stop, State#state{service_pid = undefined}};
+
+websocket_info({timeout, _, {stop, Reason}}, State) ->
+    lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
+    {stop, State};
+%% 处理关闭信号
+websocket_info({stop, Reason}, State) ->
+    lager:debug("[ws_channel] the channel will be closed with reason: ~p", [Reason]),
+    {stop, State};
+
+%% 处理其他未知消息
+websocket_info(Info, State) ->
+    lager:debug("[ws_channel] channel get unknown info: ~p", [Info]),
+    {ok, State}.
+
+%% 进程关闭事件
+terminate(Reason, _Req, State) ->
+    lager:debug("[ws_channel] channel close with reason: ~p, state is: ~p", [Reason, State]),
+    ok.
+
+%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+%% helper methods
+%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
+
+%% 注册
+handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"service_id">> := ServiceId}}, State) ->
+    case gen_channel:register(Id, ServiceId) of
+        {error, Reply} ->
+            delay_stop(10, normal),
+            {reply, {binary,  <<?PACKET_RESPONSE:8, Reply/binary>>}, State};
+        {ok, Reply, ServicePid} ->
+            delay_stop(10, normal),
+            {reply, {binary, <<?PACKET_RESPONSE:8, Reply/binary>>}, State#state{service_id = ServiceId, service_pid = ServicePid, is_registered = true}}
+    end;
+%% 请求参数
+handle_request(#{<<"id">> := Id, <<"method">> := <<"request_config">>}, State = #state{service_pid = ServicePid, is_registered = true}) ->
+    {ok, ConfigJson} = efka_service:request_config(ServicePid),
+    Packet = gen_channel:json_result(Id, ConfigJson),
+    Reply = <<?PACKET_RESPONSE:8, Packet/binary>>,
+    {reply, {binary, Reply}, State};
+
+%% 数据项
+handle_request(#{<<"id">> := 0, <<"method">> := <<"metric_data">>,
+    <<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"route_key">> := RouteKey, <<"metric">> := Metric}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
+    efka_service:metric_data(ServicePid, DeviceUUID, RouteKey, Metric),
+    {ok, State};
+
+%% Event事件
+handle_request(#{<<"id">> := 0, <<"method">> := <<"event">>, <<"params">> := #{<<"event_type">> := EventType, <<"body">> := Body}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
+    efka_service:send_event(ServicePid, EventType, Body),
+    {ok, State};
+
+%% 订阅事件
+handle_request(#{<<"id">> := 0, <<"method">> := <<"subscribe">>, <<"params">> := #{<<"topic">> := Topic}}, State = #state{is_registered = true}) ->
+    efka_subscription:subscribe(Topic, self()),
+    {ok, State}.
+
+delay_stop(Timeout, Reason) when is_integer(Timeout) ->
+    erlang:start_timer(Timeout, self(), {stop, Reason}).

apps/efka/src/websocket_server/ws_server.erl

@@ -0,0 +1,34 @@
+%%%-------------------------------------------------------------------
+%%% @author anlicheng
+%%% @copyright (C) 2025, <COMPANY>
+%%% @doc
+%%%
+%%% @end
+%%% Created : 26. 8月 2025 14:28
+%%%-------------------------------------------------------------------
+-module(ws_server).
+-author("anlicheng").
+
+%% API
+-export([start_server/0]).
+
+start_server() ->
+    {ok, Props} = application:get_env(efka, ws_server),
+    Acceptors = proplists:get_value(acceptors, Props, 50),
+    MaxConnections = proplists:get_value(max_connections, Props, 10240),
+    Backlog = proplists:get_value(backlog, Props, 1024),
+    Port = proplists:get_value(port, Props),
+
+    Dispatcher = cowboy_router:compile([
+        {'_', [{"/ws", ws_channel, []}]}
+    ]),
+
+    TransOpts = [
+        {port, Port},
+        {num_acceptors, Acceptors},
+        {backlog, Backlog},
+        {max_connections, MaxConnections}
+    ],
+    {ok, Pid} = cowboy:start_clear(ws_listener, TransOpts, #{env => #{dispatch => Dispatcher}}),
+
+    lager:debug("[efka_app] websocket server start at: ~p, pid is: ~p", [Port, Pid]).

config/sys.config

@@ -2,12 +2,19 @@
   {efka, [
     {root_dir, "/usr/local/code/efka"},
 
-    {dets_dir, "/tmp/db/"},
+    {dets_dir, "/usr/local/code/tmp/dets/"},
 
     {tcp_server, [
       {port, 18088}
     ]},
 
+    {ws_server, [
+      {port, 18080},
+      {acceptors, 10},
+      {max_connections, 1024},
+      {backlog, 256}
+    ]},
+
     {tls_server, [
       {host, "localhost"},
       {port, 443}

message_pb.proto

@@ -19,7 +19,7 @@ message AuthReply {
 // service_id主动订阅消息, 基于广播通讯
 message Pub {
     string topic = 1;
-    string content = 2;
+    bytes content = 2;
 }
 
 ///// 服务器主动推送的消息
@@ -46,7 +46,7 @@ message FetchTaskLog {
 // 需要响应, 云端主动发起的调用; 提供给用户
 message Invoke {
     string service_id = 1;
-    string payload = 2;
+    bytes payload = 2;
     uint32 timeout = 3;
 }
 
@@ -63,8 +63,9 @@ message PushServiceConfig {
 message Data {
     string service_id = 1;
     string device_uuid = 2;
+    string route_key = 3;
     // measurement[,tag_key=tag_value...] field_key=field_value[,field_key2=field_value2...] [timestamp]
-    string metric = 3;
+    bytes metric = 4;
 }
 
 //#{<<"adcode">> => 0,<<"boot_time">> => 18256077,<<"city">> => <<>>,

rebar.config

@@ -4,6 +4,7 @@
     {jiffy, ".*", {git, "https://github.com/davisp/jiffy.git", {tag, "1.1.2"}}},
     {gpb, ".*", {git, "https://github.com/tomas-abrahamsson/gpb.git", {tag, "4.20.0"}}},
     {jiffy, ".*", {git, "https://github.com/davisp/jiffy.git", {tag, "1.1.1"}}},
+    {cowboy, ".*", {git, "https://github.com/ninenines/cowboy.git", {tag, "2.10.0"}}},
     {parse_trans, ".*", {git, "https://github.com/uwiger/parse_trans", {tag, "3.0.0"}}},
     {lager, ".*", {git,"https://github.com/erlang-lager/lager.git", {tag, "3.9.2"}}}
 ]}.