-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) || <<N:32, T:N/binary>> <= 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>>.
