anlicheng/sdlan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

简化广播逻辑

Browse Source
This commit is contained in:
anlicheng 2026-01-15 11:48:41 +08:00
parent e2fd4f18cb
commit 1ae54ab276
1 changed files with 51 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

83
apps/sdlan/src/sdlan_network.erl
View File

@ -353,14 +353,14 @@ handle_cast({forward, Sock, SrcMac, DstMac, Packet}, State = #state{network_id =
error ->
PacketBytes = byte_size(Packet),
%% 消息广播
maps:foreach(fun(Mac, #endpoint{hole = Hole}) ->
case {Mac =/= SrcMac, Hole} of
{true, #hole{peer = {NatIp, NatPort}}} ->
broadcast(fun(#endpoint{hole = Hole}) ->
case Hole of
#hole{peer = {NatIp, NatPort}} ->
gen_udp:send(Sock, NatIp, NatPort, Packet);
_ ->
ok
end
end, Endpoints),
end, [SrcMac], Endpoints),
%% client和stun之间必须有心跳机制保持nat映射可用，并且通过服务转发的udp包肯定可以到达对端的nat
lager:debug("[sdlan_network] broadcast data networkd_id: ~p, src_mac: ~p, dst_mac: ~p",
@ -401,7 +401,15 @@ handle_cast({update_hole, ClientId, Mac, Peer, NatType, V6Info}, State = #state{
mac = Mac,
ip = Ip
}),
broadcast(?PACKET_EVENT_NAT_CHANGED, NatChangedEvent, Mac, Endpoints);
broadcast(fun(#endpoint{channel_pid = ChannelPid}) ->
case is_process_alive(ChannelPid) of
true ->
sdlan_channel:send_event(ChannelPid, ?PACKET_EVENT_NAT_CHANGED, NatChangedEvent);
false ->
ok
end
end, [Mac], Endpoints);
false ->
ok
end,
@ -423,9 +431,22 @@ handle_info({timeout, _, flow_report_ticker}, State = #state{network_id = Networ
catch sdlan_api:network_forward_report(NetworkId, ForwardBytes),
{noreply, State#state{forward_bytes = 0}};
handle_info({'EXIT', _Pid, shutdown}, State = #state{network_id = NetworkId, endpoints = UsedMap}) ->
handle_info({'EXIT', _Pid, shutdown}, State = #state{network_id = NetworkId, endpoints = Endpoints}) ->
lager:warning("[sdlan_network] network: ~p, get shutdown message", [NetworkId]),
broadcast_shutdown(UsedMap),
NetworkShutdownEvent = sdlan_pb:encode_msg(#sdl_network_shutdown_event {
message = <<"Network shutdown">>
}),
broadcast(fun(#endpoint{channel_pid = ChannelPid}) ->
case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of
true ->
sdlan_channel:send_event(ChannelPid, ?PACKET_EVENT_NETWORK_SHUTDOWN, NetworkShutdownEvent),
sdlan_channel:stop(ChannelPid, normal);
false ->
ok
end
end, Endpoints),
{stop, shutdown, State};
%% Channel进程退出, hole里面的数据也需要清理
handle_info({'DOWN', _MRef, process, ChannelPid, Reason}, State = #state{network_id = NetworkId, endpoints = Endpoints}) ->
@ -442,12 +463,20 @@ handle_info({'DOWN', _MRef, process, ChannelPid, Reason}, State = #state{network
State :: #state{}) -> term()).
terminate(Reason, #state{network_id = NetworkId, endpoints = Endpoints}) ->
lager:debug("[sdlan_network] network: ~p, will terminate with reason: ~p", [NetworkId, Reason]),
broadcast_shutdown(Endpoints),
%% 整个网络下的设备都需要重新连接
maps:foreach(fun(_, #endpoint{channel_pid = ChannelPid, monitor_ref = MRef}) ->
NetworkShutdownEvent = sdlan_pb:encode_msg(#sdl_network_shutdown_event {
message = <<"Network shutdown">>
}),
broadcast(fun(#endpoint{channel_pid = ChannelPid, monitor_ref = MRef}) ->
is_reference(MRef) andalso demonitor(MRef),
is_process_alive(ChannelPid) andalso sdlan_channel:stop(ChannelPid, normal)
end, Endpoints),
case is_pid(ChannelPid) andalso is_process_alive(ChannelPid) of
true ->
sdlan_channel:send_event(ChannelPid, ?PACKET_EVENT_NETWORK_SHUTDOWN, NetworkShutdownEvent),
sdlan_channel:stop(ChannelPid, normal);
false ->
ok
end
end, Endpoints),
ok.
%% @private
@ -502,30 +531,20 @@ maybe_close_channel(#endpoint{channel_pid = ChannelPid0, monitor_ref = MRef0}) -
maybe_close_channel(_) ->
ok.
-spec broadcast(EventType :: integer(), Event :: binary(), ExcludeMac :: binary(), UsedMap :: map()) -> no_return().
broadcast(EventType, Event, ExcludeMac, UsedMap) when is_map(UsedMap), is_binary(ExcludeMac), is_integer(EventType), is_binary(Event) ->
maps:foreach(fun(Mac, #endpoint{channel_pid = ChannelPid}) ->
case is_process_alive(ChannelPid) andalso ExcludeMac /= Mac of
true ->
sdlan_channel:send_event(ChannelPid, EventType, Event);
false ->
ok
end
end, UsedMap).
-spec broadcast(Fun :: binary(), Endpoints :: map()) -> no_return().
broadcast(Fun, Endpoints) when is_function(Fun, 1), is_map(Endpoints) ->
broadcast(Fun, [], Endpoints).
broadcast_shutdown(UsedMap) when is_map(UsedMap) ->
maps:foreach(fun(_, #endpoint{channel_pid = ChannelPid}) ->
case is_process_alive(ChannelPid) of
-spec broadcast(Fun :: binary(), ExcludeMacs :: [binary()], Endpoints :: map()) -> no_return().
broadcast(Fun, ExcludeMacs, Endpoints) when is_function(Fun, 1), is_map(Endpoints), is_list(ExcludeMacs) ->
maps:foreach(fun(Mac, Endpoint) ->
case lists:member(Mac, ExcludeMacs) of
true ->
NetworkShutdownEvent = sdlan_pb:encode_msg(#sdl_network_shutdown_event {
message = <<"Network shutdown">>
}),
sdlan_channel:send_event(ChannelPid, ?PACKET_EVENT_NETWORK_SHUTDOWN, NetworkShutdownEvent),
sdlan_channel:stop(ChannelPid, normal);
ok;
false ->
ok
Fun(Endpoint)
end
end, UsedMap).
end, Endpoints).
%% 解析IpAddr: <<"192.168.172/24">>
-spec parse_ipaddr(IpAddr0 :: binary()) -> {IpAddr :: binary(), MaskLen :: integer()}.