commit 007d55bd38b60e2d7dca65eeaafdf02d733ae9a2
author Brad Anderson <brad@cloudant.com> Tue Aug 17 22:04:15 2010 -0400
committer Adam Kocoloski <adam@cloudant.com> Fri Aug 27 16:52:26 2010 -0400
tree b1ea0fa5ca7849f014de1df3be716dddf4b55845
parent cd563bee7d845d9bd2d969f2a7cb72f94bc5fc72

split some rexi utilities out from fabric

ebin/rexi.app

diff --git a/ebin/rexi.app b/ebin/rexi.app
index 620c886..5c944b7 100644
--- a/ebin/rexi.app
+++ b/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,[]}}

src/rexi_utils.erl

diff --git a/src/rexi_utils.erl b/src/rexi_utils.erl
new file mode 100644
index 0000000..7918395
--- /dev/null
+++ b/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.