%%%------------------------------------------------------------------- %%% @doc Behaviour for supervised streaming RPC providers. %%% %%% `advertise_stream/5' on the raw SDK takes a bare handler fun %%% invoked as `Handler(StreamPid, Args)' in a transient process %%% spawned per inbound STREAM_OPEN (see the internal %%% macula_station_link advertise_stream/5 — "this link spawns a %%% server-side macula_stream and dispatches Handler(StreamPid, Args) %%% in a transient process"). This is the provider-side counterpart to %%% `macula_stream_sink': each inbound stream starts one supervised %%% `macula_streamer' child (under a `simple_one_for_one' factory this %%% module owns), threading state through `Module:init/1' and %%% `Module:handle_open/2', and publishing `streaming.started_v1' / %%% `streaming.completed_v1' mesh facts around the stream's lifetime. %%% %%% Sending is push-based and driven from outside the callback: once %%% `Module:handle_open/2' has done whatever registration it needs %%% (e.g. stashing `self()' in a registry keyed by some connection id), %%% any process holding this streamer's pid can call `send/2,3' / %%% `close/1' on it. This module does not prescribe the discovery %%% mechanism. %%% %%% For `client_stream' mode — a consumer pushing chunks INTO the %%% provider, e.g. a batch upload — export the optional %%% `handle_chunk/2' callback (mirroring `macula_stream_sink''s %%% consumer-side callback exactly) and this module drives the same %%% linked-reader `recv/2' loop for you, on the provider side. A %%% `server_stream'-mode module that doesn't export it is unaffected. %%% %%% A `client_stream' provider that also needs to hand the consumer a %%% terminal result (not just accept chunks) exports the optional %%% `handle_eof/1' callback: called once, when the consumer's own %%% `close_send/1' surfaces here as end-of-stream, in place of the %%% default unconditional `{stop, normal, State}'. Returning %%% `{reply, Result, NewState}' sets the stream's terminal reply %%% (`macula_stream:set_reply/2' for `{ok, Value}', `set_error/2' for %%% `{error, Reason}') so the consumer's own `macula:await_reply/1,2' %%% unblocks with it, before stopping. A module that doesn't export %%% `handle_eof/1' keeps the exact prior behavior — no reply is ever %%% set, eof just stops the stream. %%% %%% This is the general-purpose RPC streaming feature (`call_stream/5', %%% `advertise_stream/5', e.g. a `logs.tail_v1'-style procedure) — %%% unrelated to content sharing's own chunked-transfer protocol; see %%% `macula_feeder' / `macula_download' for that. %%% %%% == Cancel == %%% %%% Stopping this gen_server for any non-`normal' reason (a crash, the %%% underlying stream dying, `Module:handle_open/2'/`handle_chunk/2' %%% returning a non-normal stop) sends the peer an explicit %%% `macula_stream:abort/3' STREAM_ERROR, not just a graceful close — %%% the peer learns the transfer was cancelled/failed instead of %%% mistaking it for an ordinary end-of-stream. A `normal' stop closes %%% both sides cleanly instead. %%% %%% == Direct-dial == %%% %%% `advertise/5,6' registers the handler with the pool's advertise- %%% gossip mechanism only — nothing published lets a caller on another %%% station find this procedure without a route having propagated %%% between the two stations first. `advertise_direct/6,7' does that %%% AND publishes a signed `procedure_advertisement' DHT record naming %%% this pool's currently-connected station as the server — the exact %%% same record type and publish function `macula_response:advertise_direct/6,7' %%% uses for plain RPC (a `procedure_advertisement' does not distinguish %%% RPC from streaming), so a caller using %%% `macula_stream_sink:start_link_direct/5,6' can resolve and dial %%% here directly, in one hop, regardless of whether the two stations %%% have a routing edge between them. %%% %%% == Stream I/O == %%% %%% A streamer advertises its procedure with `advertise_stream', a %%% function of arity 6, `macula:advertise_stream/6' by default, which %%% gets the procedure's `auth' policy and no other option; %%% `advertise_direct/6,7' publishes its DHT record with %%% `publish_advertisement', `macula_direct_dial:publish_advertisement/5' %%% by default; and each streamer announces its facts with %%% `fact_publish', `macula:publish/4' by default. Each streamer runs its %%% stream on seven `macula_stream:stream_io()' functions, `recv/2', %%% `send/3', `close_send/1', `close/1', `abort/3', `set_reply/2' and %%% `set_error/2', which are `macula:recv/2' and the `macula_stream' ones %%% by default. `advertise/6' and `advertise_direct/7' take all of these %%% in their options, the seven as `stream_io', checked by %%% `macula_stream:stream_io/2', and refuse a function of another arity %%% with `function_clause' before anything is advertised. %%% %%% == Example == %%% %%% ``` %%% -module(log_tailer_provider). %%% -behaviour(macula_streamer). %%% -export([init/1, handle_open/2]). %%% %%% init(Registry) -> {ok, Registry}. %%% %%% handle_open(#{topic := Topic}, Registry) -> %%% Registry ! {tailer_ready, Topic, self()}, %%% {ok, Registry}. %%% ''' %%% %%% ``` %%% {ok, _Sup} = macula_streamer:advertise(Pool, Realm, %%% <<"logs.tail_v1">>, log_tailer_provider, self()). %%% %%% %% elsewhere, once the provider has announced its pid: %%% ok = macula_streamer:send(TailerPid, <<"a log line\n">>). %%% ''' %%% %%% A `client_stream'-mode provider exports `handle_chunk/2' instead, %%% and never calls `send/2,3' itself — the consumer is the one %%% pushing: %%% %%% ``` %%% -module(batch_upload_provider). %%% -behaviour(macula_streamer). %%% -export([init/1, handle_open/2, handle_chunk/2]). %%% %%% init(Parent) -> {ok, {Parent, []}}. %%% %%% handle_open(_StreamArgs, State) -> {ok, State}. %%% %%% handle_chunk(Chunk, {Parent, Acc}) -> %%% {noreply, {Parent, [Chunk | Acc]}}. %%% ''' %%% %%% ``` %%% {ok, _Sup} = macula_streamer:advertise(Pool, Realm, %%% <<"bulk.ingest">>, batch_upload_provider, self(), %%% #{mode => client_stream}). %%% ''' %%% @end %%%------------------------------------------------------------------- -module(macula_streamer). -behaviour(gen_server). -include_lib("kernel/include/logger.hrl"). -export([advertise/5, advertise/6, advertise_direct/6, advertise_direct/7, unadvertise/3]). -export([send/2, send/3, close/1]). -export([start_link/8]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2]). -export_type([advertise_stream/0, publish_advertisement/0, advertise_opts/0]). -callback init(Args :: term()) -> {ok, State :: term()} | {stop, Reason :: term()}. -callback handle_open(StreamArgs :: term(), State :: term()) -> {ok, NewState :: term()} | {stop, Reason :: term(), NewState :: term()}. -callback handle_chunk(Chunk :: term(), State :: term()) -> {noreply, NewState :: term()} | {stop, Reason :: term(), NewState :: term()}. -callback handle_eof(State :: term()) -> {noreply, NewState :: term()} | {reply, {ok, term()} | {error, term()}, NewState :: term()} | {stop, Reason :: term(), NewState :: term()}. -callback terminate(Reason :: term(), State :: term()) -> any(). -optional_callbacks([terminate/2, handle_chunk/2, handle_eof/1]). -define(STREAMING_STARTED, <<"streaming.started_v1">>). -define(STREAMING_COMPLETED, <<"streaming.completed_v1">>). -define(RECV_TIMEOUT, 30_000). -define(CANCEL_CODE, <<"cancelled">>). -type advertise_stream() :: fun((macula:pool(), macula:realm(), macula:procedure(), macula_stream:mode(), fun((pid(), term()) -> ok), map()) -> ok | {error, term()}). -type publish_advertisement() :: fun((macula:pool(), macula:realm(), macula:procedure(), macula_node_keys:node_key(), map()) -> ok | {error, term()}). -type advertise_opts() :: #{advertise_stream => advertise_stream(), publish_advertisement => publish_advertisement(), fact_publish => macula_lifetime_announcer:publish(), stream_io => macula_stream:stream_io(), atom() => term()}. %% What each streamer runs its stream on and announces its facts with. -type functions() :: #{stream_io := macula_stream:stream_io(), fact_publish := macula_lifetime_announcer:publish()}. -record(tstate, { module :: module(), pool :: macula:pool(), realm :: macula:realm(), announce :: boolean(), io :: macula_stream:stream_io(), fact_publish :: macula_lifetime_announcer:publish(), stream_id :: binary(), stream :: pid(), reader :: pid() | undefined, user :: term() }). %% @doc Advertise `Procedure' on `Pool'/`Realm'. Starts a private %% factory supervisor for per-stream provider children and registers %% a dispatch handler with `macula:advertise_stream/5,6'. Returns the %% supervisor pid so the caller can supervise it (or ignore it). -spec advertise(macula:pool(), macula:realm(), macula:procedure(), module(), term()) -> {ok, pid()} | {error, term()}. advertise(Pool, Realm, Procedure, Module, Args) -> advertise(Pool, Realm, Procedure, Module, Args, #{}). %% @doc As `advertise/5'. `Opts' may include `announce' (default %% `true'), `mode' (default `server_stream'), `auth' (the procedure's %% auth policy, default `open', see `macula:advertise_stream/6'), and %% `reuse_sup' — an %% existing supervisor pid (as returned by a prior `advertise/5,6' %% call) to register the handler again with, without starting a new %% factory supervisor. Use this for a periodic re-advertise (see %% `advertise_direct/6,7''s own doc) — calling plain %% `advertise/5,6' on a timer would leak one orphaned supervisor per %% tick, since each call otherwise starts a fresh one. The functions a %% streamer runs on come from `Opts' too; see "Stream I/O" above. -spec advertise(macula:pool(), macula:realm(), macula:procedure(), module(), term(), advertise_opts()) -> {ok, pid()} | {error, term()}. advertise(Pool, Realm, Procedure, Module, Args, Opts) -> AdvertiseStream = arity_6(maps:get(advertise_stream, Opts, fun macula:advertise_stream/6)), Functions = functions(Opts), Sup = existing_or_new_sup(maps:get(reuse_sup, Opts, undefined)), Announce = maps:get(announce, Opts, true), Mode = maps:get(mode, Opts, server_stream), Handler = fun(StreamPid, StreamArgs) -> dispatch(Sup, Module, Pool, Realm, Announce, Args, Functions, StreamPid, StreamArgs) end, %% The advertise function gets the procedure's auth policy, when the %% options give one, and no other option. case AdvertiseStream(Pool, Realm, Procedure, Mode, Handler, maps:with([auth], Opts)) of ok -> {ok, Sup}; {error, Reason} -> {error, Reason} end. %% The functions each streamer runs on, from the options or else the %% defaults. Stream functions macula_stream:stream_io/2 does not accept, %% or a fact_publish of another arity, are refused with function_clause, %% in the caller, before anything is advertised. functions(Opts) -> StreamIo = macula_stream:stream_io(default_stream_io(), maps:get(stream_io, Opts, undefined)), #{stream_io => StreamIo, fact_publish => arity_4(maps:get(fact_publish, Opts, fun macula:publish/4))}. default_stream_io() -> #{recv => fun macula:recv/2, controlling_process => fun macula_stream:controlling_process/2, send => fun macula_stream:send/3, close_send => fun macula_stream:close_send/1, close => fun macula_stream:close/1, abort => fun macula_stream:abort/3, set_reply => fun macula_stream:set_reply/2, set_error => fun macula_stream:set_error/2}. %% The options the advertisement publish gets: all but the functions. without_functions(Opts) -> maps:without([advertise_stream, publish_advertisement, fact_publish, stream_io], Opts). arity_4(Fun) when is_function(Fun, 4) -> Fun. arity_5(Fun) when is_function(Fun, 5) -> Fun. arity_6(Fun) when is_function(Fun, 6) -> Fun. %% See `macula_response:existing_or_new_sup/1' for why a dead `reuse_sup' %% pid must fall through to a fresh one rather than being handed to %% `dispatch/9' as-is. existing_or_new_sup(Pid) when is_pid(Pid) -> existing_or_new_sup(Pid, erlang:is_process_alive(Pid)); existing_or_new_sup(undefined) -> new_sup(). existing_or_new_sup(Pid, true) -> Pid; existing_or_new_sup(_Pid, false) -> new_sup(). new_sup() -> {ok, Sup} = macula_streamer_sup:start_link(), Sup. %% @doc As `advertise/5', and additionally publishes a signed %% `procedure_advertisement' DHT record naming this pool's connected %% station as the server, so `macula_stream_sink:start_link_direct/5,6' %% can resolve and dial here directly. `NodeIdentity' signs it and must be %% the node identity key `Pool' was started with: a caller targets that %% node_id, and the station knows the pool's connection by it. %% %% The DHT publish is best-effort: if it fails, the handler is still %% advertised and reachable via the ordinary pooled path — direct-dial %% callers just won't be able to resolve it until a later publish %% succeeds. "Best-effort" still means the failure is logged, not %% silently discarded — a caller that only ever calls this once (never %% retries) has no other way to learn its handler is pooled-only, and %% "a later publish succeeds" cannot happen if nothing ever tries again. -spec advertise_direct(macula:pool(), macula:realm(), macula:procedure(), module(), term(), macula_node_keys:node_key()) -> {ok, pid()} | {error, term()}. advertise_direct(Pool, Realm, Procedure, Module, Args, NodeIdentity) -> advertise_direct(Pool, Realm, Procedure, Module, Args, NodeIdentity, #{}). %% @doc As `advertise_direct/6', with `Opts' forwarded BOTH to %% `advertise/6' (so `mode'/`announce'/`reuse_sup' and the functions %% apply here too, e.g. `mode => client_stream') and, without the %% functions, to the advertisement publish, `publish_advertisement' in %% `Opts' or `macula_direct_dial:publish_advertisement/5' (e.g. %% `authorization', the provider authorization an org namespaced %% procedure needs): each side reads only the keys it recognizes, so one %% `Opts' map serves both. %% `reuse_sup' matters here specifically: the procedure's DHT record %% expires with its TTL, and callers reach the provider only through %% that record, so the provider republishes it — a periodic %% re-advertise with `reuse_sup => Sup' (the pid this function %% returned the first time) registers the handler again and %% republishes the DHT record without leaking a new supervisor per %% tick. `cert_chain', a 10.x option `authorization' replaces, is refused %% with `{error, {removed_option, cert_chain}}' before the handler is %% registered. -spec advertise_direct(macula:pool(), macula:realm(), macula:procedure(), module(), term(), macula_node_keys:node_key(), advertise_opts()) -> {ok, pid()} | {error, term()}. advertise_direct(Pool, Realm, Procedure, Module, Args, NodeIdentity, Opts) -> advertise_direct_unless_removed(macula_direct_dial:removed_option(advertise, Opts), Pool, Realm, Procedure, Module, Args, NodeIdentity, Opts). advertise_direct_unless_removed(none, Pool, Realm, Procedure, Module, Args, NodeIdentity, Opts) -> PublishAdvertisement = arity_5(maps:get(publish_advertisement, Opts, fun macula_direct_dial:publish_advertisement/5)), case advertise(Pool, Realm, Procedure, Module, Args, Opts) of {ok, Sup} -> log_publish_result( PublishAdvertisement(Pool, Realm, Procedure, NodeIdentity, without_functions(Opts)), Procedure), {ok, Sup}; {error, _} = Error -> Error end; advertise_direct_unless_removed(Removed, _Pool, _Realm, _Procedure, _Module, _Args, _NodeIdentity, _Opts) -> {error, Removed}. log_publish_result(ok, _Procedure) -> ok; log_publish_result({error, Reason}, Procedure) -> ?LOG_WARNING("[macula_streamer] direct-dial advertisement publish " "failed for ~s: ~p -- handler stays reachable via the " "pooled path only until a later publish succeeds", [Procedure, Reason]). %% @doc Stop advertising. Does not stop the factory supervisor %% returned by `advertise/5,6' — callers that want to tear it down %% should `exit(Sup, shutdown)' themselves. -spec unadvertise(macula:pool(), macula:realm(), macula:procedure()) -> ok. unadvertise(Pool, Realm, Procedure) -> macula:unadvertise_stream(Pool, Realm, Procedure). dispatch(Sup, Module, Pool, Realm, Announce, Args, Functions, StreamPid, StreamArgs) -> case supervisor:start_child(Sup, [Module, Pool, Realm, Announce, Args, StreamPid, StreamArgs, Functions]) of {ok, Pid} -> hand_stream_to(Functions, StreamPid, Pid); {error, _Reason} -> ok end. %% @private The process running this dispatch owns the stream, and the %% streamer it started takes the stream over, so the stream ends when the %% streamer ends rather than when this dispatch returns. hand_stream_to(#{stream_io := #{controlling_process := HandOver}}, StreamPid, StreamerPid) -> _ = HandOver(StreamPid, StreamerPid), ok. %% @doc Send a chunk out on the stream this streamer owns. -spec send(pid(), binary()) -> ok | {error, term()}. send(Pid, Chunk) -> gen_server:call(Pid, {send, Chunk}). %% @doc As `send/2', with an explicit encoding. -spec send(pid(), binary() | term(), macula_stream:encoding()) -> ok | {error, term()}. send(Pid, Chunk, Encoding) -> gen_server:call(Pid, {send, Chunk, Encoding}). %% @doc Close the send side of the stream. -spec close(pid()) -> ok. close(Pid) -> gen_server:call(Pid, close). %% @private -spec start_link(module(), macula:pool(), macula:realm(), boolean(), term(), pid(), term(), functions()) -> {ok, pid()} | {error, term()}. start_link(Module, Pool, Realm, Announce, InitArgs, StreamPid, StreamArgs, Functions) -> gen_server:start_link(?MODULE, {Module, Pool, Realm, Announce, InitArgs, StreamPid, StreamArgs, Functions}, []). %%%=================================================================== %%% gen_server callbacks %%%=================================================================== %% @private init({Module, Pool, Realm, Announce, InitArgs, StreamPid, StreamArgs, Functions}) -> process_flag(trap_exit, true), case Module:init(InitArgs) of {ok, UserState} -> open(Module, Pool, Realm, Announce, Functions, StreamPid, StreamArgs, UserState); {stop, Reason} -> {stop, Reason} end. open(Module, Pool, Realm, Announce, #{stream_io := StreamIo, fact_publish := FactPublish}, StreamPid, StreamArgs, UserState) -> case Module:handle_open(StreamArgs, UserState) of {ok, NewUserState} -> link(StreamPid), Reader = maybe_spawn_reader(Module, StreamIo, StreamPid), StreamId = crypto:strong_rand_bytes(16), publish(Announce, FactPublish, Pool, Realm, ?STREAMING_STARTED, #{stream_id => StreamId}), {ok, #tstate{module = Module, pool = Pool, realm = Realm, announce = Announce, io = StreamIo, fact_publish = FactPublish, stream_id = StreamId, stream = StreamPid, reader = Reader, user = NewUserState}}; {stop, Reason, _NewUserState} -> abort_rejected_stream(StreamIo, Reason, StreamPid), {stop, Reason} end. %% @private A rejected open (`handle_open/2' returning `{stop, Reason, _}') %% never links `StreamPid', so `terminate/2' never runs on it and the peer %% that opened the stream would otherwise be stranded until its own `recv' %% timeout. Abort it explicitly so the peer gets an immediate signal %% instead of silence, naming the reason and carrying none of its terms. abort_rejected_stream(#{abort := Abort}, Reason, StreamPid) -> Message = macula_reason_name:text(Reason), try Abort(StreamPid, ?CANCEL_CODE, Message) catch _:_ -> ok end. %% @private For `client_stream'-mode providers that export %% `handle_chunk/2': spawn the same linked-reader `recv/2' loop %% `macula_stream_sink' drives on the consumer side, applied here to %% the provider's own stream. A `server_stream'-mode module has no %% reason to export `handle_chunk/2', so this is a no-op for it. maybe_spawn_reader(Module, #{recv := Recv}, Stream) -> case erlang:function_exported(Module, handle_chunk, 2) of true -> spawn_reader(Recv, Stream); false -> undefined end. spawn_reader(Recv, Stream) -> Parent = self(), spawn_link(fun() -> reader_loop(Parent, Recv, Stream) end). reader_loop(Parent, Recv, Stream) -> dispatch_recv(Recv(Stream, ?RECV_TIMEOUT), Parent, Recv, Stream). dispatch_recv({chunk, Data}, Parent, Recv, Stream) -> Parent ! {stream_item, Data}, reader_loop(Parent, Recv, Stream); dispatch_recv({data, Data}, Parent, Recv, Stream) -> Parent ! {stream_item, Data}, reader_loop(Parent, Recv, Stream); dispatch_recv(eof, Parent, _Recv, _Stream) -> Parent ! stream_eof; dispatch_recv({error, Reason}, Parent, _Recv, _Stream) -> Parent ! {stream_error, Reason}. %% @private handle_call({send, Chunk}, _From, #tstate{io = #{send := Send}, stream = Stream} = State) -> {reply, Send(Stream, Chunk, raw), State}; handle_call({send, Chunk, Encoding}, _From, #tstate{io = #{send := Send}, stream = Stream} = State) -> {reply, Send(Stream, Chunk, Encoding), State}; handle_call(close, _From, #tstate{io = #{close_send := CloseSend}, stream = Stream} = State) -> {reply, CloseSend(Stream), State}; handle_call(_Request, _From, State) -> {reply, {error, unsupported}, State}. %% @private handle_cast(_Msg, State) -> {noreply, State}. %% @private handle_info({stream_item, Data}, #tstate{module = Module, user = User} = State) -> deliver(Module:handle_chunk(Data, User), State); handle_info(stream_eof, State) -> handle_eof(State); handle_info({stream_error, Reason}, State) -> {stop, Reason, State}; handle_info({'EXIT', Reader, Reason}, #tstate{reader = Reader} = State) when Reason =/= normal -> {stop, {reader_crashed, Reason}, State}; handle_info({'EXIT', Stream, Reason}, #tstate{stream = Stream} = State) -> {stop, Reason, State}; %% The stream's session ended (`macula_stream:controlling_process/2'): a %% streamer has nothing left to serve, whether or not its module would stop %% by itself. handle_info({macula_stream, ended, Stream, closed}, #tstate{stream = Stream} = State) -> {stop, normal, State}; handle_info({macula_stream, ended, Stream, _How}, #tstate{stream = Stream} = State) -> {stop, {shutdown, session_ended}, State}; handle_info(_Msg, State) -> {noreply, State}. deliver({noreply, NewUser}, State) -> {noreply, State#tstate{user = NewUser}}; deliver({stop, Reason, NewUser}, State) -> {stop, Reason, State#tstate{user = NewUser}}. %% @private Default (no `handle_eof/1' exported): unchanged prior %% behavior, eof just stops the stream. Otherwise gives the callback %% one last chance to set a terminal reply before stopping. handle_eof(#tstate{module = Module, user = User} = State) -> case erlang:function_exported(Module, handle_eof, 1) of true -> deliver_eof(Module:handle_eof(User), State); false -> {stop, normal, State} end. deliver_eof({noreply, NewUser}, State) -> {stop, normal, State#tstate{user = NewUser}}; deliver_eof({reply, {ok, Value}, NewUser}, #tstate{io = #{set_reply := SetReply}, stream = Stream} = State) -> _ = SetReply(Stream, Value), {stop, normal, State#tstate{user = NewUser}}; deliver_eof({reply, {error, Reason}, NewUser}, #tstate{io = #{set_error := SetError}, stream = Stream} = State) -> _ = SetError(Stream, Reason), {stop, normal, State#tstate{user = NewUser}}; deliver_eof({stop, Reason, NewUser}, State) -> {stop, Reason, State#tstate{user = NewUser}}. %% @private terminate(Reason, #tstate{module = Module, pool = Pool, realm = Realm, announce = Announce, io = StreamIo, fact_publish = FactPublish, stream_id = StreamId, stream = Stream, reader = Reader, user = User}) -> stop_reader(Reader), finish_stream(StreamIo, Reason, Stream), publish(Announce, FactPublish, Pool, Realm, ?STREAMING_COMPLETED, outcome_fields(#{stream_id => StreamId}, Reason)), maybe_terminate(Module, Reason, User). stop_reader(undefined) -> ok; stop_reader(Reader) -> unlink(Reader), exit(Reader, kill). %% @private A `normal' reason closes both sides cleanly. Anything else %% (a crash, the underlying stream dying, a non-normal stop from %% `handle_open/2'/`handle_chunk/2') sends the peer an explicit %% `STREAM_ERROR' abort, whose message is the reason's name, instead of %% leaving it to infer cancellation from the connection simply going %% away. `Stream' may already be dead by the time this runs (e.g. its %% own exit is what triggered this termination), which is harmless and %% caught below. finish_stream(#{close := Close}, normal, Stream) -> try Close(Stream) catch _:_ -> ok end; finish_stream(#{abort := Abort}, Reason, Stream) -> Message = macula_reason_name:text(Reason), try Abort(Stream, ?CANCEL_CODE, Message) catch _:_ -> ok end. outcome_fields(Base, normal) -> Base#{outcome => completed}; outcome_fields(Base, Reason) -> Base#{outcome => failed, reason => Reason}. maybe_terminate(Module, Reason, User) -> case erlang:function_exported(Module, terminate, 2) of true -> Module:terminate(Reason, User); false -> ok end. publish(false, _FactPublish, _, _, _, _) -> ok; publish(true, FactPublish, Pool, Realm, Topic, Payload) -> _ = FactPublish(Pool, Realm, Topic, Payload), ok.