%%% vim:ts=4:sw=4:et %%%---------------------------------------------------------------------------- %%% Created: 2015-02-12 %%%---------------------------------------------------------------------------- -module(file_log_reader). -moduledoc """ Periodically read an append-only log file and parse newly added data. The user controls the interval in msec how often to check for file modifications. When new data is appended to file it triggers invocation of the user-defined parsing function that deliminates the file, and the result is delivered to the consumer by calling the consumer callback function. The log reader can be started as a `gen_server` or can be controlled synchronously by using `init/3`, `run/1`, and `close/1` methods. Author: Serge Aleynikov Copyright: 2015 Serge Aleynikov """. -author('saleyn@gmail.com'). -behaviour(gen_server). %%% TODO: convert into behavior!!! %% API -export([start_link/3, start_link/4, start/3, start/4, stop/1, position/1, pstate/1, update_pstate/3]). -export([init/3, run/1, close/1]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -ifdef(TEST). -define(D(Fmt,Args), io:format(Fmt, Args)). -include_lib("eunit/include/eunit.hrl"). -else. -define(D(Fmt,Args), ok). -endif. -define(MAX_READ_SIZE, 32*1024*1024). -type consumer() :: fun((Msg :: any() | {'$end_of_file', string(), Res :: ok | {error|throw|exit, Reason::any(), Stacktrace::list()}}, Pos::integer(), State::any()) -> NewState::any()). -doc """ Details: - `pos` — Start reading from this position (default: 0) - `end_pos` — Read until this position and stop. If provided and file position reaches `end_pos`, the consumer() callback given to the reader will be called as: `Consumer({'$end_of_file', Filename::string(), Result}, Pos::integer(), State)` where `Result` is `ok` or `{error|exit|exception, Error::any(), StackTrace}` if an error occured. - `max_size` — Maximum chunk size to read from file in a single pass (default: 32M). - `timeout` — Number of milliseconds between successive file scanning (default: 1000) - `retry_sec` — Number of seconds between successive retries upon failure to open the market data file passed to one of the `start*/{3,4}` functions (default: 15). The value of 0 means that the file must exist or else the process won't start. - `parser` — Is the function to be called when the next chunk is read from file. The function must return: - `{ok, Msg, Tail, State}` — invoke `Consumer` callback passing it the parsed message `Msg`, and continue parsing the `Tail` binary - `{incomplete, State}` — the data contains no complete messages - wait until there's more - `{skip, Tail, State}` — disregard input and continue parsing `Tail` without calling `Consumer` callback - `pstate` — Initial value of the parser state or a functor `fun((File::string() Consumer::consumer(), Options::options()) -> PState::any())` - `pstate_update` — Update function of the parser state. Called when the user invokes `update_pstate/3` """. -type options() :: [ {pos, StartPos::integer()} | {end_pos, ReadUntilPos::integer() | eof} | {max_size, MaxReadSize::integer() | eof} | {timeout, MSec::integer()} | {retry_sec, Sec::integer()} | {parser, fun((Data::binary(), ParserState::any()) -> {ok, Msg::any(), Tail::binary(), NewParserState::any()} | {incomplete, NewParserState::any()} | {skip, Tail::binary(), NewParserState::any()}) | {Mod::atom(), Fun::atom()}} | {pstate, fun((File::string(), consumer(), Options::list()) -> any()) | any()} | {pstate_update, fun((Option::atom(), Value::any(), PState::any()) -> {ok, NewPState::any()} | {error, any()})}]. -export_type([consumer/0, options/0]). -record(state, { consumer :: consumer() , tref :: reference() , fd :: port() , file :: string() , pos = 0 :: integer() | eof , end_pos :: integer() | eof | undefined % Read until this pos (inclusive) , max_size :: integer() , timeout :: integer() , parser :: {atom(), atom()} | fun((binary(), any()) -> {any(), binary(), any()}) , pstate :: any() , pstate_update :: fun((any()) -> any()) , part_size = 0 :: integer() % Sz of incomple unparsed chunk at eof , done = false :: false | ok | % Processing completion indicator {error|exception|exit, Reason::any(), Stacktrace::any()} , incompl_count :: integer() % Counter of incomplete message reads }). %%%---------------------------------------------------------------------------- %%% External API %%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------- -doc """ To be called by the supervisor in order to start the server. If init/1 fails with Reason, the function returns `{error,Reason}`. If init/1 returns `{stop,Reason}` or ignore, the process is terminated and the function returns `{error,Reason}` or ignore, respectively. See: `start_link/3`. """. %%----------------------------------------------------------------------------- -spec start_link(atom(), string(), consumer(), options()) -> {ok, pid()} | ignore | {error, any()}. start_link(RegName, File, Consumer, Options) when is_atom(RegName), is_list(File), is_function(Consumer,3), is_list(Options) -> gen_server:start_link({local, RegName}, ?MODULE, [File, Consumer, Options], []). %%----------------------------------------------------------------------------- -doc """ Process `File` by calling `Consumer` callback on every delimited message. Message delimination is handled by the `{parser, Parser}` option. `Consumer` function gets called iteratively with the following arguments: - `(Msg, Pos::integer(), State)` — `Msg` is what the parser function returned. `Pos` is current file position following the `Msg`. `State` is current value of parser state that is initialized by `{pstate, PState}` option given to the `start_link/{3,4}` function - `({'$end_of_file', Filename::string(), Result}, Pos::integer(), PState)` — This call happens when end of file condition is reached (see definition of `consumer()` type) `Consumer` can end processing normally without reaching the end of file by throwing `{eof, PState}` exception. """. %%----------------------------------------------------------------------------- -spec start_link(string(), consumer(), options()) -> {ok, pid()} | ignore | {error, any()}. start_link(File, Consumer, Options) when is_list(File), is_function(Consumer,3), is_list(Options) -> gen_server:start_link(?MODULE, [File, Consumer, Options], []). %%----------------------------------------------------------------------------- -doc "Start the server outside of supervision tree.". %%----------------------------------------------------------------------------- -spec start(atom(), string(), consumer(), options()) -> {ok, pid()} | {error, any()}. start(RegName, File, Consumer, Options) when is_atom(RegName), is_list(File), is_function(Consumer,3), is_list(Options) -> gen_server:start({local, RegName}, ?MODULE, [File, Consumer, Options], []). -spec start(string(), consumer(), options()) -> {ok, pid()} | {error, any()}. start(File, Consumer, Options) when is_list(File), is_function(Consumer,3), is_list(Options) -> gen_server:start(?MODULE, [File, Consumer, Options], []). %%----------------------------------------------------------------------------- -doc "Report last processed file position/size.". %%----------------------------------------------------------------------------- -spec position(pid() | atom()) -> {ok, Position::integer()}. position(Pid) -> gen_server:call(Pid, position). %%----------------------------------------------------------------------------- -doc "Return current parser state (`{pstate, any()}` initialization option).". %%----------------------------------------------------------------------------- -spec pstate(pid() | atom()) -> {ok, any()}. pstate(Pid) -> gen_server:call(Pid, pstate). %%----------------------------------------------------------------------------- -doc "Update parser state.". %%----------------------------------------------------------------------------- -spec update_pstate(pid(), Option::atom(), Value::any()) -> {ok, State::any()}. update_pstate(Pid, Option, Value) when is_atom(Option) -> gen_server:call(Pid, {update_pstate, Option, Value}). %%----------------------------------------------------------------------------- -doc "Stop the server.". %%----------------------------------------------------------------------------- -spec stop(pid() | atom()) -> ok. stop(Pid) -> gen_server:call(Pid, stop). %%----------------------------------------------------------------------------- -doc """ When using file processor without gen_server, use this function to initialize the state, and then call run/1. """. %%----------------------------------------------------------------------------- -spec init(string(), consumer(), options()) -> {ok, #state{}}. init(File, Consumer, Options) when is_list(File), is_list(Options) -> try Offset = proplists:get_value(pos, Options, 0), EndPos = case proplists:get_value(end_pos, Options) of M when is_integer(M) -> M; eof -> eof; undefined -> undefined; Other2 -> throw({invalid_option, {end_pos, Other2}}) end, MaxSize = proplists:get_value(max_size, Options, ?MAX_READ_SIZE), Parser = proplists:get_value(parser, Options), RetryS = proplists:get_value(retry_sec, Options, 15), Timeout = proplists:get_value(timeout, Options, 1000), PStUpd = proplists:get_value(pstate_update,Options), PStUpd =:= undefined orelse is_function(PStUpd,3) orelse throw({badarg, pstate_update}), State = #state{file=File, consumer=Consumer, parser=Parser, pos=Offset, end_pos=EndPos, max_size=MaxSize, timeout=Timeout, incompl_count=0, pstate_update=PStUpd}, {ok, S} = try_open_file(1, RetryS, File, State), PState = case proplists:get_value(pstate, Options) of F when is_function(F, 3) -> F(File, Consumer, Options); Other -> Other end, {ok, S#state{pstate=PState}} catch throw:ignore -> ignore; T:E:STrace -> {stop, {T, E, STrace}} end. %%----------------------------------------------------------------------------- -doc "Process file from given position `Pos` to `EndPos` (or `eof`).". %%----------------------------------------------------------------------------- -spec run(#state{}) -> #state{}. run(#state{fd=FD, pos=Pos, end_pos=EndPos, max_size=MaxChunkSz, done=false} = S) -> MaxPos = Pos+MaxChunkSz, NextPos = if is_integer(EndPos) -> min(MaxPos, EndPos); true -> MaxPos end, Size = NextPos - Pos, if Size > 0 -> case file:pread(FD, Pos, Size) of {ok, Data} when byte_size(Data) > S#state.part_size -> try S1 = process_chunk(Data, S), run(S1) catch %% This is a special case - the consumer callback may throw %% an `eof' exception to force the normal end of processing: throw:{eof, PState} when EndPos =:= eof -> end_of_file(ok, S#state{end_pos=Pos, pstate=PState}); throw:{eof, PState} -> end_of_file(ok, S#state{pstate=PState}); %% Some other error occured - abort: E:W:STrace -> end_of_file({E, W, STrace}, S) end; {ok, _Data} -> schedule_timer(S); eof when EndPos =:= eof -> % Reached the requested end of input - notify consumer: end_of_file(ok, S#state{end_pos=Pos}); eof -> % No more data is currently available - wait until there's more. schedule_timer(S) end; true -> % Reached the requested end of input - notify consumer: end_of_file(ok, S#state{end_pos=Pos}) end; run(#state{} = S) -> S. end_of_file(Result, #state{pos=Pos, consumer=Callback, file=F, pstate=PState} = S) -> PS1 = Callback({'$end_of_file', F, Result}, Pos, PState), S#state{pstate = PS1, part_size=0, done=Result}. schedule_timer(#state{timeout=Timeout, fd=FD} = S) -> TRef = erlang:send_after(Timeout, self(), {check_md_files_timer, FD}), S#state{tref=TRef}. %%----------------------------------------------------------------------------- -doc "Close file processor (use this method when not using gen_server)". %%----------------------------------------------------------------------------- close(#state{fd=FD} = State) -> file:close(FD), State#state{fd=undefined}. %%%---------------------------------------------------------------------------- %%% Callback functions from gen_server %%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------- %% @private -doc "Initiates the server". %%----------------------------------------------------------------------------- -spec init(list()) -> {ok, #state{}} | {ok, #state{}, Timeout :: integer() | hibernate} | ignore | {stop, any()}. init([File, Consumer, Options]) -> try init(File, Consumer, Options) catch _:What -> {stop, What} end. %%----------------------------------------------------------------------------- %% @private -doc "Handling call messages". %%----------------------------------------------------------------------------- -spec handle_call(any(), From::tuple(), #state{}) -> {reply, Reply::any(), #state{}} | {reply, Reply::any(), #state{}, Timeout::integer() | hibernate} | {noreply, #state{}} | {noreply, #state{}, Timeout::integer() | hibernate} | {stop, Reason::any(), Reply::any(), #state{}} | {stop, Reason::any(), #state{}}. handle_call(position, _From, #state{pos = Pos} = State) -> {reply, {ok, Pos}, State}; handle_call(pstate, _From, #state{pstate = PState} = State) -> {reply, {ok, PState}, State}; handle_call({update_pstate, Option, Value}, _From, #state{} = State) -> #state{pstate_update=Fun, pstate=PState} = State, if is_function(Fun, 3) -> case Fun(Option, Value, PState) of {ok, NewPState} -> {reply, {ok, NewPState}, State#state{pstate=NewPState}}; {error, Reason} -> {reply, {error, Reason}, State} end; true -> {reply, {error, pstate_update_not_implemented}, State} end; handle_call(stop, _From, State) -> {stop, normal, ok, State}; handle_call(Request, _From, State) -> {stop, {unknown_call, Request}, State}. %%----------------------------------------------------------------------------- %% @private -doc "Handling cast messages". %%----------------------------------------------------------------------------- -spec handle_cast(any(), #state{}) -> {noreply, #state{}} | {noreply, #state{}, Timeout::integer() | hibernate} | {stop, Reason::any(), #state{}}. handle_cast(Msg, State) -> {stop, {unknown_cast, Msg}, State}. %%----------------------------------------------------------------------------- %% @private -doc "Handling all non call/cast messages". %%----------------------------------------------------------------------------- -spec handle_info(any(), #state{}) -> {noreply, #state{}} | {noreply, #state{}, Timeout::integer() | hibernate} | {stop, Reason::any(), #state{}}. handle_info({check_md_files_timer, _FD}, State) -> S = #state{done=Done} = run(State), case Done of false -> {noreply, S, hibernate}; ok -> {stop, normal, S}; {E, Reason, Stacktrace} when E=:=error; E=:=exception; E=:=exit -> erlang:raise(E, Reason, Stacktrace) end; handle_info({open_file_timer, Attempt, RetrySec, File}, State) -> {ok, State1} = try_open_file(Attempt, RetrySec, File, State), {noreply, State1}; handle_info(_Info, State) -> {noreply, State}. %%----------------------------------------------------------------------------- %% @private -doc """ This function is called by a gen_server 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_server terminates with Reason. The return value is ignored. """. %%----------------------------------------------------------------------------- -spec terminate(any(), #state{}) -> ok. terminate(_Reason, #state{}) -> ok. %%----------------------------------------------------------------------------- %% @private -doc "Convert process state when code is changed". %%----------------------------------------------------------------------------- -spec code_change(any(), #state{}, any()) -> #state{}. code_change(_OldVsn, State, _Extra) -> {ok, State}. %%%---------------------------------------------------------------------------- %%% Internal functions %%%---------------------------------------------------------------------------- try_open_file(Attempt, RetrySec, File, #state{} = State) when is_list(File) -> case file:open(File, [read,raw,binary,read_ahead]) of {ok,FD} -> {ok,End} = file:position(FD, eof), Pos = case State#state.pos of N when is_integer(N) -> N; eof -> End; Other1 -> throw({invalid_option, {pos, Other1}}) end, case Attempt of 1 -> ok; _ -> error_logger:info_msg ("~w: file ~s was opened successfully\n", [?MODULE, File]) end, {ok, schedule_timer(State#state{fd=FD, pos=Pos})}; {error, Reason} when Attempt =:= 1, RetrySec > 0 -> error_logger:warning_msg( "~w: failed to open file ~s (will retry in ~ws): ~s", [?MODULE, File, RetrySec, file:format_error(Reason)]), Msg = {open_file_timer, Attempt+1, RetrySec, File}, erlang:send_after(RetrySec*1000, self(), Msg), {ok, State}; {error, _Reason} when RetrySec > 0 -> Msg = {open_file_timer, Attempt+1, RetrySec, File}, erlang:send_after(RetrySec*1000, self(), Msg), {ok, State}; {error, Reason} -> throw({cannot_open_file, File, Reason}) end. process_chunk(<<>>, S) -> S; process_chunk(Data, #state{part_size=N} = S) when N > 0, byte_size(Data) =< N -> %% Last message in the file was found to be incomplete on the %% previous processing iteration. Current read attempt didn't read any more %% of new bytes, so do nothing and wait until there's more data available S#state{incompl_count=S#state.incompl_count+1}; process_chunk(Data, #state{pos=Pos, consumer=OnMsg, parser=P, pstate=PS} = S) -> case parse(P, Data, PS) of {ok, Msg, <<>>, PS1} -> %% The Data was fully parsed, invoke the callback and try to see if %% more data was appended to the file NPos = Pos + byte_size(Data), PS2 = OnMsg(Msg, NPos, PS1), run(S#state{pos=NPos, pstate=PS2, part_size=0}); {ok, Msg, Tail, PS1} when is_binary(Tail) -> %% A chunk of Data up to the Tail was fully parsed, invoke the callback %% and continue parsing the Tail NPos = Pos + (byte_size(Data) - byte_size(Tail)), PS2 = OnMsg(Msg, NPos, PS1), process_chunk(Tail, S#state{pos=NPos, pstate=PS2, part_size=0}); {incomplete, PS1} -> I = S#state.incompl_count, S#state{pstate=PS1, part_size=byte_size(Data), incompl_count=I+1}; {skip, Tail, PS1} when is_binary(Tail) -> N = byte_size(Data) - byte_size(Tail), process_chunk(Tail, S#state{pos=Pos + N, pstate=PS1, part_size=0}) end. parse(Fun, Data, State) when is_function(Fun, 2) -> Fun(Data, State); parse({M,F}, Data, State) -> M:F(Data, State); parse(undefined, Data, State) -> {ok, Data, <<>>, State}. %%%---------------------------------------------------------------------------- %%% Test functions %%%---------------------------------------------------------------------------- -ifdef(EUNIT). async_read_file_test() -> File = "/tmp/test-file-reader.log", Self = self(), N = 10, %% Producer of data begin_produce(File, N, false), %% Consumer of data User = spawn_link(fun() -> ?assertEqual(ok, wait(N, 1)), Self ! done end), Prsr = fun(Bin, PState) -> case binary:split(Bin, <<"\n">>) of [<<"abc5">>, Tail] -> {skip, Tail, PState+1}; [<<"abc1">>=_M,_Tail] when PState =:= 9 -> {incomplete, PState+1}; [Msg, Tail] -> {ok, Msg, Tail, PState+1} end end, Opts = [{timeout, 10}, {parser, Prsr}, {pstate, 0}], Consume = fun(Msg, _Pos, State) -> User ! Msg, State end, {ok, Pid} = start_link(File, Consume, Opts), ?assertEqual(done, receive done -> done after 1000 -> timeout end), {ok, PState} = pstate(Pid), ?assertEqual(9, PState), ?assertEqual(ok, stop(Pid)), file:delete(File). premature_end_test() -> %% This test illustrates how to abort parsing gracefully in the middle %% of the input file processing. File = "/tmp/test-file-reader.log", Self = self(), N = 10, %% Producer of data begin_produce(File, N, false), %% Consumer of data User = spawn_link(fun() -> ?assertEqual(ok, wait(N, 6)), Self ! done end), erase(eof_called), Prsr = fun(Bin, PState) -> [Msg, Tail] = binary:split(Bin, <<"\n">>), {ok, Msg, Tail, PState+1} end, Opts = [{timeout, 10}, {parser, Prsr}, {pstate, 0}], Consume = fun (<<"abc5">>, _Pos, State) -> %%?debugFmt("Msg <> state ~w", [State]), ?assertEqual(6, State), throw({eof, State}); ({'$end_of_file', _File, Reason}, _Pos, State) -> ?assertEqual(6, State), ?assertEqual(File, _File), ?assertEqual(ok, Reason), Self ! eof_called, State; (Msg, _Pos, State) -> %%?debugFmt("Msg ~p state ~w", [Msg, State]), User ! Msg, State end, {ok, Pid} = start_link(File, Consume, Opts), monitor(process, Pid), ?assertEqual(ok, receive eof_called -> put(eof_called, true), ok after 1000 -> timeout end), ?assertEqual(done, receive done -> done after 1000 -> timeout end), ?assertEqual(ok, receive {'DOWN', _Ref, process, Pid, normal} -> ok after 1000 -> timeout end), ?assertEqual(true, get(eof_called)), erase(eof_called), file:delete(File). begin_produce(File, N, Wait) -> file:delete(File), Self = self(), %% Produce data spawn_link(fun() -> Res = file:open(File, [write,raw,binary]), Self ! continue, ?assertMatch({ok,_}, Res), {ok, F} = Res, (fun Loop(0,_FF) -> ok; Loop(I, FF) -> Bin = <<"abc", (integer_to_binary(I))/binary, $\n>>, ok = file:write(FF, Bin), timer:sleep(50), Loop(I-1, FF) end)(N, F), if Wait -> Self ! end_producer; true -> ok end end), receive continue -> ok end, if Wait -> receive end_producer -> ok end; true -> ok end. wait(I, I) -> % Got all messages ok; wait(5, I) -> % <<"abc5">> is skipped by the parser wait(4, I); wait(I, End) -> Expect = <<"abc", (integer_to_binary(I))/binary>>, receive Expect -> wait(I-1, End); Other -> ?debugFmt("Unexpected message: ~p", [Other]), %{unexpected_msg, Other} wait(I-1, End) %after 1000 -> % timeout end. consume({'$end_of_file', _F, ok}, Pos, PState) when is_integer(Pos), is_integer(PState) -> put(count_end, get(count_end)+1), put(end_pos, Pos), PState; consume(Msg, _Pos, PState) when is_binary(Msg), is_integer(PState) -> Bin = <<"abc", (integer_to_binary(PState))/binary>>, ?assertEqual(Msg, Bin), PState-1. sync_read_file_test() -> File = "/tmp/test-file-reader.log", N = 10, file:delete(File), %% Produce data begin_produce(File, N, true), %% Consume data Prsr = fun(Bin, PState) -> [Msg, Tail] = binary:split(Bin, <<"\n">>), {ok, Msg, Tail, PState} end, put(count_end, 0), %% (1) Initialize file reader (fun() -> {ok, State} = init(File, fun consume/3, [{parser, Prsr}, {pstate, N}]), %% Execute file reader synchronously State1 = run(State), ?assertEqual(0, State1#state.pstate), %% Close file reader close(State1) end)(), ?assertEqual(0, get(count_end)), %% (2) Now test reading from the end of file (fun() -> {ok, State} = init(File, fun consume/3, [{pos, eof}, {parser, Prsr}, {pstate, N}]), %% Execute file reader synchronously State1 = run(State), ?assertEqual(10, State1#state.pstate), %% Close file reader close(State1) end)(), ?assertEqual(0, get(count_end)), %% (3) Now test reading from the 6th position (fun() -> {ok, State} = init(File, fun consume/3, [{pos, 6}, {parser, Prsr}, {pstate, N-1}]), %% Execute file reader synchronously State1 = run(State), ?assertEqual(0, State1#state.pstate), %% Close file reader close(State1) end)(), ?assertEqual(0, get(count_end)), %% (4) Now test reading only 11 bytes: <<"abc10\nabc9\n">> (fun() -> {ok, State} = init(File, fun consume/3, [{end_pos, 11}, {parser, Prsr}, {pstate, N}]), %% Execute file reader synchronously State1 = run(State), ?assertEqual(8, State1#state.pstate), %% Close file reader close(State1) end)(), ?assertEqual(1, get(count_end)), ?assertEqual(11, get(end_pos)), %% (5) Now test reading from the end of file till the end of file (fun() -> {ok, State} = init(File, fun consume/3, [{pos, eof}, {end_pos, eof}, {parser, Prsr}, {pstate, N}]), %% Execute file reader synchronously State1 = run(State), ?assertEqual(51, State1#state.pos), ?assertEqual(51, State1#state.end_pos), ?assertEqual(N, State1#state.pstate), %% Close file reader close(State1) end)(), ?assertEqual(2, get(count_end)), ?assertEqual(51, get(end_pos)), erase(count_end), erase(end_pos), file:delete(File). -endif. %%%. %%% vim: set filetype=erlang tabstop=4 foldmarker=%%%',%%%. foldmethod=marker: