-module(eavro_rpc_proto). -compile(export_all). -include("eavro.hrl"). -export([parse_protocol_file/1, parse_protocol/1, encode_call/3, decode_call/2, encode_response/2, encode_handshake_response/1, decode_frame_sequences/1, decode_handshake_response/1, decode_handshake_request/1, decode_call_response/2]). %% Parses Avro Protocol Definition from JSON file (*.avpr). -spec parse_protocol_file(Filename :: file:filename()) -> #avro_proto{}. parse_protocol_file(Filename) -> {ok, Json} = file:read_file(Filename), parse_protocol(Json). %% %% Parses Avro Protocol Definition from JSON formatted text.. %% -spec parse_protocol(AvprJson :: binary()) -> #avro_proto{}. parse_protocol(Json) -> Jsx = [{_,_}|_] = jsx:decode(Json), Prop = fun(Name, Props) -> case proplists:get_value(Name, Props) of undefined -> exit({expected, Name}); Val -> Val end end, Messages = Prop(<<"messages">>, Jsx), {Types, Context} = eavro:parse_types(Prop(<<"types">>, Jsx), dict:new()), Name = Prop(<<"protocol">>, Jsx), Ns = Prop(<<"namespace">>, Jsx), ParseMessage = fun({MName, M}) -> Req = Prop(<<"request">>, M), Res = Prop(<<"response">>, M), ResolveType = fun(ArgNameOrDef) -> try {TypeParsed, _Context1} = eavro:parse_type(ArgNameOrDef, Context), TypeParsed catch _:_ when is_binary(ArgNameOrDef) -> ArgName = binary_to_atom(ArgNameOrDef,latin1), hd([_] = [ T || T <- Types, case T of #avro_record{name = ArgName} -> true; #avro_enum{name = ArgName} -> true; #avro_fixed{name = ArgName} -> true; _ -> false end]) end end, MArgs = [ ResolveType(Prop(<<"type">>, Arg)) || Arg <- Req ], #avro_message{ name = MName, args = MArgs, return = ResolveType(Res) } end, #avro_proto{ ns = Ns, name = Name, types = Types, messages = lists:map(ParseMessage, Messages), json = Json }. %% %% Encoding/decoding calls. %% encode_call(Proto, MessageName, Args) when is_atom(MessageName) -> encode_call(Proto, atom_to_binary(MessageName,latin1), Args); encode_call(#avro_proto{} = Proto, MessageName, Args) when is_binary(MessageName), is_list(Args) -> #avro_message{ name = MessageNameBin, args = Types } = get_message(Proto, MessageName), length(Args) == length(Types) orelse exit(bad_message_arity), [eavro:encode(schema_Meta(), []), eavro:encode(string, MessageNameBin), [ eavro:encode(Type, Arg) || {Type, Arg} <- lists:zip(Types, Args) ] ]. decode_call(#avro_proto{} = Proto, Buff) -> {_Meta, Buff1} = eavro:decode(schema_Meta(), Buff), {MessageNameBin, Buff2} = eavro:decode(string, Buff1), Msg = #avro_message{ args = Types } = get_message(Proto, MessageNameBin), {ArgsR, Buff3} = lists:foldl( fun(T, {Vals, B}) -> {V, B1} = eavro:decode(T,B), {[V|Vals], B1} end, {[], Buff2}, Types), { {Msg, lists:reverse(ArgsR)}, Buff3}. encode_response(#avro_proto{} = Proto, MessageName, Result) -> M = get_message(Proto, MessageName), encode_response(M, Result). encode_response(#avro_message{return = RetType }, {Status, Ret}) -> [eavro:encode(schema_Meta(), []), case Status of ok -> [<<0>>, eavro:encode(RetType, Ret)]; error -> [<<1>>, eavro:encode( [string, RetType], {RetType, Ret} )] end]. encode_handshake_response(#avro_proto{json = Json})-> eavro:encode( schema_HandshakeResponse(), ['CLIENT', {string, Json}, {schema_MD5(), erlang:md5(Json)}, {schema_Meta(), []}]). get_message(#avro_proto{ messages = Messages}, MessageName) -> case lists:keyfind(if is_atom(MessageName) -> atom_to_binary(MessageName, latin1); true -> MessageName end, #avro_message.name, Messages) of #avro_message{} = Msg -> Msg; false -> exit({unsupported_proto_message, MessageName}) end. decode_response(Proto, MessageName, Buff) -> #avro_message{return = RetType } = get_message(Proto, MessageName), [{Ser, Buff1} | Buffs] = decode_frame_sequences(Buff), {HeaderResponse, Buff2} = decode_handshake_response(Buff1), {HeaderResponse, [ {S, decode_call_response(RetType, B)} || {S,B} <- [ {Ser, Buff2} | Buffs ] ] }. decode_call_response(RetType, Buff) -> {_Meta, <>} = eavro:decode(schema_Meta(), Buff), case ErrFlag of 0 -> {Rsp, <<>>} = eavro:decode(RetType,Buff1), {ok, Rsp}; 1 -> {Err, <<>>} = eavro:decode([string, RetType],Buff1), {error, Err} end. test_data() -> [<<0:32>>,<<2:32>>,<<3:32,1,2,3>>,<<4:32,1,2,3,4>>]. decode_frame_sequences(<>) -> continue_decode_frame_sequences(Serial, decode_frame_sequence(N, Tail, [])); decode_frame_sequences(Buff) -> {cont, fun(Buff1) -> decode_frame_sequences(<>) end}. continue_decode_frame_sequences(Serial, {cont, Cont}) -> {cont, fun(Buff1) -> continue_decode_frame_sequences(Serial, Cont(Buff1)) end}; continue_decode_frame_sequences(Serial, {Arr, Tail2}) -> case decode_frame_sequences(Tail2) of {cont, _} = Cont -> { [ {Serial, Arr} ], Cont}; {Arrs, Cont} -> { [ {Serial, Arr} | Arrs], Cont} end. decode_frame_sequence(0, Buff, Acc) -> {lists:reverse(Acc), Buff}; decode_frame_sequence(N, <> = Buff, Acc) -> case Tail of <> -> decode_frame_sequence(N - 1, Tail1, [F|Acc]); _ -> {cont, fun(Buff1) -> decode_frame_sequence( N, <>, Acc) end} end; decode_frame_sequence(N, Buff, Acc) -> {cont, fun(Buff1) -> decode_frame_sequence( N, <>, Acc) end}. decode_handshake_response(Buff) -> eavro:decode(schema_HandshakeResponse(), Buff). decode_handshake_request(Buff) -> eavro:decode(schema_HandshakeRequest(), Buff). %% %% Avro Protocol Handshake Schema's %% schema_HandshakeRequest() -> #avro_record{ name = 'HandshakeRequest', fields = [{clientHash, schema_MD5()}, {clientProtocol, [null, string]}, {serverHash, schema_MD5()}, {meta, [null,schema_Meta()]}]}. schema_HandshakeResponse() -> #avro_record{ name = 'HandshakeResponse', fields = [{match, #avro_enum{ name = 'HandshakeMatch', symbols = ['BOTH', 'CLIENT', 'NONE']}}, {serverProtocol, [null, string]}, {serverHash, [null, schema_MD5()]}, {meta, [null, schema_Meta()]}]}. schema_Meta() -> #avro_map{ values = bytes }. schema_MD5() -> #avro_fixed{ name = 'MD5', size = 16}.