-module(rnt). -export([start/0, define/3, invoke/1, evaluate/2]). start() -> logger:remove_handler(default), logger:add_handler(default, logger_std_h, #{config => #{type => standard_error}}), put(port, open_port({fd, 0, 1}, [binary, {packet, 4}, eof])), loop(). loop() -> serve(frame()), loop(). frame() -> Port = get(port), receive {Port, {data, Frame}} -> Frame; {Port, eof} -> halt() end. send(Frame) -> port_command(get(port), Frame). answering(F) -> try F() catch Class:Reason -> [$e, io_lib:format("~p:~p", [Class, Reason])] end. %% What RNT asks of the node. serve(<<$h>>) -> send(answering(fun() -> [M:rnt_init() || {M, _} <- code:all_loaded(), erlang:function_exported(M, rnt_init, 0)], <<$o>> end)); serve(<<$i, Id:64, M:16, Module:M/binary, F:16, Function:F/binary, B:32, Beam:B/binary>>) -> put({poll, Id}, fun(Limit) -> Name = load(binary_to_atom(Module), Beam), (apply(Name, binary_to_existing_atom(Function), []))(Limit) end); serve(<<$f, Id:64, Limit:32>>) -> send(answering(fun() -> fetch(Id, Limit) end)); serve(<<$c, Id:64>>) -> erase({poll, Id}). load(Name, Beam) -> {ok, {Name, Md5}} = beam_lib:md5(Beam), case erlang:module_loaded(Name) andalso Name:module_info(md5) =:= Md5 of true -> Name; false -> code:purge(Name), {module, Name} = code:load_binary(Name, atom_to_list(Name) ++ ".beam", Beam), Name end. fetch(Id, Limit) -> case get({poll, Id}) of done -> batch([], true); Relation -> {Tuples, Rest} = Relation(Limit), put({poll, Id}, Rest), batch(Tuples, Rest =:= done) end. batch(Tuples, Exhausted) -> [$b, case Exhausted of true -> 1; false -> 0 end | [[<<(byte_size(Term)):32>>, Term] || Tuple <- Tuples, Term <- [term_to_binary(Tuple)]]]. %% What the node asks of RNT. define(Path, Evaluator, Source) -> request([$D, path(Path), sized(Evaluator), Source]). invoke(Path) -> relation(request([$X, path(Path)])). evaluate(Evaluator, Source) -> relation(request([$E, sized(Evaluator), Source])). relation(Id) -> fun(Limit) -> {Tuples, Exhausted} = request(<<$N, Id:64, Limit:32>>), {Tuples, case Exhausted of true -> done; false -> relation(Id) end} end. request(Frame) -> send(Frame), reply(). reply() -> case frame() of <<$o>> -> ok; <<$r, Id:64>> -> Id; <<$b, Exhausted, Tuples/binary>> -> {[binary_to_term(T) || <> <= Tuples], Exhausted =:= 1}; <<$e, Message/binary>> -> error({rnt, Message}); Frame -> serve(Frame), reply() end. path(Segments) -> [<<(length(Segments)):16>> | [sized(S) || S <- Segments]]. sized(Bytes) -> <<(byte_size(Bytes)):16, Bytes/binary>>.