%%% ---------------------------------------------------------------------------- %%% @author Jim Rosenblum %%% @copyright (C) 2015, Jim Rosenblum %%% @doc %%% %%% JC Protocol 1.0 %%% A binary-encoded, string protocol used to provide socket-based %%% interoperability with JC. Incoming messages are string representations %%% of a tuple. Responses are JSON. %%% %%% The protocol defines three message types: CONNECT, CLOSE and COMMAND all of %%% which are binary strings consisting of a header, indicating the size of the %%% message in bytes, follwed by the actual message itself. The header is an %%% 8-byte, big endian, unsigned integer indicating the size of the message in %%% bytes. %%% %%% A RESPONSE is structured the same as messages - 8-byte size header followed %%% by the content of the response. %%% %%% The CONNECT command initiates a session, %%% %%% ```M = <<"{connect,{version,\"1.0\"}}">>''' %%% %%% The byte size is 26, so the CONNECT message is: %%% %%% ```<<26:8/integer-unit:8, M/binary>> = %%% <<0,0,0,0,0,0,0,26,123,99,111,110,110,101,99,116,44,123,118,101,114, %%% 115,105,111,110,44,32,34,49,46,48,34,125,125>> ''' %%% %%% The server will respond to a CONNECT command with either an error or %%% the appropriately encoded version of {\"version\":\"1.0\"} %%% %%% ``` <<17:8/integer-unit:8, <<"{\"version\":1.0}">> = %%% <0,0,0,0,0,0,0,17,123,34,118,101,114,115,105,111,110,34, %%% 58,34,49,46,48,34,125>> ''' %%% %%% %%% The CLOSE command closes the socket, ending the session %%% %%% ```M = <<"{close}">>''' %%% %%% Size is 7 so the CLOSE message is: %%% ```<0,0,0,0,0,0,0,7,123,99,108,111,115,101,125>> ''' %%% %%% %%% COMMAND messages are string versions of the tuple-messages which %%% {@link jc_bridge. jc_bridge} uses, only without the self() parameter. For %%% example the jc_brdige message, {self(), {put, Map, Key, Value}} becomes %%% {put, Map, Key, Value} %%% %%% The RESPONSE will be an appropriately encoded, binary version of a JSON %%% response representing the Erlang return value. %%% %%% A client session might look as follows: %%% %%% ```client:send("{put, evs, 1, \"a string value\"}") %%% ==> <<"{\"ok\":1}">> %%% %%% client:send("{get, evs, 1}"), %%% ==> <<"{\"ok\": \"a string value\"}">>''' %%% %%% %%% @end %%% Created : 25 Aug 2015 by Jim Rosenblum %%% ---------------------------------------------------------------------------- -module(jc_protocol). -behaviour(gen_server). -behaviour(ranch_protocol). %% module API -export([start_link/4]). %% gen_server callbacks -export([init/1, init/4, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(SERVER, ?MODULE). -define(TIMEOUT, infinity). -record(jc_p, {socket, trans :: 'jc_protocol', connected = false :: boolean(), acc = undefined :: undefined | binary(), size = undefined :: undefined | non_neg_integer(), version = <<"1.0">> :: binary()}). -type jc_p() :: #jc_p{}. %%% ============================================================================ %%% Modue API %%% ============================================================================ %% ----------------------------------------------------------------------------- %% Start the gen_server which will handle the socket. %% -spec start_link(ranch:ref(), any(), jc_protocol, any()) -> {ok,pid()}. start_link(Ref, Socket, Trans, Opts) -> proc_lib:start_link(?MODULE, init, [Ref, Socket, Trans, Opts]). %%% ============================================================================ %%% gen_server callbacks %%% ============================================================================ %% ----------------------------------------------------------------------------- %% This function is never called. Its defined so that the gen_server contract %% is fufilled. %% -spec init([]) -> {ok, undefined}. init([]) -> {ok, undefined}. %% ----------------------------------------------------------------------------- %% Set up the socket and convert the process into a gen_server. %% -spec init(ranch:ref(),any(),jc_protocol|undefined,[Port::integer()]) -> any(). init(Ref, S, T, _Opts = [Port]) -> ok = proc_lib:init_ack({ok, self()}), ok = ranch:accept_ack(Ref), ok = T:setopts(S, [{active, once}, {packet, 0}]), lager:info("~p(~p): up and listening on port ~p.", [?MODULE, self(), Port]), gen_server:enter_loop(?MODULE, [], #jc_p{socket = S, trans = T, connected = false, size = undefined, acc = <<>>}, ?TIMEOUT). %% ----------------------------------------------------------------------------- %% @private Handle info messages: Socket messages and jc_psub subscription %% messages. %% -spec handle_info(any(), jc_p()) -> {noreply, jc_p()} | {stop, normal, jc_p()}. handle_info({tcp, S, Data}, State = #jc_p{socket=S, trans=T})-> T:setopts(S, [{active, once}]), NewState = handle_data(Data, State), {noreply, NewState, ?TIMEOUT}; handle_info({tcp_closed, _Socket}, State) -> {stop, normal, State}; handle_info({tcp_error, _, Reason}, State) -> {stop, Reason, State}; handle_info(timeout, State) -> {stop, normal, State}; handle_info(Msg, State = #jc_p{socket=S, trans = T}) -> T:send(S, marshal(Msg)), {noreply, State, ?TIMEOUT}. %% ----------------------------------------------------------------------------- %% @private Hande call messages. %% -spec handle_call(term(), {pid(), _}, jc_p()) -> {reply, ok, jc_p()}. handle_call(Request, _From, State) -> lager:warning("~p: unrecognized handle_call request: ~p.", [?MODULE, Request]), {reply, ok, State}. %% ----------------------------------------------------------------------------- %% @private Handle cast messages. %% -spec handle_cast(any(), jc_p()) -> {noreply, jc_p()}. handle_cast(Msg, State) -> lager:warning("~p: unexpected cast message: ~p.", [?MODULE, Msg]), {noreply, State}. %% ----------------------------------------------------------------------------- %% @private Terminate server. %% -spec terminate(any(), jc_p()) -> any(). terminate(Reason, _State) -> lager:warning("~p: terminated with reason: ~p.", [?MODULE, Reason]), ok. %% ----------------------------------------------------------------------------- %% @private Convert process state when code is changed. %% -spec code_change(term(), jc_p(), any()) -> {ok, jc_p()}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%% ============================================================================ %%% Internal functions %%% ============================================================================ %% ----------------------------------------------------------------------------- %% @private Top-level handler for socket data, returns a new state. %% -spec handle_data(Data::binary(), OldState::jc_p()) -> NewState::jc_p(). handle_data(Data, #jc_p{trans = T, socket = S, connected = C} = State) -> try handle(Data, State) catch throw:{fatal, F} -> T:send(S, marshal({error, {protocol_error, F}})), self() ! {tcp_closed, S}, State; _:Error -> T:send(S, marshal({error, Error})), State#jc_p{acc = <<>>, size = undefined, connected=C} end. %% ----------------------------------------------------------------------------- %% Get Size Header, Get size number of bytes, Execute Command, Repeat. When no %% more bytes, save what has been retrieved thus far in the State and return. %% -spec handle(Data::binary(), OldState::jc_p()) -> NewState::jc_p(). handle(<<>>, State) -> State; handle(<>,#jc_p{size=undefined, acc = <<>>}=S)-> case byte_size(C) of CSize when CSize >= Size -> <> = C, S2 = execute(Command, S), handle(Remainder, S2); Less -> S#jc_p{size = Size - Less, acc = C} end; handle(Data, #jc_p{size = undefined, acc = Acc} = S) -> NewData = <>, case byte_size(NewData) of NSize when NSize >= 8 -> <> = NewData, handle(Remainder, S#jc_p{size=Size, acc = <<>>}); _ -> S#jc_p{acc = NewData, size = undefined} end; handle(Data, #jc_p{acc = Acc, size = Size} = S) when size /= undefined-> case byte_size(Data) of BSize when BSize >= Size -> <> = Data, Command = <>, S2 = execute(Command, S), handle(Remainder, S2#jc_p{acc = <<>>, size = undefined}); Less -> S#jc_p{size = Size - Less, acc = <>} end. %% ----------------------------------------------------------------------------- %% Execute the command. % execute(Command, #jc_p{trans=T, socket=S, connected=false} = State)-> case evaluate(Command) of open_session -> lager:debug("~p (~p): connected with version: ~p", [?MODULE, self(), <<"1.0">>]), T:send(S, marshal({version, <<"1.0">>})), State#jc_p{connected = true}; _ -> throw({fatal, missing_connect}) end; execute(Command, #jc_p{trans=T, socket = S} = State) -> case evaluate(Command) of close_session -> T:send(S, marshal({close, 1.0})), self() ! {tcp_closed, S}; {command, R} -> jc_bridge:do(R), State; _ -> throw(bad_command) end. % ============================================================================== % Utility functions % ============================================================================== % ------------------------------------------------------------------------------ % Given a string representation of a tuple, convert it to an Erlang tuple, make % strings binary and determine the cache action implied by the Command. % evaluate(Command) -> try {ok,Scanned,_} = erl_scan:string(binary_to_list(<>)), {ok,Parsed} = erl_parse:parse_exprs(strings_to_binary(Scanned, [])), determine_action(Parsed) catch _:_ -> throw(command_syntax) end. determine_action([{tuple,1,[{atom,1,close}]}]) -> close_session; determine_action([{tuple,1,[{atom,1,connect},{tuple,1, [{atom,1,version}, {bin,1, [{bin_element,1, {string,1,"1.0"}, default,default}]}]}]}]) -> open_session; determine_action([{tuple,1,_}] = AST) -> {value, R, _} = erl_eval:exprs(AST, []), {command, R}; determine_action(_) -> {error, badarg}. % Marshal the message: make it binary JSON and package it according to the % protocol. marshal(Message) -> Result = try to_json(Message) catch _:_ -> to_json({error, jc_result_to_json}) end, package(Result). to_json({ok,{H, M}}) -> jsone:encode([{hits, encode_keys(H)}, {misses, M}], [{float_format, [{decimals, 10}, compact]}]); to_json({ok, [{_,_}|_Vs]=Lst}) -> jsone:encode(encode_keys(Lst), [{float_format, [{decimals, 10}, compact]}]); to_json({{active, _}=A,{configured, _}=C}) -> jsone:encode([A,C], [{float_format, [{decimals, 10}, compact]}]); to_json({size, Sizes}) -> Os = lists:map(fun({A,B,C}) -> {A, [B,C]} end, Sizes), jsone:encode(Os, [{float_format, [{decimals, 10}, compact]}]); to_json({uptime, [{up_at, U}, {now, N},{up_time, _UT}]}) -> jsone:encode([{uptime, [{up_at, list_to_binary(U)}, {now, list_to_binary(N)}]}], [{float_format, [{decimals, 10}, compact]}]); to_json({X, Y}) -> jsone:encode([{X,Y}], [{float_format, [{decimals, 10}, compact]}]); to_json(M) -> jsone:encode(M, [{float_format, [{decimals, 10}, compact]}]). % JC Keys can be anytime, but json needs object keys to be strings, so % {k,v} needs to be embedded in a json \"key\" : key, \"value\":value % object. encode_keys(KVList) -> lists:map(fun({K,V}) -> {[{key, K},{value, V}]} end, KVList). % Make binary pay-load with size header. package(Message) -> Size = byte_size(Message), <>. % Walk the scanned list of tokens making strings binary strings strings_to_binary([], Acc) -> lists:reverse(Acc); strings_to_binary([{string, _, _}=S|Tl], Acc) -> strings_to_binary(Tl, [{'>>',1}, S, {'<<',1}|Acc]); strings_to_binary([Hd|Tl],Acc) -> strings_to_binary(Tl, [Hd|Acc]).