nitrogen_core

% vim: sw=4 ts=4 et ft=erlang
% Nitrogen Web Framework for Erlang
% Copyright (c) 2008-2010 Rusty Klophaus
% See MIT-LICENSE for licensing information.

-module (action_comet).
-include("wf.hrl").
-compile(export_all).
-define (COMET_INTERVAL, 10 * 1000).
-define (ONE_SECOND, 1000).
-define (TEN_SECONDS, 10 * 1000).
-define (TWENTY_SECONDS, 20 * 1000).

%% TODO: This whole module needs to be spliy up into several gen_servers. One
%% for Guardians, one for Accumulators, and probably one for pools, too.  There
%% are so many naked bangs, it's a little hard to follow.


% Comet and polling/continuations are now handled using Nitrogen's asynchronous
% processing scheme. This allows you to fire up an asynchronous process with the
% #comet { fun=AsyncFunction } action.
%
% TERMINOLOGY:

% AsyncFunction - An Erlang function that executes in the background. The function
% generates Actions that are then sent to the browser via the accumulator. In addition
% each AsyncFunction is part of one (and only one) pool. The pool name provides a way
% to identify previously spawned processes, much like a Pid. Messages sent to the pool
% are distributed to all AsyncFunction processes in that pool.
%
% Pool - A pool contains one or more running AsyncFunctions. Any messages sent to the pool
% are distributed to all processes within the pool. Pools can either have a local
% or global scope. Local scope means that the pool applies only to the current
% series of page requests by a user. Global means that the pool applies to
% the entire system. Global pools provide the foundation for chat applets and 
% other interactive/multi-user software.
%
% Series - A series of requests to a Nitrogen resource. A series consists of 
% the first request plus any postbacks by the same visitor in the same browser
% window, and any connected websockets from that initial request.
% 
% Accumulator - There is one accumulator per series. The accumulator holds
% Nitrogen actions generated by AsyncFunctions, and is checked at the end
% of each Nitrogen request for anything that should be sent to the browser.
%
% AsyncGuardian - There is one AsyncGuardian for each AsyncFunction. The
% Guardian is responsible for informing the Accumulator when an AsyncFunction
% dies, and vice versa.

%%% INTERFACE %%%

%% NOTE: For comet/1, comet/2, and comet_global/2, there was a bug that would
%% cause the process to loop forever without being killed in the event that
%% wf:comet was called without javascript being loaded.  This bug is documented
%% here: https://github.com/nitrogen/nitrogen/issues/37
%%
%% The "Easy" fix would be to change those functions to start the comet call
%% inside the postback by changing function=Pid to function=F. Then we would
%% be guaranteed that the accumulator would be watching the pid and can kill it
%% on its own. But the drawback to this is that the comet function itself would
%% not start until the postback is received, meaning the function wouldn't be started
%% until javascript is loaded on the client and the postback received by the server.
%% This may be too much latency, so I've introduced a couple of hacks to make sure that
%% the pid is killed if the accumulator is never started (meaning never posted back).

%% @doc Convenience method to start a comet process.
-spec comet(comet_function()) -> {ok, pid()}.
comet(F) -> 
    comet(F, default).

%% @doc Convenience method to start a comet process.
-spec comet(CometFunction :: comet_function(), Pool :: term()) -> {ok, pid()}.
comet(CometFunction, Pool) ->
    comet(CometFunction, Pool, local).

%% @doc Convenience method to start a comet process with global pool.
-spec comet_global(CometFunction :: comet_function(), Pool :: term()) -> {ok, pid()}.
comet_global(CometFunction, Pool) ->
    comet(CometFunction, Pool, global).

-spec comet(CometFunction :: comet_function(), Pool :: term(), Scope :: local | global) -> {ok, pid()}.
comet(CometFunction, Pool, Scope) ->
    Pid = spawn_with_context(CometFunction, undefined, page),
    wf:wire(page, page, #comet { function=Pid, pool=Pool, scope=Scope }),
    {ok, Pid}.

status(Pid) ->
    {dictionary, PD} = erlang:process_info(Pid, dictionary), 
    Context = proplists:get_value(context, PD),
    PC = Context#context.page_context,
    Seriesid = PC#page_context.series_id,
    %?PRINT({seriesid, Seriesid}),
    {ok, AccPid} = get_accumulator_pid_no_start(Seriesid),
    AccPid ! info.

%% @doc Gather all wired actions, and send to the accumulator.
flush() ->
    %% If we've managed to establish a websocket connection after the comet
    %% process started, this will convert the current connection to use
    %% websockets. Then instead of returning the function calls from the comet
    %% request, it'll just send the actions to the websocket and the comet
    %% request will die peacefully.
    determine_async_mode(),

    %% First, let's render actions from the action queue into a binary for easy
    %% passing around with messages. Doing this step here will ensure the
    %% integrity of the action queue processing is maintained.
    {ok, Javascript} = wf_render_actions:render_action_queue(),
    JSBin = wf:to_binary(Javascript),

    AsyncMode = wf_context:async_mode(),
    inner_flush(AsyncMode, JSBin),
    ok.

inner_flush({websocket, Pid}, JSBin) ->
    flush_websocket(Pid, JSBin);

inner_flush(_, JSBin) ->
    flush_non_websocket(JSBin).

flush_websocket(Pid, JSBin) ->
    %% If there are any latent actions from an old comet process that
    %% was upgraded to a websocket, this will clear the actions and add
    %% them to the response. It's also safe to let the websocket
    %% process render them.
    AccumulatorActions = get_actions(),

    %% And we send both the "old commands" from the accumulator, and
    %% the new ones we just processed.
    Pid ! {comet_actions, [AccumulatorActions, JSBin]}.

flush_non_websocket(JSBin) ->
    SeriesID = wf_context:series_id(),
    {ok, AccumulatorPid} = get_accumulator_pid(SeriesID),
    %% Now we send that binary to the accumulator to be pulled.
    AccumulatorPid!{add_actions, JSBin}.

%% @doc Send a message to all processes in the specified local pool.
send(Pool, Message) ->
    inner_send(Pool, local, Message).

%% @doc Send a message to all processes in the specified global pool.
send_global(Pool, Message) ->
    inner_send(Pool, global, Message).

%%% - ACTION - %%%

render_action(Record) -> 
    % If the pool is undefined, then give it a random value.
    Record1 = case Record#comet.pool == undefined of
        true -> Record#comet { pool=erlang:make_ref() };
        false -> Record
    end,

    % This will immediately trigger a postback to event/1 below.
    #event {
        delegate=?MODULE,
        postback={spawn_async_function, Record1}
    }.



% This event is called to start a Nitrogen async loop.
% In the process of starting the function, it will create
% an accumulator and a pool if they don't already exist.
event({spawn_async_function, #comet{pool=Pool,
                                    scope=Scope,
                                    function=Function,
                                    dying_message=DyingMessage,
                                    reconnect_actions=ReconActions}}) ->
    SeriesID = wf_context:series_id(),
    
    % Get or start the accumulator process, which is used to hold any Nitrogen
    % Actions that are generated by async processes.
    {ok, AccumulatorPid} = get_accumulator_pid(SeriesID),

    % Get or start the pool process, which is a distributor that sends Erlang
    % messages to the running async function.
    {ok, PoolPid} = get_pool_pid(SeriesID, Pool, Scope), 

    % Create a process for the AsyncFunction...
    FunctionPid = create_comet_process(Function, ReconActions),

    % Create a process for the AsyncGuardian...
    GuardianFun = fun() ->
        guardian_process(FunctionPid, AccumulatorPid, PoolPid, DyingMessage)
    end,
    GuardianPid = erlang:spawn(GuardianFun),

    % Register the function with the accumulator and the pool.
    AccumulatorPid!{add_guardian, GuardianPid},
    PoolPid!{add_process, FunctionPid},

    % Only start the async event loop if it has not already been started...
    Actions = [
        <<"if (!document.comet_started) { document.comet_started=true; ">>,
            make_async_event(0),
        <<" }">>
    ],
    wf:wire(page, page, Actions);


% This clause is the heart of async functions. It is first triggered by the
% event/1 function above, and then continues to trigger itself in a loop, but
% in different ways depending on whether the page is doing comet-based or
% polling-based background updates.
%
% To update the page, the function gathers actions in the accumulator and wires
% both the actions and the looping event.
event(start_async) ->
    SeriesID = wf_context:series_id(),
    case get_accumulator_pid_no_start(SeriesID) of
        undefined ->
            wf:wire("document.comet_started=false; Nitrogen.$set_disconnected_no_notice(true); Nitrogen.$reconnect_system();");
        {ok, _Pid} ->
            case effective_async_mode() of
                {websocket, Pid} ->
                    start_async_websocket(Pid);
                comet ->
                    start_async_comet();
                {poll, Interval} ->
                    start_async_polling(Interval)
            end
    end;

event({maybe_reconnect, ReconTag, ReconActions}) ->
    Pid = get_process_tag_pid(ReconTag),
    maybe_reconnect(Pid, ReconTag, ReconActions).

maybe_reconnect(Pid, ReconTag, ReconActions) ->
    case is_pid(Pid) andalso is_process_alive(Pid) of
        true ->
            %% Our comet process is still alive, so we don't even need to run the
            %% "reconnection" events, we just renew the lease, and hijack the comet
            %% process to use the new connection. 
            set_lease(?COMET_INTERVAL + ?TEN_SECONDS),

            %% We also need to requeue up the reconnection postback, since each
            %% reconnection attempt is only run a single time.
            wire_reconnection(ReconTag, ReconActions);
        false ->
            %% Our comet process is dead, so let's run our reconnection event
            %% (which may restart it, or which may just say something like
            %% "Your process died due to inactivity")
            ?WF_IF(ReconActions, wf:wire(ReconActions))
    end.

start_async_websocket(Pid) ->
    %% Because we're using websocket, we don't have to do as much.
    %% We can get away with calling an async_event and sending it to
    %% the browser to serve as a ping and also to make sure the
    %% connection isn't hung.
    Event = make_async_event(?TEN_SECONDS),
    Pid ! {comet_actions, Event},
    register_websocket_pid_with_accumulator(Pid),
    set_lease(?COMET_INTERVAL + ?TEN_SECONDS).

start_async_comet() ->
    % Tell the accumulator to stay alive until
    % we call back, with some padding...
    set_lease(?COMET_INTERVAL + ?TEN_SECONDS),

    % Start the polling postback...
    Actions = get_actions_blocking(?COMET_INTERVAL),
    Event = make_async_event(0),
    wf:wire(page, page, [Actions, Event]),

    % Renew the lease, because the blocking call
    % could have used up a significant amount of time.
    set_lease(?COMET_INTERVAL + ?TEN_SECONDS).

start_async_polling(Interval) ->
    % Tell the accumulator to stay alive until
    % we call back, with some padding.
    set_lease(Interval + ?TEN_SECONDS),

    % Start the polling postback...
    Actions = get_actions(),
    Event = make_async_event(Interval),
    wf:wire(page, page, [Actions, Event]).

create_comet_process(Pid, _) when is_pid(Pid) ->
    notify_accumulator_checker(Pid),
    Pid;
create_comet_process(Function, ReconActions) ->
    Pid = spawn_with_context(Function, ReconActions, postback),
    notify_accumulator_checker(Pid),
    Pid.

determine_async_mode() ->
    %% Reconnections happen, and as such, we need to make sure our reconnected
    %% websocket is properly updated.
    case get_websocket_pid_from_accumulator() of
        undefined ->
            ok;
        {ok, Pid} -> 
            case is_process_alive(Pid) of
                true -> 
                    wf_context:async_mode({websocket, Pid});
                false ->
                    wf_context:async_mode(comet)
            end
    end.

effective_async_mode() ->
    case wf_context:async_mode() of
        {websocket, Pid} ->
            case is_process_alive(Pid) of
                true -> {websocket, Pid};
                false -> comet
            end;
        Other -> Other
    end.

%% - POOL - %%

% Retrieve a Pid from the process_registry for the specified pool.
% A pool can either be local or global. In a local pool, messages sent
% to the pool are only sent to async processes for one browser window.
% In a global pool, messages sent to the pool are sent to all processes
% in the pool across the entire system. This is useful for multi-user applications.
get_pool_pid(SeriesID, Pool, Scope) ->
    PoolID = case Scope of
        local  -> {Pool, SeriesID};
        global -> {Pool, global}
    end,
    PoolFun = fun() -> pool_loop([]) end,
    {ok, _Pid} = process_registry_handler:get_pid({async_pool, PoolID}, PoolFun).

% The pool loop keeps track of the AsyncFunction processes in a pool, 
% and is responsible for distributing messages to all processes in the pool.
pool_loop(Processes) -> 
    receive
        {add_process, JoinPid} ->
            case lists:member(JoinPid, Processes) of
                true -> pool_loop(Processes);
                false ->
                    erlang:monitor(process, JoinPid),
                    case Processes of
                        [] -> JoinPid!'INIT';
                        _  -> [Pid!{'JOIN', JoinPid} || Pid <- Processes]
                    end,
                    pool_loop([JoinPid|Processes])
            end;
        {'DOWN', _, process, LeavePid, _} ->
            [Pid!{'LEAVE', LeavePid} || Pid <- Processes],
            %?PRINT({connected_pid_left, LeavePid}),
            NewProcesses = Processes -- [LeavePid],
            case NewProcesses == [] of 
                false -> pool_loop(NewProcesses);
                true  -> erlang:exit({pool_loop, exiting_empty_pool})
            end;

        Message ->
            [Pid!Message || Pid <- Processes],
            pool_loop(Processes)

    after ?TWENTY_SECONDS ->
        NewProcesses = [X || X <- Processes, is_remote_process_alive(X)],
        case NewProcesses == [] of
            true  -> erlang:exit({pool_loop, exiting_empty_pool});
            false -> pool_loop(NewProcesses)
        end
    end.



%% - ACCUMULATOR - %%

% Retrieve a Pid from the process registry for the specified Series.
get_accumulator_pid(SeriesID) ->
    AccumulatorFun = fun() -> accumulator_loop([], [], none, undefined, undefined, []) end,
    Key = {async_accumulator, SeriesID},
    {ok, _Pid} = process_registry_handler:get_pid(Key, AccumulatorFun).

get_accumulator_pid_no_start(SeriesID) ->
    Key = {async_accumulator, SeriesID},
    case process_registry_handler:get_pid(Key) of
        {ok, Pid} when is_pid(Pid) -> {ok, Pid};
        undefined -> undefined
    end.


% The accumulator_loop keeps track of guardian processes within a pool,
% and gathers actions from the various AsyncFunctions in order 
% to send it the page when the actions are requested.
accumulator_loop(Guardians, Actions, Waiting, TimerRef, WebsocketPid, ProcessTags) ->
    %?PRINT({accululator_loop, self()}),
    receive
        {add_guardian, Pid} ->
            erlang:monitor(process, Pid),
            accumulator_loop([Pid|Guardians], Actions, Waiting, TimerRef, WebsocketPid, ProcessTags);

        {'DOWN', _, process, Pid, _} ->
            accumulator_loop(Guardians -- [Pid], Actions, Waiting, TimerRef, WebsocketPid, ProcessTags);

        {add_actions, NewActions} ->
            case is_remote_process_alive(Waiting) of
                true -> 
                    Waiting!{actions, [NewActions|Actions]},
                    accumulator_loop(Guardians, [], none, TimerRef, WebsocketPid, ProcessTags);
                false ->
                    accumulator_loop(Guardians, [NewActions|Actions], none, TimerRef, WebsocketPid, ProcessTags)
            end;

        {get_actions_blocking, Pid} when Actions == [] ->
            accumulator_loop(Guardians, [], Pid, TimerRef, WebsocketPid, ProcessTags);

        {get_actions_blocking, Pid} when Actions /= [] ->
            Pid!{actions, lists:reverse(Actions)},
            accumulator_loop(Guardians, [], none, TimerRef, WebsocketPid, ProcessTags);

        {get_actions, Pid} ->
            Pid!{actions, lists:reverse(Actions)},
            accumulator_loop(Guardians, [], none, TimerRef, WebsocketPid, ProcessTags);

        {set_lease, LengthInMS} ->
            timer:cancel(TimerRef),
            {ok, NewTimerRef} = timer:send_after(LengthInMS, {maybe_die, LengthInMS}),
            accumulator_loop(Guardians, Actions, Waiting, NewTimerRef, WebsocketPid, ProcessTags);

        {websocket_pid, NewWebsocketPid} ->
            accumulator_loop(Guardians, Actions, Waiting, TimerRef, NewWebsocketPid, ProcessTags);

        {get_websocket_pid, Pid} ->
            Pid!{websocket_pid, WebsocketPid},
            accumulator_loop(Guardians, Actions, Waiting, TimerRef, WebsocketPid, ProcessTags);

        {add_process_tag, ReconTag, FunPid} ->
            NewProcessTags0 = proplists:delete(ReconTag, ProcessTags),
            NewProcessTags = [{ReconTag, FunPid} | NewProcessTags0],
            accumulator_loop(Guardians, Actions, Waiting, TimerRef, WebsocketPid, NewProcessTags);

        {get_process_tag_pid, Pid, ReconTag} ->
            TaggedPid = case lists:keyfind(ReconTag, 1, ProcessTags) of
                false -> undefined;
                {_, P} -> P
            end,
            Pid!{process_tag_pid, TaggedPid},
            accumulator_loop(Guardians, Actions, Waiting, TimerRef, WebsocketPid, ProcessTags);
            

        {maybe_die, LengthInMS} -> 
            %% If we're using websockets and the websocket is alive, let's not
            %% die, instead, we just renew the lease again. This allows us to
            %% verify that the connection is still alive internally in order to
            %% kill the guardian process.
            %% If we're not using websockets, or the websocket pid is dead,
            %% then die.
            case is_pid(WebsocketPid) andalso erlang:is_process_alive(WebsocketPid) of
                true ->
                    {ok, NewTimerRef} = timer:send_after(LengthInMS, {maybe_die, LengthInMS}),
                    accumulator_loop(Guardians, Actions, Waiting, NewTimerRef, WebsocketPid, ProcessTags);
                false ->
                    % Guardian_process will detect that we've died and update
                    % the pool.
                    maybe_kill_waiting(Waiting),
                    erlang:exit({accumulator_loop, exiting_lease_expired})
            end;
        info ->
            Status = [{guardians, Guardians}, {actions, Actions}, {timer_ref, TimerRef}, {websocket_pid, WebsocketPid}, {process_tags, ProcessTags}],
            ?PRINT({accumulator_status, Status}),
            accumulator_loop(Guardians, Actions, Waiting, TimerRef, WebsocketPid, ProcessTags);

        Other ->
            ?PRINT({accumulator_loop, unhandled_event, Other}),
            accumulator_loop(Guardians, Actions, Waiting, TimerRef, WebsocketPid, ProcessTags)

    after ?TWENTY_SECONDS ->
        %% If we have no TimerRef, then the browser never performed
        %% the callback, so kill this process after 20 seconds.
        case TimerRef == undefined of
            true  ->
                maybe_kill_waiting(Waiting),
                erlang:exit({accumulator_loop, timeout});
            false -> accumulator_loop(Guardians, Actions, Waiting, TimerRef, WebsocketPid, ProcessTags)
        end
    end.

maybe_kill_waiting(Pid) when is_pid(Pid) ->
    Pid ! killed_when_accumulator_died;
maybe_kill_waiting(_) ->
    ok.


% The guardian process monitors the running AsyncFunction and
% the running Accumulator. If either one dies, then send 
% DyingMessage to the pool, and end.
guardian_process(FunctionPid, AccumulatorPid, PoolPid, DyingMessage) ->
    erlang:monitor(process, FunctionPid),
    erlang:monitor(process, AccumulatorPid),
    erlang:monitor(process, PoolPid),   
    receive
        {'DOWN', _, process, FunctionPid, _} ->
            % The AsyncFunction process has died. 
            % Communicate dying_message to the pool and exit.
            case DyingMessage of
                undefined -> ignore;
                _ -> PoolPid!DyingMessage
            end,
            exit(async_function_died);

        {'DOWN', _, process, AccumulatorPid, _} -> 
            % The accumulator process has died. 
            % Communicate dying_message to the pool, 
            % kill the AsyncFunction process, and exit.
            case DyingMessage of
                undefined -> ignore;
                _ -> PoolPid!DyingMessage
            end,
            erlang:exit(FunctionPid, async_die),
            erlang:exit({guardian_process, exiting_accumulator_died});

        {'DOWN', _, process, PoolPid, {pool_loop, exiting_empty_pool}} ->
            ?PRINT({empty_pool, PoolPid, {function_pid, FunctionPid}}),
            %% An empty pool is safe to kill without making a big deal of it
            erlang:exit({guardian_process, exiting_empty_pool});

        {'DOWN', _, process, PoolPid, Info} ->
            % The pool should never die on us.
            ?PRINT({unexpected_pool_death, Info}),
            erlang:exit({guardian_process, exiting_pool_died});

        Other ->
            ?PRINT({FunctionPid, AccumulatorPid, PoolPid}),
            ?PRINT({guardian_process, unhandled_event, Other}),
            guardian_process(FunctionPid, AccumulatorPid, PoolPid, DyingMessage)
    end.


%%% PRIVATE FUNCTIONS %%%

%% @doc Mode can be either atoms 'page' or 'postback'.
%% page means it's spawned from wf:comet(), meaning that the accumulator and whatnot
%% is not started until after the comet postback. So we have to run a checker to fix it
%% postbak means it's spawned from the postback event already along with the accumulator
%% and everything. So no need to check for it.
spawn_with_context(Function, ReconActions, Mode) when is_function(Function) ->
    spawn_with_context({undefined, Function, undefined}, ReconActions, Mode);
spawn_with_context({Name, Function}, ReconActions, Mode) ->
    spawn_with_context({Name, Function, undefined}, ReconActions, Mode);
spawn_with_context({Name, Function, Msg}, ReconActions, Mode) ->
    WrappedCometFun = create_wrapped_comet_fun(Function, Mode, ReconActions),
    Pid = start_comet_process(Name, Msg, WrappedCometFun),
    start_accumulator_check_timer(Mode, Pid),
    Pid.

create_wrapped_comet_fun(Function, Mode, ReconActions) ->
    Context = wf_context:context(),
    ReconTag = "recon" ++ wf:to_list(?WF_RAND_UNIFORM(1, 9999999999999999999999999999999999)),
    fun() ->
        wf_context:context(Context),
        wf_context:type(comet),
        wf_context:clear_action_queue(),
        register_process_tag(ReconTag, self()),
        wire_reconnection(ReconTag, ReconActions),
        flush(),
        run_comet_fun_maybe_with_mode(Function, Mode),
        wire_cancel_reconnection(ReconTag),
        flush()
    end.
   
wire_reconnection(ReconTag, ReconActions) ->
    Postback = {maybe_reconnect, ReconTag, ReconActions},
    Actions = #event{postback=Postback, delegate=?MODULE},
    Action = [<<"Nitrogen.$register_system_reconnection_event('">>,ReconTag,<<"', function() {">>, Actions, <<"}); ">>],
    wf:wire(page, page, Action).

wire_cancel_reconnection(undefined) ->
    ok;
wire_cancel_reconnection(ReconTag) ->
    Action = [<<"Nitrogen.$cancel_system_reconnection_event('">>,ReconTag,<<"');">>],
    wf:defer(page, page, Action).

start_comet_process(undefined, _, WrappedCometFun) ->
    erlang:spawn(WrappedCometFun);
start_comet_process(Name, Msg, WrappedCometFun) ->
    SeriesID = wf_context:series_id(),
    Key = {SeriesID, Name},
    {ok, Pid} = process_registry_handler:get_pid(Key, WrappedCometFun),
    maybe_send_reconnection_message(Pid, Msg),
    Pid.

run_comet_fun_maybe_with_mode(Function, Mode) ->
    case erlang:fun_info(Function, arity) of
        {arity, 1} -> Function(Mode);
        {arity, 0} -> Function()
    end.

%% When the accumulator 
maybe_send_reconnection_message(_Pid, undefined) ->
    ok;
maybe_send_reconnection_message(Pid, Msg) ->
    Pid ! Msg.

start_accumulator_check_timer(page, Pid) ->
    %% This function gets called before we've verified that the browser
    %% is even doing JS, and without JS, this will never get assigned 
    %% an accumulator.  So here's a hack to make sure that the accumulator
    %% exists after 20 seconds, and if not, it simply kills the pid
    start_accumulator_check_timer(Pid);
start_accumulator_check_timer(_, _) ->
    do_nothing.

%% If P is a pid, then it was initiated on the page
%% and that means we have to kill the checker pid that's
%% waiting to kill it for us.  Hacky, I know. Real
%% spaghetti-like. I don't particularly like it, but it
%% fixes the bug.
notify_accumulator_checker(Pid) ->
    case get_accumulator_checker_pid(Pid) of
        {ok, CheckerPid} when is_pid(CheckerPid) ->
            CheckerPid ! accumulator_started;
        _ ->
            ok
    end.

get_accumulator_checker_pid(Pid) ->
    Key = get_accumulator_checker_key(Pid),
    process_registry_handler:get_pid(Key).

get_accumulator_checker_key(Pid) ->
    _Key = {accumulator_check_timer,Pid}.

start_accumulator_check_timer(Pid) ->
    Key = get_accumulator_checker_key(Pid),
    CheckFunction = fun() ->
        receive
            accumulator_started -> ok
        after ?TWENTY_SECONDS + ?TEN_SECONDS ->
            %% The page never posted back at all (Hey, we gave it 30 seconds),
            %% and so the accumulator loop doesn't exist for this pid, and so
            %% we have nothing to check for it. So let's just kill the pid
            exit(Pid,accumulator_never_started)
        end
    end,

    {ok, _CheckPid} = process_registry_handler:get_pid(Key,CheckFunction).


inner_send(Pool, Scope, Message) ->
    % ?PRINT({Pool, Scope, Message}),
    SeriesID = wf_context:series_id(),
    {ok, PoolPid} = get_pool_pid(SeriesID, Pool, Scope),
    PoolPid!Message,
    ok.

% Get actions from accumulator. If there are no actions currently in the
% accumulator, then [] is immediately returned.
get_actions() ->
    SeriesID = wf_context:series_id(),
    {ok, Pid} = get_accumulator_pid(SeriesID),
    get_actions_from_accumulator(Pid).

get_actions_from_accumulator(AccumulatorPid) ->
    AccumulatorPid!{get_actions, self()},
    receive
        {actions, X} -> X
    after 100 ->
        []
    end.

get_actions_and_register_new_websocket_pid(WebsocketPid) ->
    SeriesID = wf_context:series_id(),
    case get_accumulator_pid_no_start(SeriesID) of
        undefined -> [];
        {ok, Pid} ->
            Pid ! {websocket_pid, WebsocketPid},
            get_actions_from_accumulator(Pid)
    end.

register_process_tag(ReconTag, Pid) ->
    SeriesID = wf_context:series_id(),
    {ok, AccPid} = get_accumulator_pid(SeriesID),
    AccPid!{add_process_tag, ReconTag, Pid}.

% Get actions from accumulator in a blocking fashion. If there are no actions
% currently in the accumulator, then this blocks for up to Timeout milliseconds.
% This works by telling Erlang to send a dummy 'add_actions' command to the accumulator
% that will be executed when the timeout expires.
get_actions_blocking(Timeout) ->
    SeriesID = wf_context:series_id(),
    {ok, AccumulatorPid} = get_accumulator_pid(SeriesID),
    AccumulatorNode = node(AccumulatorPid),
    TimerRef = rpc:call(AccumulatorNode, erlang, send_after, [Timeout, AccumulatorPid, {add_actions, []}]),
    HardTimeout = Timeout + 10000,

    AccumulatorPid!{get_actions_blocking, self()},
    receive 
        {actions, X} -> erlang:cancel_timer(TimerRef), X;           
        Other -> ?PRINT({unhandled_event, Other}), []
    after HardTimeout ->
        %?PRINT({no_comet_actions_checking_pid, AccumulatorPid}),
        case erlang:is_process_alive(AccumulatorPid) of
            true ->
                %?PRINT({is_alive_relooping}),
                erlang:cancel_timer(TimerRef),
                get_actions_blocking(Timeout);
            false ->
                erlang:cancel_timer(TimerRef),
                ?PRINT({accumulator_is_dead,AccumulatorPid}),
                []
         end
    end.

set_lease(LengthInMS) ->
    SeriesID = wf_context:series_id(),
    {ok, AccumulatorPid} = get_accumulator_pid(SeriesID),
    AccumulatorPid!{set_lease, LengthInMS}.

register_websocket_pid_with_accumulator(WebsocketPid) ->
    SeriesID = wf_context:series_id(),
    {ok, AccumulatorPid} = get_accumulator_pid(SeriesID),
    AccumulatorPid!{websocket_pid, WebsocketPid}.

get_websocket_pid_from_accumulator() ->
    SeriesID = wf_context:series_id(),
    {ok, AccumulatorPid} = get_accumulator_pid(SeriesID),
    AccumulatorPid!{get_websocket_pid, self()},
    receive
        {websocket_pid, undefined} -> undefined;
        {websocket_pid, WebsocketPid} -> {ok, WebsocketPid}
    after
        10 -> undefined
    end.

get_process_tag_pid(ReconTag) ->
    SeriesID = wf_context:series_id(),
    case get_accumulator_pid_no_start(SeriesID) of
        undefined -> undefined;
        {ok, AccPid} ->
            AccPid ! {get_process_tag_pid, self(), ReconTag},
            receive
                {process_tag_pid, Pid} -> Pid
            after 10
                -> undefined
            end
    end.

% Convenience function to return an #event that will call event(start_async) above.
make_async_event(Interval) ->
    #event { type=system, delay=Interval, delegate=?MODULE, postback=start_async }.

% Return true if the process is alive, accounting for processes on other nodes. 
is_remote_process_alive(Pid) when is_pid(Pid) ->
    Node = node(Pid),
    case Node==node() of
        true ->
            erlang:is_process_alive(Pid);
        false ->
            pong == net_adm:ping(node(Pid))
                andalso rpc:call(node(Pid), erlang, is_process_alive, [Pid])
    end;
is_remote_process_alive(_) ->
    false.