%%%------------------------------------------------------------------- %%% @author anlicheng %%% @copyright (C) 2026, %%% @doc %%% QUIC transport for sdlan sessions. %%% @end %%%------------------------------------------------------------------- -module(sdlan_quic_transport). -author("anlicheng"). -include("sdlan.hrl"). -include("sdlan_pb.hrl"). -behaviour(gen_statem). %% 心跳包监测机制 -define(STREAM_ACTIVE_N, 100). %% API -export([start_link/2]). -export([accept_stream/1, send_event/2, command/4, stop/2, debug_info/1]). %% gen_statem callbacks -export([init/1, handle_event/4, terminate/3, code_change/4, callback_mode/0]). -record(state, { conn :: quicer:connection_handle(), %% 最大包大小 max_packet_size = 16384, %% 心跳间隔 heartbeat_sec = 15, stream_active_n = ?STREAM_ACTIVE_N, stream :: undefined | quicer:stream_handle(), %% 累积器，用于处理协议framing的解析 buf = <<>>, session :: sdlan_session:state(), frames_recv = 0, bytes_recv = 0, close_reason = undefined }). %%%=================================================================== %%% API %%%=================================================================== -spec send_event(Pid :: pid(), Event :: binary()) -> no_return(). send_event(Pid, ProtobufEvent) when is_pid(Pid), is_binary(ProtobufEvent) -> gen_statem:cast(Pid, {send_event, ProtobufEvent}). -spec command(Pid :: pid(), Ref :: reference(), ReceiverPid :: pid(), {Tag :: atom(), SubCommand :: any()}) -> no_return(). command(Pid, Ref, ReceiverPid, SubCommand) when is_pid(Pid), is_pid(ReceiverPid) -> gen_statem:cast(Pid, {command, Ref, ReceiverPid, SubCommand}). accept_stream(Pid) when is_pid(Pid) -> gen_statem:cast(Pid, accept_stream). -spec stop(Pid :: pid(), Reason :: term()) -> ok. stop(Pid, Reason) when is_pid(Pid) -> gen_statem:stop(Pid, Reason, 2000). debug_info(Pid) when is_pid(Pid) -> gen_statem:call(Pid, debug_info). %% @doc Creates a gen_statem process which calls Module:init/1 to %% initialize. To ensure a synchronized start-up procedure, this %% function does not return until Module:init/1 has returned. start_link(Conn, Limits) when is_list(Limits) -> gen_statem:start_link(?MODULE, [Conn, Limits], []). %%%=================================================================== %%% gen_statem callbacks %%%=================================================================== %% @private %% @doc Whenever a gen_statem is started using gen_statem:start/[3,4] or %% gen_statem:start_link/[3,4], this function is called by the new %% process to initialize. init([Conn, Limits]) -> MaxPacketSize = proplists:get_value(max_packet_size, Limits, 16384), HeartbeatSec = proplists:get_value(heartbeat_sec, Limits, 15), StreamActiveN = proplists:get_value(stream_active_n, Limits, ?STREAM_ACTIVE_N), Session = sdlan_session:new(HeartbeatSec), {ok, initializing, #state{ conn = Conn, max_packet_size = MaxPacketSize, heartbeat_sec = HeartbeatSec, stream_active_n = StreamActiveN, session = Session }}. %% @private %% @doc This function is called by a gen_statem when it needs to find out %% the callback mode of the callback module. callback_mode() -> handle_event_function. %% @private %% @doc If callback_mode is handle_event_function, then whenever a %% gen_statem receives an event from call/2, cast/2, or as a normal %% process message, this function is called. handle_event(cast, accept_stream, initializing, State = #state{conn = Conn, stream_active_n = StreamActiveN}) -> logger:debug("[sdlan_quic_transport] call do_init of conn: ~p", [Conn]), case quicer:async_accept_stream(Conn, #{active => StreamActiveN}) of {ok, _} -> {next_state, waiting_stream, State}; {error, Reason} -> {stop, {accept_stream_failed, Reason}, State} end; %% 处理收到的quic消息 handle_event(info, {quic, dgram_state_changed, Conn, Opts = #{dgram_send_enabled := true}}, _, State = #state{conn = Conn}) -> logger:debug("[sdlan_quic_transport] dgram_state_changed, opts: ~p", [Opts]), {keep_state, State}; handle_event(info, {quic, new_stream, Stream, Opts}, waiting_stream, State = #state{max_packet_size = MaxPacketSize, heartbeat_sec = HeartbeatSec}) -> logger:debug("[sdlan_quic_transport] call new_stream: ~p, opts: ~p", [Stream, Opts]), %% 发送欢迎消息 quic_send(Stream, sdlan_session:welcome_packet(MaxPacketSize, HeartbeatSec)), logger:debug("[sdlan_quic_transport] get stream: ~p, send welcome", [Stream]), {next_state, initialized, State#state{stream = Stream}}; handle_event(info, {quic, new_stream, Stream, Opts}, _StateName, State) -> logger:warning("[sdlan_quic_transport] reject unexpected stream: ~p, opts: ~p", [Stream, Opts]), quicer:close_stream(Stream, 1000), {keep_state, State}; handle_event(info, {quic, stream_closed, Stream, Props}, _StateName, State = #state{stream = Stream}) -> expected_stop({stream_closed, Props}, State); handle_event(info, {quic, peer_send_shutdown, Stream, _Props}, _StateName, State = #state{stream = Stream}) -> expected_stop(peer_send_shutdown, State); handle_event(info, {quic, peer_send_aborted, Stream, ErrorCode}, _StateName, State = #state{stream = Stream}) -> expected_stop({peer_send_aborted, ErrorCode}, State); handle_event(info, {quic, peer_receive_aborted, Stream, ErrorCode}, _StateName, State = #state{stream = Stream}) -> expected_stop({peer_receive_aborted, ErrorCode}, State); handle_event(info, {quic, send_shutdown_complete, Stream, _Props}, _StateName, State = #state{stream = Stream}) -> expected_stop(connection_shutdown, State); handle_event(info, {quic, passive, Stream, _Props}, _StateName, State = #state{stream = Stream, stream_active_n = StreamActiveN}) -> ok = quicer:setopt(Stream, active, StreamActiveN), {keep_state, State}; handle_event(info, {quic, closed, Conn, Props}, _StateName, State = #state{conn = Conn}) -> expected_stop({connection_closed, Props}, State); handle_event(info, {quic, transport_shutdown, Conn, Props}, _StateName, State = #state{conn = Conn}) -> expected_stop({transport_shutdown, Props}, State); handle_event(info, {quic, shutdown, Conn, ErrorCode}, _StateName, State = #state{conn = Conn}) -> expected_stop({connection_shutdown_by_peer, ErrorCode}, State); %% 处理quicer相关的信息, 需要转换成内部能够识别的frame消息 handle_event(info, {quic, Data, Stream, _Props}, _StateName, State = #state{stream = Stream, buf = Buf, max_packet_size = MaxPacketSize, bytes_recv = BytesRecv, frames_recv = FramesRecv}) when is_binary(Data) -> case decode_frames(<>, MaxPacketSize) of {error, Reason} -> {stop, Reason, State}; {ok, NBuf, Frames} -> Actions = [{next_event, internal, {frame, Frame}} || Frame <- Frames], {keep_state, State#state{buf = NBuf, bytes_recv = BytesRecv + byte_size(Data), frames_recv = FramesRecv + length(Frames)}, Actions} end; %% 处理内部的包消息 handle_event(internal, {frame, Frame}, StateName, State = #state{stream = Stream, session = Session}) -> case sdlan_session:handle_frame(Frame, Session) of {ok, NSession, Packets} -> send_packets(Stream, Packets), {next_state, next_state_name(StateName, NSession), State#state{session = NSession}}; {stop, Reason, NSession, Packets} -> send_packets(Stream, Packets), {stop, Reason, State#state{session = NSession}} end; handle_event(info, {timeout, TimerRef, ping_ticker}, _StateName, State = #state{session = Session}) -> case sdlan_session:handle_timeout(TimerRef, Session) of {ok, NSession} -> {next_state, next_state_name(registered, NSession), State#state{session = NSession}}; {stop, Reason, NSession} -> expected_stop(Reason, State#state{session = NSession}) end; %% 发送指令信息 handle_event(cast, {send_event, Event}, _StateName, State = #state{stream = Stream, session = Session}) -> case sdlan_session:send_event(Event, Session) of {ok, NSession, Packets} -> send_packets(Stream, Packets), {keep_state, State#state{session = NSession}}; {error, not_registered} -> keep_state_and_data end; %% 发送命令信息 handle_event(cast, {command, Ref, ReceiverPid, SubCommand}, _StateName, State = #state{stream = Stream, session = Session}) -> case sdlan_session:command(Ref, ReceiverPid, SubCommand, Session) of {ok, NSession, Packets} -> send_packets(Stream, Packets), {keep_state, State#state{session = NSession}}; {error, not_registered} -> keep_state_and_data end; handle_event({call, From}, debug_info, StateName, State) -> {keep_state, State, [{reply, From, debug_info(StateName, State)}]}; handle_event(info, {'EXIT', _, _}, _StateName, State) -> expected_stop(connection_closed, State); handle_event(EventType, Info, StateName, State) -> logger:notice("[sdlan_quic_transport] state: ~p, state_name: ~p, event_type: ~p, info: ~p", [State, StateName, EventType, Info]), keep_state_and_data. %% @private %% @doc This function is called by a gen_statem 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_statem terminates with %% Reason. The return value is ignored. terminate(Reason, _StateName, _State = #state{conn = Conn, stream = Stream, session = Session, close_reason = CloseReason}) -> Stream /= undefined andalso quicer:close_stream(Stream, 1000), quicer:close_connection(Conn), logger:notice("[sdlan_quic_transport] terminate closed with reason: ~p, close_reason: ~p", [Reason, CloseReason]), sdlan_session:close(Session), ok. %% @private %% @doc Convert process state when code is changed code_change(_OldVsn, StateName, State = #state{}, _Extra) -> {ok, StateName, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== %% 有2种情况 %% 1. 收到了多个完整的请求 %% 2. 不完整，则不处理 -spec decode_frames(Buf :: binary(), MaxPacketSize :: integer()) -> {ok, RestBin::binary(), Frames :: list()} | {error, Reason :: any()}. decode_frames(Buf, MaxPacketSize) when is_binary(Buf) -> decode_frames0(Buf, MaxPacketSize, []). decode_frames0(<>, MaxPacketSize, _Frames) when Len > MaxPacketSize -> {error, frame_too_large}; decode_frames0(<>, MaxPacketSize, Frames) -> decode_frames0(Rest, MaxPacketSize, [Frame|Frames]); decode_frames0(Rest, _MaxPacketSize, Frames) -> {ok, Rest, lists:reverse(Frames)}. -spec quic_send(Stream :: quicer:stream_handle(), Packet :: binary()) -> no_return(). quic_send(Stream, Packet) when is_binary(Packet) -> Len = byte_size(Packet), true = Len =< 65535, case quicer:send(Stream, <>) of {ok, _} -> incr_counter(quic_frames_sent, 1), incr_counter(quic_bytes_sent, Len + 2), ok; {error, Reason} -> exit({quic_send_failed, Reason}) end. send_packets(Stream, Packets) -> lists:foreach(fun(Packet) -> quic_send(Stream, Packet) end, Packets). expected_stop(Reason, State) -> logger:notice("[sdlan_quic_transport] expected close: ~p", [Reason]), {stop, normal, State#state{close_reason = Reason}}. next_state_name(_StateName, Session) -> sdlan_session:state_name(Session). debug_info(StateName, #state{ session = Session, frames_recv = FramesRecv, bytes_recv = BytesRecv, stream_active_n = StreamActiveN, heartbeat_sec = HeartbeatSec }) -> ProcInfo = maps:from_list(process_info(self(), [message_queue_len, memory, reductions])), SessionInfo = sdlan_session:debug_info(Session), maps:merge(SessionInfo, ProcInfo#{ state => StateName, session => SessionInfo, frames_recv => FramesRecv, frames_sent => get_counter(quic_frames_sent), bytes_recv => BytesRecv, bytes_sent => get_counter(quic_bytes_sent), stream_active_n => StreamActiveN, heartbeat_sec => HeartbeatSec }). incr_counter(Key, Inc) -> erlang:put(Key, get_counter(Key) + Inc). get_counter(Key) -> case erlang:get(Key) of undefined -> 0; Value when is_integer(Value) -> Value end.