Merge pull request #59 from synrc/tkit

Tkit 2

README.md

@@ -1,6 +1,8 @@
 KVS: Abstract Chain Database
 ============================
 
+tkit
+
 [![Actions Status](https://github.com/synrc/kvs/workflows/mix/badge.svg)](https://github.com/synrc/kvs/actions)
 [![Build Status](https://travis-ci.com/synrc/kvs.svg?branch=master)](https://travis-ci.com/synrc/kvs)
 [![Hex pm](http://img.shields.io/hexpm/v/kvs.svg?style=flat)](https://hex.pm/packages/kvs)

include/stream.hrl

@@ -2,7 +2,7 @@
 -define(STREAM_HRL, true).
 -include("kvs.hrl").
 -include("cursors.hrl").
--define(STREAM, [top/1, bot/1, next/1, prev/1, drop/1, take/1, append/2, cut/2, feed/1,
+-define(STREAM, [top/1, bot/1, next/1, prev/1, drop/1, take/1, append/2, feed/1,
                  load_reader/1, writer/1, reader/1, save/1, add/1, remove/2]).
 
 -spec top(#reader{})  -> #reader{}.
@@ -19,5 +19,4 @@
 -spec add(#writer{}) -> #writer{}.
 -spec append(tuple(),term()) -> any().
 -spec remove(tuple(),term()) -> integer().
--spec cut(term(),term()) -> {ok,non_neg_integer()} | {error, not_found}.
 -endif.

src/kvs.erl

@@ -9,7 +9,7 @@
 -include("cursors.hrl").
 -include("kvs.hrl").
 -include("backend.hrl").
--export([dump/0,metainfo/0,ensure/1,seq_gen/0,fold/6,fold/7,head/1,head/2,fetch/2,fetch/3,feed/2]).
+-export([dump/0,metainfo/0,ensure/1,seq_gen/0,fold/6,fold/7,head/1,head/2,feed/2]).
 -export(?API).
 -export(?STREAM).
 -compile(export_all).
@@ -145,8 +145,3 @@ remove(Rec,Feed) -> remove(Rec,Feed,#kvs{mod=dba(),st=kvs_stream()}).
 remove(Rec,Feed, #kvs{st=Mod}) -> Mod:remove(Rec,Feed).
 head(Key) -> case (kvs:take((kvs:reader(Key))#reader{args=1}))#reader.args of [X] -> X; [] -> [] end.
 head(Key,Count) -> (kvs:take((kvs:reader(Key))#reader{args=Count,dir=1}))#reader.args.
-fetch(Table, Key) -> fetch(Table, Key, []).
-fetch(Table, Key, Default) -> case get(Table, Key) of
-                                        {ok, Value} -> Value;
-                                        _ -> Default
-                                  end.

src/layers/kvs_stream.erl

@@ -107,8 +107,3 @@ append(Rec,Feed) ->
    case kvs:get(Name,Id) of
         {ok,_}    -> Id;
         {error,_} -> kvs:save(kvs:add((kvs:writer(Feed))#writer{args=Rec})), Id end.
-
-cut(_Feed,Id) ->
-   case kvs:get(writer,Id) of
-        {ok,#writer{count=N}} -> {ok,N};
-        {error,_} -> {error,not_found} end.

src/stores/kvs_rocks.erl

@@ -4,8 +4,8 @@
 -include("metainfo.hrl").
 -include_lib("stdlib/include/qlc.hrl").
 -export(?BACKEND).
--export([ref/0,cut/8,next/8,prev/8,prev2/8,next2/8,bt/1,key/2,key/1,fd/1]).
--export([seek_it/1, move_it/3]).
+-export([ref/0,bt/1,key/2,key/1,fd/1]).
+-export([seek_it/1, move_it/3, take_it/4]).
 
 e(X,Y)     -> element(X,Y).
 bt([])     -> [];
@@ -16,10 +16,21 @@ tb(T) when is_atom(T) -> atom_to_binary(T, utf8);
 tb(T) when is_binary(T) -> T;
 tb(T)      -> term_to_binary(T).
 
+fmt([]) -> [];
+fmt(K) ->
+  B = tb(K),
+  S = if byte_size(B) > 1 -> 2;true -> byte_size(B) end,
+  B1 = binary:split(B,[<<"/">>,<<"//">>], [{scope,{0,S}},trim_all]),
+  B2 = case B1 of [] -> <<>>;[X|_] -> X end,
+  S1 = if byte_size(B2) > 1 -> -2; true -> 0 end,
+  B3 = binary:split(B2,[<<"/">>,<<"//">>], [{scope,{byte_size(B2),S1}},trim_all]),
+  B4 = case B3 of [] -> <<>>;[X1|_] -> X1 end,
+  B4.
+
 key(R)     when is_tuple(R) andalso tuple_size(R) > 1 -> key(e(1,R), e(2,R));
 key(R)     -> key(R,[]).
 key(Tab,R) when is_tuple(R) andalso tuple_size(R) > 1 -> key(Tab, e(2,R));
-key(Tab,R) -> iolist_to_binary([lists:join(<<"/">>, lists:flatten([<<>>, tb(Tab), tb(R)]))]).
+key(Tab,R) -> iolist_to_binary([lists:join(<<"/">>, lists:flatten([<<>>, fmt(Tab), fmt(R)]))]).
 
 fd(Key) ->
   B = lists:reverse(binary:split(tb(Key), [<<"/">>, <<"//">>], [global, trim_all])),
@@ -30,20 +41,27 @@ o(<<>>,FK,_,_) -> {ok,FK,[],[]};
 o(Key,FK,Dir,Fx) ->
   S = size(FK),
 
-  Sheaf = fun (F,K,H,V,Acc) when binary_part(K,{0,S}) == FK -> {F(H,Dir),H,[V|Acc]};
-                                              (_,K,H,V,Acc) -> close_it(H),
-                                                               throw({ok,fd(K),bt(V),[bt(A1)||A1<-Acc]}) end,
+  Infotech = fun (F,K,H,V,Acc) when binary_part(K,{0,S}) == FK -> {F(H,Dir),H,[V|Acc]};
+                 (_,K,H,V,Acc) -> close_it(H),
+                                  throw({ok,fd(K),bt(V),[bt(A1)||A1<-Acc]}) end,
+  Privat = fun(F,K,V,H) -> case F(H,prev) of
+      {ok,K1,V1} when binary_part(K,{0,S}) == FK -> {{ok,K1,V1},H,[V]};
+      {ok,K1,V1} -> Infotech(F,K1,H,V1,[]);
+      E -> E
+  end end,
 
   It = fun(F,{ok,H})            -> {F(H,{seek,Key}),H};
-          (F,{{ok,K,V},H})      -> Sheaf(F,K,H,V,[]);
-          (F,{{ok,K,V},H,A})    -> Sheaf(F,K,H,V,A);
+          (F,{{ok,K,V},H})
+              when Dir =:= prev -> Privat(F,K,V,H);
+          (F,{{ok,K,V},H})      -> Infotech(F,K,H,V,[]);
+          (F,{{ok,K,V},H,A})    -> Infotech(F,K,H,V,A);
           (_,{{error,_},H,Acc}) -> {{ok,[],[]},H,Acc};
           (F,{R,O})             -> F(R,O);
           (F,H)                 -> F(H) end,
   catch case lists:foldl(It, {ref(),[]}, Fx) of
-    {{ok,K,Bin},_,A} when binary_part(K,{0,S}) == FK  -> {ok,fd(K),bt(Bin),[bt(A1)||A1<-A]};
-    {{ok,K,Bin},_,_}                                  -> {ok,fd(K),bt(Bin),[]};
-    {{ok,K,Bin},_}                                    -> {ok,fd(K),bt(Bin),[]}
+    {{ok,K,Bin},_,A}  -> {ok,fd(K), bt(Bin),[bt(A1)||A1<-A]};
+    {{ok,K,Bin},_}    -> {ok,fd(K), bt(Bin),[]};
+    {{error,_},_,Acc} -> {ok,fd(FK),bt(shd(Acc)),[bt(A1) ||A1<-Acc]}
   end.
 
 start()    -> ok.
@@ -63,6 +81,10 @@ index(_,_,_) -> [].
 close_it(H) -> try rocksdb:iterator_close(H) catch error:badarg -> ok end.
 seek_it(K) -> o(K,K,ok,[fun rocksdb:iterator/2,fun rocksdb:iterator_move/2]).
 move_it(K,FK,Dir) -> o(K,FK,Dir,[fun rocksdb:iterator/2,fun rocksdb:iterator_move/2,fun rocksdb:iterator_move/2]).
+take_it(Key,FK,Dir,N) when is_integer(N) andalso N >= 0 ->
+  o(Key,FK,Dir,[fun rocksdb:iterator/2,fun rocksdb:iterator_move/2] ++
+               lists:map(fun(_) -> fun rocksdb:iterator_move/2 end,lists:seq(1,N)));
+take_it(Key,FK,Dir,_) -> take_it(Key,FK,Dir,0).
 
 get(Tab, Key) ->
     case rocksdb:get(ref(), key(Tab,Key), []) of
@@ -74,48 +96,12 @@ put(Record) -> rocksdb:put(ref(), key(Record), term_to_binary(Record), [{sync,tr
 delete(Feed, Id) -> rocksdb:delete(ref(), key(Feed,Id), []).
 
 count(_) -> 0.
-all(R) -> {ok,I} = rocksdb:iterator(ref(), []),
-           Key = key(R),
-           First = rocksdb:iterator_move(I, {seek,Key}),
-           lists:reverse(next(I,Key,size(Key),First,[],[],-1,0)).
 
-next(I,Key,S,A,X,T,N,C) -> {_,L} = next2(I,Key,S,A,X,T,N,C), L.
-prev(I,Key,S,A,X,T,N,C) -> {_,L} = prev2(I,Key,S,A,X,T,N,C), L.
+all(R) -> kvs_st:feed(R).
 
 shd([]) -> [];
 shd(X) -> hd(X).
 
-next2(_,Key,_,_,X,T,N,C) when C == N -> {shd(lists:reverse(T)),T};
-next2(I,Key,S,{ok,A,X},_,T,N,C) -> next2(I,Key,S,A,X,T,N,C);
-next2(_,Key,_,{error,_},X,T,_,_) -> {shd(lists:reverse(T)),T};
-next2(I,Key,S,A,X,T,N,C) when size(A) > S ->
-     case binary:part(A, 0, S) of Key ->
-          next2(I, Key, S, rocksdb:iterator_move(I, next), [], [bt(X)|T], N, C + 1);
-          _ -> {shd(lists:reverse(T)),T} end;
-next2(_,Key,_,{ok,A,_},X,T,_,_) -> {bt(X),T};
-next2(_,Key,_,_,X,T,_,_) -> {shd(lists:reverse(T)),T}.
-
-prev2(_,Key,_,_,X,T,N,C) when C == N -> {bt(X),T};
-prev2(I,Key,S,{ok,A,X},_,T,N,C) -> prev2(I,Key,S,A,X,T,N,C);
-prev2(_,Key,_,{error,_},X,T,_,_) -> {bt(X),T};
-prev2(I,Key,S,A,X,T,N,C) when size(A) > S ->
-     case binary:part(A, 0, S) of Key ->
-          prev2(I, Key, S, rocksdb:iterator_move(I, prev), [], [bt(X)|T], N, C + 1);
-          _ -> {shd(lists:reverse(T)),T} end;
-prev2(_,Key,_,{ok,A,_},X,T,_,_) -> {bt(X),T};
-prev2(_,Key,_,_,X,T,_,_) -> {bt(X),T}.
-
-cut(_,_,_,_,_,_,N,C) when C == N -> C;
-cut(I,Key,S,{ok,A,X},_,T,N,C) -> prev(I,Key,S,A,X,T,N,C);
-cut(_,___,_,{error,_},_,_,_,C) -> C;
-cut(I,Key,S,A,_,_,N,C) when size(A) > S ->
-     case binary:part(A,0,S) of Key ->
-          rocksdb:delete(ref(), A, []),
-          Next = rocksdb:iterator_move(I, prev),
-          cut(I,Key, S, Next, [], A, N, C + 1);
-                                  _ -> C end;
-cut(_,_,_,_,_,_,_,C) -> C.
-
 seq(_,_) ->
   case os:type() of
        {win32,nt} -> {Mega,Sec,Micro} = erlang:timestamp(), integer_to_list((Mega*1000000+Sec)*1000000+Micro);

src/stores/kvs_st.erl

@@ -4,7 +4,7 @@
 -include("stream.hrl").
 -include("metainfo.hrl").
 -export(?STREAM).
--import(kvs_rocks, [key/2, key/1, bt/1, ref/0, fd/1, seek_it/1, move_it/3]).
+-import(kvs_rocks, [key/2, key/1, bt/1, ref/0, fd/1, seek_it/1, move_it/3, take_it/4]).
 
 % section: kvs_stream prelude
 
@@ -14,77 +14,30 @@ c4(R,V) -> se(#reader.args,  R, V).
 si(M,T) -> se(#it.id, M, T).
 id(T) -> e(#it.id, T).
 
-% section: next, prev
-feed(Feed) -> #reader{args=Args} = take((reader(Feed))#reader{args=-1}), Args.
+k(F,[]) -> key(F);
+k(_,{_,Id,SF}) -> key(SF,Id).
 
-top(#reader{feed=Feed}=C) -> #writer{count=Cn} = writer(Feed), read_it(C#reader{count=Cn},seek_it(key(Feed))).
-bot(#reader{feed=Feed}=C) -> #writer{cache=Ch, count=Cn} = writer(Feed), C#reader{cache=Ch, count=Cn, dir=1}.
-
-% iterator -> specific feed reader
-read_it(C,{ok,F,V,H}) -> C#reader{cache={e(1,V),id(V), F}, args=lists:reverse(H)};
+read_it(C,{ok,_,[],H}) -> C#reader{cache=[], args=lists:reverse(H)};
+read_it(C,{ok,F,V,H})  -> C#reader{cache={e(1,V),id(V),F}, args=lists:reverse(H)};
 read_it(C,_) -> C.
 
-next(#reader{feed=Feed,cache=I}=C) -> read_it(C,move_it(key(Feed,I),key(Feed),next)).
-prev(#reader{cache=I,feed=Feed}=C) -> read_it(C,move_it(key(Feed,I),key(Feed),prev)).
-
-% section: take, drop
-
+top(#reader{feed=Feed}=C) -> #writer{count=Cn} = writer(Feed), read_it(C#reader{count=Cn},seek_it(key(Feed))).
+bot(#reader{feed=Feed}=C) -> #writer{cache=Ch, count=Cn} = writer(Feed), C#reader{cache=Ch, count=Cn}.
+next(#reader{feed=Feed,cache=I}=C) -> read_it(C,move_it(k(Feed,I),key(Feed),next)).
+prev(#reader{cache=I,feed=Feed}=C) -> read_it(C,move_it(k(Feed,I),key(Feed),prev)).
+take(#reader{args=N,feed=Feed,cache=I,dir=1}=C) -> read_it(C,take_it(k(Feed,I),key(Feed),prev,N));
+take(#reader{args=N,feed=Feed,cache=I,dir=_}=C) -> read_it(C,take_it(k(Feed,I),key(Feed),next,N)).
 drop(#reader{args=N}=C) when N =< 0 -> C;
-drop(#reader{args=N,feed=Feed,cache=I}=C) -> (take(C#reader{dir=0}))#reader{args=[]}.
-
-%  1. Курсор всегда выставлен на следущий невычитанный элемент
-%  2. Если после вычитки курсор указывает на недавно вычитаный элемент -- это признак конца списка
-%  3. Если результат вычитки меньше требуемого значения -- это признак конца списка
-%  4. Если курсор установлен в конец списка и уже вернул его последний элемент
-%     то результат вычитки будет равным пустому списку
-
+drop(#reader{}=C) -> (take(C#reader{dir=0}))#reader{args=[]}.
 
-take(#reader{pos='end',dir=0}=C) -> C#reader{args=[]}; % 4
-take(#reader{args=N,feed=Feed,cache={T,O,_},dir=0}=C) -> take(C#reader{cache={T,O}});
-take(#reader{args=N,feed=Feed,cache={T,O},dir=0}=C) -> % 1
-   Key = key(Feed),
-   {ok,I} = rocksdb:iterator(ref(), []),
-   {ok,K,BERT} = rocksdb:iterator_move(I, {seek,key(Feed,{T,O})}),
-   {KK,Res} = kvs_rocks:next2(I,Key,size(Key),K,BERT,[],case N of -1 -> -1; J -> J + 1 end,0),
-   Last = last(KK,O,'end'),
-   case {Res,length(Res)} of
-        {[],_} -> C#reader{args=[],cache=[]};
-        {[H],  _A} when element(2,KK) == O -> C#reader{args=Res,pos=Last,cache={e(1,H),e(2,H)}}; % 2
-        {[H|_X],A} when A < N + 1 orelse N == -1 -> C#reader{args=Res,cache={e(1,H),e(2,H)},pos=Last};
-        {[H| X],A} when A == N -> C#reader{args=[bt(BERT)|X],cache={e(1,H),e(2,H)},pos=Last};
-        {[H|_X],A} when A =< N andalso Last == 'end'-> C#reader{args=Res,cache={e(1,H),e(2,H)},pos=Last};
-        {[H| X],_} -> C#reader{args=X,cache={e(1,H),e(2,H)}} end;
-
-
-take(#reader{pos=0,dir=0}=C)       -> C#reader{pos='begin',args=[]};
-take(#reader{pos='begin',dir=1}=C) -> C#reader{args=[]}; % 4
-take(#reader{pos=0,cache=[],dir=1}=C) -> C#reader{args=[]};
-
-% TODO: try to remove lists:reverse and abstract both branches
-take(#reader{args=N,feed=Feed,cache={T,O,_},dir=1}=C) -> take(C#reader{cache={T,O}});
-take(#reader{args=N,feed=Feed,cache={T,O},dir=1}=C) -> % 1
-   Key = key(Feed),
-   {ok,I} = rocksdb:iterator(ref(), []),
-   {ok,K,BERT} = rocksdb:iterator_move(I, {seek,key(Feed,{T,O})}),
-   {KK,Res} = kvs_rocks:prev2(I,Key,size(Key),K,BERT,[],case N of -1 -> -1; J -> J + 1 end,0),
-   Last = last(KK,O,'begin'),
-   case {lists:reverse(Res),length(Res)} of
-        {[],_} -> C#reader{args=[],cache=[]};
-        {[H],_} when element(2,KK) == O -> C#reader{args=Res,pos=Last,cache={e(1,H),e(2,H)}}; % 2
-        {[_|_],A} when A < N - 1 orelse N == -1 -> [HX|_] = Res, C#reader{args=Res,cache={e(1,HX),e(2,HX)},pos=Last};
-        {[_|X],A} when A == N -> [HX|_] = Res, C#reader{args=[bt(BERT)|X],cache={e(1,HX),e(2,HX)},pos=Last};
-        {[_|_],A} when A =< N andalso Last == 'begin'-> [HX|_] = Res, C#reader{args=lists:reverse(Res),cache={e(1,HX),e(2,HX)},pos=Last};
-        {[_|_],_} -> [HX|TL] = Res, C#reader{args=lists:reverse(TL),cache={e(1,HX),e(2,HX)}} end.
-
-last(KK,O,Atom) ->
-   Last = case KK of
-      [] -> Atom;
-      _ when element(2,KK) == O -> Atom;
-      _ -> 0
-   end,
-   Last.
-
-% new, save, load, up, down, top, bot
+feed(Feed) -> feed(fun(#reader{}=R) -> take(R#reader{args=4}) end, top(reader(key(Feed))),[]).
+feed(F,#reader{cache=C1}=R,Acc) ->
+  #reader{args=A, cache=Ch, feed=Feed} = R1 = F(R),
+  case Ch of
+    C1 -> Acc ++ A;
+    {_,_,K} when binary_part(K,{0,byte_size(Feed)}) == Feed -> feed(F, R1, Acc ++ A);
+    _ -> Acc ++ A
+  end.
 
 load_reader(Id) ->
     case kvs:get(reader,Id) of
@@ -94,10 +47,7 @@ load_reader(Id) ->
 writer(Id) -> case kvs:get(writer,Id) of {ok,W} -> W; {error,_} -> #writer{id=Id} end.
 reader(Id) -> case kvs:get(writer,Id) of
   {ok,#writer{id=Feed, count=Cn}} ->
-    {ok,I} = rocksdb:iterator(ref(), []),
-    {ok,F1,BERT} = rocksdb:iterator_move(I, {seek,key(Feed)}),
-    F = bt(BERT),
-    #reader{id=kvs:seq([],[]),feed=Id,count=Cn,cache={e(1,F),e(2,F),fd(F1)}};
+    read_it(#reader{id=kvs:seq([],[]),feed=key(Feed),count=Cn},seek_it(key(Feed)));
   {error,_} -> save(#writer{id=Id}), reader(Id) end.
 save(C) -> NC = c4(C,[]), kvs:put(NC), NC.
 
@@ -106,33 +56,24 @@ save(C) -> NC = c4(C,[]), kvs:put(NC), NC.
 add(#writer{args=M}=C) when element(2,M) == [] -> add(si(M,kvs:seq([],[])),C);
 add(#writer{args=M}=C) -> add(M,C).
 
-add(M,#writer{id=Feed,count=S}=C) -> NS=S+1, raw_append(M,Feed), C#writer{cache={e(1,M),e(2,M),fd(Feed)},count=NS}.
+add(M,#writer{id=Feed,count=S}=C) -> NS=S+1, raw_append(M,Feed), C#writer{cache={e(1,M),e(2,M),key(Feed)},count=NS}.
 
 remove(Rec,Feed) ->
    kvs:ensure(#writer{id=Feed}),
    W = #writer{count=C, cache=Ch} = kvs:writer(Feed),
-   Ch1 = case {e(1,Rec),e(2,Rec)} of Ch -> Ch;_ -> [] end, % need to keep reference for next element
+   Ch1 = case {e(1,Rec),e(2,Rec),key(Feed)} of Ch -> Ch;_ -> [] end, % need to keep reference for next element
    case kvs:delete(Feed,id(Rec)) of
         ok -> Count = C - 1,
               save(W#writer{count = Count, cache=Ch1}),
               Count;
          _ -> C end.
 
-raw_append(M,Feed) ->
-   rocksdb:put(ref(), key(Feed,M), term_to_binary(M), [{sync,true}]).
+raw_append(M,Feed) -> rocksdb:put(ref(), key(Feed,M), term_to_binary(M), [{sync,true}]).
 
 append(Rec,Feed) ->
    kvs:ensure(#writer{id=Feed}),
    Id = e(2,Rec),
    W = writer(Feed),
    case kvs:get(Feed,Id) of
-        {ok,_} -> raw_append(Rec,Feed), save(W#writer{cache={e(1,Rec),Id,fd(Feed)},count=W#writer.count + 1}), Id;
+        {ok,_} -> raw_append(Rec,Feed), save(W#writer{cache={e(1,Rec),Id,key(Feed)},count=W#writer.count + 1}), Id;
         {error,_} -> save(add(W#writer{args=Rec})), Id end.
-
-cut(Feed,Id) ->
-    Key    = key(Feed),
-    A      = key(Feed,Id),
-    {ok,I} = rocksdb:iterator(ref(), []),
-    case rocksdb:iterator_move(I, {seek,A}) of
-         {ok,A,X} -> {ok,kvs_rocks:cut(I,Key,size(Key),A,X,[],-1,0)};
-                _ -> {error,not_found} end.

test/fd_test.exs

@@ -0,0 +1,88 @@
+ExUnit.start()
+
+defmodule Fd.Test do
+    use ExUnit.Case, async: false
+    require KVS
+    import Record
+
+    defrecord(:msg, id: [], body: [])
+
+    setup do: (on_exit(fn -> :ok = :kvs.leave();:ok = :kvs.destroy() end);:kvs.join())
+    setup kvs, do: [
+        id0: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/duck") end, :lists.seq(1,10)),
+        id1: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/luck") end, :lists.seq(1,10)),
+        id2: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/truck") end, :lists.seq(1,10))]
+
+    test "reader", kvs do
+        ltop = Enum.at(kvs[:id1],0)
+        dtop = Enum.at(kvs[:id0],0)
+        ttop = Enum.at(kvs[:id2],0)
+        assert KVS.reader(feed: "/crm/luck", count: 10, dir: 0, args: [], cache: {:msg, ^ltop, "/crm/luck"}) = :kvs.reader("/crm/luck")
+        assert KVS.reader(feed: "/crm/duck", count: 10, dir: 0, args: [], cache: {:msg, ^dtop, "/crm/duck"}) = :kvs.reader("/crm/duck")
+        assert KVS.reader(feed: "/crm/truck", count: 10, dir: 0, args: [], cache: {:msg, ^ttop, "/crm/truck"}) = :kvs.reader("/crm/truck")
+        assert KVS.reader(feed: "/crm", count: 0, dir: 0, args: [], cache: {:msg, ^dtop, "/crm/duck"}) = :kvs.reader("/crm")
+        assert KVS.reader(feed: "/noroute", count: 0, dir: 0, args: []) = :kvs.reader("/noroute")
+        assert KVS.reader(feed: "/", count: 0, dir: 0, args: [], cache: {:msg, ^dtop, "/crm/duck"}) = :kvs.reader("/")
+        assert KVS.reader(feed: "", count: 0, dir: 0, args: [], cache: []) = :kvs.reader([])
+    end
+
+    test "range", kvs do
+        ltop = Enum.at(kvs[:id1],0)
+        dtop = Enum.at(kvs[:id0],0)
+        lbot = Enum.at(kvs[:id1],9)
+
+        assert KVS.reader(feed: "/crm/luck", count: 10, dir: 0, args: [], cache: {:msg, ^ltop, "/crm/luck"}) = :kvs.top(:kvs.reader("/crm/luck"))
+        assert KVS.reader(feed: "/crm", count: 0, dir: 0, args: [], cache: {:msg, ^dtop, "/crm/duck"}) = :kvs.top(:kvs.reader("/crm"))
+        assert KVS.reader(feed: "/", count: 0, dir: 0, args: [], cache: {:msg, ^dtop, "/crm/duck"}) = :kvs.top(:kvs.reader("/"))
+
+        assert KVS.reader(feed: "/crm/luck", count: 10, dir: 0, args: [], cache: {:msg, ^lbot, "/crm/luck"}) = :kvs.bot(:kvs.reader("/crm/luck"))
+        assert KVS.reader(feed: "/crm", count: 0, dir: 0, args: [], cache: []) = :kvs.bot(:kvs.reader("/crm"))
+        assert KVS.reader(feed: "/", count: 0, dir: 0, args: [], cache: []) = :kvs.bot(:kvs.reader("/"))
+    end
+
+    test "next", kvs do
+        last = msg(id: Enum.at(kvs[:id1],9))
+        KVS.reader(id: rid) = :kvs.save(:kvs.top(:kvs.reader("/crm/luck")))
+        kvs[:id1] |> Enum.with_index 
+                  |> Enum.each(fn {id,9} ->
+                r = :kvs.load_reader(rid)
+                assert r1 = KVS.reader(feed: "/crm/luck", cache: c1, count: 10, dir: 0, args: [^last]) = :kvs.next(r)
+                assert KVS.reader(args: [], feed: "/crm/luck", cache: c1) = :kvs.save(r1)
+            {id,i} -> 
+                v = msg(id: Enum.at(kvs[:id1],i))
+                c = Enum.at(kvs[:id1],i+1)
+                r = :kvs.load_reader(rid)
+                assert r1 = KVS.reader(feed: "/crm/luck", cache: {:msg,^c,"/crm/luck"}, count: 10, dir: 0, args: [^v]) = :kvs.next(r)
+                assert KVS.reader(args: [], feed: "/crm/luck", cache: {:msg,^c,"/crm/luck"}) = :kvs.save(r1)
+            end)
+        r = :kvs.load_reader(rid)
+        assert r == :kvs.next(r)
+        assert r == KVS.reader(:kvs.next(:kvs.bot(r)), args: [])
+    end
+
+    test "prev", kvs do
+        out = Enum.at(kvs[:id0],9)
+        KVS.reader(id: rid) = :kvs.save(:kvs.bot(:kvs.reader("/crm/luck")))
+        ids = kvs[:id1] |> Enum.reverse
+        ids |> Enum.with_index
+            |> Enum.each(fn {id,9} ->
+                r = :kvs.load_reader(rid)
+                v = msg(id: Enum.at(ids, 9))
+                assert r1 = KVS.reader(feed: "/crm/luck", cache: {:msg, ^out, "/crm/duck"}, count: 10, args: [^v]) = :kvs.prev(r)
+                assert KVS.reader(args: [], feed: "/crm/luck", cache: c1) = :kvs.save(r1)
+            {id,i} ->
+                r = :kvs.load_reader(rid)
+                v = msg(id: Enum.at(ids, i))
+                c = Enum.at(ids, i+1)
+                assert r1 = KVS.reader(feed: "/crm/luck", cache: {:msg, ^c, "/crm/luck"}, count: 10, args: [^v]) = :kvs.prev(r)
+                assert KVS.reader(args: [], feed: "/crm/luck", cache: {:msg, ^c, "/crm/luck"}) = :kvs.save(r1)
+            end)
+        r = :kvs.load_reader(rid)
+        assert r = :kvs.prev(r)
+        assert r = KVS.reader(:kvs.prev(:kvs.top(r)), args: [])                 
+    end
+
+
+    defp log(x), do: IO.puts '#{inspect(x)}'
+    defp log(m, x), do: IO.puts '#{m} #{inspect(x)}'
+end

test/old_test.exs

@@ -1,326 +0,0 @@
-ExUnit.start()
-
-defmodule OLD.Test do
-  use ExUnit.Case, async: true
-  require KVS
-
-  setup do: (on_exit(fn -> :ok = :kvs.leave();:ok = :kvs.destroy() end);:kvs.join())
-
-  test "basic" do
-    id1 = "/basic/one"
-    id2 = "/basic/two"
-    x = 5
-    :kvs.save(:kvs.writer(id1))
-    :kvs.save(:kvs.writer(id2))
-
-    :lists.map(
-      fn _ ->
-        :kvs.save(:kvs.add(KVS.writer(:kvs.writer(id1), args: {:"$msg", [], [], [], [], []})))
-      end,
-      :lists.seq(1, x)
-    )
-
-    :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, id2) end, :lists.seq(1, x))
-    r1 = :kvs.save(:kvs.reader(id1))
-    r2 = :kvs.save(:kvs.reader(id2))
-    x1 = :kvs.take(KVS.reader(:kvs.load_reader(KVS.reader(r1, :id)), args: 20))
-    x2 = :kvs.take(KVS.reader(:kvs.load_reader(KVS.reader(r2, :id)), args: 20))
-    b = :kvs.feed(id1)
-
-    case :application.get_env(:kvs, :dba_st, :kvs_st) do
-      :kvs_st ->
-        c = :kvs.all("/basic/two")
-        assert :lists.reverse(c) == KVS.reader(x2, :args)
-      _ ->
-        # mnesia doesn't support `all` over feeds (only for tables)
-        []
-    end
-
-    assert KVS.reader(x1, :args) == b
-
-    assert length(KVS.reader(x1, :args)) == length(KVS.reader(x2, :args))
-    assert x == length(b)
-  end
-
-  test "sym" do
-    id = {:sym, :kvs.seq([], [])}
-    :kvs.save(:kvs.writer(id))
-    x = 5
-
-    :lists.map(
-      fn
-        z ->
-          :kvs.remove(KVS.writer(z, :cache), id)
-      end, :lists.map(
-        fn _ ->
-          :kvs.save(:kvs.add(KVS.writer(:kvs.writer(id), args: {:"$msg", [], [], [], [], []})))
-        end,
-        :lists.seq(1, x)
-      )
-    )
-
-    {:ok, KVS.writer(count: 0)} = :kvs.get(:writer, id)
-  end
-
-  test "take" do
-    feed = :partial
-    x = 5
-    :kvs.save(:kvs.writer(feed))
-    :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, feed) end, :lists.seq(1, x))
-    KVS.reader(id: rid) = :kvs.save(:kvs.reader(feed))
-    t = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 20))
-    b = :kvs.feed(feed)
-    #: mnesia
-    assert KVS.reader(t, :args) == b
-  end
-
-  test "take back full" do
-    log(:st, "take back full")
-    feed = :partial
-    x = 5
-    :kvs.save(:kvs.writer(feed))
-    :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, feed) end, :lists.seq(1, x))
-    KVS.reader(id: rid) = :kvs.save(:kvs.reader(feed))
-    t = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 5))
-    :kvs.save(KVS.reader(t, dir: 1))
-    log("t:", t)
-    n = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 5))
-    b = :kvs.feed(feed)
-    log("n:", n)
-    assert KVS.reader(n, :args) == KVS.reader(t, :args)
-    assert KVS.reader(t, :args) == b
-    log(:end, "take back full")
-  end
-
-  test "partial take back" do
-    feed = :partial
-    x = 3
-    p = 2
-    :kvs.save(:kvs.writer(feed))
-    :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, feed) end, :lists.seq(1, x))
-    KVS.reader(id: rid) = :kvs.save(:kvs.reader(feed))
-    t = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    :kvs.save(KVS.reader(t, dir: 1))
-    n = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p + 1))
-    assert KVS.reader(t, :args) == tl(KVS.reader(n, :args))
-  end
-
-  test "partial full bidirectional" do
-    log(:st, "partial full bidirectional")
-    feed = :partial
-    x = 5
-    p =2
-    :kvs.save(:kvs.writer(feed))
-    :lists.map(fn _ -> :kvs.append({:"$msg", :kvs.seq([],[]), [], [], [], []}, feed) end, :lists.seq(1, x))
-    r = :kvs.save(:kvs.reader(feed))
-    rid = KVS.reader(r, :id)
-    t1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
-    z1 = KVS.reader(t1, :args)
-    IO.inspect :kvs.all(feed)
-    r = :kvs.save(t1)
-    log("next t1:", t1)
-
-    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z2 = KVS.reader(t2, :args)
-    r = :kvs.save(t2)
-    log("next t2:", t2)
-
-    t3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z3 = KVS.reader(t3, :args)
-    :kvs.save(KVS.reader(t3, dir: 1, pos: 0))
-    log("next t3:", t3)
-
-    n1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    nz1 = KVS.reader(n1, :args)
-    :kvs.save n1
-    log("prev n1:", n1)
-
-    n2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    nz2 = KVS.reader(n2, :args)
-    :kvs.save n2
-    log("prev n2:", n2)
-
-    n3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    nz3 = KVS.reader(n3, :args)
-    log("prev n3:", n3)
-
-    assert z3 ++ z2 ++ z1 == nz1 ++ nz2 ++ nz3
-    log(:end, "partial full bidirectional")
-  end
-
-  test "test bidirectional (new)" do
-    log(:st, "test bidirectional (new)")
-    feed = :partial
-    x = 6
-    p = 3
-    :kvs.save(:kvs.writer(feed))
-    :lists.map(fn _ -> :kvs.append({:"$msg", :kvs.seq([],[]), [], [], [], []}, feed) end, :lists.seq(1, x))
-    r = :kvs.save(:kvs.reader(feed))
-    rid = KVS.reader(r, :id)
-    IO.inspect :kvs.all(feed)
-
-    t1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
-    z1 = KVS.reader(t1, :args)
-    r = :kvs.save(t1)
-    log("next t1:", t1)
-
-    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z2 = KVS.reader(t2, :args)
-    r = :kvs.save(t2)
-    log("next t2:", t2)
-
-    t3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z3 = KVS.reader(t3, :args)
-    :kvs.save(KVS.reader(t3, dir: 1, pos: 0))
-    log("next t3:", t3)
-
-    assert z3 == []
-
-    n1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    nz1 = KVS.reader(n1, :args)
-    :kvs.save n1
-    log("prev n1:", n1)
-
-    n2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    nz2 = KVS.reader(n2, :args)
-    :kvs.save n2
-    log("prev n2:", n2)
-
-    n3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    nz3 = KVS.reader(n3, :args)
-    :kvs.save(KVS.reader(n3, dir: 0))
-    log("prev n3:", n3)
-
-    assert nz3 == []
-
-    t4 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
-    z4 = KVS.reader(t4, :args)
-    r = :kvs.save(t4)
-    log("next t4:", t4)
-
-    assert length(z4) == p
-    log(:end, "test bidirectional (new)")
-  end
-
-  test "partial take forward full" do
-    log(:st, "partial take forward full")
-    feed = :partial
-    x = 7
-    :kvs.save(:kvs.writer(feed))
-    :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, feed) end, :lists.seq(1, x))
-    KVS.reader(id: rid) = :kvs.save(:kvs.reader(feed))
-    p = 3
-    IO.inspect :kvs.all(feed)
-
-    t1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z1 = KVS.reader(t1, :args)
-    :kvs.save(t1)
-    log("next t1:", t1)
-
-    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z2 = KVS.reader(t2, :args)
-    :kvs.save(t2)
-    log("next t2:", t2)
-
-    t3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z3 = KVS.reader(t3, :args)
-    :kvs.save(t3)
-    log("next t3:", t3)
-
-    assert length(z3) == 1
-    assert :lists.reverse(z1) ++ :lists.reverse(z2) ++ z3 == :kvs.all(:partial)
-    log(:end, "partial take forward full")
-  end
-
-  test "take with empy" do
-    log(:st, "take with empy")
-    feed = :partial
-    x = 6
-    p = 3
-    :kvs.save(:kvs.writer(feed))
-    :lists.map(fn _ -> :kvs.append({:"$msg", :kvs.seq([],[]), [], [], [], []}, feed) end, :lists.seq(1, x))
-    r = :kvs.save(:kvs.reader(feed))
-    IO.inspect :kvs.all(feed)
-    rid = KVS.reader(r, :id)
-    t1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
-    z1 = KVS.reader(t1, :args)
-    r = :kvs.save(t1)
-    log("next t1:", t1)
-
-    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z2 = KVS.reader(t2, :args)
-    r = :kvs.save(t2)
-    log("next t2:", t2)
-
-    t3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z3 = KVS.reader(t3, :args)
-    r = :kvs.save(t3)
-    log("next t3:", t3)
-    assert  z3 == []
-
-    KVS.reader(id: tid) = :kvs.save(KVS.reader(t3, dir: 1, pos: 0))
-    n1 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p))
-    nz1 = KVS.reader(n1, :args)
-    :kvs.save n1
-    log("prev b1:", n1)
-
-    n2 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p))
-    nz2 = KVS.reader(n2, :args)
-    :kvs.save n2
-    log("prev b2:", n2)
-
-    assert z2 ++ z1 == nz1 ++ nz2
-    log(:end, "take with empy")
-  end
-
-  test "test prev" do
-    log(:st, "test prev")
-    feed = :partial
-    x = 6
-    p = 3
-    :kvs.save(:kvs.writer(feed))
-    :lists.map(fn _ -> :kvs.append({:"$msg", :kvs.seq([],[]), [], [], [], []}, feed) end, :lists.seq(1, x))
-    r = :kvs.save(:kvs.reader(feed))
-    rid = KVS.reader(r, :id)
-    IO.inspect :kvs.all(feed)
-
-    t1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
-    z1 = KVS.reader(t1, :args)
-    r = :kvs.save(t1)
-    log("next z1:", z1)
-
-    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    z2 = KVS.reader(t2, :args)
-
-    KVS.reader(id: tid) = :kvs.save(KVS.reader(t2, dir: 1, pos: 0))
-    log("next z2:", z2)
-
-    n1 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p))
-    nz1 = tl(:lists.reverse(KVS.reader(n1, :args)))
-    :kvs.save(n1)
-    log("prev nz1:", nz1)
-
-    n2 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p))
-    nz2 = KVS.reader(n2, :args)
-    :kvs.save n2
-    log("prev n2:", n2)
-
-    assert length(nz2) == p
-    assert nz2 == z1
-
-    n3 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p))
-    nz3 = KVS.reader(n3, :args)
-    :kvs.save(KVS.reader(n3, dir: 0))
-    log("prev nz3:", nz3)
-
-    assert nz3 = []
-
-    log(:end, "test prev")
-
-  end
-
-  def log(x,cursor) do
-     IO.inspect {x,cursor}
-  end
-end
-

test/sc_test.exs

@@ -0,0 +1,58 @@
+ExUnit.start()
+
+defmodule Sc.Test do
+    use ExUnit.Case, async: false
+    require KVS
+    import Record
+    @moduledoc """
+        refined old scenarios
+    """
+
+    defrecord(:msg, id: [], body: [])
+
+    setup do: (on_exit(fn -> :ok = :kvs.leave();:ok = :kvs.destroy() end);:kvs.join())
+    setup kvs, do: [
+        id0: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/duck") end, :lists.seq(1,10)),
+        id1: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/luck") end, :lists.seq(1,10)),
+        id2: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/truck") end, :lists.seq(1,10)),
+        id3: :lists.map(fn _ -> :kvs.save(:kvs.add(KVS.writer(:kvs.writer(:sym),
+                                        args: msg(id: :kvs.seq([],[]))))) end, :lists.seq(1,10))]
+    test "basic", kvs do
+        KVS.reader(id: rid1) = :kvs.save(:kvs.reader("/crm/luck"))
+        KVS.reader(id: rid2) = :kvs.save(:kvs.reader("/crm/truck"))
+        x1 = :kvs.take(KVS.reader(:kvs.load_reader(rid1), args: 20))
+        x2 = :kvs.take(KVS.reader(:kvs.load_reader(rid2), args: 20))
+        b = :kvs.feed("/crm/luck")
+        assert 10 == length(b)
+        assert :kvs.all("/crm/truck") == KVS.reader(x2, :args)
+        assert KVS.reader(x1, :args) == b
+        assert length(KVS.reader(x1, :args)) == length(KVS.reader(x2, :args))
+    end
+
+    test "sym",kvs do
+        KVS.writer(args: last) = Enum.at(kvs[:id3],-1)
+        {:ok, KVS.writer(id: :sym, count: 10, cache: last)} = :kvs.get(:writer, :sym)
+    end
+
+    test "take back full" do
+        feed = "/crm/duck"
+        KVS.reader(id: rid) = :kvs.save(:kvs.reader(feed))
+        t = KVS.reader(args: a1) = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 10))
+        assert a1 == :kvs.feed(feed)
+        :kvs.save(KVS.reader(t, dir: 1))
+        KVS.reader(args: a2) = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 10))
+        assert :lists.reverse(a2) == :kvs.feed(feed)
+    end
+
+    test "partial take back" do
+        log "all",:kvs.all("/crm/luck")
+        KVS.reader(id: rid) = :kvs.save(:kvs.reader("/crm/luck"))
+        r = KVS.reader(args: t) = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 2))
+        :kvs.save(KVS.reader(r, dir: 1))
+        KVS.reader(args: n) = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 3))
+        assert :lists.reverse(t) == tl(n)
+    end
+
+    defp log(x), do: IO.puts '#{inspect(x)}'
+    defp log(m, x), do: IO.puts '#{m} #{inspect(x)}'
+end

test/st_test.exs

@@ -1,11 +1,12 @@
 ExUnit.start()
 
-defmodule ST.Test do
+defmodule St.Test do
     use ExUnit.Case, async: false
     import Record
     require KVS
 
     defrecord(:msg, id: [], body: [])
+
     setup do: (on_exit(fn -> :ok = :kvs.leave();:ok = :kvs_rocks.destroy() end);:kvs.join())
     setup kvs, do: [
         ids: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), :feed) end, :lists.seq(1,10)),
@@ -13,27 +14,6 @@ defmodule ST.Test do
         id1: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/personal/Реєстратор А1/in/mail") end, :lists.seq(1,10)),
         id2: :lists.map(fn _ -> :kvs.append(msg(id: :kvs.seq([],[])), "/crm/personal/Реєстратор А1/in/doc") end, :lists.seq(1,10))]
 
-    test "al0", kvs, do: assert kvs[:ids] |> Enum.map(&msg(id: &1)) == :kvs.all(:feed)
-    test "al1", kvs, do: assert (kvs[:id0] ++ kvs[:id2] ++ kvs[:id1]) |> Enum.map(&msg(id: &1)) == :kvs.all("/crm/personal/Реєстратор А1/in")
-    
-    #: old behaviour is reversed ? 
-    test "fe0", kvs, do: assert kvs[:ids] |> Enum.reverse |> Enum.map(&msg(id: &1)) == :kvs.feed(:feed)
-
-    #: real cache {:feed, :msg, id}
-    test "top",  kvs, do: (r0=:kvs.reader(:feed); assert KVS.reader(r0, cache: {:msg, Enum.at(kvs[:ids],0), "/feed"}, dir: 0) == :kvs.top(r0))
-    test "bot",  kvs, do: (r0=:kvs.reader(:feed); assert KVS.reader(r0, cache: {:msg, Enum.at(kvs[:ids],9), "/feed"}, dir: 1) == :kvs.bot(r0))
-    
-    test "next", kvs do
-        KVS.reader(id: rid) = :kvs.save(:kvs.top(:kvs.reader(:feed)))
-        kvs[:ids] |> Enum.each(&assert(KVS.reader(cache: {:msg,&1,"/feed"}) = :kvs.next(:kvs.load_reader(rid))))
-    end
-
-    test "prev", kvs do
-        KVS.reader(id: rid) = :kvs.save(:kvs.bot(:kvs.reader(:feed)))
-        kvs[:ids] |> Enum.reverse |> Enum.each(&assert KVS.reader(cache: {:msg,&1,"/feed"}) = :kvs.prev(:kvs.load_reader(rid)))
-    end
-   
-
     test "take-ø", kvs do
         r = KVS.reader() = :kvs.reader("/empty-feed")
         assert r1 = KVS.reader(feed: "/empty-feed", args: []) = :kvs.take(KVS.reader(r, args: 1))
@@ -52,21 +32,90 @@ defmodule ST.Test do
         assert KVS.reader(feed: "/empty-feed", args: []) = :kvs.take(KVS.reader(rs1, args: 0, dir: 1))
     end
 
+    test "take-0", kvs do
+        feed = "/crm/personal/Реєстратор А1/in/doc"
+        assert r = KVS.reader(id: rid, args: []) = :kvs.reader(feed)
+
+        assert KVS.reader(id: ^rid, feed: ^feed, args: []) = :kvs.take(KVS.reader(r, args: 0, dir: 0))
+        assert KVS.reader(id: ^rid, feed: ^feed, args: []) = :kvs.take(KVS.reader(r, args: -1, dir: 0))
+        
+        assert r1 = KVS.reader(id: ^rid, feed: ^feed, args: a01) = :kvs.take(KVS.reader(r,  args: 10, dir: 0))
+        assert kvs[:id2] |> Enum.map(&msg(id: &1)) == a01
+        assert KVS.reader(id: ^rid, feed: ^feed, args: []) = :kvs.take(KVS.reader(r1, args: 10, dir: 0))
+
+        assert KVS.reader(id: ^rid, feed: ^feed, args: af) = :kvs.take(KVS.reader(r, args: 100, dir: 0))
+        assert kvs[:id2] |> Enum.map(&msg(id: &1)) == af
+
+        assert r2 = KVS.reader(id: ^rid, feed: ^feed, args: a03) = :kvs.take(KVS.reader(r,  args: 3, dir: 0))
+        assert Enum.take(kvs[:id2],3) |> Enum.map(&msg(id: &1)) == a03
+
+        assert KVS.reader(id: ^rid, feed: ^feed, args: a07) = :kvs.take(KVS.reader(r2, args: 7, dir: 0))
+        assert Enum.drop(kvs[:id2],3) |> Enum.map(&msg(id: &1)) == a07
+
+        assert KVS.reader(id: ^rid, feed: ^feed, args: a17) = :kvs.take(KVS.reader(r2, args: 100, dir: 0))
+        assert Enum.drop(kvs[:id2],3) |> Enum.map(&msg(id: &1)) == a17
+    end
+
+    test "take-1", kvs do
+        feed = "/crm/personal/Реєстратор А1/in/mail"
+        top = Enum.at(kvs[:id1],0)
+        bot = Enum.at(kvs[:id1],9)
+        tpm = Enum.take(kvs[:id1],1) |> Enum.map(&msg(id: &1))
+
+        assert r = KVS.reader(id: rid, args: [], cache: {:msg, ^top, ^feed}, count: 10) = :kvs.reader(feed)
+        assert r = :kvs.top(r)
+        assert KVS.reader(id: ^rid, feed: ^feed, args: ^tpm, dir: 1) = :kvs.take(KVS.reader(r, args: 1, dir: 1))
+        assert KVS.reader(id: ^rid, feed: ^feed, args: ^tpm, dir: 1) = :kvs.take(KVS.reader(r, args: 100, dir: 1))
+
+        assert r1 = KVS.reader(feed: ^feed, count: 10, args: [], cache: {:msg,^bot,^feed}) = :kvs.bot(r)
+        
+        assert r2 = KVS.reader(feed: ^feed, count: 10, args: a01) = :kvs.take(KVS.reader(r1, args: 5, dir: 1))
+        x01 = Enum.drop(kvs[:id1],5) |> Enum.map(&msg(id: &1)) |> Enum.reverse
+        assert x01 == a01
+
+        assert r3 = KVS.reader(feed: ^feed, count: 10, args: a02) = :kvs.take(KVS.reader(r2, args: 10, dir: 1))
+        x02 = Enum.take(kvs[:id1],5) |> Enum.map(&msg(id: &1)) |> Enum.reverse
+        assert x02 == a02
+
+        assert KVS.reader(feed: ^feed, count: 10, args: []) = :kvs.take(KVS.reader(r3, args: 20, dir: 1))
+    end
+
     test "drop", kvs do
-        assert r = KVS.reader(id: rid, args: [], cache: c0) = :kvs.save(:kvs.reader(:feed))
-        assert r1 = KVS.reader(id: ^rid, feed: :feed, args: []) = :kvs.drop(KVS.reader(r,  args: 10, dir: 0))
+        feed = "/feed"
+        assert r = KVS.reader(id: rid, args: []) = :kvs.save(:kvs.reader(:feed))
+        assert r1 = KVS.reader(id: ^rid, feed: ^feed, args: []) = :kvs.drop(KVS.reader(r,  args: 10, dir: 0))
 
         kvs[:ids] |> Enum.map(&msg(id: &1)) 
-                  |> Enum.each(&assert(KVS.reader(id: rid, feed: :feed, args: [], cache: &1) = :kvs.save(:kvs.drop(KVS.reader(:kvs.load_reader(rid), args: 1, dir: 0)))))
-
-        assert r2 = KVS.reader(id: ^rid, feed: :feed, args: [], cache: c1) = :kvs.drop(KVS.reader(r, args: 1, dir: 0))
-        assert {:msg, Enum.at(kvs[:ids], 1)} == c1
-        assert r3 = KVS.reader(id: ^rid, feed: :feed, args: [], cache: c2) = :kvs.drop(KVS.reader(r2, args: 5, dir: 0))
-        assert {:msg, Enum.at(kvs[:ids], 6)} == c2
-        assert r4 = KVS.reader(id: ^rid, feed: :feed, args: [], cache: c3) = :kvs.drop(KVS.reader(r1, args: 100))
-        assert {:msg, Enum.at(kvs[:ids],9)} == c3
+                  |> Enum.each(&assert(
+                    KVS.reader(id: rid, feed: ^feed, args: [], cache: &1) =
+                        :kvs.save(:kvs.drop(KVS.reader(:kvs.load_reader(rid), args: 1, dir: 0)))))
+
+        assert r2 = KVS.reader(id: ^rid, feed: ^feed, args: [], cache: c1) = :kvs.drop(KVS.reader(r, args: 1, dir: 0))
+        assert {:msg, Enum.at(kvs[:ids], 1), feed} == c1
+
+        assert KVS.reader(id: ^rid, feed: ^feed, args: [], cache: c2) = :kvs.drop(KVS.reader(r2, args: 5, dir: 0))
+        assert {:msg, Enum.at(kvs[:ids], 6), feed} == c2
+        assert KVS.reader(id: ^rid, feed: ^feed, args: []) = :kvs.drop(KVS.reader(r1, args: 100))
     end
 
+    test "feed",kvs do
+        docs  = :kvs.all("/crm/personal/Реєстратор А1/in/doc")
+        ducks = :kvs.all("/crm/personal/Реєстратор А1/in/directory/duck")
+        mail  = :kvs.all("/crm/personal/Реєстратор А1/in/mail")
+        total = :kvs.all("/crm/personal/Реєстратор А1/in")
+
+        assert ducks++docs++mail == total
+
+        assert docs  = :kvs.feed("/crm/personal/Реєстратор А1/in/doc")
+        assert ducks = :kvs.feed("/crm/personal/Реєстратор А1/in/directory/duck")
+        assert mail  = :kvs.feed("/crm/personal/Реєстратор А1/in/mail")
+        assert total = :kvs.feed("/crm/personal/Реєстратор А1/in")
+
+        assert :kvs.feed("/crm/personal/Реєстратор А1/in/directory/duck") 
+            ++ :kvs.feed("/crm/personal/Реєстратор А1/in/doc")
+            ++ :kvs.feed("/crm/personal/Реєстратор А1/in/mail") == :kvs.feed("/crm/personal/Реєстратор А1/in")
+    end 
+
     defp log(x), do: IO.puts '#{inspect(x)}'
     defp log(m, x), do: IO.puts '#{m} #{inspect(x)}'