rexi_example.erl

-module(rexi_example).

-export([setup/0, go/0, oh/0]).

setup() ->
    % assuming dist. erlang already set up across nodes
    rpc:multicall(rexi, start, []).

%% submit the jobs to individual workers (a little verbose for illustration)
go() ->
    {M,F,A} = {rexi_example, oh, []},
    Workers = lists:map(fun(Node) ->
        {rexi:cast(Node, {M,F,A})} % tuple is for keyfind in rexi_utils
    end, nodes()),
    KeyPos = 1, % place in tuple for keyfind
    Acc0 = {length(Workers), []},
    GlobalTimeout = 5000,
    PerMsgTimeout = infinity,
    rexi_utils:recv(Workers, KeyPos, fun handle_message/3, Acc0,
                    GlobalTimeout, PerMsgTimeout).

%% handles messages from individual workers
handle_message({rexi_DOWN, _, _, _}, _Worker, Acc0) ->
    Acc0;
handle_message({rexi_EXIT, Reason}, _Worker, Acc0) ->
    io:format("rexi_EXIT: ~p", [Reason]),
    Acc0;
handle_message({ok, Msg}, _Worker, {C0, Acc0}) ->
    % good msg, so decrement counter and add Msg to Acc0
    {C1, Acc1} = {C0 - 1, [Msg | Acc0]},
    case C0 of
    1 -> {stop, Acc1};
    _ -> {ok, {C1, Acc1}}
    end;
handle_message(Msg, _Worker, Acc0) ->
    io:format("bad response: ~p", [Msg]),
    Acc0.

%% RPC endpoint
oh() ->
    rexi:reply({ok, hai}).