util v1.1.5

View Source file_log_reader (util v1.1.5)

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.

Link to this section Summary

Types

consumer/0
options/0
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

Functions

close(State)
Close file processor (use this method when not using gen_server)
init(File, Consumer, Options)
When using file processor without gen_server, use this function to initialize the state, and then call run/1.
position(Pid)
Report last processed file position/size.
pstate(Pid)
Return current parser state ({pstate, any()} initialization option).
run(State)
Process file from given position Pos to EndPos (or eof).
start(File, Consumer, Options)
start(RegName, File, Consumer, Options)
Start the server outside of supervision tree.
start_link(File, Consumer, Options)
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.
start_link(RegName, File, Consumer, Options)
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 also: start_link/3.

stop(Pid)
Stop the server.
update_pstate(Pid, Option, Value)
Update parser state.

Link to this section Types

Link to this type

consumer/0

View Source 
-type consumer() ::
    fun((Msg ::
             any() |
             {'$end_of_file',
              string(),
              Res :: ok | {error | throw | exit, Reason :: any(), Stacktrace :: list()}},
         Pos :: integer(),
         State :: any()) ->
            NewState :: any()).
Link to this type

options/0

View Source 
-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

Link to this section Functions

Link to this function

close(State)

View Source
Close file processor (use this method when not using gen_server)
Link to this function

init(File, Consumer, Options)

View Source 
-spec init(string(), consumer(), options()) -> {ok, #state{}}.
When using file processor without gen_server, use this function to initialize the state, and then call run/1.
Link to this function

position(Pid)

View Source 
-spec position(pid() | atom()) -> {ok, Position :: integer()}.
Report last processed file position/size.
Link to this function

pstate(Pid)

View Source 
-spec pstate(pid() | atom()) -> {ok, any()}.
Return current parser state ({pstate, any()} initialization option).
Link to this function

run(State)

View Source 
-spec run(#state{}) -> #state{}.
Process file from given position Pos to EndPos (or eof).
Link to this function

start(File, Consumer, Options)

View Source 
-spec start(string(), consumer(), options()) -> {ok, pid()} | {error, any()}.
Link to this function

start(RegName, File, Consumer, Options)

View Source 
-spec start(atom(), string(), consumer(), options()) -> {ok, pid()} | {error, any()}.
Start the server outside of supervision tree.
Link to this function

start_link(File, Consumer, Options)

View Source 
-spec start_link(string(), consumer(), options()) -> {ok, pid()} | ignore | {error, any()}.
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.
Link to this function

start_link(RegName, File, Consumer, Options)

View Source 
-spec start_link(atom(), string(), consumer(), options()) -> {ok, pid()} | ignore | {error, any()}.
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 also: start_link/3.

Link to this function

stop(Pid)

View Source 
-spec stop(pid() | atom()) -> ok.
Stop the server.
Link to this function

update_pstate(Pid, Option, Value)

View Source 
-spec update_pstate(pid(), Option :: atom(), Value :: any()) -> {ok, State :: any()}.
Update parser state.