254 lines
11 KiB
Erlang
254 lines
11 KiB
Erlang
%%%-------------------------------------------------------------------
|
|
%%% @author anlicheng
|
|
%%% @copyright (C) 2025, <COMPANY>
|
|
%%% @doc
|
|
%%%
|
|
%%% @end
|
|
%%% Created : 30. 4月 2025 09:22
|
|
%%%-------------------------------------------------------------------
|
|
-module(efka_tcp_channel).
|
|
-author("anlicheng").
|
|
|
|
-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]).
|
|
|
|
-define(SERVER, ?MODULE).
|
|
|
|
%% 消息类型
|
|
|
|
%% 服务注册
|
|
-define(PACKET_REQUEST, 16#01).
|
|
%% 消息响应
|
|
-define(PACKET_RESPONSE, 16#02).
|
|
%% 上传数据
|
|
-define(PACKET_PUSH, 16#03).
|
|
|
|
-record(state, {
|
|
packet_id = 1,
|
|
socket :: gen_tcp:socket(),
|
|
service_id :: undefined | binary(),
|
|
service_pid :: undefined | pid(),
|
|
is_registered = false :: boolean(),
|
|
|
|
%% 请求响应的对应关系, #{packet_id => {ReceiverPid, Ref}}
|
|
inflight = #{}
|
|
}).
|
|
|
|
%%%===================================================================
|
|
%%% 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()}).
|
|
start_link(Socket) ->
|
|
gen_server:start_link(?MODULE, [Socket], []).
|
|
|
|
%%%===================================================================
|
|
%%% 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([Socket]) ->
|
|
ok = inet:setopts(Socket, [{active, true}]),
|
|
lager:debug("[tcp_channel] get new socket: ~p", [Socket]),
|
|
{ok, #state{socket = Socket}}.
|
|
|
|
%% @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(_Request, _From, State = #state{}) ->
|
|
{reply, ok, 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({push_config, Ref, ReceiverPid, ConfigJson}, State = #state{socket = Socket, packet_id = PacketId, inflight = Inflight}) ->
|
|
PushConfig = #{<<"id">> => PacketId, <<"method">> => <<"push_config">>, <<"params">> => #{<<"config">> => ConfigJson, <<"timeout">> => 5}},
|
|
Packet = jiffy:encode(PushConfig, [force_utf8]),
|
|
ok = gen_tcp:send(Socket, <<?PACKET_PUSH:8, Packet/binary>>),
|
|
|
|
{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, <<"timeout">> => 5}},
|
|
Packet = jiffy:encode(PushConfig, [force_utf8]),
|
|
|
|
ok = gen_tcp:send(Socket, <<?PACKET_PUSH:8, Packet/binary>>),
|
|
{noreply, State#state{packet_id = next_packet_id(PacketId), inflight = maps:put(PacketId, {ReceiverPid, Ref}, Inflight)}};
|
|
|
|
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{}}).
|
|
%% 处理client主动的请求
|
|
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
|
|
{ok, NewState} ->
|
|
{noreply, NewState};
|
|
{stop, Reason, NewState} ->
|
|
{stop, Reason, NewState}
|
|
end;
|
|
|
|
%% 处理client的响应
|
|
handle_info({tcp, Socket, <<?PACKET_RESPONSE:8, Data/binary>>}, State = #state{socket = Socket, inflight = Inflight}) ->
|
|
Response = jiffy:decode(Data, [return_maps]),
|
|
NInflight = reply(Response, Inflight),
|
|
|
|
{noreply, State#state{inflight = NInflight}};
|
|
|
|
handle_info(Info, State = #state{}) ->
|
|
lager:debug("[tcp_channel] get info: ~p", [Info]),
|
|
{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
|
|
%%%===================================================================
|
|
|
|
%% 注册
|
|
handle_request(#{<<"id">> := Id, <<"method">> := <<"register">>, <<"params">> := #{<<"service_id">> := ServiceId}}, State = #state{socket = Socket}) ->
|
|
case efka_service:get_pid(ServiceId) of
|
|
undefined ->
|
|
lager:warning("[efka_tcp_channel] service_id: ~p, not running", [ServiceId]),
|
|
Packet = json_error(Id, -1, <<"service not running">>),
|
|
ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Packet/binary>>),
|
|
{stop, normal, State};
|
|
Pid when is_pid(Pid) ->
|
|
case efka_service:attach_channel(Pid, self()) of
|
|
ok ->
|
|
Packet = json_result(Id, <<"ok">>),
|
|
ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Packet/binary>>),
|
|
{ok, State#state{service_id = ServiceId, service_pid = Pid, 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),
|
|
ok = gen_tcp:send(Socket, <<?PACKET_RESPONSE:8, Packet/binary>>),
|
|
{ok, State};
|
|
|
|
%% 数据项
|
|
handle_request(#{<<"id">> := 0, <<"method">> := <<"metric_data">>, <<"params">> := #{<<"device_uuid">> := DeviceUUID, <<"metric">> := Metric}}, State = #state{service_pid = ServicePid, is_registered = true}) ->
|
|
efka_service:metric_data(ServicePid, DeviceUUID, 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}.
|
|
|
|
%% 采用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]).
|
|
|
|
-spec reply(Resp :: map(), Inflight :: map()) -> NInflight :: map().
|
|
reply(Resp = #{<<"id">> := Id, <<"result">> := Result}, Inflight) ->
|
|
case maps:take(Id, Inflight) of
|
|
error ->
|
|
lager:warning("[tcp_channel] get unknown publish response message: ~p, packet_id: ~p", [Resp, Id]),
|
|
Inflight;
|
|
{{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,
|
|
NInflight
|
|
end;
|
|
reply(Resp = #{<<"id">> := Id, <<"error">> := #{<<"code">> := _Code, <<"message">> := Error}}, Inflight) ->
|
|
case maps:take(Id, Inflight) of
|
|
error ->
|
|
lager:warning("[tcp_channel] get unknown publish response message: ~p, packet_id: ~p", [Resp, Id]),
|
|
Inflight;
|
|
{{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,
|
|
NInflight
|
|
end. |