-module(enet_peer). -behaviour(gen_statem). -include("enet_peer.hrl"). -include("enet_commands.hrl"). -include("enet_protocol.hrl"). %% API -export([ start_link/2, disconnect/1, disconnect_now/1, channels/1, channel/2, recv_incoming_packet/4, send_command/2, get_connect_id/1, get_mtu/1, get_name/1, get_peer_id/1, get_pool/1, get_pool_worker_id/1 ]). %% gen_statem callbacks -export([ init/1, callback_mode/0, terminate/3, code_change/4 ]). %% gen_statem state functions -export([ connecting/3, acknowledging_connect/3, acknowledging_verify_connect/3, verifying_connect/3, connected/3, disconnecting/3 ]). -record(state, { local_port, ip, port, remote_peer_id = undefined, peer_id, incoming_session_id = 16#FF, outgoing_session_id = 16#FF, incoming_bandwidth = 0, outgoing_bandwidth = 0, window_size = ?MAX_WINDOW_SIZE, packet_throttle_interval = ?PEER_PACKET_THROTTLE_INTERVAL, packet_throttle_acceleration = ?PEER_PACKET_THROTTLE_ACCELERATION, packet_throttle_deceleration = ?PEER_PACKET_THROTTLE_DECELERATION, outgoing_reliable_sequence_number = 1, incoming_unsequenced_group = 0, outgoing_unsequenced_group = 1, unsequenced_window = 0, connect_id, host, channel_count, channels, worker, connect_fun }). %%============================================================== %% Connection handshake %%============================================================== %% %% %% state client server state %% (init) * %% | connect %% |------------------->* (init) %% 'connecting' | | 'acknowledging connect' %% | ack connect | %% |<-------------------| %% 'acknowledging | | %% verify connect' | | %% | verify connect | %% |<-------------------| %% | | 'verifying connect' %% | ack verify connect | %% |------------------->| %% 'connected' | | 'connected' %% | | %% %% %%============================================================== %%============================================================== %% Disconnect procedure %%============================================================== %% %% %% state client server state %% peer peer %% | | %% 'connected' | | 'connected' %% | disconnect | %% |------------------->| %% 'disconnecting' | | %% | ack disconnect | %% |<-------------------| %% (exit) | | (exit) %% * * %% %% %%============================================================== %%%=================================================================== %%% API %%%=================================================================== start_link(LocalPort, Peer) -> gen_statem:start_link(?MODULE, [LocalPort, Peer], []). disconnect(Peer) -> gen_statem:cast(Peer, disconnect). disconnect_now(Peer) -> gen_statem:cast(Peer, disconnect_now). channels(Peer) -> gen_statem:call(Peer, channels). channel(Peer, ID) -> gen_statem:call(Peer, {channel, ID}). recv_incoming_packet(Peer, FromIP, SentTime, Packet) -> gen_statem:cast(Peer, {incoming_packet, FromIP, SentTime, Packet}). send_command(Peer, {H, C}) -> gen_statem:cast(Peer, {outgoing_command, {H, C}}). get_connect_id(Peer) -> gproc:get_value({p, l, connect_id}, Peer). get_mtu(Peer) -> gproc:get_value({p, l, mtu}, Peer). get_name(Peer) -> gproc:get_value({p, l, name}, Peer). get_peer_id(Peer) -> gproc:get_value({p, l, peer_id}, Peer). get_pool(Peer) -> gen_statem:call(Peer, pool). get_pool_worker_id(Peer) -> gen_statem:call(Peer, pool_worker_id). %%%=================================================================== %%% gen_statem callbacks %%%=================================================================== init([LocalPort, P = #enet_peer{handshake_flow = local}]) -> %% %% The client application wants to connect to a remote peer. %% %% - Send a Connect command to the remote peer (use peer ID) %% - Start in the 'connecting' state %% #enet_peer{ peer_id = PeerID, ip = IP, port = Port, name = Ref, host = Host, channels = N, connect_fun = ConnectFun } = P, enet_pool:connect_peer(LocalPort, Ref), gproc:reg({n, l, {enet_peer, Ref}}), gproc:reg({p, l, name}, Ref), gproc:reg({p, l, peer_id}, PeerID), S = #state{ host = Host, local_port = LocalPort, ip = IP, port = Port, peer_id = PeerID, channel_count = N, connect_fun = ConnectFun }, {ok, connecting, S}; init([LocalPort, P = #enet_peer{handshake_flow = remote}]) -> %% %% A remote peer wants to connect to the client application. %% %% - Start in the 'acknowledging_connect' state %% - Handle the received Connect command %% #enet_peer{ peer_id = PeerID, ip = IP, port = Port, name = Ref, host = Host, connect_fun = ConnectFun } = P, enet_pool:connect_peer(LocalPort, Ref), gproc:reg({n, l, {enet_peer, Ref}}), gproc:reg({p, l, name}, Ref), gproc:reg({p, l, peer_id}, PeerID), S = #state{ host = Host, local_port = LocalPort, ip = IP, port = Port, peer_id = PeerID, connect_fun = ConnectFun }, {ok, acknowledging_connect, S}. callback_mode() -> [state_functions, state_enter]. %%% %%% Connecting state %%% connecting(enter, _OldState, S) -> %% %% Sending the initial Connect command. %% #state{ host = Host, channel_count = ChannelCount, ip = IP, port = Port, peer_id = PeerID, incoming_session_id = IncomingSessionID, outgoing_session_id = OutgoingSessionID, packet_throttle_interval = PacketThrottleInterval, packet_throttle_acceleration = PacketThrottleAcceleration, packet_throttle_deceleration = PacketThrottleDeceleration, outgoing_reliable_sequence_number = SequenceNr } = S, IncomingBandwidth = enet_host:get_incoming_bandwidth(Host), OutgoingBandwidth = enet_host:get_outgoing_bandwidth(Host), MTU = enet_host:get_mtu(Host), gproc:reg({p, l, mtu}, MTU), <> = crypto:strong_rand_bytes(4), {ConnectH, ConnectC} = enet_command:connect( PeerID, IncomingSessionID, OutgoingSessionID, ChannelCount, MTU, IncomingBandwidth, OutgoingBandwidth, PacketThrottleInterval, PacketThrottleAcceleration, PacketThrottleDeceleration, ConnectID, SequenceNr ), HBin = enet_protocol_encode:command_header(ConnectH), CBin = enet_protocol_encode:command(ConnectC), Data = [HBin, CBin], {sent_time, SentTime} = enet_host:send_outgoing_commands(Host, Data, IP, Port), ChannelID = 16#FF, ConnectTimeout = make_resend_timer( ChannelID, SentTime, SequenceNr, ?PEER_TIMEOUT_MINIMUM, Data ), NewS = S#state{ outgoing_reliable_sequence_number = SequenceNr + 1, connect_id = ConnectID }, {keep_state, NewS, [ConnectTimeout]}; connecting(cast, {incoming_command, {H, C = #acknowledge{}}}, S) -> %% %% Received an Acknowledge command in the 'connecting' state. %% %% - Verify that the acknowledge is correct %% - Change state to 'acknowledging_verify_connect' %% #command_header{channel_id = ChannelID} = H, #acknowledge{ received_reliable_sequence_number = SequenceNumber, received_sent_time = SentTime } = C, CanceledTimeout = cancel_resend_timer(ChannelID, SentTime, SequenceNumber), {next_state, acknowledging_verify_connect, S, [CanceledTimeout]}; connecting({timeout, {_ChannelID, _SentTime, _SequenceNumber}}, _, S) -> logger:debug("connection timeout"), {stop, timeout, S}; connecting(EventType, EventContent, S) -> handle_event(EventType, EventContent, S). %%% %%% Acknowledging Connect state %%% acknowledging_connect(enter, _OldState, S) -> {keep_state, S}; acknowledging_connect(cast, {incoming_command, {_H, C = #connect{}}}, S) -> %% %% Received a Connect command. %% %% - Verify that the data is sane (TODO) %% - Send a VerifyConnect command (use peer ID) %% - Start in the 'verifying_connect' state %% #connect{ outgoing_peer_id = RemotePeerID, incoming_session_id = _IncomingSessionID, outgoing_session_id = _OutgoingSessionID, mtu = MTU, window_size = WindowSize, channel_count = ChannelCount, incoming_bandwidth = IncomingBandwidth, outgoing_bandwidth = OutgoingBandwidth, packet_throttle_interval = PacketThrottleInterval, packet_throttle_acceleration = PacketThrottleAcceleration, packet_throttle_deceleration = PacketThrottleDeceleration, connect_id = ConnectID, data = _Data } = C, #state{ host = Host, ip = IP, port = Port, peer_id = PeerID, incoming_session_id = IncomingSessionID, outgoing_session_id = OutgoingSessionID, outgoing_reliable_sequence_number = SequenceNr } = S, gproc:reg({p, l, mtu}, MTU), HostChannelLimit = enet_host:get_channel_limit(Host), HostIncomingBandwidth = enet_host:get_incoming_bandwidth(Host), HostOutgoingBandwidth = enet_host:get_outgoing_bandwidth(Host), {VCH, VCC} = enet_command:verify_connect( C, PeerID, IncomingSessionID, OutgoingSessionID, HostChannelLimit, HostIncomingBandwidth, HostOutgoingBandwidth, SequenceNr ), HBin = enet_protocol_encode:command_header(VCH), CBin = enet_protocol_encode:command(VCC), Data = [HBin, CBin], {sent_time, SentTime} = enet_host:send_outgoing_commands(Host, Data, IP, Port, RemotePeerID), ChannelID = 16#FF, VerifyConnectTimeout = make_resend_timer( ChannelID, SentTime, SequenceNr, ?PEER_TIMEOUT_MINIMUM, Data ), NewS = S#state{ remote_peer_id = RemotePeerID, %% channels = Channels, connect_id = ConnectID, incoming_bandwidth = IncomingBandwidth, outgoing_bandwidth = OutgoingBandwidth, window_size = WindowSize, packet_throttle_interval = PacketThrottleInterval, packet_throttle_acceleration = PacketThrottleAcceleration, packet_throttle_deceleration = PacketThrottleDeceleration, outgoing_reliable_sequence_number = SequenceNr + 1, channel_count = ChannelCount }, {next_state, verifying_connect, NewS, [VerifyConnectTimeout]}; acknowledging_connect({timeout, {_ChannelID, _SentTime, _SequenceNr}}, _, S) -> logger:debug("acknowledgment timeout"), {stop, timeout, S}; acknowledging_connect(EventType, EventContent, S) -> handle_event(EventType, EventContent, S). %%% %%% Acknowledging Verify Connect state %%% acknowledging_verify_connect(enter, _OldState, S) -> {keep_state, S}; acknowledging_verify_connect( cast, {incoming_command, {_H, C = #verify_connect{}}}, S ) -> %% %% Received a Verify Connect command in the 'acknowledging_verify_connect' %% state. %% %% - Verify that the data is correct %% - Add the remote peer ID to the Peer Table %% - Notify worker that we are connected %% - Change state to 'connected' %% #verify_connect{ outgoing_peer_id = RemotePeerID, incoming_session_id = _IncomingSessionID, outgoing_session_id = _OutgoingSessionID, mtu = RemoteMTU, window_size = WindowSize, channel_count = RemoteChannelCount, incoming_bandwidth = IncomingBandwidth, outgoing_bandwidth = OutgoingBandwidth, packet_throttle_interval = ThrottleInterval, packet_throttle_acceleration = ThrottleAcceleration, packet_throttle_deceleration = ThrottleDeceleration, connect_id = ConnectID } = C, %% %% TODO: Calculate and validate Session IDs %% #state{channel_count = LocalChannelCount} = S, LocalMTU = get_mtu(self()), case S of #state{ %% --- %% Fields below are matched against the values received in %% the Verify Connect command. %% --- window_size = WindowSize, incoming_bandwidth = IncomingBandwidth, outgoing_bandwidth = OutgoingBandwidth, packet_throttle_interval = ThrottleInterval, packet_throttle_acceleration = ThrottleAcceleration, packet_throttle_deceleration = ThrottleDeceleration, connect_id = ConnectID %% --- } when LocalChannelCount =:= RemoteChannelCount, LocalMTU =:= RemoteMTU -> NewS = S#state{remote_peer_id = RemotePeerID}, {next_state, connected, NewS}; _Mismatch -> {stop, connect_verification_failed, S} end; acknowledging_verify_connect(EventType, EventContent, S) -> handle_event(EventType, EventContent, S). %%% %%% Verifying Connect state %%% verifying_connect(enter, _OldState, S) -> {keep_state, S}; verifying_connect(cast, {incoming_command, {H, C = #acknowledge{}}}, S) -> %% %% Received an Acknowledge command in the 'verifying_connect' state. %% %% - Verify that the acknowledge is correct %% - Notify worker that a new peer has been connected %% - Change to 'connected' state %% #command_header{channel_id = ChannelID} = H, #acknowledge{ received_reliable_sequence_number = SequenceNumber, received_sent_time = SentTime } = C, CanceledTimeout = cancel_resend_timer(ChannelID, SentTime, SequenceNumber), {next_state, connected, S, [CanceledTimeout]}; verifying_connect(EventType, EventContent, S) -> handle_event(EventType, EventContent, S). %%% %%% Connected state %%% connected(enter, _OldState, S) -> #state{ local_port = LocalPort, ip = IP, port = Port, remote_peer_id = RemotePeerID, connect_id = ConnectID, channel_count = N, connect_fun = ConnectFun } = S, true = gproc:mreg(p, l, [ {connect_id, ConnectID}, {remote_host_port, Port}, {remote_peer_id, RemotePeerID} ]), ok = enet_disconnector:set_trigger(LocalPort, RemotePeerID, IP, Port), Channels = start_channels(N), PeerInfo = #{ ip => IP, port => Port, peer => self(), channels => Channels, connect_id => ConnectID }, case start_worker(ConnectFun, PeerInfo) of {error, Reason} -> {stop, {worker_init_error, Reason}, S}; {ok, Worker} -> _Ref = monitor(process, Worker), [ enet_channel_srv:set_worker(C, Worker) || C <- maps:values(Channels) ], NewS = S#state{ channels = Channels, worker = Worker }, SendTimeout = reset_send_timer(), RecvTimeout = reset_recv_timer(), {keep_state, NewS, [SendTimeout, RecvTimeout]} end; connected(cast, {incoming_command, {_H, #ping{}}}, S) -> %% %% Received PING. %% %% - Reset the receive-timer %% RecvTimeout = reset_recv_timer(), {keep_state, S, [RecvTimeout]}; connected(cast, {incoming_command, {H, C = #acknowledge{}}}, S) -> %% %% Received an Acknowledge command. %% %% - Verify that the acknowledge is correct %% - Reset the receive-timer %% #command_header{channel_id = ChannelID} = H, #acknowledge{ received_reliable_sequence_number = SequenceNumber, received_sent_time = SentTime } = C, CanceledTimeout = cancel_resend_timer(ChannelID, SentTime, SequenceNumber), RecvTimeout = reset_recv_timer(), {keep_state, S, [CanceledTimeout, RecvTimeout]}; connected(cast, {incoming_command, {_H, C = #bandwidth_limit{}}}, S) -> %% %% Received Bandwidth Limit command. %% %% - Set bandwidth limit %% - Reset the receive-timer %% #bandwidth_limit{ incoming_bandwidth = IncomingBandwidth, outgoing_bandwidth = OutgoingBandwidth } = C, #state{host = Host} = S, HostOutgoingBandwidth = enet_host:get_outgoing_bandwidth(Host), WSize = case {IncomingBandwidth, HostOutgoingBandwidth} of {0, 0} -> ?MAX_WINDOW_SIZE; {0, H} -> ?MIN_WINDOW_SIZE * H div ?PEER_WINDOW_SIZE_SCALE; {P, 0} -> ?MIN_WINDOW_SIZE * P div ?PEER_WINDOW_SIZE_SCALE; {P, H} -> ?MIN_WINDOW_SIZE * min(P, H) div ?PEER_WINDOW_SIZE_SCALE end, NewS = S#state{ incoming_bandwidth = IncomingBandwidth, outgoing_bandwidth = OutgoingBandwidth, window_size = max(?MIN_WINDOW_SIZE, min(?MAX_WINDOW_SIZE, WSize)) }, RecvTimeout = reset_recv_timer(), {keep_state, NewS, [RecvTimeout]}; connected(cast, {incoming_command, {_H, C = #throttle_configure{}}}, S) -> %% %% Received Throttle Configure command. %% %% - Set throttle configuration %% - Reset the receive-timer %% #throttle_configure{ packet_throttle_interval = Interval, packet_throttle_acceleration = Acceleration, packet_throttle_deceleration = Deceleration } = C, NewS = S#state{ packet_throttle_interval = Interval, packet_throttle_acceleration = Acceleration, packet_throttle_deceleration = Deceleration }, RecvTimeout = reset_recv_timer(), {keep_state, NewS, [RecvTimeout]}; connected(cast, {incoming_command, {H, C = #unsequenced{}}}, S) -> %% %% Received Send Unsequenced command. %% %% - Forward the command to the relevant channel %% - Reset the receive-timer %% #command_header{channel_id = ChannelID} = H, #state{channels = #{ChannelID := Channel}} = S, ok = enet_channel_srv:recv_unsequenced(Channel, {H, C}), RecvTimeout = reset_recv_timer(), {keep_state, S, [RecvTimeout]}; connected(cast, {incoming_command, {H, C = #unreliable{}}}, S) -> %% %% Received Send Unreliable command. %% %% - Forward the command to the relevant channel %% - Reset the receive-timer %% #command_header{channel_id = ChannelID} = H, #state{channels = #{ChannelID := Channel}} = S, ok = enet_channel_srv:recv_unreliable(Channel, {H, C}), RecvTimeout = reset_recv_timer(), {keep_state, S, [RecvTimeout]}; connected(cast, {incoming_command, {H, C = #reliable{}}}, S) -> %% %% Received Send Reliable command. %% %% - Forward the command to the relevant channel %% - Reset the receive-timer %% #command_header{channel_id = ChannelID} = H, #state{channels = #{ChannelID := Channel}} = S, ok = enet_channel_srv:recv_reliable(Channel, {H, C}), RecvTimeout = reset_recv_timer(), {keep_state, S, [RecvTimeout]}; connected(cast, {incoming_command, {_H, #disconnect{}}}, S) -> %% %% Received Disconnect command. %% %% - Notify worker application %% - Stop %% #state{ worker = Worker, local_port = LocalPort, ip = IP, port = Port, remote_peer_id = RemotePeerID, connect_id = ConnectID } = S, ok = enet_disconnector:unset_trigger(LocalPort, RemotePeerID, IP, Port), Worker ! {enet, disconnected, remote, self(), ConnectID}, {stop, normal, S}; connected(cast, {outgoing_command, {H, C = #unsequenced{}}}, S) -> %% %% Sending an Unsequenced, unreliable command. %% %% - TODO: Increment total data passed through peer %% - Increment outgoing_unsequenced_group %% - Set unsequenced_group on command to outgoing_unsequenced_group %% - Queue the command for sending %% - Reset the send-timer %% #state{ host = Host, ip = IP, port = Port, remote_peer_id = RemotePeerID, outgoing_unsequenced_group = Group } = S, C1 = C#unsequenced{group = Group}, HBin = enet_protocol_encode:command_header(H), CBin = enet_protocol_encode:command(C1), Data = [HBin, CBin], {sent_time, _SentTime} = enet_host:send_outgoing_commands(Host, Data, IP, Port, RemotePeerID), NewS = S#state{outgoing_unsequenced_group = Group + 1}, SendTimeout = reset_send_timer(), {keep_state, NewS, [SendTimeout]}; connected(cast, {outgoing_command, {H, C = #unreliable{}}}, S) -> %% %% Sending a Sequenced, unreliable command. %% %% - Forward the encoded command to the host %% - Reset the send-timer %% #state{ host = Host, ip = IP, port = Port, remote_peer_id = RemotePeerID } = S, HBin = enet_protocol_encode:command_header(H), CBin = enet_protocol_encode:command(C), Data = [HBin, CBin], {sent_time, _SentTime} = enet_host:send_outgoing_commands(Host, Data, IP, Port, RemotePeerID), SendTimeout = reset_send_timer(), {keep_state, S, [SendTimeout]}; connected(cast, {outgoing_command, {H, C = #reliable{}}}, S) -> %% %% Sending a Sequenced, reliable command. %% %% - Forward the encoded command to the host %% - Reset the send-timer %% #state{ host = Host, ip = IP, port = Port, remote_peer_id = RemotePeerID } = S, #command_header{ channel_id = ChannelID, reliable_sequence_number = SequenceNr } = H, HBin = enet_protocol_encode:command_header(H), CBin = enet_protocol_encode:command(C), Data = [HBin, CBin], {sent_time, SentTime} = enet_host:send_outgoing_commands(Host, Data, IP, Port, RemotePeerID), SendReliableTimeout = make_resend_timer( ChannelID, SentTime, SequenceNr, ?PEER_TIMEOUT_MINIMUM, Data ), SendTimeout = reset_send_timer(), {keep_state, S, [SendReliableTimeout, SendTimeout]}; connected(cast, disconnect, State) -> %% %% Disconnecting. %% %% - Queue a Disconnect command for sending %% - Change state to 'disconnecting' %% #state{ host = Host, ip = IP, port = Port, remote_peer_id = RemotePeerID } = State, {H, C} = enet_command:sequenced_disconnect(), HBin = enet_protocol_encode:command_header(H), CBin = enet_protocol_encode:command(C), Data = [HBin, CBin], enet_host:send_outgoing_commands(Host, Data, IP, Port, RemotePeerID), {next_state, disconnecting, State}; connected(cast, disconnect_now, State) -> %% %% Disconnecting immediately. %% %% - Stop %% {stop, normal, State}; connected({timeout, {ChannelID, SentTime, SequenceNr}}, Data, S) -> %% %% A resend-timer was triggered. %% %% - TODO: Keep track of number of resends %% - Resend the associated command %% - Reset the resend-timer %% - Reset the send-timer %% #state{ host = Host, ip = IP, port = Port, remote_peer_id = RemotePeerID } = S, enet_host:send_outgoing_commands(Host, Data, IP, Port, RemotePeerID), NewTimeout = make_resend_timer( ChannelID, SentTime, SequenceNr, ?PEER_TIMEOUT_MINIMUM, Data ), SendTimeout = reset_send_timer(), {keep_state, S, [NewTimeout, SendTimeout]}; connected({timeout, recv}, ping, S) -> %% %% The receive-timer was triggered. %% %% - Stop %% logger:debug("ping timeout"), {stop, normal, S}; connected({timeout, send}, ping, S) -> %% %% The send-timer was triggered. %% %% - Send a PING %% - Reset the send-timer %% #state{ host = Host, ip = IP, port = Port, remote_peer_id = RemotePeerID } = S, {H, C} = enet_command:ping(), HBin = enet_protocol_encode:command_header(H), CBin = enet_protocol_encode:command(C), Data = [HBin, CBin], {sent_time, _SentTime} = enet_host:send_outgoing_commands(Host, Data, IP, Port, RemotePeerID), SendTimeout = reset_send_timer(), {keep_state, S, [SendTimeout]}; connected(EventType, EventContent, S) -> handle_event(EventType, EventContent, S). %%% %%% Disconnecting state %%% disconnecting(enter, _OldState, S) -> #state{ local_port = LocalPort, ip = IP, port = Port, remote_peer_id = RemotePeerID } = S, ok = enet_disconnector:unset_trigger(LocalPort, RemotePeerID, IP, Port), {keep_state, S}; disconnecting(cast, {incoming_command, {_H, _C = #acknowledge{}}}, S) -> %% %% Received an Acknowledge command in the 'disconnecting' state. %% %% - Verify that the acknowledge is correct %% - Notify worker application %% - Stop %% #state{ worker = Worker, connect_id = ConnectID } = S, Worker ! {enet, disconnected, local, self(), ConnectID}, {stop, normal, S}; disconnecting(cast, {incoming_command, {_H, _C}}, S) -> {keep_state, S}; disconnecting(EventType, EventContent, S) -> handle_event(EventType, EventContent, S). %%% %%% terminate %%% terminate(_Reason, _StateName, #state{local_port = LocalPort}) -> Name = get_name(self()), enet_pool:disconnect_peer(LocalPort, Name), ok. %%% %%% code_change %%% code_change(_OldVsn, StateName, State, _Extra) -> {ok, StateName, State}. %%%=================================================================== %%% Internal functions %%%=================================================================== handle_event(cast, {incoming_packet, FromIP, SentTime, Packet}, S) -> %% %% Received an incoming packet of commands. %% %% - Split and decode the commands from the binary %% - Send the commands as individual events to ourselves %% #state{host = Host, port = Port} = S, {ok, Commands} = enet_protocol_decode:commands(Packet), lists:foreach( fun ({H = #command_header{please_acknowledge = 0}, C}) -> %% %% Received command that does not need to be acknowledged. %% %% - Send the command to self for handling %% gen_statem:cast(self(), {incoming_command, {H, C}}); ({H = #command_header{please_acknowledge = 1}, C}) -> %% %% Received a command that should be acknowledged. %% %% - Acknowledge the command %% - Send the command to self for handling %% {AckH, AckC} = enet_command:acknowledge(H, SentTime), HBin = enet_protocol_encode:command_header(AckH), CBin = enet_protocol_encode:command(AckC), RemotePeerID = case C of #connect{} -> C#connect.outgoing_peer_id; #verify_connect{} -> C#verify_connect.outgoing_peer_id; _ -> S#state.remote_peer_id end, {sent_time, _AckSentTime} = enet_host:send_outgoing_commands( Host, [HBin, CBin], FromIP, Port, RemotePeerID ), gen_statem:cast(self(), {incoming_command, {H, C}}) end, Commands ), {keep_state, S#state{ip = FromIP}}; handle_event({call, From}, channels, S) -> {keep_state, S, [{reply, From, S#state.channels}]}; handle_event({call, From}, pool, S) -> {keep_state, S, [{reply, From, S#state.local_port}]}; handle_event({call, From}, pool_worker_id, S) -> WorkerID = enet_pool:worker_id(S#state.local_port, get_name(self())), {keep_state, S, [{reply, From, WorkerID}]}; handle_event({call, From}, {channel, ID}, S) -> #state{channels = #{ID := Channel}} = S, {keep_state, S, [{reply, From, Channel}]}; handle_event(info, {'DOWN', _, process, O, _}, S = #state{worker = O}) -> {stop, worker_process_down, S}. start_channels(N) -> IDs = lists:seq(0, N - 1), Channels = lists:map( fun(ID) -> {ok, Channel} = enet_channel_srv:start_link(ID, self()), {ID, Channel} end, IDs ), maps:from_list(Channels). make_resend_timer(ChannelID, SentTime, SequenceNumber, Time, Data) -> {{timeout, {ChannelID, SentTime, SequenceNumber}}, Time, Data}. cancel_resend_timer(ChannelID, SentTime, SequenceNumber) -> {{timeout, {ChannelID, SentTime, SequenceNumber}}, cancel}. reset_recv_timer() -> {{timeout, recv}, 2 * ?PEER_PING_INTERVAL, ping}. reset_send_timer() -> {{timeout, send}, ?PEER_PING_INTERVAL, ping}. start_worker({Module, Fun, Args}, PeerInfo) -> erlang:apply(Module, Fun, Args ++ [PeerInfo]); start_worker(ConnectFun, PeerInfo) -> ConnectFun(PeerInfo).