platybelodon

-module(platybelodon).
-behaviour(gen_server).
-include("stomp_frame.hrl").
-include_lib("kernel/include/logger.hrl").

-export([start_link/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3, handle_continue/2]).

-define(DEFAULT_HEARTBEAT_RECEIVE_TIMEOUT_TOLERANCE, 5000).
-define(DEFAULT_HEARTBEAT_SEND_MARGIN, 1000).

start_link(Opts) ->
    gen_server:start_link(?MODULE, Opts, []).

init(Opts = #{host := _Host, port := _Port, callback_handler := _CallbackHandler}) -> 
    {ok, maps:merge(Opts, #{heartbeats => [0,0], is_connected => false}), {continue, connect}}.

handle_call(Msg, _From, State) ->
    logger:warning("Unexpected call: ~p~n",[Msg]),
    {noreply, State}.

handle_cast(Msg, State) ->
    logger:warning("Unexpected cast: ~p~n",[Msg]),
    {noreply, State}.

handle_info(DisconnectMessage = {platybelodon, disconnect, _Type}, State = #{is_connected := true, callback_handler := CallbackHandler}) ->
    CallbackHandler ! DisconnectMessage,
    {noreply, State};

handle_info({platybelodon, disconnect, _Type}, State) ->
    {noreply, State};

handle_info({platybelodon, frames, [Frame = #stomp_frame{command = connected}]}, State = #{callback_handler := CallbackHandler, sender_pid := SenderPid, receiver_pid:= ReceiverPid, heartbeats := ClientHeartbeats}) ->
    ServerHeartbeats = heart_beat_from_frame(Frame),
    [HeartbeatSend, HeartbeatReceive] = NegotiatedHeartbeats = negotiate_heartbeats(ClientHeartbeats, ServerHeartbeats),

    ReceiverPid ! {set_heartbeat, receive_timeout(HeartbeatReceive)},
    SenderPid ! {set_heartbeat, send_timeout(HeartbeatSend)},

    CallbackHandler ! {platybelodon, frames, {[Frame], SenderPid}},
    {noreply, State#{heartbeats := NegotiatedHeartbeats}};

handle_info({platybelodon, frames, []}, State) ->
    {noreply, State};

handle_info({platybelodon, frames, Frames}, State = #{callback_handler := CallbackHandler, sender_pid := SenderPid}) ->
    CallbackHandler ! {platybelodon, frames, {Frames, SenderPid}},
    {noreply, State};

handle_info(Frame = #stomp_frame{command=_Command}, State = #{sender_pid := SenderPid}) when _Command=:=stomp; _Command=:=connect ->
    Heartbeats = heart_beat_from_frame(Frame),
    SenderPid ! Frame,
    {noreply, State#{heartbeats := Heartbeats}};

handle_info(Msg, State) ->
    logger:warning("Unexpected message: ~p~n",[Msg]),
    {noreply, State}.

handle_continue(connect, State = #{host := Host, port := Port, callback_handler := CallbackHandler}) ->
    case gen_tcp:connect(Host, Port, [binary, {active, false}, {keepalive, true}], 10_000) of
        {ok, Socket} ->
            inet:setopts(Socket,
            [
                {buffer, 16#400_000},
                {recbuf, 16#400_000},
                {sndbuf, 16#400_000},
                {send_timeout, 10#10_000},
                {send_timeout_close, true}
            ]),
            SenderPid = platybelodon_network:start_send_link(self(), Socket),
            ReceiverPid = platybelodon_network:start_receive_link(self(), Socket),

            gen_tcp:controlling_process(Socket, ReceiverPid),

            CallbackHandler ! {platybelodon, connect, {ok, self()}},

            {noreply, maps:merge(State, #{
                is_connected => true,
                socket => Socket,
                sender_pid => SenderPid,
                receiver_pid => ReceiverPid
            })};

        {error, Error} ->
            CallbackHandler ! {platybelodon, connect, {error, Error}},
            {noreply, State}
    end.

terminate(normal, _State) ->
    ok.

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

heart_beat_from_frame(#stomp_frame{command = _Command, headers = Headers}) when _Command =:= connect; _Command =:= connected; _Command =:= stomp ->
    case platybelodon_frame:headers_to_map(Headers) of
        #{<<"heart-beat">> := HeartbeatsRaw} ->
            [_X, _Y] = [binary_to_integer(Heartbeat) || Heartbeat <- binary:split(HeartbeatsRaw, <<",">>)];
        _ ->
            [0, 0]
    end.

negotiate_heartbeats([Cx, Cy], [Sx, Sy]) ->
    [heartbeat_max(Cx, Sy), heartbeat_max(Cy, Sx)].

heartbeat_max(_, 0) ->
    0;

heartbeat_max(0, _) ->
    0;

heartbeat_max(A, B) ->
    max(A, B).

receive_timeout(0) ->
    infinity;

receive_timeout(Cy) ->
    Cy + ?DEFAULT_HEARTBEAT_RECEIVE_TIMEOUT_TOLERANCE.

send_timeout(0) ->
    infinity;

send_timeout(Cx) when Cx > ?DEFAULT_HEARTBEAT_SEND_MARGIN + 100 ->
    Cx - ?DEFAULT_HEARTBEAT_SEND_MARGIN;

send_timeout(_Cx) ->
    500.