fix transport
This commit is contained in:
parent
ce2e8a4d60
commit
2c88842012
@ -170,30 +170,30 @@ handle_event(info, {quic, Data, Stream, _Props}, _StateName,
|
||||
end;
|
||||
|
||||
%% 处理内部的包消息
|
||||
handle_event(internal, {frame, Frame}, StateName, State = #state{stream = Stream, session = Session}) ->
|
||||
handle_event(internal, {frame, Frame}, _StateName, State = #state{stream = Stream, session = Session}) ->
|
||||
case sdlan_session:handle_frame(Frame, Session) of
|
||||
{ok, NSession, Packets} ->
|
||||
{ok, NStateName, NSession, Packets} ->
|
||||
send_packets(Stream, Packets),
|
||||
{next_state, next_state_name(StateName, NSession), State#state{session = NSession}};
|
||||
{stop, Reason, NSession, Packets} ->
|
||||
{next_state, NStateName, State#state{session = NSession}};
|
||||
{stop, Reason, _NStateName, 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} ->
|
||||
{ok, NStateName, NSession} ->
|
||||
{next_state, NStateName, State#state{session = NSession}};
|
||||
{stop, Reason, _NStateName, 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} ->
|
||||
{ok, NStateName, NSession, Packets} ->
|
||||
send_packets(Stream, Packets),
|
||||
{keep_state, State#state{session = NSession}};
|
||||
{next_state, NStateName, State#state{session = NSession}};
|
||||
{error, not_registered} ->
|
||||
keep_state_and_data
|
||||
end;
|
||||
@ -201,9 +201,9 @@ handle_event(cast, {send_event, Event}, _StateName, State = #state{stream = Stre
|
||||
%% 发送命令信息
|
||||
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} ->
|
||||
{ok, NStateName, NSession, Packets} ->
|
||||
send_packets(Stream, Packets),
|
||||
{keep_state, State#state{session = NSession}};
|
||||
{next_state, NStateName, State#state{session = NSession}};
|
||||
{error, not_registered} ->
|
||||
keep_state_and_data
|
||||
end;
|
||||
@ -272,9 +272,6 @@ 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,
|
||||
|
||||
@ -11,7 +11,7 @@
|
||||
-include("sdlan_pb.hrl").
|
||||
|
||||
%% API
|
||||
-export([new/1, state_name/1, handle_frame/2, handle_timeout/2]).
|
||||
-export([new/1, handle_frame/2, handle_timeout/2]).
|
||||
-export([send_event/2, command/4, close/1, debug_info/1]).
|
||||
-export([welcome_packet/2]).
|
||||
-export([test_rules/2]).
|
||||
@ -68,10 +68,6 @@ test_rules(SrcIdentityId, DstIdentityId) when is_integer(SrcIdentityId), is_inte
|
||||
logger:debug("[sdlan_session] test_rules policy_request src_identity_id: ~p, dst_identity_id: ~p, rules: ~p", [SrcIdentityId, DstIdentityId, Rules]),
|
||||
iolist_to_binary(lists:map(fun({Proto, Port}) -> <<Proto:8, Port:16>> end, Rules)).
|
||||
|
||||
-spec state_name(Session :: #session{}) -> initialized | registered.
|
||||
state_name(#session{status = Status}) ->
|
||||
Status.
|
||||
|
||||
-spec welcome_packet(MaxPacketSize :: integer(), HeartbeatSec :: integer()) -> binary().
|
||||
welcome_packet(MaxPacketSize, HeartbeatSec) ->
|
||||
Ipv6Assist = case application:get_env(sdlan, ipv6_assist_info) of
|
||||
@ -93,8 +89,8 @@ welcome_packet(MaxPacketSize, HeartbeatSec) ->
|
||||
<<?PACKET_WELCOME, WelcomePkt/binary>>.
|
||||
|
||||
-spec handle_frame(Frame :: binary(), Session :: #session{}) ->
|
||||
{ok, NewSession :: #session{}, Packets :: [binary()]} |
|
||||
{stop, Reason :: term(), NewSession :: #session{}, Packets :: [binary()]}.
|
||||
{ok, StateName :: initialized | registered, NewSession :: #session{}, Packets :: [binary()]} |
|
||||
{stop, Reason :: term(), StateName :: initialized | registered, NewSession :: #session{}, Packets :: [binary()]}.
|
||||
handle_frame(<<?PACKET_REGISTER_SUPER, Body/binary>>, Session = #session{status = initialized}) ->
|
||||
handle_register_super(Body, Session);
|
||||
|
||||
@ -110,7 +106,7 @@ handle_frame(<<?PACKET_QUERY_INFO, Body/binary>>, Session = #session{status = re
|
||||
v4_info = undefined,
|
||||
v6_info = undefined
|
||||
}),
|
||||
{ok, Session, [<<?PACKET_PEER_INFO, EmptyResponse/binary>>]};
|
||||
ok_result(Session, [<<?PACKET_PEER_INFO, EmptyResponse/binary>>]);
|
||||
{ok, {NatPeer = {{Ip0, Ip1, Ip2, Ip3}, NatPort}, NatType}, V6Info} ->
|
||||
logger:debug("[sdlan_session] query_info src_mac is: ~p, dst_mac: ~p, nat_peer: ~p",
|
||||
[sdlan_util:format_mac(SrcMac), sdlan_util:format_mac(DstMac), NatPeer]),
|
||||
@ -124,7 +120,7 @@ handle_frame(<<?PACKET_QUERY_INFO, Body/binary>>, Session = #session{status = re
|
||||
},
|
||||
v6_info = V6Info
|
||||
}),
|
||||
{ok, Session, [<<?PACKET_PEER_INFO, PeerInfo/binary>>]}
|
||||
ok_result(Session, [<<?PACKET_PEER_INFO, PeerInfo/binary>>])
|
||||
end;
|
||||
|
||||
%% arp查询
|
||||
@ -139,7 +135,7 @@ handle_frame(<<?PACKET_ARP_REQUEST, Body/binary>>, Session = #session{status = r
|
||||
origin_ip = OriginIp,
|
||||
context = Context
|
||||
}),
|
||||
{ok, Session, [<<?PACKET_ARP_RESPONSE, EmptyArpResponsePkt/binary>>]};
|
||||
ok_result(Session, [<<?PACKET_ARP_RESPONSE, EmptyArpResponsePkt/binary>>]);
|
||||
{ok, Mac} ->
|
||||
logger:debug("[sdlan_session] network: ~p, arp_request target_ip: ~p, mac: ~p", [NetworkId, sdlan_util:int_to_ipv4(TargetIp), sdlan_util:format_mac(Mac)]),
|
||||
ArpResponsePkt = sdlan_pb:encode_msg(#'SDLArpResponse'{
|
||||
@ -148,7 +144,7 @@ handle_frame(<<?PACKET_ARP_REQUEST, Body/binary>>, Session = #session{status = r
|
||||
origin_ip = OriginIp,
|
||||
context = Context
|
||||
}),
|
||||
{ok, Session, [<<?PACKET_ARP_RESPONSE, ArpResponsePkt/binary>>]}
|
||||
ok_result(Session, [<<?PACKET_ARP_RESPONSE, ArpResponsePkt/binary>>])
|
||||
end;
|
||||
|
||||
handle_frame(<<?PACKET_POLICY_REQUEST, Body/binary>>, Session = #session{status = registered, network_pid = NetworkPid}) when is_pid(NetworkPid) ->
|
||||
@ -169,7 +165,7 @@ handle_frame(<<?PACKET_POLICY_REQUEST, Body/binary>>, Session = #session{status
|
||||
else _ ->
|
||||
[]
|
||||
end,
|
||||
{ok, Session, Packets};
|
||||
ok_result(Session, Packets);
|
||||
|
||||
%% 处理命令的响应逻辑
|
||||
handle_frame(<<?PACKET_COMMAND_ACK, Body/binary>>, Session = #session{status = registered, pending_commands = PendingCommands}) ->
|
||||
@ -184,46 +180,48 @@ handle_frame(<<?PACKET_COMMAND_ACK, Body/binary>>, Session = #session{status = r
|
||||
false ->
|
||||
ok
|
||||
end,
|
||||
{ok, Session#session{pending_commands = RestPendingCommands}, []}
|
||||
ok_result(Session#session{pending_commands = RestPendingCommands}, [])
|
||||
else _ ->
|
||||
{ok, Session, []}
|
||||
ok_result(Session, [])
|
||||
end;
|
||||
|
||||
handle_frame(<<?PACKET_PING>>, Session = #session{ping_counter = PingCounter, client_id = ClientId}) ->
|
||||
logger:warning("[sdlan_session] get ping: ~p", [ClientId]),
|
||||
{ok, Session#session{ping_counter = PingCounter + 1}, [<<?PACKET_PONG>>]};
|
||||
ok_result(Session#session{ping_counter = PingCounter + 1}, [<<?PACKET_PONG>>]);
|
||||
|
||||
%% 取消注册
|
||||
handle_frame(<<?PACKET_UNREGISTER>>, Session = #session{status = registered, client_id = ClientId, mac = Mac, network_pid = NetworkPid}) when is_pid(NetworkPid) ->
|
||||
logger:warning("[sdlan_session] unregister client_id: ~p", [ClientId]),
|
||||
sdlan_network:unregister(NetworkPid, ClientId, Mac),
|
||||
{stop, normal, Session, []};
|
||||
stop_result(normal, Session, []);
|
||||
|
||||
handle_frame(Frame, Session) ->
|
||||
logger:notice("[sdlan_session] unexpected frame: ~p, status: ~p", [Frame, Session#session.status]),
|
||||
{ok, Session, []}.
|
||||
ok_result(Session, []).
|
||||
|
||||
-spec handle_timeout(TimerRef :: reference(), Session :: #session{}) ->
|
||||
{ok, NewSession :: #session{}} | {stop, Reason :: term(), NewSession :: #session{}}.
|
||||
{ok, StateName :: initialized | registered, NewSession :: #session{}} |
|
||||
{stop, Reason :: term(), StateName :: initialized | registered, NewSession :: #session{}}.
|
||||
handle_timeout(TimerRef, Session = #session{ping_timer = TimerRef, client_id = ClientId, ping_counter = PingCounter}) ->
|
||||
case PingCounter > 0 of
|
||||
true ->
|
||||
{ok, schedule_ping(Session#session{ping_counter = 0, ping_timer = undefined})};
|
||||
ok_timeout_result(schedule_ping(Session#session{ping_counter = 0, ping_timer = undefined}));
|
||||
false ->
|
||||
logger:debug("[sdlan_session] client_id: ~p, ping losted", [ClientId]),
|
||||
{stop, heartbeat_timeout, Session#session{ping_counter = 0, ping_timer = undefined}}
|
||||
stop_timeout_result(heartbeat_timeout, Session#session{ping_counter = 0, ping_timer = undefined})
|
||||
end;
|
||||
handle_timeout(_TimerRef, Session) ->
|
||||
{ok, Session}.
|
||||
ok_timeout_result(Session).
|
||||
|
||||
-spec send_event(Event :: binary(), Session :: #session{}) -> {ok, NewSession :: #session{}, Packets :: [binary()]} | {error, not_registered}.
|
||||
-spec send_event(Event :: binary(), Session :: #session{}) ->
|
||||
{ok, StateName :: initialized | registered, NewSession :: #session{}, Packets :: [binary()]} | {error, not_registered}.
|
||||
send_event(Event, Session = #session{status = registered}) when is_binary(Event) ->
|
||||
{ok, Session, [<<?PACKET_EVENT, Event/binary>>]};
|
||||
send_event(_Event, _State) ->
|
||||
ok_result(Session, [<<?PACKET_EVENT, Event/binary>>]);
|
||||
send_event(_Event, _Session) ->
|
||||
{error, not_registered}.
|
||||
|
||||
-spec command(Ref :: reference(), ReceiverPid :: pid(), {Tag :: atom(), SubCommand :: any()}, Session :: #session{}) ->
|
||||
{ok, NewSession :: #session{}, Packets :: [binary()]} | {error, not_registered}.
|
||||
{ok, StateName :: initialized | registered, NewSession :: #session{}, Packets :: [binary()]} | {error, not_registered}.
|
||||
command(Ref, ReceiverPid, SubCommand, Session = #session{status = registered, pkt_id = PktId, pending_commands = PendingCommands, client_id = ClientId})
|
||||
when is_reference(Ref), is_pid(ReceiverPid) ->
|
||||
CommandPkt = sdlan_pb:encode_msg(#'SDLCommand'{
|
||||
@ -232,9 +230,9 @@ command(Ref, ReceiverPid, SubCommand, Session = #session{status = registered, pk
|
||||
}),
|
||||
logger:debug("[sdlan_session] client_id: ~p, will send Command: ~p", [ClientId, SubCommand]),
|
||||
|
||||
{ok, Session#session{pkt_id = PktId + 1, pending_commands = maps:put(PktId, {Ref, ReceiverPid}, PendingCommands)},
|
||||
[<<?PACKET_COMMAND, CommandPkt/binary>>]};
|
||||
command(_Ref, _ReceiverPid, _SubCommand, _State) ->
|
||||
ok_result(Session#session{pkt_id = PktId + 1, pending_commands = maps:put(PktId, {Ref, ReceiverPid}, PendingCommands)},
|
||||
[<<?PACKET_COMMAND, CommandPkt/binary>>]);
|
||||
command(_Ref, _ReceiverPid, _SubCommand, _Session) ->
|
||||
{error, not_registered}.
|
||||
|
||||
-spec close(Session :: #session{}) -> ok.
|
||||
@ -331,17 +329,17 @@ handle_register_super(Body, Session) ->
|
||||
ip = Ip,
|
||||
offline_cb = OfflineCb
|
||||
}),
|
||||
{ok, NSession, [<<?PACKET_REGISTER_SUPER_ACK, RegisterSuperAck/binary>>]};
|
||||
ok_result(NSession, [<<?PACKET_REGISTER_SUPER_ACK, RegisterSuperAck/binary>>]);
|
||||
undefined ->
|
||||
logger:warning("[sdlan_session] client_id: ~p, register get error: network not found", [ClientId]),
|
||||
{stop, normal, Session, [register_nak_reply(?NAK_INTERNAL_FAULT, <<"Internal Error">>)]}
|
||||
stop_result(normal, Session, [register_nak_reply(?NAK_INTERNAL_FAULT, <<"Internal Error">>)])
|
||||
end;
|
||||
{ok, #{<<"error">> := #{<<"code">> := Code, <<"message">> := Message}}} ->
|
||||
logger:warning("[sdlan_session] network_id: ~p, client_id: ~p, register get error: ~ts, error_code: ~p", [NetworkId, ClientId, Message, Code]),
|
||||
{stop, normal, Session, [register_nak_reply(Code, Message)]};
|
||||
stop_result(normal, Session, [register_nak_reply(Code, Message)]);
|
||||
{error, Reason} ->
|
||||
logger:warning("[sdlan_session] network_id: ~p, client_id: ~p, register get error: ~p", [NetworkId, ClientId, Reason]),
|
||||
{stop, normal, Session, [register_nak_reply(?NAK_NETWORK_FAULT, <<"Network Error">>)]}
|
||||
stop_result(normal, Session, [register_nak_reply(?NAK_NETWORK_FAULT, <<"Network Error">>)])
|
||||
end.
|
||||
|
||||
-spec register_nak_reply(ErrorCode :: integer(), ErrorMsg :: binary()) -> binary().
|
||||
@ -364,6 +362,18 @@ get_rules(SrcIdentityId, DstIdentityId) when is_integer(SrcIdentityId), is_integ
|
||||
schedule_ping(Session = #session{heartbeat_sec = HeartbeatSec}) ->
|
||||
Session#session{ping_timer = erlang:start_timer(heartbeat_ms(HeartbeatSec), self(), ping_ticker)}.
|
||||
|
||||
ok_result(Session = #session{status = StateName}, Packets) ->
|
||||
{ok, StateName, Session, Packets}.
|
||||
|
||||
stop_result(Reason, Session = #session{status = StateName}, Packets) ->
|
||||
{stop, Reason, StateName, Session, Packets}.
|
||||
|
||||
ok_timeout_result(Session = #session{status = StateName}) ->
|
||||
{ok, StateName, Session}.
|
||||
|
||||
stop_timeout_result(Reason, Session = #session{status = StateName}) ->
|
||||
{stop, Reason, StateName, Session}.
|
||||
|
||||
heartbeat_ms(HeartbeatSec) when is_integer(HeartbeatSec), HeartbeatSec > 0 ->
|
||||
HeartbeatSec * 1000;
|
||||
heartbeat_ms(_) ->
|
||||
|
||||
@ -71,13 +71,13 @@ handle_event(info, {handshake, Ref, Transport, Socket, Timeout}, handshaking,
|
||||
{stop, {ssl_handshake_failed, Reason}, State}
|
||||
end;
|
||||
|
||||
handle_event(info, {ssl, Socket, Frame}, StateName,
|
||||
handle_event(info, {ssl, Socket, Frame}, _StateName,
|
||||
State = #state{socket = Socket, transport = Transport, session = Session}) when is_binary(Frame) ->
|
||||
case sdlan_session:handle_frame(Frame, Session) of
|
||||
{ok, NSession, Packets} ->
|
||||
{ok, NStateName, NSession, Packets} ->
|
||||
send_packets(Transport, Socket, Packets),
|
||||
{next_state, next_state_name(StateName, NSession), State#state{session = NSession}};
|
||||
{stop, Reason, NSession, Packets} ->
|
||||
{next_state, NStateName, State#state{session = NSession}};
|
||||
{stop, Reason, _NStateName, NSession, Packets} ->
|
||||
send_packets(Transport, Socket, Packets),
|
||||
{stop, Reason, State#state{session = NSession}}
|
||||
end;
|
||||
@ -90,18 +90,18 @@ handle_event(info, {ssl_error, Socket, Reason}, _StateName, State = #state{socke
|
||||
|
||||
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} ->
|
||||
{ok, NStateName, NSession} ->
|
||||
{next_state, NStateName, State#state{session = NSession}};
|
||||
{stop, Reason, _NStateName, NSession} ->
|
||||
expected_stop(Reason, State#state{session = NSession})
|
||||
end;
|
||||
|
||||
%% 发送事件信息
|
||||
handle_event(cast, {send_event, Event}, _StateName, State = #state{socket = Socket, transport = Transport, session = Session}) ->
|
||||
case sdlan_session:send_event(Event, Session) of
|
||||
{ok, NSession, Packets} ->
|
||||
{ok, NStateName, NSession, Packets} ->
|
||||
send_packets(Transport, Socket, Packets),
|
||||
{keep_state, State#state{session = NSession}};
|
||||
{next_state, NStateName, State#state{session = NSession}};
|
||||
{error, not_registered} ->
|
||||
keep_state_and_data
|
||||
end;
|
||||
@ -109,9 +109,9 @@ handle_event(cast, {send_event, Event}, _StateName, State = #state{socket = Sock
|
||||
%% 发送命令信息
|
||||
handle_event(cast, {command, Ref, ReceiverPid, SubCommand}, _StateName, State = #state{socket = Socket, transport = Transport, session = Session}) ->
|
||||
case sdlan_session:command(Ref, ReceiverPid, SubCommand, Session) of
|
||||
{ok, NSession, Packets} ->
|
||||
{ok, NStateName, NSession, Packets} ->
|
||||
send_packets(Transport, Socket, Packets),
|
||||
{keep_state, State#state{session = NSession}};
|
||||
{next_state, NStateName, State#state{session = NSession}};
|
||||
{error, not_registered} ->
|
||||
keep_state_and_data
|
||||
end;
|
||||
@ -155,9 +155,6 @@ expected_stop(Reason, State) ->
|
||||
logger:notice("[sdlan_ssl_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, heartbeat_sec = HeartbeatSec}) ->
|
||||
ProcInfo = maps:from_list(process_info(self(), [message_queue_len, memory, reductions])),
|
||||
SessionInfo = sdlan_session:debug_info(Session),
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user