-module(i3_client_protocol). -moduledoc false. -export([ connect/1, checkin/1, handle_message/3, handle_message/4, handle_info/2, 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(#{buffer := Buffer} = State) when is_binary(Buffer) -> activate(State, Buffer). handle_message(Type, Payload, State) -> handle_message(Type, Payload, fun(_, _) -> ok end, State). 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, State) -> handle_info(Msg, fun(_, _) -> ok end, State). handle_info(Msg, Publish, State) -> case handle_socket(Msg, State) of {data, Data} -> handle_data(State, Publish, Data); {disconnect, _, _} = Disconnect -> Disconnect; unknown -> {unknown, State} 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, _} = handle_msg(State0, Publish0, Msg), activate(State1, <<>>); {ok, Msg, Buffer1} -> {State1, Publish1} = handle_msg(State0, Publish0, Msg), handle_data(State1, Publish1, Buffer1); {disconnect, _, _} = Disconnect -> Disconnect 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, State1}; {disconnect, _, _} = Disconnect -> Disconnect end. handle_msg(State, Publish, {event, Type, Payload}) -> Publish(i3_client_codec:decode_event_type(Type), Payload), {State, Publish}; 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, <>); {error, Reason} -> disconnect(State, <<"recv">>, Reason, iolist_to_binary(Buffer0)) end. activate(State, <<>> = Buffer) -> maybe ok ?= setopts(State, [{active, once}], Buffer), {ok, State#{buffer := active_once}} end; activate(#{sock := Sock} = State, Buffer) -> self() ! {tcp, Sock, Buffer}, {ok, 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(<>) -> {ok, i3_client_codec:decode_msg(T, P), R}; msg_decode(<>) -> {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]).