From e46dc8d2bb184c660ec4967502ad23bd0eeb0774 Mon Sep 17 00:00:00 2001 From: anlicheng <244108715@qq.com> Date: Sat, 9 May 2026 18:07:13 +0800 Subject: [PATCH] fix heartbeat --- src/iot_sup.erl | 9 +++ src/transport/udp/udp_server.erl | 115 ++++++++++++++++++++++++++----- 2 files changed, 105 insertions(+), 19 deletions(-) diff --git a/src/iot_sup.erl b/src/iot_sup.erl index 0d15f30..fccda27 100644 --- a/src/iot_sup.erl +++ b/src/iot_sup.erl @@ -37,6 +37,15 @@ init([]) -> modules => ['iot_container_task_sup'] }, + #{ + id => 'udp_server', + start => {'udp_server', start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => ['udp_server'] + }, + #{ id => endpoint_sup_sup, start => {'endpoint_sup_sup', start_link, []}, diff --git a/src/transport/udp/udp_server.erl b/src/transport/udp/udp_server.erl index 26fa9bf..a61a854 100644 --- a/src/transport/udp/udp_server.erl +++ b/src/transport/udp/udp_server.erl @@ -1,33 +1,110 @@ %%%------------------------------------------------------------------- -%%% @author aresei -%%% @copyright (C) 2023, +%%% @author anlicheng +%%% @copyright (C) 2026, %%% @doc %%% %%% @end -%%% Created : 31. 8月 2023 13:48 +%%% Created : 09. 5月 2026 18:02 %%%------------------------------------------------------------------- -module(udp_server). --author("aresei"). +-author("anlicheng"). -%% API --export([start_link/2, loop/2]). +-behaviour(gen_server). -define(HEARTBEAT_VERSION, 1). -define(HEARTBEAT_NONCE_BYTES, 16). -define(HEARTBEAT_MAC_BYTES, 32). -start_link(Transport, Peer) -> - {ok, spawn_link(?MODULE, loop, [Transport, Peer])}. +%% API +-export([start_link/0]). -loop(Transport = {udp, Server, _Sock}, Peer) -> - receive - {datagram, Server, Packet} -> - handle_heartbeat_packet(Packet, Peer), - loop(Transport, Peer); - Other -> - logger:warning("[udp_server] ignore unknown message from peer: ~p, message: ~p", [Peer, Other]), - loop(Transport, Peer) - end. +%% gen_server callbacks +-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). + +-define(SERVER, ?MODULE). + +-record(state, { + socket :: gen_udp:socket() +}). + +%%%=================================================================== +%%% API +%%%=================================================================== + +%% @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, UdpServerProps} = application:get_env(iot, udp_server), + Port = proplists:get_value(port, UdpServerProps), + {ok, Socket} = gen_udp:open(Port, [binary, {active, true}]), + {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(_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({udp, Socket, Ip, Port, Packet}, State = #state{socket = Socket}) -> + handle_heartbeat_packet(Packet, {Ip, Port}), + {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 +%%%=================================================================== -spec handle_heartbeat_packet(binary(), term()) -> ok. handle_heartbeat_packet(Packet, Peer) when is_binary(Packet) -> @@ -50,9 +127,9 @@ handle_heartbeat_packet(Packet, Peer) when is_binary(Packet) -> {ok, UUID :: binary(), Timestamp :: integer(), Payload :: binary(), Mac :: binary()} | error. decode_heartbeat_packet(Packet = <>) - when UUIDLen > 0 -> + when UUIDLen > 0 -> PayloadSize = byte_size(Packet) - ?HEARTBEAT_MAC_BYTES, <> = Packet, {ok, UUID, Timestamp, Payload, Mac}; decode_heartbeat_packet(_Packet) -> - error. + error. \ No newline at end of file