Packages
dqe
0.3.8
0.4.15
0.4.14
0.4.13
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.25
0.3.24
0.3.23
0.3.22
0.3.21
0.3.20
0.3.19
0.3.18
0.3.17
0.3.16
0.3.15
0.3.14
0.3.13
0.3.12
0.3.11
0.3.10
0.3.9
0.3.8
0.3.7
0.3.6
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.2.2
0.2.1
0.2.0
0.1.36
0.1.35
0.1.34
0.1.33
0.1.22
0.1.21
0.1.20
0.1.10
0.1.9
DalmatinerDB query engine
Current section
Files
Jump to
Current section
Files
src/dqe_get_fun.erl~
-module(dqe_get_fun).
-behaviour(dflow).
-export([init/1, describe/1, start/2, emit/3, done/2]).
-record(state, {
bucket :: binary(),
key :: binary(),
start :: non_neg_integer(),
count :: non_neg_integer(),
get_chunk :: pos_integer(),
dqe_fun :: atom(),
chunk :: pos_integer(),
acc = <<>> :: binary(),
fun_state :: dqe_fun:fun_state()
}).
init([Fun, FunState, Start, Count, Resolution, Bucket, Key]) ->
{ok, Chunk} = application:get_env(dqe, get_chunk),
init([Start, Count, Resolution, Bucket, Key, Chunk]);
init([Start, Count, Resolution, Bucket, KeyL, Get_Chunk]) when is_list(KeyL)->
Key = dproto:metric_from_list(KeyL),
init([Start, Count, Resolution, Bucket, Key, Get_Chunk]);
init([Fun, FunState, Start, Count, _Resolution, Bucket, Key, Get_Chunk]) ->
Chunk = Fun:chunk(FunState),
{ok, #state{start = Start, count = Count, bucket = Bucket, key = Key,
get_chunk = Get_Chunk, dqe_fun = Fun, chunk = Chunk,
fun_state = FunState}, []}.
describe(#state{bucket = Bucket, key = Key}) ->
[Bucket, "/", Key].
start(run, State = #state{count = 0}) ->
{done, State};
start(run,
State = #state{start = Start, count = Count, get_chunk = Get_Chunk,
bucket = Bucket, key = Key, acc = Acc}) when
Count >= Get_Chunk ->
%% We do a bit of cheating here this allows us to loop.
State1 = State#state{start = Start + Get_Chunk, count = Count - Get_Chunk},
case ddb_connection:get(Bucket, Key, Start, Get_Chunk) of
{error, _Error} ->
{done, State};
{ok, <<>>} ->
dflow:start(self(), run),
{emit, mmath_bin:realize(mmath_bin:empty(Get_Chunk)), State1};
{ok, Data} ->
dflow:start(self(), run),
{emit, mmath_bin:realize(Data), State1}
end;
start(run, State = #state{start = Start, count = Count,
bucket = Bucket, key = Key}) ->
case ddb_connection:get(Bucket, Key, Start, Count) of
{error, _Error} ->
{done, State};
{ok, <<>>} ->
do(State, mmath_bin:empty(Count), done);
{done, mmath_bin:realize(), State};
{ok, Data} ->
do(State, mmath_bin:empty(Data), done)
end.
do(State, Data, Action) ->
%%<<Acc/binary, mmath_bin:empty(Count)>>
{Action, Data, State}.
emit(_Child, _Data, State) ->
{ok, State}.
done(_, State) ->
{done, State}.