-module(gleam_otp_process_external). % Channels -export([open_channel/1, close_channel/1]). % Receivers -export([make_receiver/0, include_channel/3, include_process_monitor/3, include_port_monitor/3, include_process_exit/3, include_system/2, include_bare/2, include_all_exits/2, include_all/2, remove_timeout/1, set_timeout/2, run_receiver/1, flush_receiver/1, flush_other/2]). % % import Gleam records % -include("gen/src/gleam@otp@process_Channel.hrl"). -include("gen/src/gleam@otp@process_Exit.hrl"). -include("gen/src/gleam@otp@process_PortDown.hrl"). -include("gen/src/gleam@otp@process_PortMonitor.hrl"). -include("gen/src/gleam@otp@process_ProcessDown.hrl"). -include("gen/src/gleam@otp@process_ProcessMonitor.hrl"). -include("gen/src/gleam@otp@process_StatusInfo.hrl"). % % Guards % -define(is_n_tuple(Term, N), (is_tuple(Term) andalso tuple_size(Term) =:= N)). -define(is_record(Tag, Arity, Term), (?is_n_tuple(Term, Arity + 1) andalso element(1, Term) =:= Tag)). -define(is_system_msg(Term), ?is_record(system, 2, Term)). -define(is_monitor_msg(Term), ?is_record('DOWN', 4, Term)). -define(is_exit_msg(Term), ?is_record('EXIT', 2, Term)). -define(is_gen_reply(Term), (is_tuple(Term) andalso tuple_size(Term) =:= 2 andalso is_reference(element(1, Term)))). -define(is_special_msg(Term), (?is_system_msg(Term) orelse ?is_monitor_msg(Term) orelse ?is_exit_msg(Term) orelse ?is_gen_reply(Term))). % % Channels % currently_open_channels() -> get('$gleam_open_channels'). update_channels(Fn) -> case currently_open_channels() of undefined -> Fn(#{}); Refs -> Fn(Refs) end. open_channel(Ref) -> update_channels(fun(Refs) -> maps:put(Ref, [], Refs) end). close_channel(Ref) -> update_channels(fun(Refs) -> maps:remove(Ref, Refs) end). % % Receivers % -record(receiver, {system, bare, all, all_exits, exit_pids, refs, timeout, flush_other}). make_receiver() -> #receiver{timeout = 5000, system = undefined, bare = undefined, all = undefined, flush_other = false, all_exits = undefined, exit_pids = #{}, refs = #{}}. receiver_include(Receiver, Ref, Fn) -> Refs = maps:put(Ref, Fn, Receiver#receiver.refs), Receiver#receiver{refs = Refs}. include_channel(Receiver, Channel, Fn) -> receiver_include(Receiver, Channel#channel.reference, Fn). include_process_monitor(Receiver, Monitor, Fn) -> receiver_include(Receiver, Monitor#process_monitor.reference, Fn). include_port_monitor(Receiver, Monitor, Fn) -> receiver_include(Receiver, Monitor#port_monitor.reference, Fn). include_process_exit(Receiver, Pid, Fn) -> Pids = maps:put(Pid, Fn, Receiver#receiver.exit_pids), Receiver#receiver{exit_pids = Pids}. get_map_msg(Map, Key, Msg) -> Fn = maps:get(Key, Map), {ok, Fn(Msg)}. run_receiver(Receiver) -> #receiver{timeout = Timeout, system = System, bare = Bare, refs = Refs, all = All, flush_other = FlushOther, exit_pids = ExitPids} = Receiver, OpenChannels = currently_open_channels(), receive % Message on closed channels are discarded {Ref, _} when not is_map_key(Ref, OpenChannels) -> % TODO: shrink timeout if time has passed run_receiver(Receiver); {Ref, Msg} when is_map_key(Ref, Refs) -> get_map_msg(Refs, Ref, Msg); {'DOWN', Ref, process, Pid, Reason} when is_map_key(Ref, Refs) -> get_map_msg(Refs, Ref, #process_down{pid = Pid, reason = Reason}); {'DOWN', Ref, port, Port, Reason} when is_map_key(Ref, Refs) -> get_map_msg(Refs, Ref, #port_down{port = Port, reason = Reason}); {'EXIT', Pid, Reason} when is_map_key(Pid, ExitPids) -> get_map_msg(ExitPids, Pid, #exit{pid = Pid, reason = Reason}); {system, From, Request} when is_function(System) -> {ok, System(system_msg(From, Request))}; Msg when (not ?is_special_msg(Msg)) andalso is_function(Bare) -> {ok, Bare(Msg)}; Msg when is_function(All) -> {ok, All(Msg)}; _ when FlushOther -> % TODO: shrink timeout if time has passed run_receiver(Receiver) after Timeout -> {error, nil} end. flush_receiver(Receiver, N) -> #receiver{system = System, all = All, bare = Bare, refs = Refs, exit_pids = ExitPids} = Receiver, OpenChannels = currently_open_channels(), receive % Messages on closed channels are _always_ discarded {Ref, _} when not is_map_key(Ref, OpenChannels) -> flush_receiver(Receiver, N); % Don't count these messages {Ref, _} when is_map_key(Ref, Refs) -> flush_receiver(Receiver, N + 1); {'DOWN', Ref, _, _, _} when is_map_key(Ref, Refs) -> flush_receiver(Receiver, N + 1); {'EXIT', Pid, _} when is_map_key(Pid, ExitPids) -> flush_receiver(Receiver, N + 1); {system, _, _} when is_function(System) -> flush_receiver(Receiver, N + 1); Msg when (not ?is_special_msg(Msg)) andalso is_function(Bare) -> flush_receiver(Receiver, N + 1); _ when is_function(All) -> flush_receiver(Receiver, N + 1) after 0 -> N end. set_timeout(Receiver, Timeout) -> Receiver#receiver{timeout = Timeout}. remove_timeout(Receiver) -> Receiver#receiver{timeout = infinity}. flush_receiver(Receiver) -> flush_receiver(Receiver, 0). include_system(Receiver, Fn) -> Receiver#receiver{system = Fn}. include_bare(Receiver, Fn) -> Receiver#receiver{bare = Fn}. include_all(Receiver, Fn) -> Receiver#receiver{all = Fn}. include_all_exits(Receiver, Fn) -> Receiver#receiver{all_exits = Fn}. flush_other(Receiver, FlushOther) -> Receiver#receiver{flush_other = FlushOther}. system_msg(From = {Pid, Ref}, Msg) -> Send = fun(X) -> system_reply(Msg, From, X) end, {Msg, #channel{pid = Pid, reference = Ref, send = Send}}. system_reply(Msg, From, Reply) -> case Msg of resume -> gen:reply(From, ok); suspend -> gen:reply(From, ok); get_state -> gen:reply(From, Reply); get_status -> gen:reply(From, process_status(Reply)) end. process_status(Status) -> #status_info{mode = Mode, parent = Parent, debug_state = Debug, state = State, mod = Mod} = Status, Data = [ get(), Mode, Parent, Debug, [{header, "Status for Gleam actor " ++ pid_to_list(self())}, {data, [{'Status', Mode}, {'Parent', Parent}, {'State', State}]}] ], {status, self(), {module, Mod}, Data}.