%%%=================================================================== %%% @author Mathieu Kerjouan %%% @copyright 2018, Mathieu Kerjouan %%% @version 0.1.0 %%% @title Infinite Bitstring Stream State Machine %%% @doc = Introduction = %%% %%% This code was mainly created to solve a problem of high %%% throughput and dynamic parsing over infinite transaction %%% stream. %%% %%% stream library use standard OTP function to start and %%% stop process: %%% %%% ``` {ok, Pid} = stream:start(). %%% ``` %%% %%% In this example, we'll see how to simulate a flow %%% of random bytes coming from timer module. First thing %%% we need to start timer module: %%% %%% ``` timer:start(). %%% ``` %%% %%% We can now start our random byte generator. stream:timer/3 %%% is a wrapper around timer:apply_interface/3 function, giving %%% us the possibility to call a function from any kind of module: %%% %%% ``` {ok, Ref} = stream:timer(Pid, 1000, crypto:rand_bytes(10)). %%% ``` %%% %%% We can now see what we have currently on our stream buffer: %%% %%% ``` {Stream, Next} = stream:copy(Pid, 8). %%% ``` %%% %%% Next is a lambda function which give us the possibility to %%% navigate through our buffer: %%% %%% ``` {Stream2, Next2} = Next(). %%% {Stream3, Next3} = Next2(). %%% ``` %%% %%% copy function doesn't alter the current stream, and to %%% when you recall it with Next2 function, you'll be automatically %%% shifted based on the pattern size. %%% %%% Alternatively, cut function alter the stream, cutting them in %%% parts. %%% %%% ``` {Cut1, CNext} = cut(Pid, 8). %%% {Cut2, CNext2} = CNext(). %%% ``` %%% %%% We can now stop our FSM. %%% %%% ``` ok = stream:stop(Pid). %%% ``` %%% @end %%%=================================================================== -module(stream). -behaviour(gen_statem). -export([start/0, start/1, start/2]). -export([start_link/0, start_link/1, start_link/2]). -export([stop/1]). -export([callback_mode/0, init/1, code_change/4, terminate/3]). -export([bit/3, byte/3, custom/3]). -export([handle_cast/2, handle_call/3, handle_info/2]). -export([call/2, call/3, cast/2, info/2]). -export([cut/2, cut/3]). -export([input/2]). -export([copy/2, copy/3]). -export([map/3, map/4]). -export([timer/3]). -include_lib("eunit/include/eunit.hrl"). -record(state, { buffer = <<>> :: bitstring() , mode = bit , size = 1 , insufficient = undefined :: term() }). %%-------------------------------------------------------------------- %% @doc wrappers around generic start functions. %% %% ``` {ok, Pid} = stream:start(). %% {ok, Pid2} = stream:start(StreamArgs). %% {ok, Pid3} = stream:start(StreamArgs, FSM_Opts). %% ``` %% @end %%-------------------------------------------------------------------- -spec start() -> {ok, pid()}. start() -> start([], []). -spec start( Args :: list()) -> {ok, pid()}. start(Args) -> start(Args, []). -spec start( Args :: list() , Opts :: list()) -> {ok, pid()}. start(Args, Opts) -> gen_statem:start(?MODULE, Args, Opts). start_0001_test() -> {ok, Pid} = start(), ?assertEqual(true, is_process_alive(Pid)), ?assertEqual(ok, stop(Pid)). start_1001_test() -> {ok, Pid} = start([]), ?assertEqual(true, is_process_alive(Pid)), ?assertEqual(ok, stop(Pid)). start_2001_test() -> {ok, Pid} = start([], []), ?assertEqual(true, is_process_alive(Pid)), ?assertEqual(ok, stop(Pid)). %%-------------------------------------------------------------------- %% @doc wrappers around generic start_link functions. %% %% ``` {ok, Pid} = stream:start_link(). %% {ok, Pid2} = stream:start_link(StreamArgs). %% {ok, Pid3} = stream:start_link(StreamArgs, FSM_Opts). %% ``` %% @end %%-------------------------------------------------------------------- -spec start_link() -> {ok, pid()}. start_link() -> start_link([], []). -spec start_link( Args :: list()) -> {ok, pid()}. start_link(Args) -> start_link(Args, []). -spec start_link( Args :: list() , Opts :: list()) -> {ok, pid()}. start_link(Args, Opts) -> gen_statem:start_link(?MODULE, Args, Opts). start_link_0001_test() -> {ok, Pid} = start_link(), ?assertEqual(true, is_process_alive(Pid)), ?assertEqual(ok, stop(Pid)). start_link_1001_test() -> {ok, Pid} = start_link([]), ?assertEqual(true, is_process_alive(Pid)), ?assertEqual(ok, stop(Pid)). start_link_2001_test() -> {ok, Pid} = start_link([], []), ?assertEqual(true, is_process_alive(Pid)), ?assertEqual(ok, stop(Pid)). %%-------------------------------------------------------------------- %% @doc wrappers around genenric stop function. %% %% ``` ok = stream:stop(Pid). %% ``` %% @end %%-------------------------------------------------------------------- stop(Pid) -> gen_statem:stop(Pid). %%-------------------------------------------------------------------- %% @doc we use state functions as callback mode. Each fsm state %% is currently linked on the default pattern size extracted %% from buffer (bit, byte and custom). %% @end %%-------------------------------------------------------------------- -spec callback_mode() -> atom(). callback_mode() -> state_functions. %%-------------------------------------------------------------------- %% @doc standard gen_statem otp initialization. %% @end %%-------------------------------------------------------------------- -spec init( Args :: list()) -> {ok, atom(), #state{}}. init(Args) -> Mode = proplists:get_value(mode, Args, bit), mode(Mode). %%-------------------------------------------------------------------- %% @doc initial mode switching, return a tuple. %% @end %%-------------------------------------------------------------------- -spec mode( Mode :: atom() | {customer, non_neg_integer()}) -> {ok, atom(), #state{}}. mode(byte) -> {ok, bit, #state{ mode = byte, size = 8}}; mode({custom, Size}) -> {ok, custom, #state{ mode = custom, size = Size}}; mode(_) -> {ok, bit, #state{ mode = bit, size = 1 }}. %%-------------------------------------------------------------------- %% @doc code_change callback. Currently not supported. %% @end %%-------------------------------------------------------------------- code_change(_,_,_,_) -> ok. %%-------------------------------------------------------------------- %% @doc terminate callback. Currently not supported. %% @end %%-------------------------------------------------------------------- terminate(_,_,_) -> ok. %%-------------------------------------------------------------------- %% @doc bit state is the main state of stream process. %% @end %%-------------------------------------------------------------------- % insert data at the end of the buffer bit(cast, {input, Bitstring}, Data) -> handle_cast({input, Bitstring}, Data); % create a new process based on current buffer bit({call, _From}, split, Data) -> {keep_state, Data}; % create a new process, based on current buffer with a shift bit({call, _From}, {split, _Shift}, Data) -> {keep_state, Data}; % apply a function to one element of buffer bit({call, From}, {map, Size, Function}, Data) -> lmap(From, Data, Size, Function); % apply a function to one element of buffer with a shifting bit({call, From}, {map, Size, Shift, Function}, Data) -> lmap(From, Data, Size, Shift, Function); % copy a part of the buffer bit({call, From}, {copy, Size}, Data) -> lcopy(From, Data, Size); % copy a part of the buffer with a shift bit({call, From}, {copy, Size, Shift}, Data) -> lcopy(From, Data, Size, Shift); % cut a part of the buffer (altering it) bit({call, From}, {cut, Size}, Data ) -> lcut(From, Data, Size); % cut a part of the buffer with a shift (altering it) bit({call, From}, {cut, Size, Shift}, Data) -> lcut(From, Data, Size, Shift); bit(cast, {mode, bit}, Data) -> {keep_state, Data}; bit(cast, {mode, byte}, Data) -> {next_state, byte, Data#state{ size = 8 }}; bit(cast, {mode, custom, Size}, Data) -> {next_state, custom, Data#state{ size = Size }}; bit(_Method, _Event, Data) -> {keep_state, Data}. %%-------------------------------------------------------------------- %% @doc byte state, get 8 bits long pattern from buffer. %% @end %%-------------------------------------------------------------------- byte(cast, {mode, bit}, Data) -> {next_state, bit, Data#state{ size = 1 }}; byte(cast, {mode, byte}, Data) -> {keep_state, Data}; byte(cast, {mode, custom, Size}, Data) -> {next_state, custom, Data#state{ size = Size }}; byte(cast, Event, Data) -> handle_cast(Event, Data); byte(_Method, _Event, Data) -> {keep_state, Data}. %%-------------------------------------------------------------------- %% @doc custom state, set your own pattern size. %% @end %%-------------------------------------------------------------------- custom(cast, {mode, bit}, Data) -> {next_state, bit, Data#state{ size = 1 }}; custom(cast, {mode, byte}, Data) -> {next_state, custom, Data#state{ size = 8 }}; custom(cast, {mode, custom, Size}, Data) -> {keep_state, Data#state{ size = Size }}; custom(cast, Event, Data) -> handle_cast(Event, Data); custom(_Method, _Event, Data) -> {keep_state, Data}. %%-------------------------------------------------------------------- %% @doc common cast shared across states. %% @end %%-------------------------------------------------------------------- % if we don't have enough data, insufficient is set and % contain the function and the caller. We can update our stream on % demand and check when its good. handle_cast( {input, Stream} , #state{ insufficient = Resource } = Data) when Resource =/= undefined -> {_, Data2} = linput(Data, Stream), case Resource of {From, {cut, Size}} -> lcut(From, Data2, Size); {From, {cut, Size, Shift}} -> lcut(From, Data2, Size, Shift); {From, {copy, Size}} -> lcopy(From, Data2, Size); {From, {copy, Size, Shift}} -> lcopy(From, Data2, Size, Shift); {From, {map, Size, Fun}} -> lmap(From, Data2, Size, Fun) end; % standard input handle_cast({input, Stream}, Data) -> linput(Data, Stream); handle_cast({input, Stream, _Opts}, Data) -> linput(Data, Stream); handle_cast(_Event, Data) -> {keep_state, Data}. %%-------------------------------------------------------------------- %% @doc common call shared across states. %% @end %%-------------------------------------------------------------------- handle_call(From, _Event, Data) -> {keep_state, Data, [{reply, From, ok}]}. %%-------------------------------------------------------------------- %% @doc common info shared across states. %% @end %%-------------------------------------------------------------------- handle_info(_Event, Data) -> {keep_state, Data}. %%==================================================================== %% Internal function %%==================================================================== %%-------------------------------------------------------------------- %% @doc lerror function return standard error reply when something %% goes wrong. %% @end %%-------------------------------------------------------------------- -type error_reply() :: [{reply, term(), {error, term()}}]. -type lerror() :: {keep_state, #state{}, error_reply()}. -spec lerror( Data :: #state{} , From :: term() , Reason :: term()) -> Reply :: lerror(). lerror(Data, From, Reason) -> { keep_state , Data , [{reply, From, {error, Reason}}] }. %%-------------------------------------------------------------------- %% @doc lreply is a function helper to generate gen_statem reply %% data structure, containing callee and infinite stream %% as tuple. %% @end %%-------------------------------------------------------------------- -type ok_reply() :: {bitstring(), function()}. -type lreply() :: [{reply, term(), ok_reply()}]. -spec lreply( From :: term() , Head :: bitstring() , Function :: atom() , Args :: list()) -> Reply :: lreply(). lreply(From, Head, Function, Args) -> Reply = {Head, continuator(self(), Function, Args)}, [{reply, From, Reply}]. %%-------------------------------------------------------------------- %% @doc linput got external bitstring and put it at the end of %% buffer, in standard case, data of current state stored %% as record. %% @end %%-------------------------------------------------------------------- -spec linput( Data :: #state{} , Input :: bitstring()) -> {keep_state, #state{}}. linput(#state{ buffer = Buffer } = Data, Input) -> { keep_state , Data#state{ buffer = <> } }. %%-------------------------------------------------------------------- %% @doc lcut cut the first buffer part, altering data structure, %% and reply it to caller. %% @end %%-------------------------------------------------------------------- -spec lcut( From :: term() , Data :: #state{} , Size :: non_neg_integer()) -> Reply :: lreply() | lerror(). lcut(From, #state{ buffer = Buffer } = Data, Size) -> try <> = Buffer, { keep_state , Data#state{ buffer = Rest , insufficient = undefined } , lreply(From, Pattern, cut, [Size]) } catch error:{badmatch, Reason} -> io:format("catch it~n"), { keep_state , Data#state{ insufficient = {From, {cut, Size} } } } end. %%-------------------------------------------------------------------- %% @doc same function, with a shifting. %% @end %%-------------------------------------------------------------------- -spec lcut( From :: term() , Data :: #state{} , Size :: non_neg_integer() , Shift :: non_neg_integer() ) -> Reply :: lreply() | lerror(). lcut(From, #state{ buffer = Buffer } = Data, Size, Shift) -> try <> = Buffer, { keep_state , Data#state{ buffer = <> , insufficient = undefined } , lreply(From, Pattern, cut, [Size, Shift]) } catch error:{badmatch, Reason} -> io:format("catch it~n"), { keep_state , Data#state{ insufficient = {From, {cut, Size, Shift} } } } end. %%-------------------------------------------------------------------- %% @doc lcopy copy a part of the current stream without altering %% it and reply it to the caller. %% @end %%-------------------------------------------------------------------- -spec lcopy( From :: term() , Data :: #state{} , Size :: non_neg_integer()) -> Reply :: lreply() | lerror(). lcopy(From, Data, Size) -> lcopy(From, Data, Size, 0). %%-------------------------------------------------------------------- %% @doc same function, with a shifting. %% @end %%-------------------------------------------------------------------- -spec lcopy( From :: term() , Data :: #state{} , Size :: non_neg_integer() , Shift :: non_neg_integer()) -> Reply :: lreply() | lerror(). lcopy(From, #state{ buffer = Buffer } = Data, Size, Shift) -> try <<_:Shift/bitstring, Head:Size/bitstring, _/bitstring>> = Buffer, { keep_state , Data , lreply(From, Head, copy, [Size, Shift+Size]) } catch error:{badmatch, Reason} -> lerror(Data, From, Reason) end. %%-------------------------------------------------------------------- %% @doc lmap copy a part of the stream and apply a function on it %% without altering the state. %% @end %%-------------------------------------------------------------------- -spec lmap( From :: pid() , Data :: #state{} , Size :: non_neg_integer() , Fun :: function()) -> Reply :: lreply() | lerror(). lmap(From, Data, Size, Fun) -> lmap(From, Data, Size, 0, Fun). %%-------------------------------------------------------------------- %% @doc lmap copy a part of the stream and apply a function on it %% without altering the state. %% @end %%-------------------------------------------------------------------- -spec lmap( From :: pid() , Data :: #state{} , Size :: non_neg_integer() , Shift :: non_neg_integer() , Fun :: function()) -> Reply :: lreply() | lerror(). lmap(From, #state{ buffer = Buffer } = Data, Size, Shift, Fun) -> try <<_:Shift, Head:Size/bitstring, _/bitstring>> = Buffer, { keep_state , Data , lreply(From, Fun(Head), map, [Size, Shift+Size, Fun]) } catch error:{badmatch,Reason} -> lerror(Data, From, Reason) end. %%-------------------------------------------------------------------- %% @doc continuator function is an helper function to generate %% custom lambda function with defined value on it. %% @end %%-------------------------------------------------------------------- -spec continuator( Pid :: pid() , Function :: atom() , List :: list()) -> function(). continuator(Pid, map, [Size, Shift, Fun]) -> fun() -> map(Pid, Size, Shift, Fun) end; continuator(Pid, cut, [Size]) -> fun() -> cut(Pid, Size) end; continuator(Pid, cut, [Size, Shift]) -> fun() -> cut(Pid, Size, Shift) end; continuator(Pid, copy, [Size]) -> fun() -> copy(Pid, Size) end; continuator(Pid, copy, [Size, Shift]) -> fun() -> copy(Pid, Size, Shift+8) end. %%==================================================================== %% API %%==================================================================== %%-------------------------------------------------------------------- %% @doc input is a cast and will always return ok except if process %% isn't present. %% %% ``` ok = stream:input(Pid, <<"mystream">>. %% ok = stream:input(Pid, <<"mystream2">>. %% ``` %% %% @end %%-------------------------------------------------------------------- -spec input( Pid :: pid() , Stream :: bitstring()) -> ok. input(Pid, Stream) -> cast(Pid, {input, Stream}). input_0001_test() -> {ok, Pid} = start_link(), input(Pid, <<"test">>), {bit, #state{ buffer = Buffer }} = sys:get_state(Pid), ?assertEqual(<<"test">>, Buffer). input_0002_test() -> {ok, Pid} = start_link(), input(Pid, <<"test">>), input(Pid, <<"1234">>), {bit, #state{ buffer = Buffer }} = sys:get_state(Pid), ?assertEqual(<<"test1234">>, Buffer), ok = stop(Pid). %%-------------------------------------------------------------------- %% @doc cut alter current buffer stored in infinite stream FSM. This %% function will return current cutted pattern from buffer to %% the caller. We can do it based on first value or with %% defined shift. %% %% ``` {Stream, Next} = stream:cut(Pid, 8). %% {Stream2, Next2} = Next(). %% ``` %% @end %%-------------------------------------------------------------------- -spec cut( Pid :: pid() , Size :: non_neg_integer()) -> {bitstring(), function()}. cut(Pid, Size) -> call(Pid, {cut, Size}). cut_0001_test() -> {ok, Pid} = start_link(), input(Pid, <<"test">>), {T, _Fun} = cut(Pid, 8), ?assertEqual(<<"t">>, T), {bit, #state{ buffer = Buffer }} = sys:get_state(Pid), ?assertEqual(<<"est">>, Buffer), ok = stop(Pid). cut_0002_test() -> {ok, Pid} = start_link(), % initialize bitstring input(Pid, <<"test">>), % first cut see cut_0001_test function {_T, Fun} = cut(Pid, 8), % second cut {T2, Fun2} = Fun(), {bit, #state{ buffer = Buffer }} = sys:get_state(Pid), ?assertEqual(<<"st">>, Buffer), ?assertEqual(<<"e">>, T2), % second cut {T3, Fun3} = Fun2(), {bit, #state{ buffer = Buffer2 }} = sys:get_state(Pid), ?assertEqual(<<"t">>, Buffer2), ?assertEqual(<<"s">>, T3), % third cut {T4, _Fun4} = Fun3(), {bit, #state{ buffer = Buffer3 }} = sys:get_state(Pid), ?assertEqual(<<>>, Buffer3), ?assertEqual(<<"t">>, T4), ok = stop(Pid). cut_0003_test() -> {ok, Pid} = start_link(), timer(Pid, 10, <<"a">>), {T, _Fun} = cut(Pid, 4*8), ?assertEqual(<<"aaaa">>, T). -spec cut( Pid :: pid() , Size :: non_neg_integer() , Shift :: non_neg_integer()) -> {bitstring(), function()}. cut(Pid, Size, Shift) -> call(Pid, {cut, Size, Shift}). cut_1001_test() -> {ok, Pid} = start_link(), input(Pid, <<"test">>), {T, _Fun} = cut(Pid, 8, 8), ?assertEqual(<<"e">>, T), {bit, #state{ buffer = Buffer }} = sys:get_state(Pid), ?assertEqual(<<"tst">>, Buffer), ok = stop(Pid). cut_1003_test() -> {ok, Pid} = start_link(), timer(Pid, 10, <<"a">>), {T, _Fun} = cut(Pid, 3*8, 8), ?assertEqual(<<"aaa">>, T), ok = stop(Pid). %%-------------------------------------------------------------------- %% @doc copy retrieve a copied value of the current buffer managed %% by infinite stream FSM. Next value is based on the Size*Shift, %% where Size is the size of the pattern and Shift corresponding %% to the offset. When lambda function returned is executed you %% will automatically add shifted value. %% %% ``` {Stream, Next} = stream:copy(Pid, 8). %% {Stream2, Next2} = Next(). %% ``` %% %% @end %%-------------------------------------------------------------------- -spec copy( Pid :: pid() , Size :: non_neg_integer()) -> {bitstring(), function()}. copy(Pid, Size) -> call(Pid, {copy, Size}). copy_0001_test() -> {ok, Pid} = start(), ok = input(Pid, <<"thisisatest">>), {Out, _Fun} = copy(Pid, 8), ?assertEqual(<<"t">>, Out), stop(Pid). -spec copy( Pid :: pid() , Size :: non_neg_integer() , Shift :: non_neg_integer()) -> {bitstring(), function()}. copy(Pid, Size, Shift) -> call(Pid, {copy, Size, Shift}). copy_1001_test() -> {ok, Pid} = start(), ok = input(Pid, <<"thisisatest">>), {Out, _Fun} = copy(Pid, 8, 0), ?assertEqual(<<"t">>, Out), stop(Pid). %%-------------------------------------------------------------------- %% @doc map function will copy a value from current buffer and apply %% a function on this value (returning it to caller). This %% function doesn't alter the current buffer. %% %% ``` Fun = fun(Pattern) -> {ok, Pattern}. %% {ModifiedPattern, Next} = stream:map(Pid, 8, Fun). %% {ModifiedPattern2, Next2} = Next(). %% ``` %% @end %%-------------------------------------------------------------------- -spec map( Pid :: pid() , Size :: non_neg_integer() , Fun :: function()) -> {bitstring(), function()}. map(Pid, Size, Fun) -> call(Pid, {map, Size, Fun}). map_0001_test() -> {ok, Pid} = start(), ok = input(Pid, <<"thisisatest">>), {Term, _Fun} = map(Pid, 8, fun(A) -> {A, true} end), ?assertEqual({<<"t">>, true}, Term), ok = stop(Pid). map_0002_test() -> {ok, Pid} = start(), ok = input(Pid, <<"thisisatest">>), {Term, Fun} = map(Pid, 8, fun(A) -> {A, true} end), ?assertEqual({<<"t">>, true}, Term), {Term2, _Fun2} = Fun(), ?assertEqual({<<"h">>, true}, Term2), ok = stop(Pid). -spec map( Pid :: pid() , Size :: non_neg_integer() , Shift :: non_neg_integer() , Fun :: function()) -> {bitstring(), function()}. map(Pid, Size, Shift, Fun) -> call(Pid, {map, Size, Shift, Fun}). map_1001_test() -> {ok, Pid} = start(), ok = input(Pid, <<"thisisatest">>), {Term, _Fun} = map(Pid, 8, 0, fun(A) -> {A, true} end), ?assertEqual({<<"t">>, true}, Term), ok = stop(Pid). -spec map( Pid :: pid() , Size :: non_neg_integer() , Shift :: non_neg_integer() , Fun :: function() , Opts :: term() ) -> {bitstring(), function()}. map(Pid, Size, Shift, Fun, Opts) -> call(Pid, {map, Size, Shift, Fun}). %%-------------------------------------------------------------------- %% @doc local wrapper around gen_statem:call function %% @end %%-------------------------------------------------------------------- call(Pid, Event) -> call(Pid, Event, infinity). call(Pid, Event, Timeout) -> gen_statem:call(Pid, Event, Timeout). %%-------------------------------------------------------------------- %% @doc local wrapper around gen_statem:cast function %% @end %%-------------------------------------------------------------------- cast(Pid, Event) -> gen_statem:cast(Pid, Event). %%-------------------------------------------------------------------- %% @doc local wrapper around gen_statem:info function %% @end %%-------------------------------------------------------------------- info(Pid, Event) -> gen_statem:cast(Pid, Event). %%-------------------------------------------------------------------- %% @doc Run a timer, sending stream at specific time interval. %% This function require timer application and is mainly %% used for testing purpose. %% %% ``` {ok, Ref} = stream:timer(Pid, 1000, <<"mydata">>). %% ``` %% @end %%-------------------------------------------------------------------- -spec timer( Pid :: pid() , Time :: non_neg_integer() , Stream :: bitstring()) -> {ok, {interval, reference()}}. timer(Pid, Time, Bitstring) -> timer:apply_interval(Time, stream, input, [Pid, Bitstring]).