Packages

dqe

0.2.1
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
Readme
53 Versions
10 Dependencies
Dependants
Files
Activity
dqe src dqe_collect.erl
Raw

src/dqe_collect.erl 

-module(dqe_collect).



-behaviour(dflow).



-export([init/1, describe/1, start/2, emit/3, done/2]).



-record(state, {

          acc :: {binary(), binary()} | binary(),

          resolution :: pos_integer()

         }).



init([SubQ]) when not is_list(SubQ) ->

    {ok, #state{}, SubQ}.



describe(_) ->

    "collect".



start({_Start, _Count}, State) ->

    {ok, State}.



%% When we get the first data we can calculate both the applied

%% time and the upwards resolution.

emit(_C, {realized, {Name, Data, Resolution}},

     State = #state{resolution = undefined}) ->

    {ok, State#state{resolution = Resolution, acc = {Name, Data}}};



emit(_C, {realized, {Data, Resolution}}, State = #state{resolution = undefined}) ->

    {ok, State#state{resolution = Resolution, acc = Data}};



emit(_C, {realized, {Name, Data, _R}}, State = #state{acc = {Name, Acc}}) ->

    {ok, State#state{acc = {Name, <<Acc/binary, Data/binary>>}}};



emit(_C, {realized, {Data, _R}}, State = #state{acc = Acc}) ->

    {ok, State#state{acc = <<Acc/binary, Data/binary>>}}.



done(_Child, State = #state{resolution = undefined}) ->

    {done, State};



done(_Child, State = #state{resolution = Resolution, acc = {Name, Acc}}) ->

    {done, {points, {Name, mmath_bin:derealize(Acc), Resolution}}, State#state{acc = <<>>}};



done(_Child, State = #state{resolution = Resolution, acc = Acc}) ->

    {done, {points, {mmath_bin:derealize(Acc), Resolution}}, State#state{acc = <<>>}}.