enet

-module(enet_channel_srv).
-behaviour(gen_server).

-include("enet_commands.hrl").

% API
-export([
    start_link/2,
    stop/1,
    set_worker/2,
    recv_unsequenced/2,
    send_unsequenced/2,
    recv_unreliable/2,
    send_unreliable/2,
    recv_reliable/2,
    send_reliable/2
]).

% required gen_server callbacks
-export([
    init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2,
    code_change/3
]).

% enet_uint16 ENetPeer::outgoingReliableSequenceNumber
-define(ENET_MAX_SEQ_INDEX, 65536).
% ENET_PEER_RELIABLE_WINDOW_SIZE = 0x1000,
-define(ENET_PEER_RELIABLE_WINDOW_SIZE, 16#1000).

% records
-record(state, {
    id,
    peer,
    worker,
    incoming_reliable_sequence_number = 1,
    incoming_unreliable_sequence_number = 1,
    outgoing_reliable_sequence_number = 1,
    outgoing_unreliable_sequence_number = 1,
    %% reliableWindows [ENET_PEER_RELIABLE_WINDOWS] (uint16 * 16 = 32 bytes)
    reliable_window = [],
    sys_parent,
    sys_debug
}).

-spec start_link(term(), term()) -> Result when
    Result ::
        {ok, Pid :: pid()}
        | ignore
        | {error, Reason :: term()}.
start_link(ID, Peer) ->
    % start an anonymous gen_server
    gen_server:start_link(?MODULE, [ID, Peer], []).

-spec stop(pid()) -> ok.
stop(Channel) ->
    gen_server:stop(Channel).

set_worker(Channel, Worker) ->
    gen_server:cast(Channel, {set_worker, Worker}).

recv_unsequenced(Channel, {H, C}) ->
    gen_server:cast(Channel, {recv_unsequenced, {H, C}}).

-spec send_unsequenced(pid(), term()) -> ok.
send_unsequenced(Channel, Data) ->
    gen_server:cast(Channel, {send_unsequenced, Data}).

recv_unreliable(Channel, {H, C}) ->
    gen_server:cast(Channel, {recv_unreliable, {H, C}}).

-spec send_unreliable(pid(), term()) -> ok.
send_unreliable(Channel, Data) ->
    gen_server:cast(Channel, {send_unreliable, Data}).

recv_reliable(Channel, {H, C}) ->
    gen_server:cast(Channel, {recv_reliable, {H, C}}).

-spec send_reliable(pid(), term()) -> ok.
send_reliable(Channel, Data) ->
    gen_server:cast(Channel, {send_reliable, Data}).

% gen_server callbacks

init([ID, Peer]) ->
    S0 = #state{
        id = ID,
        peer = Peer
    },
    {ok, S0}.

handle_call(_Msg, _From, S0) ->
    {reply, ok, S0}.

handle_cast({set_worker, Worker}, S0) ->
    S1 = S0#state{worker = Worker},
    {noreply, S1};
handle_cast(
    {recv_unsequenced, {#command_header{unsequenced = 1}, C = #unsequenced{}}},
    S0
) ->
    Worker = S0#state.worker,
    ID = S0#state.id,
    Worker ! {enet, ID, C},
    {noreply, S0};
handle_cast({send_unsequenced, Data}, S0) ->
    ID = S0#state.id,
    Peer = S0#state.peer,
    {H, C} = enet_command:send_unsequenced(ID, Data),
    ok = enet_peer:send_command(Peer, {H, C}),
    {noreply, S0};
handle_cast({recv_unreliable, {#command_header{}, C = #unreliable{sequence_number = N}}}, S0) ->
    Expected = S0#state.incoming_unreliable_sequence_number,
    if
        % The guard is more complex because we need to account for wrapped
        % sequence numbers. 
        N < Expected; N - Expected >= ?ENET_MAX_SEQ_INDEX/2  ->
            %% Data is old - drop it and continue.
            logger:debug("Discard outdated packet. Recv: ~p. Expect: ~p", [N, Expected]),
            {noreply, S0};
        % N is equal to or greater than the expected packet. Dispatch it.
        true ->
            Worker = S0#state.worker,
            ID = S0#state.id,
            Worker ! {enet, ID, C},
            NextSeq = maybe_wrap(N+1),
            S1 = S0#state{incoming_unreliable_sequence_number = NextSeq},
            {noreply, S1}
    end;
handle_cast({send_unreliable, Data}, S0) ->
    ID = S0#state.id,
    Peer = S0#state.peer,
    N = S0#state.outgoing_unreliable_sequence_number,
    {H, C} = enet_command:send_unreliable(ID, N, Data),
    ok = enet_peer:send_command(Peer, {H, C}),
    S1 = S0#state{outgoing_reliable_sequence_number = maybe_wrap(N + 1)},
    {noreply, S1};
handle_cast({recv_reliable, _Data}, S0 = #state{reliable_window = W}) when
    length(W) > ?ENET_PEER_RELIABLE_WINDOW_SIZE
->
    {stop, out_of_sync, S0};
handle_cast({recv_reliable, {#command_header{reliable_sequence_number = N}, C = #reliable{}}}, S0) ->
    Expected = S0#state.incoming_reliable_sequence_number,
    if
        % These guards are more complex because we need to account for wrapped
        % sequence numbers. 
        N > Expected; N - Expected =< -?ENET_MAX_SEQ_INDEX/2  ->
            logger:debug("Buffer ahead-of-sequence packet. Recv: ~p. Expect: ~p.", [N, Expected]),
            ReliableWindow0 = S0#state.reliable_window,
            S1 = S0#state{reliable_window = [{N, C} | ReliableWindow0]},
            {noreply, S1};
        N < Expected; N - Expected >= ?ENET_MAX_SEQ_INDEX/2  ->
            logger:debug("Discard outdated packet. Recv: ~p. Expect: ~p", [N, Expected]),
            {noreply, S0};
        % Must be equal, and so dispatch:
        true ->
            ID = S0#state.id,
            % Dispatch this packet
            Worker = S0#state.worker,
            Worker ! {enet, ID, C},
            % Dispatch any buffered packets
            Window = S0#state.reliable_window,
            SortedWindow = wrapped_sort(Window),
            {NextSeq, NewWindow} = dispatch(N, SortedWindow, ID, Worker),
            S1 = S0#state{incoming_reliable_sequence_number = NextSeq, reliable_window = NewWindow},
            {noreply, S1}
    end;
handle_cast({send_reliable, Data}, S0) ->
    ID = S0#state.id,
    Peer = S0#state.peer,
    N = S0#state.outgoing_reliable_sequence_number,
    {H, C} = enet_command:send_reliable(ID, N, Data),
    ok = enet_peer:send_command(Peer, {H, C}),
    S1 = S0#state{outgoing_reliable_sequence_number = maybe_wrap(N + 1)},
    {noreply, S1};
handle_cast(Msg, S0) ->
    logger:debug("Unhandled message: ~p", [Msg]),
    logger:debug("Current state: ~p", [S0]),
    {noreply, S0}.

handle_info(Msg, S0) ->
    logger:debug("Got unhandled msg: ~p", [Msg]),
    {noreply, S0}.

terminate(_Reason, _S0) ->
    ok.

code_change(_OldVsn, S0, _Extra) ->
    {ok, S0}.

% Internal
%

-spec dispatch(pos_integer(), list(), pos_integer(), pid()) ->
    {pos_integer(), list()}.
dispatch(CurSeq, [], _ChannelID, _Worker) ->
    NextSeq = maybe_wrap(CurSeq + 1),
    {NextSeq, []};
dispatch(CurSeq, Window = [{Seq1, D1} | RemainingWindow], ChannelID, Worker) ->
    % If the buffered item comes immediately after the current sequence number,
    % dispatch the next packet.
    NextSeq = maybe_wrap(CurSeq + 1),
    case NextSeq == Seq1 of
        true ->
            % Dispatch the packet
            logger:debug("Dispatching queued packet ~p", [Seq1]),
            Worker ! {enet, ChannelID, D1},
            dispatch(NextSeq, RemainingWindow, ChannelID, Worker);
        _ ->
            % The first packet in the window is not the one we're looking for,
            % so just return.
            {NextSeq, Window}
    end.

maybe_wrap(Seq) ->
    % Must wrap at 16-bits.
    (Seq) rem ?ENET_MAX_SEQ_INDEX.

wrapped_sort(List) ->
    % Keysort while preserving order through 16-bit integer wrapping.
    F = fun({A, _}, {B, _}) ->
        Compare = B - A,
        if
            Compare > 0, Compare =< ?ENET_MAX_SEQ_INDEX / 2 ->
                true;
            Compare < 0, Compare =< -?ENET_MAX_SEQ_INDEX / 2 ->
                true;
            true ->
                false
        end
    end,
    lists:sort(F, List).