comet_request_handler.erl

comet(Req) ->
    Body = Req:recv_body(),
    io:format("~nBody: ~p~n", [Body]),
    Socket = Req:get(socket),
    inet:setopts(Socket, [{active, once}]),
    Response = connection:handle_json(Body),
    inet:setopts(Socket, [{active, false}]),
    io:format("~nSending Response: ~s~n", [Response]),
    Req:ok({"application/json", [], Response}).

connection_fsm.erl

%%%-------------------------------------------------------------------
%%% File    : connection.erl
%%% Author  : Jon Rosebaugh <[email protected]>
%%% Description : 
%%%
%%% Created : 28 Jan 2010 by Jon Rosebaugh <[email protected]>
%%%-------------------------------------------------------------------
-module(connection).

-behaviour(gen_fsm).

-define(NO_REQUEST_TIMEOUT, 45000).
-define(REQUEST_WAIT_TIME, 60000).

%% API
-export([handle_json/1, handle_packet/2]).

%% gen_fsm callbacks
-export([init/1, waiting/2, waiting/3, have_packet/2, have_packet/3, have_request/2, have_request/3, handle_event/3,
	 handle_sync_event/4, handle_info/3, terminate/3, code_change/4]).

-record(state, {connectionid, waitingpid, waitingpackets, dietimer}).

whereis_name(ConnectionID) ->
    global:whereis_name({?MODULE, ConnectionID}).

%%====================================================================
%% API
%%====================================================================

% if there is no sid, the response has to include a sid.

handle_json(Body) ->
    {struct, Props} = mochijson2:decode(Body),
    ResponseProps = case proplists:is_defined(<<"sid">>, Props) of
			false ->
			    handle_setup(Props);
			true ->
			    ConnectionID = proplists:get_value(<<"sid">>, Props),
			    Rid = proplists:get_value(<<"rid">>, Props),
			    {struct, Request} = proplists:get_value(<<"request">>, Props, {struct, []}),
			    handle_request(ConnectionID, Rid, Request)
		    end,
    mochijson2:encode({struct, ResponseProps}).

handle_setup(Props) ->
    Rid = proplists:get_value(<<"rid">>, Props),
    ConnectionID = start(),
    [{<<"ack">>, Rid}, {<<"sid">>, ConnectionID}, {<<"response">>, empty_response()}].

handle_request(ConnectionID, Rid, _Request) ->
    gen_fsm:send_event(whereis_name(ConnectionID), {request, self()}),
    io:format("~nGonna Wait~n"),
    Response = receive
		   {tcp_closed, _Socket} ->
		       io:format("~nSocket Died~n"),
		       gen_fsm:send_event(whereis_name(ConnectionID), request_died),
		       who_cares;
		   {packet, Packet} ->
		       Packet
	       end,
    [{<<"ack">>, Rid}, {<<"response">>, Response}].

handle_packet(ConnectionID, Packet) ->
    gen_fsm:send_event(whereis_name(ConnectionID), {packet, Packet}).

%%--------------------------------------------------------------------
%% Function: start_link() -> ok,Pid} | ignore | {error,Error}
%% Description:Creates a gen_fsm process which calls Module:init/1 to
%% initialize. To ensure a synchronized start-up procedure, this function
%% does not return until Module:init/1 has returned.  
%%--------------------------------------------------------------------
start() ->
    ConnectionID = connectionidgenerator:getconnectionid(),
    {ok, _Pid} = gen_fsm:start({global, {?MODULE, ConnectionID}}, ?MODULE, {ConnectionID}, []),
    ConnectionID.

%%====================================================================
%% gen_fsm callbacks
%%====================================================================
%%--------------------------------------------------------------------
%% Function: init(Args) -> {ok, StateName, State} |
%%                         {ok, StateName, State, Timeout} |
%%                         ignore                              |
%%                         {stop, StopReason}                   
%% Description:Whenever a gen_fsm is started using gen_fsm:start/[3,4] or
%% gen_fsm:start_link/3,4, this function is called by the new process to 
%% initialize. 
%%--------------------------------------------------------------------
init({ConnectionID}) ->
    State = #state{connectionid=ConnectionID, waitingpid=nil, waitingpackets=queue:new(), dietimer=make_die_timer()},
    io:format("~nStarting State: ~p~n", [State]),
    {ok, waiting, State}.

%%--------------------------------------------------------------------
%% Function: 
%% state_name(Event, State) -> {next_state, NextStateName, NextState}|
%%                             {next_state, NextStateName, 
%%                                NextState, Timeout} |
%%                             {stop, Reason, NewState}
%% Description:There should be one instance of this function for each possible
%% state name. Whenever a gen_fsm receives an event sent using
%% gen_fsm:send_event/2, the instance of this function with the same name as
%% the current state name StateName is called to handle the event. It is also 
%% called if a timeout occurs. 
%%--------------------------------------------------------------------
waiting({timeout, _Ref, no_request_death}, State) ->
    io:format("~nAll these moments will be lost in time... like tears in rain...~n"),
    NewState = State#state{dietimer=nil},
    io:format("~nNew State: ~p~n", [NewState]),
    {stop, no_request_death, NewState};

waiting({request, Pid}, State) ->
    gen_fsm:cancel_timer(State#state.dietimer),
    NewState = State#state{waitingpid=Pid, dietimer=nil},
    io:format("~nNew State: ~p~n", [NewState]),
    {next_state, have_request, NewState, ?REQUEST_WAIT_TIME};

waiting({packet, NewPacket}, State) ->
    Waiting = State#state.waitingpackets,
    NewWaiting = queue:in(NewPacket, Waiting),
    NewState = State#state{waitingpackets=NewWaiting},
    io:format("~nNew State: ~p~n", [NewState]),
    {next_state, have_packet, NewState}.

have_request({request, Pid}, State) ->
    State#state.waitingpid ! {packet, empty_response()},
    NewState = State#state{waitingpid=Pid},
    io:format("~nNew State: ~p~n", [NewState]),
    {next_state, have_request, NewState};
    
have_request({packet, NewPacket}, State) ->
    State#state.waitingpid ! {packet, [NewPacket]},
    NewState = State#state{waitingpid=nil, dietimer=make_die_timer()},
    io:format("~nNew State: ~p~n", [NewState]),
    {next_state, waiting, NewState};

have_request(request_died, State) ->
    NewState = State#state{waitingpid=nil, dietimer=make_die_timer()},
    io:format("~nNew State: ~p~n", [NewState]),
    {next_state, waiting, NewState};

have_request(timeout, State) ->    
    % We've had a request for ?REQUEST_WAIT_TIME ms with no packets. Recycle it just to be safe.
    State#state.waitingpid ! {packet, empty_response()},
    NewState = State#state{waitingpid=nil, dietimer=make_die_timer()},
    io:format("~nNew State: ~p~n", [NewState]),
    {next_state, waiting, NewState}.

have_packet({timeout, _Ref, no_request_death}, State) ->
    io:format("~nAll these moments will be lost in time... like tears in rain...~n"),
    NewState = State#state{dietimer=nil},
    io:format("~nNew State: ~p~n", [NewState]),
    {stop, no_request_death, NewState};

have_packet({request, Pid}, State) ->    
    Pid ! {packet, queue:to_list(State#state.waitingpackets)},
    % Reset dietimer
    gen_fsm:cancel_timer(State#state.dietimer),
    NewState = State#state{waitingpackets=queue:new(), dietimer=make_die_timer()},
    io:format("~nNew State: ~p~n", [NewState]),
    {next_state, waiting, NewState};

have_packet({packet, NewPacket}, State) ->
    Waiting = State#state.waitingpackets,
    NewWaiting = queue:in(NewPacket, Waiting),
    NewState = State#state{waitingpackets=NewWaiting},
    io:format("~nNew State: ~p~n", [NewState]),
    {next_state, have_packet, NewState}.    

%%--------------------------------------------------------------------
%% Function:
%% state_name(Event, From, State) -> {next_state, NextStateName, NextState} |
%%                                   {next_state, NextStateName, 
%%                                     NextState, Timeout} |
%%                                   {reply, Reply, NextStateName, NextState}|
%%                                   {reply, Reply, NextStateName, 
%%                                    NextState, Timeout} |
%%                                   {stop, Reason, NewState}|
%%                                   {stop, Reason, Reply, NewState}
%% Description: There should be one instance of this function for each
%% possible state name. Whenever a gen_fsm receives an event sent using
%% gen_fsm:sync_send_event/2,3, the instance of this function with the same
%% name as the current state name StateName is called to handle the event.
%%--------------------------------------------------------------------

%% waiting(request, From, State) ->
%%     Waiting = State#state.waitingpids,
%%     NewWaiting = queue:in(From, Waiting)

waiting(_Event, _From, State) ->
    Reply = ok,
    {reply, Reply, waiting, State}.

have_packet(_Event, _From, State) ->
    Reply = ok,
    {reply, Reply, have_packet, State}.

have_request(_Event, _From, State) ->
    Reply = ok,
    {reply, Reply, have_request, State}.

%%--------------------------------------------------------------------
%% Function: 
%% handle_event(Event, StateName, State) -> {next_state, NextStateName, 
%%						  NextState} |
%%                                          {next_state, NextStateName, 
%%					          NextState, Timeout} |
%%                                          {stop, Reason, NewState}
%% Description: Whenever a gen_fsm receives an event sent using
%% gen_fsm:send_all_state_event/2, this function is called to handle
%% the event.
%%--------------------------------------------------------------------
handle_event(_Event, StateName, State) ->
    {next_state, StateName, State}.

%%--------------------------------------------------------------------
%% Function: 
%% handle_sync_event(Event, From, StateName, 
%%                   State) -> {next_state, NextStateName, NextState} |
%%                             {next_state, NextStateName, NextState, 
%%                              Timeout} |
%%                             {reply, Reply, NextStateName, NextState}|
%%                             {reply, Reply, NextStateName, NextState, 
%%                              Timeout} |
%%                             {stop, Reason, NewState} |
%%                             {stop, Reason, Reply, NewState}
%% Description: Whenever a gen_fsm receives an event sent using
%% gen_fsm:sync_send_all_state_event/2,3, this function is called to handle
%% the event.
%%--------------------------------------------------------------------
handle_sync_event(_Event, _From, StateName, State) ->
    Reply = ok,
    {reply, Reply, StateName, State}.

%%--------------------------------------------------------------------
%% Function: 
%% handle_info(Info,StateName,State)-> {next_state, NextStateName, NextState}|
%%                                     {next_state, NextStateName, NextState, 
%%                                       Timeout} |
%%                                     {stop, Reason, NewState}
%% Description: This function is called by a gen_fsm when it receives any
%% other message than a synchronous or asynchronous event
%% (or a system message).
%%--------------------------------------------------------------------
handle_info(_Info, StateName, State) ->
    {next_state, StateName, State}.

%%--------------------------------------------------------------------
%% Function: terminate(Reason, StateName, State) -> void()
%% Description:This function is called by a gen_fsm when it is about
%% to terminate. It should be the opposite of Module:init/1 and do any
%% necessary cleaning up. When it returns, the gen_fsm terminates with
%% Reason. The return value is ignored.
%%--------------------------------------------------------------------
terminate(_Reason, _StateName, _State) ->
    ok.

%%--------------------------------------------------------------------
%% Function:
%% code_change(OldVsn, StateName, State, Extra) -> {ok, StateName, NewState}
%% Description: Convert process state when code is changed
%%--------------------------------------------------------------------
code_change(_OldVsn, StateName, State, _Extra) ->
    {ok, StateName, State}.

%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------

make_die_timer() ->
    gen_fsm:start_timer(?NO_REQUEST_TIMEOUT, no_request_death).

empty_response() ->
    [{struct, []}].

connection_id_generator.erl

% This is the internals of a gen_server that provides session id generation for the entire Erlang cluster; it's registered globally so there's only one running.

initial_state() ->
    crypto:rand_uniform(100000, 900001).

increment_counter(OldValue) ->
    OldValue + crypto:rand_uniform(1000, 9001).