%%% vim:ts=4:sw=4:et %%%---------------------------------------------------------------------------- %%% @doc 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 %%% @end %%%---------------------------------------------------------------------------- %%% Created: 2015-02-12 %%%---------------------------------------------------------------------------- -module(file_log_reader). -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()). -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()})}]. %% 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'
%%
-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 %% @end %%----------------------------------------------------------------------------- -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. %% @end %%----------------------------------------------------------------------------- -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. %% @end %%----------------------------------------------------------------------------- -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. %% @end %%----------------------------------------------------------------------------- -spec position(pid() | atom()) -> {ok, Position::integer()}. position(Pid) -> gen_server:call(Pid, position). %%----------------------------------------------------------------------------- %% @doc Return current parser state (`{pstate, any()}' initialization option). %% @end %%----------------------------------------------------------------------------- -spec pstate(pid() | atom()) -> {ok, any()}. pstate(Pid) -> gen_server:call(Pid, pstate). %%----------------------------------------------------------------------------- %% @doc Update parser state. %% @end %%----------------------------------------------------------------------------- -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. %% @end %%----------------------------------------------------------------------------- -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. %% @end %%----------------------------------------------------------------------------- -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'). %% @end %%----------------------------------------------------------------------------- -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) %% @end %%----------------------------------------------------------------------------- close(#state{fd=FD} = State) -> file:close(FD), State#state{fd=undefined}. %%%---------------------------------------------------------------------------- %%% Callback functions from gen_server %%%---------------------------------------------------------------------------- %%----------------------------------------------------------------------------- %% @private %% @doc Initiates the server %% @end %%----------------------------------------------------------------------------- -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 %% @end %%----------------------------------------------------------------------------- -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 %% @end %%----------------------------------------------------------------------------- -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 %% @end %%----------------------------------------------------------------------------- -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. %% @end %%----------------------------------------------------------------------------- -spec terminate(any(), #state{}) -> ok. terminate(_Reason, #state{}) -> ok. %%----------------------------------------------------------------------------- %% @private %% @doc Convert process state when code is changed %% @end %%----------------------------------------------------------------------------- -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: