i3_client

-module(i3_client_protocol).
-moduledoc false.
-export([connect/1, checkin/2, handle_message/4, handle_info/3]).
-include("i3_client_codec.hrl").
-define(timeout, 15000).
-define(sock_opts, [{packet, raw}, {mode, binary}, {active, false}]).
connect(Opts) ->
    {path, Path} = lists:keyfind(path, 1, Opts),
    Timeout = proplists:get_value(timeout, Opts, ?timeout),
    case gen_tcp:connect({local, Path}, 0, ?sock_opts) of
        {ok, Sock} ->
            {ok, #{
                sock => Sock,
                buffer => <<"">>,
                timeout => Timeout
            }};
        {error, _Reason} = Error ->
            Error
    end.
checkin(Publish, #{buffer := Buffer} = State) when is_binary(Buffer) ->
    activate(State, Publish, Buffer).
handle_message(Type, Payload, Publish, #{buffer := Buffer} = State) ->
    maybe
        ok ?= msg_send(State#{buffer := <<"">>}, Type, Payload, Buffer),
        recv_simple(State, Publish, Buffer)
    end.
handle_info(Msg, Publish, State0) ->
    case handle_socket(Msg, State0) of
        {data, Data} ->
            handle_data(State0, Publish, Data);
        {disconnect, Reason, State1} ->
            {disconnect, Reason, Publish, State1};
        unknown ->
            {unknown, Publish, State0}
    end.
handle_socket({tcp, Sock, Data}, #{sock := Sock}) ->
    {data, Data};
handle_socket({tcp_closed, Sock}, #{sock := Sock} = State) ->
    disconnect(State, <<"async recv">>, closed);
handle_socket({tcp_error, Sock, Reason}, #{sock := Sock} = State) ->
    disconnect(State, <<"async recv">>, Reason);
handle_socket(_, _) ->
    unknown.
handle_data(#{timeout := Timeout} = State0, Publish0, Buffer0) ->
    case msg_recv(State0, Timeout, Buffer0) of
        {ok, Msg, <<>>} ->
            {State1, Publish1} = handle_msg(State0, Publish0, Msg),
            activate(State1, Publish1, <<>>);
        {ok, Msg, Buffer1} ->
            {State1, Publish1} = handle_msg(State0, Publish0, Msg),
            handle_data(State1, Publish1, Buffer1);
        {disconnect, Reason, State1} ->
            {disconnect, Reason, Publish0, State1}
    end.
msg_send(#{sock := Sock} = State, Type, Payload, Buffer) ->
    maybe
        {error, Reason} ?= gen_tcp:send(Sock, i3_client_codec:encode_msg(Type, Payload)),
        disconnect(State#{buffer := Buffer}, <<"send">>, Reason)
    end.
recv_simple(State0, Publish0, Buffer0) ->
    case msg_recv(State0, infinity, Buffer0) of
        {ok, {event, _, _} = Event, Buffer1} ->
            {State1, Publish1} = handle_msg(State0, Publish0, Event),
            recv_simple(State1, Publish1, Buffer1);
        {ok, {reply, _, _} = Reply, Buffer1} ->
            State1 = State0#{buffer := Buffer1},
            {ok, Reply, Publish0, State1};
        {disconnect, Reason, State1} ->
            {disconnect, Reason, Publish0, State1}
    end.
handle_msg(State, Publish0, {event, Type, Payload}) ->
    Publish1 = i3_client_publish:call(
        i3_client_codec:decode_event_type(Type),
        Payload,
        Publish0
    ),
    {State, Publish1};
handle_msg(State, Publish, _Msg) ->
    {State, Publish}.
msg_recv(#{sock := Sock} = State, Timeout, active_once) ->
    receive
        {tcp, Sock, Buffer} ->
            msg_recv(State, Timeout, Buffer);
        {tcp_closed, Sock} ->
            disconnect(State, <<"async recv">>, closed, active_once);
        {tcp_error, Sock, Reason} ->
            disconnect(State, <<"async recv">>, Reason, active_once)
    after Timeout ->
        disconnect(State, <<"async recv">>, timeout, active_once)
    end;
msg_recv(State, Timeout, Buffer) ->
    maybe
        {more, More} ?= msg_decode(Buffer),
        msg_recv(State, Timeout, Buffer, More)
    end.
msg_recv(#{sock := Sock} = State, Timeout, Buffer0, More) ->
    case gen_tcp:recv(Sock, More, Timeout) of
        {ok, Data} when is_binary(Buffer0) ->
            msg_recv(State, Timeout, <<Buffer0/binary, Data/binary>>);
        {error, Reason} ->
            disconnect(State, <<"recv">>, Reason, iolist_to_binary(Buffer0))
    end.
activate(State, Publish, <<>> = Buffer) ->
    maybe
        ok ?= setopts(State, [{active, once}], Buffer),
        {ok, Publish, State#{buffer := active_once}}
    end;
activate(#{sock := Sock} = State, Publish, Buffer) ->
    self() ! {tcp, Sock, Buffer},
    {ok, Publish, State}.
setopts(#{sock := Sock} = State, Opts, Buffer) ->
    maybe
        {error, Reason} ?= inet:setopts(Sock, Opts),
        disconnect(State, io_lib:format("setopts: ~p", [Opts]), Reason, Buffer)
    end.
msg_decode(<<?magic, S:32/native, T:32/native, P:S/binary, R/binary>>) ->
    {ok, i3_client_codec:decode_msg(T, P), R};
msg_decode(<<?magic, S:32/native, _:32/native, R/binary>>) ->
    {more, S - byte_size(R)};
msg_decode(Bin) when byte_size(Bin) < ?header_bytes ->
    {more, ?header_bytes - byte_size(Bin)}.
disconnect(State, Action, Reason, Buffer) ->
    disconnect(State#{buffer := Buffer}, Action, Reason).
disconnect(State, Action, Reason) ->
    {disconnect, iolist_to_binary(connection_error(Action, Reason)), State}.
connection_error(Action, Reason) when (Reason =:= closed) or (Reason =:= timeout) ->
    io_lib:format("tcp ~s: ~s", [Action, Reason]);
connection_error(Action, Reason) ->
    io_lib:format("tcp ~s: ~s - ~s", [Action, inet:format_error(Reason), Reason]).