Packages
dqe
0.1.22
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.erl
-module(dqe_get).
-behaviour(dflow).
-export([init/1, describe/1, start/2, emit/3, done/2]).
-record(state, {
bucket :: binary(),
metric :: binary(),
chunk :: pos_integer()
}).
init([Bucket, Metric]) when is_binary(Metric) ->
{ok, Chunk} = application:get_env(dqe, get_chunk),
init([Bucket, Metric, Chunk]);
init([Bucket, Metric, Chunk]) ->
{ok, #state{bucket = Bucket, metric = Metric, chunk = Chunk}, []}.
describe(#state{bucket = Bucket, metric = Metric}) ->
[Bucket, "/", Metric].
start({_Start, 0}, State) ->
{done, State};
start({Start, Count},
State = #state{bucket = Bucket, metric = Metric, chunk = Chunk}) when
Count >= Chunk ->
%% We do a bit of cheating here this allows us to loop.
case dalmatiner_connection:get(Bucket, Metric, Start, Chunk) of
{error, _Error} ->
{done, State};
{ok, Res, <<>>} ->
dflow:start(self(), {Start + Chunk, Count - Chunk}),
{emit, {mmath_bin:empty(Chunk), Res}, State};
{ok, Res, Data} ->
dflow:start(self(), {Start + Chunk, Count - Chunk}),
{emit, {mmath_bin:realize(Data), Res}, State}
end;
start({Start, Count}, State = #state{bucket = Bucket, metric = Metric}) ->
case dalmatiner_connection:get(Bucket, Metric, Start, Count) of
{error, _Error} ->
{done, State};
{ok, Res, <<>>} ->
{done, {mmath_bin:empty(Count), Res}, State};
{ok, Res, Data} ->
{done, {mmath_bin:realize(Data), Res}, State}
end.
emit(_Child, _Data, State) ->
{ok, State}.
done(_, State) ->
{done, State}.