]> granicus.if.org Git - ejabberd/commitdiff
* src/ejabberd_s2s_out.erl: Autodisconnect s2s connections which are overloaded ...
authorMickaël Rémond <mickael.remond@process-one.net>
Tue, 24 Jul 2007 09:04:09 +0000 (09:04 +0000)
committerMickaël Rémond <mickael.remond@process-one.net>
Tue, 24 Jul 2007 09:04:09 +0000 (09:04 +0000)
* src/p1_fsm.erl: Likewise.

SVN Revision: 821

ChangeLog
src/ejabberd_s2s_out.erl
src/p1_fsm.erl [new file with mode: 0644]

index c4a883836110473a3d87ebd15bbc094ae24029c2..182461b8aea35cee03f64a46685c2b3dab39dc61 100644 (file)
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,9 @@
+2007-07-24  Mickael Remond  <mickael.remond@process-one.net>
+
+       * src/ejabberd_s2s_out.erl: Autodisconnect s2s connections which
+       are overloaded (EJAB-287).
+       * src/p1_fsm.erl: Likewise.
+
 2007-07-19  Mickael Remond  <mickael.remond@process-one.net>
 
        * src/ejabberd_s2s_in.erl: Add s2s whitelist / blacklist support
        * src/mod_private_odbc.erl: Private storage support using odbc
        * src/odbc/pg.sql: Likewise
 
-2006-10-17 Mickael Remond  <mickael.remond@process-one.net>
+1999-11-30  Mickael Remond  <mickael.remond@process-one.net>
 
        * src/ejabberd_auth_ldap.erl: LDAP authentication now allows to
        match on several alternative attributes (thanks to Evgeniy
index 0effa91e3b7b3a76403a8500af2dc300ad068a14..4efb4183ec846f4606f71d17aa2b23217200397b 100644 (file)
@@ -1,12 +1,12 @@
 %%%----------------------------------------------------------------------
 %%% File    : ejabberd_s2s_out.erl
-%%% Author  : Alexey Shchepin <alexey@sevcom.net>
-%%% Purpose : 
-%%% Created :  6 Dec 2002 by Alexey Shchepin <alexey@sevcom.net>
+%%% Author  : Alexey Shchepin <alexey@process-one.net>
+%%% Purpose : Manage outgoing server-to-server connections
+%%% Created :  6 Dec 2002 by Alexey Shchepin <alexey@process-one.net>
 %%%----------------------------------------------------------------------
 
 -module(ejabberd_s2s_out).
--author('alexey@sevcom.net').
+-author('alexey@process-one.net').
 
 -behaviour(gen_fsm).
 
 -define(FSMOPTS, []).
 -endif.
 
+%% Only change this value if you now what your are doing:
+-define(FSMLIMITS,[]).
+%% -define(FSMLIMITS, [{max_queue, 2000}]).
+
 -define(STREAM_HEADER,
        "<?xml version='1.0'?>"
        "<stream:stream "
@@ -85,13 +89,14 @@ start(From, Host, Type) ->
     supervisor:start_child(ejabberd_s2s_out_sup, [From, Host, Type]).
 
 start_link(From, Host, Type) ->
-    gen_fsm:start_link(ejabberd_s2s_out, [From, Host, Type], ?FSMOPTS).
+    p1_fsm:start_link(ejabberd_s2s_out, [From, Host, Type],
+                      ?FSMLIMITS ++ ?FSMOPTS).
 
 start_connection(Pid) ->
-    gen_fsm:send_event(Pid, init).
+    p1_fsm:send_event(Pid, init).
 
 stop_connection(Pid) ->
-    gen_fsm:send_event(Pid, stop).
+    p1_fsm:send_event(Pid, stop).
 
 %%%----------------------------------------------------------------------
 %%% Callback functions from gen_fsm
@@ -105,6 +110,7 @@ stop_connection(Pid) ->
 %%          {stop, StopReason}                   
 %%----------------------------------------------------------------------
 init([From, Server, Type]) ->
+    process_flag(trap_exit, true),
     TLS = case ejabberd_config:get_local_option(s2s_use_starttls) of
              undefined ->
                  false;
@@ -269,12 +275,12 @@ wait_for_validation({xmlstreamelement, El}, StateData) ->
                {Pid, _Key, _SID} ->
                    case Type of
                        "valid" ->
-                           gen_fsm:send_event(
+                           p1_fsm:send_event(
                              Pid, {valid,
                                    StateData#state.server,
                                    StateData#state.myname});
                        _ ->
-                           gen_fsm:send_event(
+                           p1_fsm:send_event(
                              Pid, {invalid,
                                    StateData#state.server,
                                    StateData#state.myname})
@@ -525,7 +531,7 @@ reopen_socket({xmlstreamerror, _}, StateData) ->
 reopen_socket(timeout, StateData) ->
     {stop, normal, StateData};
 reopen_socket(closed, StateData) ->
-    gen_fsm:send_event(self(), init),
+    p1_fsm:send_event(self(), init),
     {next_state, open_socket, StateData}.
 
 
@@ -538,12 +544,12 @@ stream_established({xmlstreamelement, El}, StateData) ->
                {VPid, _VKey, _SID} ->
                    case VType of
                        "valid" ->
-                           gen_fsm:send_event(
+                           p1_fsm:send_event(
                              VPid, {valid,
                                     StateData#state.server,
                                     StateData#state.myname});
                        _ ->
-                           gen_fsm:send_event(
+                           p1_fsm:send_event(
                              VPid, {invalid,
                                     StateData#state.server,
                                     StateData#state.myname})
@@ -635,18 +641,6 @@ handle_info({send_element, El}, StateName, StateData) ->
                                                    timer = Timer}}
     end;
 
-%handle_info({tcp, Socket, Data}, StateName, StateData) ->
-%    xml_stream:send_text(StateData#state.xmlpid, Data),
-%    {next_state, StateName, StateData};
-%
-%handle_info({tcp_closed, Socket}, StateName, StateData) ->
-%    gen_fsm:send_event(self(), closed),
-%    {next_state, StateName, StateData};
-%
-%handle_info({tcp_error, Socket, Reason}, StateName, StateData) ->
-%    gen_fsm:send_event(self(), closed),
-%    {next_state, StateName, StateData};
-
 handle_info({timeout, Timer, _}, StateName,
            #state{timer = Timer} = StateData) ->
     {stop, normal, StateData};
diff --git a/src/p1_fsm.erl b/src/p1_fsm.erl
new file mode 100644 (file)
index 0000000..3fb5cdb
--- /dev/null
@@ -0,0 +1,683 @@
+%% ``The contents of this file are subject to the Erlang Public License,
+%% Version 1.1, (the "License"); you may not use this file except in
+%% compliance with the License. You should have received a copy of the
+%% Erlang Public License along with this software. If not, it can be
+%% retrieved via the world wide web at http://www.erlang.org/.
+%% 
+%% Software distributed under the License is distributed on an "AS IS"
+%% basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
+%% the License for the specific language governing rights and limitations
+%% under the License.
+%% 
+%% The Initial Developer of the Original Code is Ericsson Utvecklings AB.
+%% Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
+%% AB. All Rights Reserved.''
+%% 
+%% The code has been modified and improved by Process-one.
+%% Copyright 2007, Process-one
+%%
+%%  The change adds the following features:
+%%   - You can send exit(priority_shutdown) to the p1_fsm process to
+%%   terminate immediatetly. If the fsm trap_exit process flag has been
+%%   set to true, the FSM terminate function will called.
+%%   - You can pass the gen_fsm options to control resource usage.
+%%   {max_messages, N} will exit the process with priority_shutdown
+%%   - You can limit the time processing a message (TODO): If the
+%%   message processing does not return in a given period of time, the
+%%   process will be terminated.
+%% 
+%%     $Id$
+%%
+-module(p1_fsm).
+
+%%%-----------------------------------------------------------------
+%%%   
+%%% This state machine is somewhat more pure than state_lib.  It is
+%%% still based on State dispatching (one function per state), but
+%%% allows a function handle_event to take care of events in all states.
+%%% It's not that pure anymore :(  We also allow synchronized event sending.
+%%%
+%%% If the Parent process terminates the Module:terminate/2
+%%% function is called.
+%%%
+%%% The user module should export:
+%%%
+%%%   init(Args)
+%%%     ==> {ok, StateName, StateData}
+%%%         {ok, StateName, StateData, Timeout}
+%%%         ignore
+%%%         {stop, Reason}
+%%%
+%%%   StateName(Msg, StateData)
+%%%
+%%%    ==> {next_state, NewStateName, NewStateData}
+%%%        {next_state, NewStateName, NewStateData, Timeout}
+%%%        {stop, Reason, NewStateData}
+%%%              Reason = normal | shutdown | Term terminate(State) is called
+%%%
+%%%   StateName(Msg, From, StateData)
+%%%
+%%%    ==> {next_state, NewStateName, NewStateData}
+%%%        {next_state, NewStateName, NewStateData, Timeout}
+%%%        {reply, Reply, NewStateName, NewStateData}
+%%%        {reply, Reply, NewStateName, NewStateData, Timeout}
+%%%        {stop, Reason, NewStateData}
+%%%              Reason = normal | shutdown | Term terminate(State) is called
+%%%
+%%%   handle_event(Msg, StateName, StateData)
+%%%
+%%%    ==> {next_state, NewStateName, NewStateData}
+%%%        {next_state, NewStateName, NewStateData, Timeout}
+%%%        {stop, Reason, Reply, NewStateData}
+%%%        {stop, Reason, NewStateData}
+%%%              Reason = normal | shutdown | Term terminate(State) is called
+%%%
+%%%   handle_sync_event(Msg, From, StateName, StateData)
+%%%
+%%%    ==> {next_state, NewStateName, NewStateData}
+%%%        {next_state, NewStateName, NewStateData, Timeout}
+%%%        {reply, Reply, NewStateName, NewStateData}
+%%%        {reply, Reply, NewStateName, NewStateData, Timeout}
+%%%        {stop, Reason, Reply, NewStateData}
+%%%        {stop, Reason, NewStateData}
+%%%              Reason = normal | shutdown | Term terminate(State) is called
+%%%
+%%%   handle_info(Info, StateName) (e.g. {'EXIT', P, R}, {nodedown, N}, ...
+%%%
+%%%    ==> {next_state, NewStateName, NewStateData}
+%%%        {next_state, NewStateName, NewStateData, Timeout}
+%%%        {stop, Reason, NewStateData}
+%%%              Reason = normal | shutdown | Term terminate(State) is called
+%%%
+%%%   terminate(Reason, StateName, StateData) Let the user module clean up
+%%%        always called when server terminates
+%%%
+%%%    ==> the return value is ignored
+%%%
+%%%
+%%% The work flow (of the fsm) can be described as follows:
+%%%
+%%%   User module                           fsm
+%%%   -----------                          -------
+%%%     start              ----->             start
+%%%     init               <-----              .
+%%%
+%%%                                           loop
+%%%     StateName          <-----              .
+%%%
+%%%     handle_event       <-----              .
+%%%
+%%%     handle__sunc_event <-----              .
+%%%
+%%%     handle_info        <-----              .
+%%%
+%%%     terminate          <-----              .
+%%%
+%%%
+%%% ---------------------------------------------------
+
+-export([start/3, start/4,
+        start_link/3, start_link/4,
+        send_event/2, sync_send_event/2, sync_send_event/3,
+        send_all_state_event/2,
+        sync_send_all_state_event/2, sync_send_all_state_event/3,
+        reply/2,
+        start_timer/2,send_event_after/2,cancel_timer/1,
+        enter_loop/4, enter_loop/5, enter_loop/6]).
+
+-export([behaviour_info/1]).
+
+%% Internal exports
+-export([init_it/6, print_event/3,
+        system_continue/3,
+        system_terminate/4,
+        system_code_change/4,
+        format_status/2]).
+
+-import(error_logger , [format/2]).
+
+%%% Internal gen_fsm state
+%%% This state is used to defined resource control values:
+-record(limits, {max_queue}).
+
+%%% ---------------------------------------------------
+%%% Interface functions.
+%%% ---------------------------------------------------
+
+behaviour_info(callbacks) ->
+    [{init,1},{handle_event,3},{handle_sync_event,4},{handle_info,3},
+     {terminate,3},{code_change,4}];
+behaviour_info(_Other) ->
+    undefined.
+
+%%% ---------------------------------------------------
+%%% Starts a generic state machine.
+%%% start(Mod, Args, Options)
+%%% start(Name, Mod, Args, Options)
+%%% start_link(Mod, Args, Options)
+%%% start_link(Name, Mod, Args, Options) where:
+%%%    Name ::= {local, atom()} | {global, atom()}
+%%%    Mod  ::= atom(), callback module implementing the 'real' fsm
+%%%    Args ::= term(), init arguments (to Mod:init/1)
+%%%    Options ::= [{debug, [Flag]}]
+%%%      Flag ::= trace | log | {logfile, File} | statistics | debug
+%%%          (debug == log && statistics)
+%%% Returns: {ok, Pid} |
+%%%          {error, {already_started, Pid}} |
+%%%          {error, Reason}
+%%% ---------------------------------------------------
+start(Mod, Args, Options) ->
+    gen:start(?MODULE, nolink, Mod, Args, Options).
+
+start(Name, Mod, Args, Options) ->
+    gen:start(?MODULE, nolink, Name, Mod, Args, Options).
+
+start_link(Mod, Args, Options) ->
+    gen:start(?MODULE, link, Mod, Args, Options).
+
+start_link(Name, Mod, Args, Options) ->
+    gen:start(?MODULE, link, Name, Mod, Args, Options).
+
+
+send_event({global, Name}, Event) ->
+    catch global:send(Name, {'$gen_event', Event}),
+    ok;
+send_event(Name, Event) ->
+    Name ! {'$gen_event', Event},
+    ok.
+
+sync_send_event(Name, Event) ->
+    case catch gen:call(Name, '$gen_sync_event', Event) of
+       {ok,Res} ->
+           Res;
+       {'EXIT',Reason} ->
+           exit({Reason, {?MODULE, sync_send_event, [Name, Event]}})
+    end.
+
+sync_send_event(Name, Event, Timeout) ->
+    case catch gen:call(Name, '$gen_sync_event', Event, Timeout) of
+       {ok,Res} ->
+           Res;
+       {'EXIT',Reason} ->
+           exit({Reason, {?MODULE, sync_send_event, [Name, Event, Timeout]}})
+    end.
+
+send_all_state_event({global, Name}, Event) ->
+    catch global:send(Name, {'$gen_all_state_event', Event}),
+    ok;
+send_all_state_event(Name, Event) ->
+    Name ! {'$gen_all_state_event', Event},
+    ok.
+
+sync_send_all_state_event(Name, Event) ->
+    case catch gen:call(Name, '$gen_sync_all_state_event', Event) of
+       {ok,Res} ->
+           Res;
+       {'EXIT',Reason} ->
+           exit({Reason, {?MODULE, sync_send_all_state_event, [Name, Event]}})
+    end.
+
+sync_send_all_state_event(Name, Event, Timeout) ->
+    case catch gen:call(Name, '$gen_sync_all_state_event', Event, Timeout) of
+       {ok,Res} ->
+           Res;
+       {'EXIT',Reason} ->
+           exit({Reason, {?MODULE, sync_send_all_state_event,
+                          [Name, Event, Timeout]}})
+    end.
+
+%% Designed to be only callable within one of the callbacks
+%% hence using the self() of this instance of the process.
+%% This is to ensure that timers don't go astray in global
+%% e.g. when straddling a failover, or turn up in a restarted
+%% instance of the process.
+
+%% Returns Ref, sends event {timeout,Ref,Msg} after Time 
+%% to the (then) current state.
+start_timer(Time, Msg) ->
+    erlang:start_timer(Time, self(), {'$gen_timer', Msg}).
+
+%% Returns Ref, sends Event after Time to the (then) current state.
+send_event_after(Time, Event) ->
+    erlang:start_timer(Time, self(), {'$gen_event', Event}).
+
+%% Returns the remaing time for the timer if Ref referred to 
+%% an active timer/send_event_after, false otherwise.
+cancel_timer(Ref) ->
+    case erlang:cancel_timer(Ref) of
+       false ->
+           receive {timeout, Ref, _} -> 0
+           after 0 -> false 
+           end;
+       RemainingTime ->
+           RemainingTime
+    end.
+
+%% enter_loop/4,5,6
+%% Makes an existing process into a gen_fsm.
+%% The calling process will enter the gen_fsm receive loop and become a
+%% gen_fsm process.
+%% The process *must* have been started using one of the start functions
+%% in proc_lib, see proc_lib(3).
+%% The user is responsible for any initialization of the process,
+%% including registering a name for it.
+enter_loop(Mod, Options, StateName, StateData) ->
+    enter_loop(Mod, Options, StateName, StateData, self(), infinity).
+
+enter_loop(Mod, Options, StateName, StateData, ServerName = {_,_}) ->
+    enter_loop(Mod, Options, StateName, StateData, ServerName,infinity);
+enter_loop(Mod, Options, StateName, StateData, Timeout) ->
+    enter_loop(Mod, Options, StateName, StateData, self(), Timeout).
+
+enter_loop(Mod, Options, StateName, StateData, ServerName, Timeout) ->
+    Name = get_proc_name(ServerName),
+    Parent = get_parent(),
+    Debug = gen:debug_options(Options),
+    Limits= limit_options(Options),
+    loop(Parent, Name, StateName, StateData, Mod, Timeout, Debug, Limits).
+
+get_proc_name(Pid) when is_pid(Pid) ->
+    Pid;
+get_proc_name({local, Name}) ->
+    case process_info(self(), registered_name) of
+       {registered_name, Name} ->
+           Name;
+       {registered_name, _Name} ->
+           exit(process_not_registered);
+       [] ->
+           exit(process_not_registered)
+    end;
+get_proc_name({global, Name}) ->
+    case global:safe_whereis_name(Name) of
+       undefined ->
+           exit(process_not_registered_globally);
+       Pid when Pid==self() ->
+           Name;
+       _Pid ->
+           exit(process_not_registered_globally)
+    end.
+
+get_parent() ->
+    case get('$ancestors') of
+       [Parent | _] when is_pid(Parent) ->
+           Parent;
+       [Parent | _] when is_atom(Parent) ->
+           name_to_pid(Parent);
+       _ ->
+           exit(process_was_not_started_by_proc_lib)
+    end.
+
+name_to_pid(Name) ->
+    case whereis(Name) of
+       undefined ->
+           case global:safe_whereis_name(Name) of
+               undefined ->
+                   exit(could_not_find_registerd_name);
+               Pid ->
+                   Pid
+           end;
+       Pid ->
+           Pid
+    end.
+
+%%% ---------------------------------------------------
+%%% Initiate the new process.
+%%% Register the name using the Rfunc function
+%%% Calls the Mod:init/Args function.
+%%% Finally an acknowledge is sent to Parent and the main
+%%% loop is entered.
+%%% ---------------------------------------------------
+init_it(Starter, self, Name, Mod, Args, Options) ->
+    init_it(Starter, self(), Name, Mod, Args, Options);
+init_it(Starter, Parent, Name, Mod, Args, Options) ->
+    Debug = gen:debug_options(Options),
+    Limits= limit_options(Options),
+    case catch Mod:init(Args) of
+       {ok, StateName, StateData} ->
+           proc_lib:init_ack(Starter, {ok, self()}),       
+           loop(Parent, Name, StateName, StateData, Mod, infinity, Debug, Limits);
+       {ok, StateName, StateData, Timeout} ->
+           proc_lib:init_ack(Starter, {ok, self()}),       
+           loop(Parent, Name, StateName, StateData, Mod, Timeout, Debug, Limits);
+       {stop, Reason} ->
+           proc_lib:init_ack(Starter, {error, Reason}),
+           exit(Reason);
+       ignore ->
+           proc_lib:init_ack(Starter, ignore),
+           exit(normal);
+       {'EXIT', Reason} ->
+           proc_lib:init_ack(Starter, {error, Reason}),
+           exit(Reason);
+       Else ->
+           Error = {bad_return_value, Else},
+           proc_lib:init_ack(Starter, {error, Error}),
+           exit(Error)
+    end.
+
+%%-----------------------------------------------------------------
+%% The MAIN loop
+%%-----------------------------------------------------------------
+%% First we test if we have reach a defined limit ...
+loop(Parent, Name, StateName, StateData, Mod, Time, Debug, Limits) ->
+    try        
+       message_queue_len(Limits),
+       %% TODO: We can add more limit checking here...
+       process_message(Parent, Name, StateName, StateData,
+                       Mod, Time, Debug, Limits)
+    catch
+       {process_limit, Limit} ->
+           Reason = {process_limit, Limit},
+           Msg = {'EXIT', Parent, {error, {process_limit, Limit}}},
+           terminate(Reason, Name, Msg, Mod, StateName, StateData, Debug)
+    end.
+%% ... then we can process a new message:
+process_message(Parent, Name, StateName, StateData, Mod, Time, Debug, Limits) ->
+    Msg = receive
+             {'EXIT', Parent, priority_shutdown} ->
+                 {'EXIT', Parent, priority_shutdown}
+             after 0 ->
+                     receive
+                         Input ->
+                             Input
+                     after Time ->
+                             {'$gen_event', timeout}
+                     end
+             end,
+    case Msg of
+        {system, From, Req} ->
+           sys:handle_system_msg(Req, From, Parent, ?MODULE, Debug,
+                                 [Name, StateName, StateData,
+                                  Mod, Time, Limits]);
+       {'EXIT', Parent, Reason} ->
+           terminate(Reason, Name, Msg, Mod, StateName, StateData, Debug);
+       _Msg when Debug == [] ->
+           handle_msg(Msg, Parent, Name, StateName, StateData,
+                      Mod, Time, Limits);
+       _Msg ->
+           Debug1 = sys:handle_debug(Debug, {?MODULE, print_event}, 
+                                     {Name, StateName}, {in, Msg}),
+           handle_msg(Msg, Parent, Name, StateName, StateData,
+                      Mod, Time, Debug1, Limits)
+    end.
+
+%%-----------------------------------------------------------------
+%% Callback functions for system messages handling.
+%%-----------------------------------------------------------------
+%% TODO: Fix me
+system_continue(Parent, Debug, [Name, StateName, StateData,
+                               Mod, Time, Limits]) ->
+    loop(Parent, Name, StateName, StateData, Mod, Time, Debug, Limits).
+
+system_terminate(Reason, _Parent, Debug,
+                [Name, StateName, StateData, Mod, _Time, _Limits]) ->
+    terminate(Reason, Name, [], Mod, StateName, StateData, Debug).
+
+system_code_change([Name, StateName, StateData, Mod, Time, Limits],
+                  _Module, OldVsn, Extra) ->
+    case catch Mod:code_change(OldVsn, StateName, StateData, Extra) of
+       {ok, NewStateName, NewStateData} ->
+           {ok, [Name, NewStateName, NewStateData, Mod, Time, Limits]};
+       Else -> Else
+    end.
+
+%%-----------------------------------------------------------------
+%% Format debug messages.  Print them as the call-back module sees
+%% them, not as the real erlang messages.  Use trace for that.
+%%-----------------------------------------------------------------
+print_event(Dev, {in, Msg}, {Name, StateName}) ->
+    case Msg of
+       {'$gen_event', Event} ->
+           io:format(Dev, "*DBG* ~p got event ~p in state ~w~n",
+                     [Name, Event, StateName]);
+       {'$gen_all_state_event', Event} ->
+           io:format(Dev,
+                     "*DBG* ~p got all_state_event ~p in state ~w~n",
+                     [Name, Event, StateName]);
+       {timeout, Ref, {'$gen_timer', Message}} ->
+           io:format(Dev,
+                     "*DBG* ~p got timer ~p in state ~w~n",
+                     [Name, {timeout, Ref, Message}, StateName]);
+       {timeout, _Ref, {'$gen_event', Event}} ->
+           io:format(Dev,
+                     "*DBG* ~p got timer ~p in state ~w~n",
+                     [Name, Event, StateName]);
+       _ ->
+           io:format(Dev, "*DBG* ~p got ~p in state ~w~n",
+                     [Name, Msg, StateName])
+    end;
+print_event(Dev, {out, Msg, To, StateName}, Name) ->
+    io:format(Dev, "*DBG* ~p sent ~p to ~w~n"
+                  "      and switched to state ~w~n",
+             [Name, Msg, To, StateName]);
+print_event(Dev, return, {Name, StateName}) ->
+    io:format(Dev, "*DBG* ~p switched to state ~w~n",
+             [Name, StateName]).
+
+handle_msg(Msg, Parent, Name, StateName, StateData, Mod, _Time, Limits) -> %No debug here
+    From = from(Msg),
+    case catch dispatch(Msg, Mod, StateName, StateData) of
+       {next_state, NStateName, NStateData} ->     
+           loop(Parent, Name, NStateName, NStateData,
+                Mod, infinity, [], Limits);
+       {next_state, NStateName, NStateData, Time1} ->
+           loop(Parent, Name, NStateName, NStateData, Mod, Time1, [], Limits);
+        {reply, Reply, NStateName, NStateData} when From /= undefined ->
+           reply(From, Reply),
+           loop(Parent, Name, NStateName, NStateData,
+                Mod, infinity, [], Limits);
+        {reply, Reply, NStateName, NStateData, Time1} when From /= undefined ->
+           reply(From, Reply),
+           loop(Parent, Name, NStateName, NStateData, Mod, Time1, [], Limits);
+       {stop, Reason, NStateData} ->
+           terminate(Reason, Name, Msg, Mod, StateName, NStateData, []);
+       {stop, Reason, Reply, NStateData} when From /= undefined ->
+           {'EXIT', R} = (catch terminate(Reason, Name, Msg, Mod,
+                                          StateName, NStateData, [])),
+           reply(From, Reply),
+           exit(R);
+       {'EXIT', What} ->
+           terminate(What, Name, Msg, Mod, StateName, StateData, []);
+       Reply ->
+           terminate({bad_return_value, Reply},
+                     Name, Msg, Mod, StateName, StateData, [])
+    end.
+
+handle_msg(Msg, Parent, Name, StateName, StateData,
+          Mod, _Time, Debug, Limits) ->
+    From = from(Msg),
+    case catch dispatch(Msg, Mod, StateName, StateData) of
+       {next_state, NStateName, NStateData} ->
+           Debug1 = sys:handle_debug(Debug, {?MODULE, print_event}, 
+                                     {Name, NStateName}, return),
+           loop(Parent, Name, NStateName, NStateData,
+                Mod, infinity, Debug1, Limits);
+       {next_state, NStateName, NStateData, Time1} ->
+           Debug1 = sys:handle_debug(Debug, {?MODULE, print_event}, 
+                                     {Name, NStateName}, return),
+           loop(Parent, Name, NStateName, NStateData,
+                Mod, Time1, Debug1, Limits);
+        {reply, Reply, NStateName, NStateData} when From /= undefined ->
+           Debug1 = reply(Name, From, Reply, Debug, NStateName),
+           loop(Parent, Name, NStateName, NStateData,
+                Mod, infinity, Debug1, Limits);
+        {reply, Reply, NStateName, NStateData, Time1} when From /= undefined ->
+           Debug1 = reply(Name, From, Reply, Debug, NStateName),
+           loop(Parent, Name, NStateName, NStateData,
+                Mod, Time1, Debug1, Limits);
+       {stop, Reason, NStateData} ->
+           terminate(Reason, Name, Msg, Mod, StateName, NStateData, Debug);
+       {stop, Reason, Reply, NStateData} when From /= undefined ->
+           {'EXIT', R} = (catch terminate(Reason, Name, Msg, Mod,
+                                          StateName, NStateData, Debug)),
+           reply(Name, From, Reply, Debug, StateName),
+           exit(R);
+       {'EXIT', What} ->
+           terminate(What, Name, Msg, Mod, StateName, StateData, Debug);
+       Reply ->
+           terminate({bad_return_value, Reply},
+                     Name, Msg, Mod, StateName, StateData, Debug)
+    end.
+
+dispatch({'$gen_event', Event}, Mod, StateName, StateData) ->
+    Mod:StateName(Event, StateData);
+dispatch({'$gen_all_state_event', Event}, Mod, StateName, StateData) ->
+    Mod:handle_event(Event, StateName, StateData);
+dispatch({'$gen_sync_event', From, Event}, Mod, StateName, StateData) ->
+    Mod:StateName(Event, From, StateData);
+dispatch({'$gen_sync_all_state_event', From, Event},
+        Mod, StateName, StateData) ->
+    Mod:handle_sync_event(Event, From, StateName, StateData);
+dispatch({timeout, Ref, {'$gen_timer', Msg}}, Mod, StateName, StateData) ->
+    Mod:StateName({timeout, Ref, Msg}, StateData);
+dispatch({timeout, _Ref, {'$gen_event', Event}}, Mod, StateName, StateData) ->
+    Mod:StateName(Event, StateData);
+dispatch(Info, Mod, StateName, StateData) ->
+    Mod:handle_info(Info, StateName, StateData).
+
+from({'$gen_sync_event', From, _Event}) -> From;
+from({'$gen_sync_all_state_event', From, _Event}) -> From;
+from(_) -> undefined.
+
+%% Send a reply to the client.
+reply({To, Tag}, Reply) ->
+    catch To ! {Tag, Reply}.
+
+reply(Name, {To, Tag}, Reply, Debug, StateName) ->
+    reply({To, Tag}, Reply),
+    sys:handle_debug(Debug, {?MODULE, print_event}, Name,
+                    {out, Reply, To, StateName}).
+
+%%% ---------------------------------------------------
+%%% Terminate the server.
+%%% ---------------------------------------------------
+
+terminate(Reason, Name, Msg, Mod, StateName, StateData, Debug) ->
+    case catch Mod:terminate(Reason, StateName, StateData) of
+       {'EXIT', R} ->
+           error_info(R, Name, Msg, StateName, StateData, Debug),
+           exit(R);
+       _ ->
+           case Reason of
+               normal ->
+                   exit(normal);
+               shutdown ->
+                   exit(shutdown);
+               priority_shutdown ->
+                   %% Priority shutdown should be considered as
+                   %% shutdown by SASL
+                   exit(shutdown);
+               {process_limit, Limit} ->
+                   %% Priority shutdown should be considered as
+                   %% shutdown by SASL
+                   error_logger:error_msg("FSM limit reached (~p): ~p~n",
+                                          [self(), Limit]),
+                   exit(shutdown);
+               _ ->
+                   error_info(Reason, Name, Msg, StateName, StateData, Debug),
+                   exit(Reason)
+           end
+    end.
+
+error_info(Reason, Name, Msg, StateName, StateData, Debug) ->
+    Reason1 = 
+       case Reason of
+           {undef,[{M,F,A}|MFAs]} ->
+               case code:is_loaded(M) of
+                   false ->
+                       {'module could not be loaded',[{M,F,A}|MFAs]};
+                   _ ->
+                       case erlang:function_exported(M, F, length(A)) of
+                           true ->
+                               Reason;
+                           false ->
+                               {'function not exported',[{M,F,A}|MFAs]}
+                       end
+               end;
+           _ ->
+               Reason
+       end,
+    Str = "** State machine ~p terminating \n" ++
+       get_msg_str(Msg) ++
+       "** When State == ~p~n"
+        "**      Data  == ~p~n"
+        "** Reason for termination = ~n** ~p~n",
+    format(Str, [Name, get_msg(Msg), StateName, StateData, Reason1]),
+    sys:print_log(Debug),
+    ok.
+
+get_msg_str({'$gen_event', _Event}) ->
+    "** Last event in was ~p~n";
+get_msg_str({'$gen_sync_event', _Event}) ->
+    "** Last sync event in was ~p~n";
+get_msg_str({'$gen_all_state_event', _Event}) ->
+    "** Last event in was ~p (for all states)~n";
+get_msg_str({'$gen_sync_all_state_event', _Event}) ->
+    "** Last sync event in was ~p (for all states)~n";
+get_msg_str({timeout, _Ref, {'$gen_timer', _Msg}}) ->
+    "** Last timer event in was ~p~n";
+get_msg_str({timeout, _Ref, {'$gen_event', _Msg}}) ->
+    "** Last timer event in was ~p~n";
+get_msg_str(_Msg) ->
+    "** Last message in was ~p~n".
+
+get_msg({'$gen_event', Event}) -> Event;
+get_msg({'$gen_sync_event', Event}) -> Event;
+get_msg({'$gen_all_state_event', Event}) -> Event;
+get_msg({'$gen_sync_all_state_event', Event}) -> Event;
+get_msg({timeout, Ref, {'$gen_timer', Msg}}) -> {timeout, Ref, Msg};
+get_msg({timeout, _Ref, {'$gen_event', Event}}) -> Event;
+get_msg(Msg) -> Msg.
+
+%%-----------------------------------------------------------------
+%% Status information
+%%-----------------------------------------------------------------
+format_status(Opt, StatusData) ->
+    [PDict, SysState, Parent, Debug, [Name, StateName, StateData, Mod, _Time]] =
+       StatusData,
+    Header = lists:concat(["Status for state machine ", Name]),
+    Log = sys:get_debug(log, Debug, []),
+    Specfic = 
+       case erlang:function_exported(Mod, format_status, 2) of
+           true ->
+               case catch Mod:format_status(Opt,[PDict,StateData]) of
+                   {'EXIT', _} -> [{data, [{"StateData", StateData}]}];
+                   Else -> Else
+               end;
+           _ ->
+               [{data, [{"StateData", StateData}]}]
+       end,
+    [{header, Header},
+     {data, [{"Status", SysState},
+            {"Parent", Parent},
+            {"Logged events", Log},
+            {"StateName", StateName}]} |
+     Specfic].
+
+%%-----------------------------------------------------------------
+%% Resources limit management
+%%-----------------------------------------------------------------
+%% Extract know limit options
+limit_options(Options) ->
+    limit_options(Options, #limits{}).
+limit_options([], Limits) ->
+    Limits;
+%% Maximum number of messages allowed in the process message queue
+limit_options([{max_queue,N}|Options], Limits) 
+  when integer(N) ->
+    NewLimits = Limits#limits{max_queue=N},
+    limit_options(Options, NewLimits);
+limit_options([_|Options], Limits) ->
+    limit_options(Options, Limits).
+
+%% Throw max_queue if we have reach the max queue size
+%% Returns ok otherwise
+message_queue_len(#limits{max_queue = undefined}) ->
+    ok;
+message_queue_len(#limits{max_queue = MaxQueue}) ->
+    Pid = self(),
+    case process_info(Pid, message_queue_len) of
+        {message_queue_len, N} when N > MaxQueue ->
+           throw({process_limit, {max_queue, N}});
+       _ ->
+           ok
+    end.