Packages
kvs
9.4.0
13.5.22-aleph
13.4.16
13.4.15
13.4.14
13.4.13
13.3.1
13.2.28
11.9.1
10.8.3
10.8.2
10.3.0
9.9.2
9.9.1
9.9.0
9.8.0
9.7.0
9.4.8
9.4.7
9.4.6
9.4.5
9.4.4
9.4.3
9.4.2
9.4.1
9.4.0
8.12.0
8.11.2
8.11.1
8.10.4
8.10.3
8.10.2
8.10.1
8.10.0
8.5.2
8.5.1
8.5.0
8.4.1
8.4.0
8.3.1
8.3.0
7.11.5
7.9.1
7.7.0
7.1.3
7.1.2
7.1.1
6.12.11
6.12.10
6.12.9
6.12.8
6.12.7
6.12.6
6.12.5
6.12.4
6.12.3
6.12.2
6.12.1
6.12.0
6.11.2
6.11.1
6.11.0
6.10.2
6.10.1
6.10.0
6.9.2
6.9.1
6.9.0
6.7.7
6.7.6
6.7.5
6.7.4
6.7.3
6.7.2
6.7.1
6.7.0
6.6.0
2.1.0
0.12.1
retired
KVS Key-Value Store Abstraction Layer
Current section
Files
Jump to
Current section
Files
src/layers/kvs_stream.erl
-module(kvs_stream).
-description('KVS STREAM LAYER').
-include("kvs.hrl").
-include("stream.hrl").
-include("metainfo.hrl").
-export(?STREAM).
-export([metainfo/0]).
db() -> kvs:db().
% boot for sample
metainfo() -> #schema { name = kvs, tables = tables() }.
tables() -> [ #table { name = writer, fields = record_info(fields, writer) },
#table { name = reader, fields = record_info(fields, reader) } ].
% section: kvs_stream prelude
se(X,Y,Z) -> setelement(X,Y,Z).
e(X,Y) -> element(X,Y).
c4(R,V) -> se(#reader.args, R, V).
sn(M,T) -> se(#iter.next, M, T).
sp(M,T) -> se(#iter.prev, M, T).
si(M,T) -> se(#iter.id, M, T).
tab(T) -> e(1, T).
id(T) -> e(#iter.id, T).
en(T) -> e(#iter.next, T).
ep(T) -> e(#iter.prev, T).
acc(0) -> next;
acc(1) -> prev.
n({ok,R},C,P,Db) -> r(kvs:get(tab(R),en(R),#kvs{db=Db}),C,P);
n({error,X},_,_,_) -> {error,X}.
p({ok,R},C,P,Db) -> r(kvs:get(tab(R),ep(R),#kvs{db=Db}),C,P);
p({error,X},_,_,_) -> {error,X}.
r({ok,R},C,P) -> C#reader{cache={tab(R),id(R)},pos=P};
r({error,X},_,_) -> {error,X}.
w({ok,#writer{first=[]}},bot,C) -> C#reader{cache=[],pos=1,dir=0};
w({ok,#writer{first=B}},bot,C) -> C#reader{cache={tab(B),id(B)},pos=1,dir=0};
w({ok,#writer{cache=B,count=Size}},top,C) -> C#reader{cache={tab(B),id(B)},pos=Size,dir=1};
w({error,X},_,_) -> {error,X}.
% next, prev, top, bot
top(#reader{}=X) -> top(X,db()).
top(#reader{feed=F}=C,Db) -> w(kvs:get(writer,F,#kvs{db=Db}),top,C).
bot(#reader{}=X) -> bot(X,db()).
bot(#reader{feed=F}=C,Db) -> w(kvs:get(writer,F,#kvs{db=Db}),bot,C).
next(#reader{}=X) -> next(X,db()).
next(#reader{cache=[]},_) -> {error,empty};
next(#reader{cache={T,R},pos=P}=C,Db) -> n(kvs:get(T,R,#kvs{db=Db}),C,P+1,Db).
prev(#reader{}=X) -> prev(X,db()).
prev(#reader{cache=[]},_) -> {error,empty};
prev(#reader{cache={T,R},pos=P}=C,Db) -> p(kvs:get(T,R,#kvs{db=Db}),C,P-1,Db).
% take, drop, feed
drop(#reader{}=X) -> drop(X,db()).
drop(#reader{cache=[]}=C,_) -> C#reader{args=[]};
drop(#reader{dir=D,cache=B,args=N,pos=P}=C,Db) -> drop(acc(D),N,C,C,P,B,Db).
take(#reader{}=X) -> take(X,db()).
take(#reader{cache=[]}=C,_) -> C#reader{args=[]};
take(#reader{dir=D,cache=_B,args=N,pos=P}=C,Db) -> take(acc(D),N,C,C,[],P,Db).
take(_,_,{error,_},C2,R,P,_) -> C2#reader{args=lists:flatten(R),pos=P,cache={tab(hd(R)),en(hd(R))}};
take(_,0,_,C2,R,P,_) -> C2#reader{args=lists:flatten(R),pos=P,cache={tab(hd(R)),en(hd(R))}};
take(A,N,#reader{cache={T,I},pos=P}=C,C2,R,_,Db) -> take(A,N-1,?MODULE:A(C,Db),C2,[element(2,kvs:get(T,I,#kvs{db=Db}))|R],P,Db).
drop(_,_,{error,_},C2,P,B,_) -> C2#reader{pos=P,cache=B};
drop(_,0,_,C2,P,B,_) -> C2#reader{pos=P,cache=B};
drop(A,N,#reader{cache=B,pos=P}=C,C2,_,_,Db) -> drop(A,N-1,?MODULE:A(C,Db),C2,P,B,Db).
feed(Feed) -> feed(Feed,db()).
feed(Feed,Db) -> #reader{args=Args} = take((reader(Feed,Db))#reader{args=-1},Db), Args.
% new, save, load, writer, reader
load_reader (Id) -> load_reader(Id,db()).
load_reader (Id,Db) -> case kvs:get(reader,Id,#kvs{db=Db}) of {ok,C} -> C; _ -> #reader{id=[]} end.
save (C) -> save(C,db()).
save (C,Db) -> NC = c4(C,[]), kvs:put(NC,#kvs{db=Db}), NC.
writer (Id) -> writer(Id,db()).
writer (Id,Db) -> case kvs:get(writer,Id,#kvs{db=Db}) of {ok,W} -> W; {error,_} -> #writer{id=Id} end.
reader (Id) -> reader(Id,db()).
reader (Id,Db) -> case kvs:get(writer,Id,#kvs{db=Db}) of
{ok,#writer{first=[]}} -> #reader{id=kvs:seq(reader,1),feed=Id,cache=[]};
{ok,#writer{first=F}} -> #reader{id=kvs:seq(reader,1),feed=Id,cache={tab(F),id(F)}};
{error,_} -> save(#writer{id=Id},Db), reader(Id,Db) end.
% add, remove, append
add(#writer{}=X) -> add(X,db()).
add(#writer{args=M}=C,Db) when element(2,M) == [] -> add(si(M,kvs:seq(tab(M),1)),C,Db);
add(#writer{args=M}=C,Db) -> add(M,C,Db).
add(M,#writer{cache=[]}=C,Db) ->
_Id=id(M), N=sp(sn(M,[]),[]), kvs:put(N,#kvs{db=Db}),
C#writer{cache=N,count=1,first=N};
add(M,#writer{cache=V1,count=S}=C,Db) ->
{ok,V} = kvs:get(tab(V1),id(V1),#kvs{db=Db}),
N=sp(sn(M,[]),id(V)), P=sn(V,id(M)), kvs:put([N,P],#kvs{db=Db}),
C#writer{cache=N,count=S+1}.
remove(Rec,Feed) -> remove(Rec,Feed,db()).
remove(_Rec,Feed,Db) ->
{ok,W=#writer{count=Count}} = kvs:get(writer,Feed,#kvs{db=Db}),
NC = Count-1,
kvs:save(W#writer{count=NC},#kvs{db=Db}),
NC.
append(Rec,Feed) -> append(Rec,Feed,db()).
append(Rec,Feed,Db) ->
kvs:ensure(#writer{id=Feed},#kvs{db=Db}),
Name = element(1,Rec),
Id = element(2,Rec),
case kvs:get(Name,Id,#kvs{db=Db}) of
{ok,_} -> Id;
{error,_} -> kvs:save(kvs:add((kvs:writer(Feed,#kvs{db=Db}))#writer{args=Rec},#kvs{db=Db}),#kvs{db=Db}), Id end.