streams

include/kvs.hrl

@@ -2,10 +2,10 @@
 -define(KVS_HRL, true).
 
 -record(cur, {id =  [] :: term(),
-              val=  [] :: [] | tuple(),
-              dir=   0 ::  0 | 1,
               top=  [] :: [] | integer(),
-              bot=  [] :: [] | integer()}).
+              bot=  [] :: [] | integer(),
+              dir=   0 ::  0 | 1,
+              val=  [] :: [] | tuple()}).
 
 -define(ITER, id=   [] :: term(),
               next= [] :: [] | integer(),

man/kvs_stream.htm

@@ -54,6 +54,20 @@
 
 </code></figure>
 
+<p><ul>
+<li>id &mdash; Unique key of the cursor</li>
+<li>val &mdash; Cached value of current element of the list</li>
+<li>dir &mdash; 0 from top do next, 1 from bot do prev.</li>
+<li>top &mdash; The top of the list</li>
+<li>bot &mdash; The bottom of the list</li>
+</ul></p>
+
+<p><ul>
+<li>id &mdash; Unique key of the record in the list</li>
+<li>next &mdash; The next element of the list</li>
+<li>prev &mdash; The prev element of the list</li>
+</ul></p>
+
     </section>
     <section>
 
@@ -100,50 +114,168 @@ check() ->
 
 <p>Creates a KVS cursor.</p>
 
+<figure><code>
+ > kvs_stream:new().
+ {cur,1,[],0,[],[]}
+ > kvs:get(id_seq,"cur").
+ {ok,{id_seq,"cur",1}}
+ > kvs_stream:new().
+ {cur,2,[],0,[],[]}
+ > kvs:get(id_seq,"cur").
+ {ok,{id_seq,"cur",2}}
+
+</code></figure>
+
 <h4>save(#cur{}) -> #cur{}.</h4>
 
 <p>Saves the cursor to database.<?p>
 
-<h4>load() -> #ok{data::#cur{}} | #error{}.</h4>
+<figure><code>
+ > kvs:all(cur).
+ []
+ > kvs_stream:save(kvs_stream:new()).
+ {cur,3,[],0,[],[]}
+ > kvs:all(cur).
+ [{cur,3,[],0,[],[]}]
+
+</code></figure>
+
+<h4>load(Id) -> #ok{data::#cur{}} | #error{}.</h4>
 
 <p>Gets a cursor from database.</p>
 
-<h4>next(#cur{}) -> #cur{} | #error{}.</h4>
+<figure><code>
+  > kvs_stream:load(3).
+  {cur,3,[],[],0,[]}
 
-<p>Moves cursor to next. Consume data top down.</p>
+</code></figure>
+
+<h4>add(Message,#cur{}) -> #cur{}.</h4>
+
+<p>Adds message to datatabase and update cursor to new data.
+   Message is linked on next prev fields with existed data under cursor.
+   If cursor doesn't contain top or bottom value the additional
+   seek to the end is performed according to cursor direction.</p>
+
+<figure><code>
+ > rr(kvs_user).
+ [block,column,container,cur,group,id_seq,iter,iterator,kvs,
+  log,operation,person,query,schema,table,user]
+
+ > kvs_stream:save(
+   kvs_stream:add(#person{},
+   kvs_stream:load(3))).
+ #cur{id = 4,top = 1,bot = 1,dir = 0,
+      val = #person{id = 1,next = [],prev = [],mail = [],
+                    name = [],pass = [],zone = [],type = []}}
+
+</code></figure>
 
 <h4>prev(#cur{}) -> #cur{} | #error{}.</h4>
 
-<p>Moves cursor to prev. Consume data bottom up.</p>
+<p>Moves cursor to prev. Consume data bottom up.
+   Reutrn error if lists is empty, otherwise next element or last.</p>
+
+<figure><code>
+ > kvs_stream:prev(kvs_stream:load(3)).
+ {error,#cur{id = 3,top = 1,bot = 1,dir = 0,
+             val = #person{id = 1,next = [],prev = [],mail = [],
+                           name = [],pass = [],zone = [],type = []}}}
+
+</code></figure>
+
+<h4>next(#cur{}) -> #cur{} | #error{}.</h4>
+
+<p>Moves cursor to next. Consume data top down.
+   Reutrn error if lists is empty, otherwise next element or last.</p>
+
+<figure><code>
+ > kvs_stream:next(
+   kvs_stream:save(
+   kvs_stream:add(#person{},
+   kvs_stream:load(3)))).
+ #cur{id = 3,top = 2,bot = 1,dir = 0,
+      val = #person{id = 1,next = [],prev = 2,mail = [],name = [],
+                     pass = [],zone = [],type = []}}
+
+</code></figure>
 
 <h4>seek(Id,#cur{}) -> #cur{} | #error{}.</h4>
 
 <p>Moves cursor to record by its id.
    If cursor has no cached value then function returns error.</p>
 
+<figure><code>
+ >  kvs_stream:seek(2,
+    kvs_stream:save(
+    kvs_stream:add(#person{id=kvs:next_id(person,1)},
+    kvs_stream:load(3)))).
+ #cur{id = 3,top = 3,bot = 1,dir = 0,
+      val = #person{id = 2,next = 1,prev = 3,mail = [],name = [],
+                    pass = [],zone = [],type = []}}
+
+</code></figure>
+
 <h4>top(#cur{}) -> #cur{} | #error{}.</h4>
 
 <p>Moves cursor to top of the list.</p>
 
+<figure><code>
+ > kvs_stream:top(
+   kvs_stream:load(3)).
+ #cur{id = 3,top = 3,bot = 1,dir = 0,
+      val = #person{id = 3,next = 2,prev = [],mail = [],name = [],
+                    pass = [],zone = [],type = []}}
+
+</code></figure>
+
 <h4>bot(#cur{}) -> #cur{} | #error{}.</h4>
 
 <p>Moves cursor to bottom of the list.</p>
 
-<h4>add(Message,#cur{}) -> #cur{}.</h4>
+<figure><code>
+ > kvs_stream:bot(kvs_stream:load(3)).
+ #cur{id = 3,top = 3,bot = 1,dir = 0,
+      val = #person{id = 1,next = [],prev = 2,mail = [],name = [],
+                    pass = [],zone = [],type = []}}
 
-<p>Adds message to datatabase and update cursor to new data.
-   Message is linked on next prev fields with existed data under cursor.
-   If cursor doesn't contain top or bottom value the additional
-   seek to the end is performed according to cursor direction.</p>
+</code></figure>
+
+<h4>take(N,#cur{}) -> list().</h4>
+
+<p>Trying to consume N records from stream using its current value and direction.
+   Returns consumed data. Usually you seek to some position and then consume some data.</p>
+
+<figure><code>
+ > kvs_stream:take(-1,kvs_stream:load(3)).
+ [#person{id = 1,next = [],prev = 2, mail = [],name = [],
+          pass = [],zone = [],type = []},
+  #person{id = 2,next = 1,prev = 3,mail = [],name = [],
+          pass = [],zone = [],type = []},
+  #person{id = 3,next = 2,prev = [],mail = [],name = [],
+          pass = [],zone = [],type = []}]
+
+</code></figure>
+
+<p>Moves cursor to bottom of the list.</p>
 
 <h4>remove(Id,#cur{}) -> #cur{} | #error{}.</h4>
 
 <p>Removes record by id from database and unlink it from list.
-   If cursor has no cached value then function returns error.</p>
+   If cursor has no cached value then function returns error.
+   Please do not use remove, keep your data immutable :-)</p>
 
-<h4>take(N,#cur{}) -> list().</h4>
+<figure><code>
+ > kvs_stream:take(-1,
+   kvs_stream:save(
+   kvs_stream:remove(2,
+   kvs_stream:load(3)))).
+ [#person{id = 1,next = [],prev = 3,mail = [],name = [],
+          pass = [],zone = [],type = []},
+  #person{id = 3,next = 1,prev = [],mail = [],name = [],
+          pass = [],zone = [],type = []}]
 
-<p>Trying to consume N records from stream using its current value and direction. Returns consumed data.</p>
+</code></figure>
 
     </section>
     <section>

src/kvs_feed.erl

@@ -12,6 +12,7 @@ metainfo() ->  #schema{name=kvs,tables= core() ++ feeds() }.
 
 core()     -> [ #table{name=config,fields=record_info(fields,config)},
                 #table{name=log,container=true,fields=record_info(fields,log)},
+                #table{name=cur,container=feed,fields=record_info(fields,cur)},
                 #table{name=operation,container=log,fields=record_info(fields,operation)},
                 #table{name=id_seq,fields=record_info(fields,id_seq),keys=[thing]} ].
 

src/kvs_stream.erl

@@ -15,18 +15,15 @@ bot({ok,#cur{}=C})    -> bot(C);
 bot({error,X})        -> {error,X};
 bot(#cur{bot=[]}=C)   -> C#cur{val=[]};
 bot(#cur{bot=B}=C)    -> seek(B,C).
+add(M,#cur{dir=D}=C) when element(2,M) == [] -> add(dir(D),si(M,kvs:next_id(tab(M),1)),C);
 add(M,#cur{dir=D}=C)  -> add(dir(D),M,C).
 save(#cur{}=C)        -> kvs:put(C), C.
-load(K)               -> kvs:get(cur,K).
+load(K)               -> case kvs:get(cur,K) of {ok,C} -> C; E -> E end.
 next(#cur{val=[]}=C)  -> {error,[]};
 next(#cur{val=B}=C)   -> lookup(kvs:get(tab(B),en(B)),C).
 prev(#cur{val=[]}=C)  -> {error,[]};
 prev(#cur{val=B}=C)   -> lookup(kvs:get(tab(B),ep(B)),C).
-take(N,{ok,#cur{}}=C)    -> take(N,C);
-take(N,{error,X})        -> {error,X};
-take(N,#cur{dir=D}=C)    -> take(acc(D),N,C,[]).
-seek(I,  {ok,#cur{}=C})  -> seek(I,C); % accept ok/error functors is ok
-seek(I,  {error,X})      -> {error,X};
+take(N,#cur{dir=D}=C)    -> take(acc(D),N,?MODULE:(dir(D))(C),[]).
 seek(I,  #cur{val=[]}=C) -> {error,[]};
 seek(I,   #cur{val=B}=C) -> {ok,R}=kvs:get(tab(B),I), C#cur{val=R}.
 remove(I,#cur{val=[]}=C) -> {error,val};
@@ -49,10 +46,10 @@ join([L,  R],C) -> N=sp(R,id(L)), kvs:put([N,sn(L,id(R))]), C#cur{
 
 sn(M,T) -> setelement(#iter.next, M, T).
 sp(M,T) -> setelement(#iter.prev, M, T).
-el(X,T) -> element(X,T).
-tab(T)  -> element(1,T).
-id(T)   -> element(2,T).
-
+el(X,T) -> element(X, T).
+tab(T)  -> element(1, T).
+id(T)   -> element(#iter.id, T).
+si(M,T) -> setelement(#iter.id, M, T).
 en(T)   -> element(#iter.next, T).
 ep(T)   -> element(#iter.prev, T).
 dir(0)  -> top;
@@ -68,7 +65,7 @@ fix({ok,O}) -> O;
 fix(_)      -> [].
 
 lookup({ok,R},C)          -> C#cur{val=R};
-lookup({error,X},C)       -> {error,X}.
+lookup({error,X},C)       -> {error,C}.
 take(_,_,{error,_},R)     -> lists:flatten(R);
 take(_,0,_,R)             -> lists:flatten(R);
 take(A,N,#cur{val=B}=C,R) -> take(A,N-1,?MODULE:A(C),[B|R]).

src/kvs_user.erl

@@ -10,7 +10,6 @@ metainfo() ->
     #schema{name=kvs,tables=[
         #table{name=person,container=feed,fields=record_info(fields,person)},
         #table{name=group,container=feed,fields=record_info(fields,group)},
-        #table{name=cur,container=feed,fields=record_info(fields,cur)},
         #table{name=user,container=feed,fields=record_info(fields,user),keys=[email]}
     ]}.