summaryrefslogtreecommitdiff
path: root/apps/rexi
diff options
context:
space:
mode:
Diffstat (limited to 'apps/rexi')
-rw-r--r--apps/rexi/ebin/rexi.app7
-rw-r--r--apps/rexi/src/rexi_utils.erl53
2 files changed, 59 insertions, 1 deletions
diff --git a/apps/rexi/ebin/rexi.app b/apps/rexi/ebin/rexi.app
index 620c8863..5c944b72 100644
--- a/apps/rexi/ebin/rexi.app
+++ b/apps/rexi/ebin/rexi.app
@@ -1,7 +1,12 @@
{application, rexi, [
{description, "Lightweight RPC server"},
{vsn, "1.2"},
- {modules, [rexi, rexi_app, rexi_sup, rexi_monitor, rexi_server]},
+ {modules, [rexi,
+ rexi_app,
+ rexi_sup,
+ rexi_monitor,
+ rexi_server,
+ rexi_utils]},
{registered, [rexi_sup, rexi_server]},
{applications, [kernel, stdlib]},
{mod, {rexi_app,[]}}
diff --git a/apps/rexi/src/rexi_utils.erl b/apps/rexi/src/rexi_utils.erl
new file mode 100644
index 00000000..79183951
--- /dev/null
+++ b/apps/rexi/src/rexi_utils.erl
@@ -0,0 +1,53 @@
+-module(rexi_utils).
+
+-export([recv/6]).
+
+%% @doc set up the receive loop with an overall timeout
+-spec recv([any()], integer(), function(), any(), timeout(), timeout()) ->
+ {ok, any()} | timeout | {error, any()}.
+recv(Refs, Keypos, Fun, Acc0, infinity, PerMsgTO) ->
+ process_mailbox(Refs, Keypos, Fun, Acc0, nil, PerMsgTO);
+recv(Refs, Keypos, Fun, Acc0, GlobalTimeout, PerMsgTO) ->
+ TimeoutRef = erlang:make_ref(),
+ {ok, TRef} = timer:send_after(GlobalTimeout, {timeout, TimeoutRef}),
+ try
+ process_mailbox(Refs, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO)
+ after
+ timer:cancel(TRef)
+ end.
+
+process_mailbox(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
+ case process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) of
+ {ok, Acc} ->
+ process_mailbox(RefList, Keypos, Fun, Acc, TimeoutRef, PerMsgTO);
+ {stop, Acc} ->
+ {ok, Acc};
+ Error ->
+ Error
+ end.
+
+process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
+ receive
+ {timeout, TimeoutRef} ->
+ timeout;
+ {Ref, Msg} ->
+ case lists:keyfind(Ref, Keypos, RefList) of
+ false ->
+ % this was some non-matching message which we will ignore
+ {ok, Acc0};
+ Worker ->
+ Fun(Msg, Worker, Acc0)
+ end;
+ {Ref, From, Msg} ->
+ case lists:keyfind(Ref, Keypos, RefList) of
+ false ->
+ {ok, Acc0};
+ Worker ->
+ Fun(Msg, {Worker, From}, Acc0)
+ end;
+ {rexi_DOWN, _RexiMonPid, ServerPid, Reason} = Msg ->
+ io:format("rexi_DOWN ~p ~p", [ServerPid, Reason]),
+ Fun(Msg, nil, Acc0)
+ after PerMsgTO ->
+ timeout
+ end.