fix transport
This commit is contained in:
parent
1cf007f946
commit
c15a38e8c8
@ -34,7 +34,7 @@
|
||||
%% 累积器，用于处理协议framing的解析
|
||||
buf = <<>>,
|
||||
|
||||
session :: sdlan_session:state(),
|
||||
session :: sdlan_session:session(),
|
||||
|
||||
frames_recv = 0,
|
||||
bytes_recv = 0,
|
||||
|
||||
@ -15,7 +15,7 @@
|
||||
-export([send_event/2, command/4, close/1, debug_info/1]).
|
||||
-export([welcome_packet/2]).
|
||||
-export([test_rules/2]).
|
||||
-export_type([state/0]).
|
||||
-export_type([session/0]).
|
||||
|
||||
%% 心跳包监测机制
|
||||
-define(PING_TICKER, 15000).
|
||||
@ -27,7 +27,7 @@
|
||||
%% 内部错误
|
||||
-define(NAK_INTERNAL_FAULT, 5).
|
||||
|
||||
-record(state, {
|
||||
-record(session, {
|
||||
status = initialized :: initialized | registered,
|
||||
%% 心跳间隔
|
||||
heartbeat_sec = 15,
|
||||
@ -52,15 +52,15 @@
|
||||
offline_cb :: undefined | fun()
|
||||
}).
|
||||
|
||||
-type state() :: #state{}.
|
||||
-type session() :: #session{}.
|
||||
|
||||
%%%===================================================================
|
||||
%%% API
|
||||
%%%===================================================================
|
||||
|
||||
-spec new(HeartbeatSec :: integer()) -> #state{}.
|
||||
-spec new(HeartbeatSec :: integer()) -> #session{}.
|
||||
new(HeartbeatSec) ->
|
||||
#state{heartbeat_sec = HeartbeatSec}.
|
||||
#session{heartbeat_sec = HeartbeatSec}.
|
||||
|
||||
%% 测试规则函数
|
||||
test_rules(SrcIdentityId, DstIdentityId) when is_integer(SrcIdentityId), is_integer(DstIdentityId) ->
|
||||
@ -68,8 +68,8 @@ 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(State :: #state{}) -> initialized | registered.
|
||||
state_name(#state{status = Status}) ->
|
||||
-spec state_name(Session :: #session{}) -> initialized | registered.
|
||||
state_name(#session{status = Status}) ->
|
||||
Status.
|
||||
|
||||
-spec welcome_packet(MaxPacketSize :: integer(), HeartbeatSec :: integer()) -> binary().
|
||||
@ -92,13 +92,13 @@ welcome_packet(MaxPacketSize, HeartbeatSec) ->
|
||||
}),
|
||||
<<?PACKET_WELCOME, WelcomePkt/binary>>.
|
||||
|
||||
-spec handle_frame(Frame :: binary(), State :: #state{}) ->
|
||||
{ok, NewState :: #state{}, Packets :: [binary()]} |
|
||||
{stop, Reason :: term(), NewState :: #state{}, Packets :: [binary()]}.
|
||||
handle_frame(<<?PACKET_REGISTER_SUPER, Body/binary>>, State = #state{status = initialized}) ->
|
||||
handle_register_super(Body, State);
|
||||
-spec handle_frame(Frame :: binary(), Session :: #session{}) ->
|
||||
{ok, NewSession :: #session{}, Packets :: [binary()]} |
|
||||
{stop, Reason :: term(), NewSession :: #session{}, Packets :: [binary()]}.
|
||||
handle_frame(<<?PACKET_REGISTER_SUPER, Body/binary>>, Session = #session{status = initialized}) ->
|
||||
handle_register_super(Body, Session);
|
||||
|
||||
handle_frame(<<?PACKET_QUERY_INFO, Body/binary>>, State = #state{status = registered, network_pid = NetworkPid, mac = SrcMac}) when is_pid(NetworkPid) ->
|
||||
handle_frame(<<?PACKET_QUERY_INFO, Body/binary>>, Session = #session{status = registered, network_pid = NetworkPid, mac = SrcMac}) when is_pid(NetworkPid) ->
|
||||
#'SDLQueryInfo'{dst_mac = DstMac} = sdlan_pb:decode_msg(Body, 'SDLQueryInfo'),
|
||||
case sdlan_network:peer_info(NetworkPid, SrcMac, DstMac) of
|
||||
error ->
|
||||
@ -110,7 +110,7 @@ handle_frame(<<?PACKET_QUERY_INFO, Body/binary>>, State = #state{status = regist
|
||||
v4_info = undefined,
|
||||
v6_info = undefined
|
||||
}),
|
||||
{ok, State, [<<?PACKET_PEER_INFO, EmptyResponse/binary>>]};
|
||||
{ok, 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,11 +124,11 @@ handle_frame(<<?PACKET_QUERY_INFO, Body/binary>>, State = #state{status = regist
|
||||
},
|
||||
v6_info = V6Info
|
||||
}),
|
||||
{ok, State, [<<?PACKET_PEER_INFO, PeerInfo/binary>>]}
|
||||
{ok, Session, [<<?PACKET_PEER_INFO, PeerInfo/binary>>]}
|
||||
end;
|
||||
|
||||
%% arp查询
|
||||
handle_frame(<<?PACKET_ARP_REQUEST, Body/binary>>, State = #state{status = registered, network_id = NetworkId, network_pid = NetworkPid}) when is_pid(NetworkPid) ->
|
||||
handle_frame(<<?PACKET_ARP_REQUEST, Body/binary>>, Session = #session{status = registered, network_id = NetworkId, network_pid = NetworkPid}) when is_pid(NetworkPid) ->
|
||||
#'SDLArpRequest'{target_ip = TargetIp, origin_ip = OriginIp, context = Context} = sdlan_pb:decode_msg(Body, 'SDLArpRequest'),
|
||||
case sdlan_network:arp_request(NetworkPid, TargetIp) of
|
||||
error ->
|
||||
@ -139,7 +139,7 @@ handle_frame(<<?PACKET_ARP_REQUEST, Body/binary>>, State = #state{status = regis
|
||||
origin_ip = OriginIp,
|
||||
context = Context
|
||||
}),
|
||||
{ok, State, [<<?PACKET_ARP_RESPONSE, EmptyArpResponsePkt/binary>>]};
|
||||
{ok, 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,10 +148,10 @@ handle_frame(<<?PACKET_ARP_REQUEST, Body/binary>>, State = #state{status = regis
|
||||
origin_ip = OriginIp,
|
||||
context = Context
|
||||
}),
|
||||
{ok, State, [<<?PACKET_ARP_RESPONSE, ArpResponsePkt/binary>>]}
|
||||
{ok, Session, [<<?PACKET_ARP_RESPONSE, ArpResponsePkt/binary>>]}
|
||||
end;
|
||||
|
||||
handle_frame(<<?PACKET_POLICY_REQUEST, Body/binary>>, State = #state{status = registered, network_pid = NetworkPid}) when is_pid(NetworkPid) ->
|
||||
handle_frame(<<?PACKET_POLICY_REQUEST, Body/binary>>, Session = #session{status = registered, network_pid = NetworkPid}) when is_pid(NetworkPid) ->
|
||||
Packets = maybe
|
||||
#'SDLPolicyRequest'{src_identity_id = SrcIdentityId, dst_identity_id = DstIdentityId, version = Version} ?= sdlan_pb:decode_msg(Body, 'SDLPolicyRequest'),
|
||||
|
||||
@ -169,10 +169,10 @@ handle_frame(<<?PACKET_POLICY_REQUEST, Body/binary>>, State = #state{status = re
|
||||
else _ ->
|
||||
[]
|
||||
end,
|
||||
{ok, State, Packets};
|
||||
{ok, Session, Packets};
|
||||
|
||||
%% 处理命令的响应逻辑
|
||||
handle_frame(<<?PACKET_COMMAND_ACK, Body/binary>>, State = #state{status = registered, pending_commands = PendingCommands}) ->
|
||||
handle_frame(<<?PACKET_COMMAND_ACK, Body/binary>>, Session = #session{status = registered, pending_commands = PendingCommands}) ->
|
||||
maybe
|
||||
CommandAck = sdlan_pb:decode_msg(Body, 'SDLCommandAck'),
|
||||
#'SDLCommandAck'{pkt_id = PktId} ?= CommandAck,
|
||||
@ -184,47 +184,47 @@ handle_frame(<<?PACKET_COMMAND_ACK, Body/binary>>, State = #state{status = regis
|
||||
false ->
|
||||
ok
|
||||
end,
|
||||
{ok, State#state{pending_commands = RestPendingCommands}, []}
|
||||
{ok, Session#session{pending_commands = RestPendingCommands}, []}
|
||||
else _ ->
|
||||
{ok, State, []}
|
||||
{ok, Session, []}
|
||||
end;
|
||||
|
||||
handle_frame(<<?PACKET_PING>>, State = #state{ping_counter = PingCounter, client_id = ClientId}) ->
|
||||
handle_frame(<<?PACKET_PING>>, Session = #session{ping_counter = PingCounter, client_id = ClientId}) ->
|
||||
logger:warning("[sdlan_session] get ping: ~p", [ClientId]),
|
||||
{ok, State#state{ping_counter = PingCounter + 1}, [<<?PACKET_PONG>>]};
|
||||
{ok, Session#session{ping_counter = PingCounter + 1}, [<<?PACKET_PONG>>]};
|
||||
|
||||
%% 取消注册
|
||||
handle_frame(<<?PACKET_UNREGISTER>>, State = #state{status = registered, client_id = ClientId, mac = Mac, network_pid = NetworkPid}) when is_pid(NetworkPid) ->
|
||||
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, State, []};
|
||||
{stop, normal, Session, []};
|
||||
|
||||
handle_frame(Frame, State) ->
|
||||
logger:notice("[sdlan_session] unexpected frame: ~p, status: ~p", [Frame, State#state.status]),
|
||||
{ok, State, []}.
|
||||
handle_frame(Frame, Session) ->
|
||||
logger:notice("[sdlan_session] unexpected frame: ~p, status: ~p", [Frame, Session#session.status]),
|
||||
{ok, Session, []}.
|
||||
|
||||
-spec handle_timeout(TimerRef :: reference(), State :: #state{}) ->
|
||||
{ok, NewState :: #state{}} | {stop, Reason :: term(), NewState :: #state{}}.
|
||||
handle_timeout(TimerRef, State = #state{ping_timer = TimerRef, client_id = ClientId, ping_counter = PingCounter}) ->
|
||||
-spec handle_timeout(TimerRef :: reference(), Session :: #session{}) ->
|
||||
{ok, NewSession :: #session{}} | {stop, Reason :: term(), NewSession :: #session{}}.
|
||||
handle_timeout(TimerRef, Session = #session{ping_timer = TimerRef, client_id = ClientId, ping_counter = PingCounter}) ->
|
||||
case PingCounter > 0 of
|
||||
true ->
|
||||
{ok, schedule_ping(State#state{ping_counter = 0, ping_timer = undefined})};
|
||||
{ok, schedule_ping(Session#session{ping_counter = 0, ping_timer = undefined})};
|
||||
false ->
|
||||
logger:debug("[sdlan_session] client_id: ~p, ping losted", [ClientId]),
|
||||
{stop, heartbeat_timeout, State#state{ping_counter = 0, ping_timer = undefined}}
|
||||
{stop, heartbeat_timeout, Session#session{ping_counter = 0, ping_timer = undefined}}
|
||||
end;
|
||||
handle_timeout(_TimerRef, State) ->
|
||||
{ok, State}.
|
||||
handle_timeout(_TimerRef, Session) ->
|
||||
{ok, Session}.
|
||||
|
||||
-spec send_event(Event :: binary(), State :: #state{}) -> {ok, NewState :: #state{}, Packets :: [binary()]} | {error, not_registered}.
|
||||
send_event(Event, State = #state{status = registered}) when is_binary(Event) ->
|
||||
{ok, State, [<<?PACKET_EVENT, Event/binary>>]};
|
||||
-spec send_event(Event :: binary(), Session :: #session{}) -> {ok, 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) ->
|
||||
{error, not_registered}.
|
||||
|
||||
-spec command(Ref :: reference(), ReceiverPid :: pid(), {Tag :: atom(), SubCommand :: any()}, State :: #state{}) ->
|
||||
{ok, NewState :: #state{}, Packets :: [binary()]} | {error, not_registered}.
|
||||
command(Ref, ReceiverPid, SubCommand, State = #state{status = registered, pkt_id = PktId, pending_commands = PendingCommands, client_id = ClientId})
|
||||
-spec command(Ref :: reference(), ReceiverPid :: pid(), {Tag :: atom(), SubCommand :: any()}, Session :: #session{}) ->
|
||||
{ok, 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'{
|
||||
pkt_id = PktId,
|
||||
@ -232,19 +232,19 @@ command(Ref, ReceiverPid, SubCommand, State = #state{status = registered, pkt_id
|
||||
}),
|
||||
logger:debug("[sdlan_session] client_id: ~p, will send Command: ~p", [ClientId, SubCommand]),
|
||||
|
||||
{ok, State#state{pkt_id = PktId + 1, pending_commands = maps:put(PktId, {Ref, ReceiverPid}, PendingCommands)},
|
||||
{ok, Session#session{pkt_id = PktId + 1, pending_commands = maps:put(PktId, {Ref, ReceiverPid}, PendingCommands)},
|
||||
[<<?PACKET_COMMAND, CommandPkt/binary>>]};
|
||||
command(_Ref, _ReceiverPid, _SubCommand, _State) ->
|
||||
{error, not_registered}.
|
||||
|
||||
-spec close(State :: #state{}) -> ok.
|
||||
close(#state{offline_cb = OfflineCb}) ->
|
||||
-spec close(Session :: #session{}) -> ok.
|
||||
close(#session{offline_cb = OfflineCb}) ->
|
||||
%% 触发客户端的离线逻辑
|
||||
is_function(OfflineCb) andalso OfflineCb(),
|
||||
ok.
|
||||
|
||||
-spec debug_info(State :: #state{}) -> map().
|
||||
debug_info(#state{
|
||||
-spec debug_info(Session :: #session{}) -> map().
|
||||
debug_info(#session{
|
||||
status = Status,
|
||||
client_id = ClientId,
|
||||
network_id = NetworkId,
|
||||
@ -267,7 +267,7 @@ debug_info(#state{
|
||||
%%% Internal functions
|
||||
%%%===================================================================
|
||||
|
||||
handle_register_super(Body, State) ->
|
||||
handle_register_super(Body, Session) ->
|
||||
#'SDLRegisterSuper'{
|
||||
client_id = ClientId, network_id = NetworkId, mac = Mac, ip = Ip, mask_len = MaskLen,
|
||||
hostname = HostName, pub_key = PubKey, access_token = AccessToken} = sdlan_pb:decode_msg(Body, 'SDLRegisterSuper'),
|
||||
@ -322,7 +322,7 @@ handle_register_super(Body, State) ->
|
||||
<<"status">> => 0
|
||||
})
|
||||
end,
|
||||
NState = schedule_ping(State#state{
|
||||
NSession = schedule_ping(Session#session{
|
||||
status = registered,
|
||||
network_id = NetworkId,
|
||||
network_pid = NetworkPid,
|
||||
@ -331,17 +331,17 @@ handle_register_super(Body, State) ->
|
||||
ip = Ip,
|
||||
offline_cb = OfflineCb
|
||||
}),
|
||||
{ok, NState, [<<?PACKET_REGISTER_SUPER_ACK, RegisterSuperAck/binary>>]};
|
||||
{ok, NSession, [<<?PACKET_REGISTER_SUPER_ACK, RegisterSuperAck/binary>>]};
|
||||
undefined ->
|
||||
logger:warning("[sdlan_session] client_id: ~p, register get error: network not found", [ClientId]),
|
||||
{stop, normal, State, [register_nak_reply(?NAK_INTERNAL_FAULT, <<"Internal Error">>)]}
|
||||
{stop, 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, State, [register_nak_reply(Code, Message)]};
|
||||
{stop, 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, State, [register_nak_reply(?NAK_NETWORK_FAULT, <<"Network Error">>)]}
|
||||
{stop, normal, Session, [register_nak_reply(?NAK_NETWORK_FAULT, <<"Network Error">>)]}
|
||||
end.
|
||||
|
||||
-spec register_nak_reply(ErrorCode :: integer(), ErrorMsg :: binary()) -> binary().
|
||||
@ -361,8 +361,8 @@ get_rules(SrcIdentityId, DstIdentityId) when is_integer(SrcIdentityId), is_integ
|
||||
DstPolicyIds = identity_policy_ets:get_policies(DstIdentityId),
|
||||
rule_ets:get_rules(SrcPolicyIds, DstPolicyIds).
|
||||
|
||||
schedule_ping(State = #state{heartbeat_sec = HeartbeatSec}) ->
|
||||
State#state{ping_timer = erlang:start_timer(heartbeat_ms(HeartbeatSec), self(), ping_ticker)}.
|
||||
schedule_ping(Session = #session{heartbeat_sec = HeartbeatSec}) ->
|
||||
Session#session{ping_timer = erlang:start_timer(heartbeat_ms(HeartbeatSec), self(), ping_ticker)}.
|
||||
|
||||
heartbeat_ms(HeartbeatSec) when is_integer(HeartbeatSec), HeartbeatSec > 0 ->
|
||||
HeartbeatSec * 1000;
|
||||
|
||||
@ -27,7 +27,7 @@
|
||||
max_packet_size = 16384,
|
||||
heartbeat_sec = 15,
|
||||
|
||||
session :: sdlan_session:state(),
|
||||
session :: sdlan_session:session(),
|
||||
|
||||
frames_recv = 0,
|
||||
bytes_recv = 0,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user