gen_fsm for gatherer

TODO.md

@@ -1,5 +1,4 @@
 TODO
 =====
-  - Fix gatherer state.
+  - Fix problem of infinite consecutive errors.
   - Limit number of emails that can be sent in a period of time.
-  - gen_fsm for gatherer?

apps/log_monitor/src/gatherer.erl

@@ -8,21 +8,23 @@
 %%%-------------------------------------------------------------------
 -module(gatherer).
 
--behaviour(gen_server).
+-behaviour(gen_fsm).
 
 %% API
 -export([start_link/2]).
 
-%% gen_server callbacks
+%% gen_fsm callbacks
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+-export([init/1, handle_event/3, handle_sync_event/4,
-         terminate/2, code_change/3]).
+         handle_info/3, terminate/3, code_change/4]).
+-export([state_off/2, state_on/2]).
 
 -define(SERVER, ?MODULE).
 -define(DEFAULT_TIMER_TIME, 1000).
 -define(DEFAULT_SAFE_TIMER_TIME, 30000).
 
 -record(log, {file, error_regex}).
-%% -record(state, {}).
+-record(state_off, {log}).
+-record(state_on, {log, error, until}).
 
 %%%===================================================================
 %%% API
@@ -30,110 +32,156 @@
 
 %%--------------------------------------------------------------------
 %% @doc
-%% Starts the server
+%% 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.
 %%
-%% @spec start_link(File, ErrorRegex) -> {ok, Pid} |
+%% @spec start_link() -> {ok, Pid} | ignore | {error, Error}
-%%                                       ignore |
-%%                                       {error, Error}
 %% @end
 %%--------------------------------------------------------------------
 start_link(File, ErrorRegex) ->
-    gen_server:start_link(?MODULE, [#log{file = File, error_regex = ErrorRegex}], []).
+    gen_fsm:start_link(?MODULE, [File, ErrorRegex], []).
 
 %%%===================================================================
-%%% gen_server callbacks
+%%% gen_fsm callbacks
 %%%===================================================================
 
 %%--------------------------------------------------------------------
 %% @private
 %% @doc
-%% Initializes the server
+%% 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.
 %%
-%% @spec init(Args) -> {ok, State} |
+%% @spec init(Args) -> {ok, StateName, State} |
-%%                     {ok, State, Timeout} |
+%%                     {ok, StateName, State, Timeout} |
 %%                     ignore |
-%%                     {stop, Reason}
+%%                     {stop, StopReason}
 %% @end
 %%--------------------------------------------------------------------
-init([Log]) ->
+init([File, ErrorRegex]) ->
-    {ok, [off, Log]}.
+    {ok, state_off, #state_off{log = #log{file = File, error_regex = ErrorRegex}}}.
 
 %%--------------------------------------------------------------------
 %% @private
 %% @doc
-%% Handling call messages
+%% 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.
 %%
-%% @spec handle_call(Request, From, State) ->
+%% @spec state_off(Event, State) ->
-%%                                   {reply, Reply, State} |
+%%                   {next_state, NextStateName, NextState} |
-%%                                   {reply, Reply, State, Timeout} |
+%%                   {next_state, NextStateName, NextState, Timeout} |
-%%                                   {noreply, State} |
+%%                   {stop, Reason, NewState}
-%%                                   {noreply, State, Timeout} |
-%%                                   {stop, Reason, Reply, State} |
-%%                                   {stop, Reason, State}
 %% @end
 %%--------------------------------------------------------------------
-handle_call(_Request, _From, State) ->
+state_off({log_line, Text}, State = #state_off{log = Log}) ->
-    Reply = ok,
+    case isError(Text, Log#log.error_regex) of
-    {reply, Reply, State}.
-
-%%--------------------------------------------------------------------
-%% @private
-%% @doc
-%% Handling cast messages
-%%
-%% @spec handle_cast(Msg, State) -> {noreply, State} |
-%%                                  {noreply, State, Timeout} |
-%%                                  {stop, Reason, State}
-%% @end
-%%--------------------------------------------------------------------
-handle_cast(_Msg, State) ->
-    {noreply, State}.
-
-%%--------------------------------------------------------------------
-%% @private
-%% @doc
-%% Handling all non call/cast messages
-%%
-%% @spec handle_info(Info, State) -> {noreply, State} |
-%%                                   {noreply, State, Timeout} |
-%%                                   {stop, Reason, State}
-%% @end
-%%--------------------------------------------------------------------
-handle_info({log_line, Text}, [off, Log = #log{error_regex = ErrorRegex}]) ->
-    case isError(Text, ErrorRegex) of
         true ->
-            {ok, Timer} = timer:send_after(timer_time(), {timeout}),
+            Timeout = timer_time(),
-            {ok, SafeTimer} = timer:send_after(safe_timer_time(), {timeout}),
+            Now = to_milliseconds(os:timestamp()),
-            {noreply, [on, Log, Text, Timer, SafeTimer]};
+            Until = Now + Timeout,
-        false -> {noreply, [off, Log]}
+            {next_state, state_on, #state_on{log = Log, error = Text, until = Until}, Timeout};
-    end;
+        false -> {next_state, state_off, State}
-handle_info({log_line, Text}, [on, Log = #log{error_regex = ErrorRegex}, Error, Timer, SafeTimer]) ->
+    end.
-    case isError(Text, ErrorRegex) of
+
+%%--------------------------------------------------------------------
+%% @private
+%% @doc
+%% 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.
+%%
+%% @spec state_on(Event, State) ->
+%%                   {next_state, NextStateName, NextState} |
+%%                   {next_state, NextStateName, NextState, Timeout} |
+%%                   {stop, Reason, NewState}
+%% @end
+%%--------------------------------------------------------------------
+state_on({log_line, Text}, #state_on{log = Log, error = Error, until = Until}) ->
+    case isError(Text, Log#log.error_regex) of
         true ->
-            timer:cancel(Timer),
+            Timeout = timer_time(),
-            {ok, NewTimer} = timer:send_after(timer_time(), {timeout}),
+            Now = to_milliseconds(os:timestamp()),
-            {noreply, [on, Log, Error ++ Text, NewTimer, SafeTimer]};
+            Until = Now + Timeout,
+            {next_state, state_on, #state_on{log = Log, error = Error ++ Text, until = Until}, Timeout};
         false ->
-            {noreply, [on, Log, Error ++ Text, Timer, SafeTimer]}
+            Now = to_milliseconds(os:timestamp()),
+            Timeout = Until - Now,
+            {next_state, state_on, #state_on{log = Log, error = Error ++ Text, until = Until}, Timeout}
     end;
-handle_info({timeout}, [on, Log = #log{file = File}, Error, Timer, SafeTimer]) ->
+state_on(timeout, #state_on{log = Log, error = Error, until = _}) ->
-    timer:cancel(Timer),
+    mailer ! {error, Log#log.file, Error},
-    timer:cancel(SafeTimer),
+    {next_state, state_off, #state_off{log = Log}}.
-    mailer ! {error, File, Error},
-    {noreply, [off, Log]}.
 
 %%--------------------------------------------------------------------
 %% @private
 %% @doc
-%% This function is called by a gen_server when it is about to
+%% Whenever a gen_fsm receives an event sent using
-%% terminate. It should be the opposite of Module:init/1 and do any
+%% gen_fsm:send_all_state_event/2, this function is called to handle
-%% necessary cleaning up. When it returns, the gen_server terminates
+%% the event.
-%% with Reason. The return value is ignored.
 %%
-%% @spec terminate(Reason, State) -> void()
+%% @spec handle_event(Event, StateName, State) ->
+%%                   {next_state, NextStateName, NextState} |
+%%                   {next_state, NextStateName, NextState, Timeout} |
+%%                   {stop, Reason, NewState}
 %% @end
 %%--------------------------------------------------------------------
-terminate(_Reason, _State) ->
+handle_event(_Event, StateName, State) ->
+    {next_state, StateName, State}.
+
+%%--------------------------------------------------------------------
+%% @private
+%% @doc
+%% 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.
+%%
+%% @spec 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}
+%% @end
+%%--------------------------------------------------------------------
+handle_sync_event(_Event, _From, StateName, State) ->
+    Reply = ok,
+    {reply, Reply, StateName, State}.
+
+%%--------------------------------------------------------------------
+%% @private
+%% @doc
+%% This function is called by a gen_fsm when it receives any
+%% message other than a synchronous or asynchronous event
+%% (or a system message).
+%%
+%% @spec handle_info(Info,StateName,State)->
+%%                   {next_state, NextStateName, NextState} |
+%%                   {next_state, NextStateName, NextState, Timeout} |
+%%                   {stop, Reason, NewState}
+%% @end
+%%--------------------------------------------------------------------
+handle_info(_Info, StateName, State) ->
+    {next_state, StateName, State}.
+
+%%--------------------------------------------------------------------
+%% @private
+%% @doc
+%% 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.
+%%
+%% @spec terminate(Reason, StateName, State) -> void()
+%% @end
+%%--------------------------------------------------------------------
+terminate(_Reason, _StateName, _State) ->
     shutdown.
 
 %%--------------------------------------------------------------------
@@ -141,11 +189,12 @@ terminate(_Reason, _State) ->
 %% @doc
 %% Convert process state when code is changed
 %%
-%% @spec code_change(OldVsn, State, Extra) -> {ok, NewState}
+%% @spec code_change(OldVsn, StateName, State, Extra) ->
+%%                   {ok, StateName, NewState}
 %% @end
 %%--------------------------------------------------------------------
-code_change(_OldVsn, State, _Extra) ->
+code_change(_OldVsn, StateName, State, _Extra) ->
-    {ok, State}.
+    {ok, StateName, State}.
 
 %%%===================================================================
 %%% Internal functions
@@ -165,3 +214,6 @@ timer_time() ->
 safe_timer_time() ->
     {ok, ProcessingConfig} = application:get_env(log_monitor, processing_config),
     proplists:get_value(max_gathering_time, ProcessingConfig, ?DEFAULT_TIMER_TIME).
+
+to_milliseconds({Me, S, Mu}) ->
+    (Me * 1000 * 1000 * 1000) + (S * 1000) + (Mu div 1000).

apps/log_monitor/src/watcher.erl

@@ -103,7 +103,7 @@ handle_info(Msg, State = #state{file = _, supervisor = SupPid, port = Port}) ->
     case Msg of
         {Port, {data, Text}} ->
             GathererPid = gatherer_pid(SupPid),
-            GathererPid ! {log_line, binary_to_list(Text)},
+            gen_fsm:send_event(GathererPid, {log_line, binary_to_list(Text)}),
             {noreply, State};
         {Port, {exit_status, _Status}} ->
             {stop, tail_exit, State}