tests passed

src/stores/kvs_rocks.erl

@@ -4,8 +4,9 @@
 -include("metainfo.hrl").
 -include_lib("stdlib/include/qlc.hrl").
 -export(?BACKEND).
--export([ref/0,next/8,prev/8,format/1,bt/1]).
+-export([ref/0,next/8,prev/8,prev2/8,next2/8,format/1,bt/1]).
 
+bt([])     -> [];
 bt(X)      -> binary_to_term(X,[safe]).
 start()    -> ok.
 stop()     -> ok.
@@ -51,23 +52,28 @@ all(R) -> {ok,I} = rocksdb:iterator(ref(), []),
            First = rocksdb:iterator_move(I, {seek,Key}),
            lists:reverse(next(I,Key,size(Key),First,[],[],-1,0)).
 
-next(_,_,_,_,_,T,N,C) when C == N -> T;
-next(I,Key,S,{ok,A,X},_,T,N,C) -> next(I,Key,S,A,X,T,N,C);
-next(_,___,_,{error,_},_,T,_,_) -> T;
-next(I,Key,S,A,X,T,N,C) when size(A) > S ->
+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.
+
+next2(_,Key,_,_,X,T,N,C) when C == N -> {bt(X),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,_,_) -> {bt(X),T};
+next2(I,Key,S,A,X,T,N,C) when size(A) > S ->
      case binary:part(A, 0, S) of Key ->
-          next(I, Key, S, rocksdb:iterator_move(I, next), [], [bt(X)|T], N, C + 1);
-          _ -> T end;
-next(_,_,_,_,_,T,_,_) -> T.
+          next2(I, Key, S, rocksdb:iterator_move(I, next), [], [bt(X)|T], N, C + 1);
+          _ -> {hd(lists:reverse(T)),T} end;
+next2(_,Key,_,{ok,A,_},X,T,_,_) -> {bt(X),T};
+next2(_,Key,_,_,X,T,_,_) -> {bt(X),T}.
 
-prev(_,_,_,_,_,T,N,C) when C == N -> T;
-prev(I,Key,S,{ok,A,X},_,T,N,C) -> prev(I,Key,S,A,X,T,N,C);
-prev(_,___,_,{error,_},_,T,_,_) -> T;
-prev(I,Key,S,A,X,T,N,C) when size(A) > S ->
+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 ->
-          prev(I, Key, S, rocksdb:iterator_move(I, prev), [], [bt(X)|T], N, C + 1);
-          _ -> T end;
-prev(_,_,_,_,_,T,_,_) -> T.
+          prev2(I, Key, S, rocksdb:iterator_move(I, prev), [], [bt(X)|T], N, C + 1);
+          _ -> {hd(lists:reverse(T)),T} end;
+prev2(_,Key,_,{ok,A,_},X,T,_,_) -> {bt(X),T};
+prev2(_,Key,_,_,X,T,_,_) -> {bt(X),T}.
 
 seq(_,_) ->
   case os:type() of

src/stores/kvs_st.erl

@@ -63,26 +63,42 @@ drop(#reader{args=N,feed=Feed,cache=I}=C) when N > 0 ->
            lists:seq(0,N)),
    C#reader{cache=bt(element(1,element(2,Term)))}.
 
+%  1. Курсор всегда выставлен на следущий невычитанный элемент
+%  2. Если после вычитки курсор указывает на последний вычитаный элемент -- это признак конца списка
+%  3. Если результат вычитки меньше требуемого значения -- это признак конца списка
+%  4. Если курсор установлен в конец списка значит результат вычитки будет равным пустому списку
+
+take(#reader{pos='end',dir=0}=C) -> C#reader{args=[]};
 take(#reader{args=N,feed=Feed,cache={T,O},dir=0}=C) ->
    Key = list_to_binary(lists:concat(["/",kvs_rocks:format(Feed)])),
    {ok,I} = rocksdb:iterator(ref(), []),
    {ok,K,BERT} = rocksdb:iterator_move(I, {seek,feed_key({T,O},Feed)}),
-   Res = kvs_rocks:next(I,Key,size(Key),K,BERT,[],case N of -1 -> -1; J -> J + 1 end,0),
-   case {Res,length(Res) < N + 1 orelse N == -1} of
-        {[],_}    -> C#reader{args=[],cache=[]};
-        {[H|X],false} -> C#reader{args=X,cache={e(1,H),e(2,H)}};
-        {[H|X],true} -> C#reader{args=Res,cache={e(1,H),e(2,H)}} end;
-
+   {KK,Res} = kvs_rocks:next2(I,Key,size(Key),K,BERT,[],case N of -1 -> -1; J -> J + 1 end,0),
+   RevHd = hd(lists:reverse(Res)),
+   io:format("Res: ~p~n",[{KK,O,element(2,RevHd)}]),
+   case {KK,Res,length(Res)} of
+        {_,[],_}    -> C#reader{args=[],cache=[]};
+        {KK,[H|_],A} when element(2,KK) == O -> C#reader{args=Res,pos='end',cache={e(1,H),e(2,H)}};
+%        {_,[H|_],A} when element(2,H) == O -> C#reader{args=[],pos='end',cache={e(1,H),e(2,H)}};
+        {_,[H|X],A} when A < N + 1 orelse N == -1 -> C#reader{args=Res,cache={e(1,H),e(2,H)}};
+        {_,[H|X],A} when A == N -> C#reader{args=[bt(BERT)|X],cache={e(1,H),e(2,H)}};
+        {_,[H|X],_} -> C#reader{args=X,cache={e(1,H),e(2,H)}} end;
+
+take(#reader{pos='begin',dir=1}=C) -> C#reader{args=[]};
 take(#reader{args=N,feed=Feed,cache={T,O},dir=1}=C) ->
    Key = list_to_binary(lists:concat(["/",kvs_rocks:format(Feed)])),
    {ok,I} = rocksdb:iterator(ref(), []),
    {ok,K,BERT} = rocksdb:iterator_move(I, {seek,feed_key({T,O},Feed)}),
-   Res = kvs_rocks:prev(I,Key,size(Key),K,BERT,[],case N of -1 -> -1; J -> J + 1 end,0),
-   case {lists:reverse(Res),length(Res) < N -1 orelse N == -1, N == length(Res)} of
-        {[],_, _}    -> C#reader{args=[],cache=[]};
-        {[H|_], _, true} -> [HX|_] = Res, C#reader{args=Res,cache={e(1,HX),e(2,HX)}};
-        {[H|X],false, _} -> [HX|_] = Res, C#reader{args=X,cache={e(1,HX),e(2,HX)}};
-        {[H|X],true, _}  -> [HX|_] = Res, C#reader{args=Res,cache={e(1,HX),e(2,HX)}} end.
+   {KK,Res} = kvs_rocks:prev2(I,Key,size(Key),K,BERT,[],case N of -1 -> -1; J -> J + 1 end,0),
+   RevHd = hd(lists:reverse(Res)),
+   case {lists:reverse(Res),length(Res)} of
+        {[],_} -> C#reader{args=[],cache=[]};
+        {[H],A} when element(2,KK) == O -> C#reader{args=Res,pos='begin',cache={e(1,H),e(2,H)}};
+        {[H|X],A} when A < N - 1 orelse N == -1 -> [HX|_] = Res, C#reader{args=Res,cache={e(1,HX),e(2,HX)}};
+        {[H|X],A} when A == N -> [HX|TL] = Res, C#reader{args=[bt(BERT)|X],cache={e(1,HX),e(2,HX)}};
+        {[H|X],A} when A == N + 1 -> [HX|TL] = Res, C#reader{args=lists:reverse(TL),cache={e(1,HX),e(2,HX)}};
+        {[H|X],A} -> [HX|_] = Res, C#reader{args=X,cache={e(1,HX),e(2,HX)}}
+   end.
 
 % new, save, load, up, down, top, bot
 

test/test_helper.exs

@@ -66,8 +66,8 @@ defmodule BPE.Test do
     x = 5
     :kvs.save(:kvs.writer(id))
     :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, id) end, :lists.seq(1, x))
-    r = :kvs.save(:kvs.reader(id))
-    t = :kvs.take(KVS.reader(:kvs.load_reader(KVS.reader(r, :id)), args: 20))
+    KVS.reader(id: rid) = :kvs.save(:kvs.reader(id))
+    t = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 20))
     b = :kvs.feed(id)
     # mnesia
     assert KVS.reader(t, :args) == b
@@ -78,13 +78,17 @@ defmodule BPE.Test do
     x = 5
     :kvs.save(:kvs.writer(id))
     :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, id) end, :lists.seq(1, x))
-    r = :kvs.save(:kvs.reader(id))
-    t = :kvs.take(KVS.reader(:kvs.load_reader(KVS.reader(r, :id)), args: 5))
-    KVS.reader(id: tid) = :kvs.save(KVS.reader(t, dir: 1))
-    n = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: 5, dir: 1))
+    KVS.reader(id: rid) = :kvs.save(:kvs.reader(id))
+    t = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 5))
+    :kvs.save(KVS.reader(t, dir: 1))
+    IO.inspect "t:"
+    IO.inspect t
+    n = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 5))
     b = :kvs.feed(id)
+    IO.inspect "n:"
+    IO.inspect n
+    assert KVS.reader(n, :args) == KVS.reader(t, :args)
     assert KVS.reader(t, :args) == b
-    assert KVS.reader(n, :args) == :lists.reverse b
   end
 
   test "take back2" do
@@ -93,11 +97,11 @@ defmodule BPE.Test do
     p = 2
     :kvs.save(:kvs.writer(id))
     :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, id) end, :lists.seq(1, x))
-    r = :kvs.save(:kvs.reader(id))
-    t = :kvs.take(KVS.reader(:kvs.load_reader(KVS.reader(r, :id)), args: p))
-    KVS.reader(id: tid) = :kvs.save(KVS.reader(t, dir: 1))
-    n = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p, dir: 1))
-    assert KVS.reader(t, :args) == KVS.reader(n, :args)
+    KVS.reader(id: rid) = :kvs.save(:kvs.reader(id))
+    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, dir: 1))
+    assert KVS.reader(t, :args) == tl(KVS.reader(n, :args))
   end
 
   test "take back3" do
@@ -109,65 +113,117 @@ defmodule BPE.Test do
     r = :kvs.save(:kvs.reader(id))
     rid = KVS.reader(r, :id)
     t1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
-
-    z1 = :lists.reverse(KVS.reader(t1, :args))
+    z1 = KVS.reader(t1, :args)
+    IO.inspect :kvs.all(id)
     r = :kvs.save(t1)
+    IO.inspect "t1:"
+    IO.inspect t1
 
-    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
-    z2 = :lists.reverse(KVS.reader(t2, :args))
+    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
+    z2 = KVS.reader(t2, :args)
     r = :kvs.save(t2)
-
-    t3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
-    z3 = :lists.reverse(KVS.reader(t3, :args))
-
-    KVS.reader(id: tid) = :kvs.save(KVS.reader(t3, dir: 1))
-
-    n1 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p, dir: 1))
-    nz1 = :lists.reverse(KVS.reader(n1, :args))
-    KVS.reader(id: tid) = :kvs.save(KVS.reader(n1, dir: 1))
-
-    n2 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p, dir: 1))
-    nz2 = :lists.reverse(KVS.reader(n2, :args))
-    assert :lists.reverse(z1 ++ z2 ++ z3) == :lists.reverse(nz2 ++ nz1 ++ z3)
+    IO.inspect "t2:"
+    IO.inspect 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))
+    IO.inspect "t3:"
+    IO.inspect t3
+
+    n1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
+    nz1 = KVS.reader(n1, :args)
+    :kvs.save n1
+    IO.inspect "n1:"
+    IO.inspect n1
+
+    n2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
+    nz2 = KVS.reader(n2, :args)
+    :kvs.save n2
+    IO.inspect "n2:"
+    IO.inspect n2
+
+    n3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
+    nz3 = KVS.reader(n3, :args)
+    IO.inspect "n3:"
+    IO.inspect n3
+    assert z3 ++ z2 ++ z1 == nz1 ++ nz2 ++ nz3
   end
 
 
   test "partial take" do
     id = {:partial, :kvs.seq([], [])}
-    x = 5
+    x = 7
     :kvs.save(:kvs.writer(id))
     :lists.map(fn _ -> :kvs.append({:"$msg", [], [], [], [], []}, id) end, :lists.seq(1, x))
+    KVS.reader(id: rid) = :kvs.save(:kvs.reader(id))
+    p = 3
+    IO.inspect :kvs.all id
+
+    t1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
+    z1 = KVS.reader(t1, :args)
+    :kvs.save(t1)
+    IO.inspect "t1:"
+    IO.inspect t1
+
+    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
+    z2 = KVS.reader(t2, :args)
+    :kvs.save(t2)
+    IO.inspect "t2:"
+    IO.inspect t2
+
+    t3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
+    z3 = KVS.reader(t3, :args)
+    :kvs.save(t3)
+    IO.inspect "t3:"
+    IO.inspect t3
+    assert length(z3) == 1
+
+    assert :lists.reverse(z1) ++ :lists.reverse(z2) ++ z3 == :kvs.all(id)
+  end
+
+  test "take back4" do
+    id = {:partial, :kvs.seq([], [])}
+    x = 6
+    p = 3
+    :kvs.save(:kvs.writer(id))
+    :lists.map(fn _ -> :kvs.append({:"$msg", :kvs.seq([],[]), [], [], [], []}, id) end, :lists.seq(1, x))
     r = :kvs.save(:kvs.reader(id))
+    IO.inspect :kvs.all(id)
     rid = KVS.reader(r, :id)
-    p = 2
+    t1 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
+    z1 = KVS.reader(t1, :args)
+    r = :kvs.save(t1)
+    IO.inspect "t1:"
+    IO.inspect t1
 
-    cache = KVS.reader(r, :cache)
-    t = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    a = :lists.reverse(KVS.reader(t, :args))
-    z1 = a
-    r = :kvs.save(t)
-    IO.inspect({cache, r, a})
-    assert {:erlang.element(1, hd(a)), :erlang.element(2, hd(a))} == cache
-    assert length(a) == p
-
-    cache = KVS.reader(r, :cache)
-    t = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    a = :lists.reverse(KVS.reader(t, :args))
-    r = :kvs.save(t)
-    z2 = a
-    IO.inspect({cache, r, a})
-    assert {:erlang.element(1, hd(a)), :erlang.element(2, hd(a))} == cache
-    assert length(a) == p
-
-    cache = KVS.reader(r, :cache)
-    t = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p))
-    a = :lists.reverse(KVS.reader(t, :args))
-    r = :kvs.save(t)
-    z3 = a
-    assert {:erlang.element(1, hd(a)), :erlang.element(2, hd(a))} == cache
-    IO.inspect({cache, t, a})
-    assert length(a) == 1
-
-    assert :lists.reverse(z1 ++ z2 ++ z3) == :kvs.feed(id)
+    t2 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
+    z2 = KVS.reader(t2, :args)
+    r = :kvs.save(t2)
+    IO.inspect "t2:"
+    IO.inspect t2
+
+    t3 = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: p, dir: 0))
+    z3 = KVS.reader(t3, :args)
+    r = :kvs.save(t3)
+    IO.inspect "t3:"
+    IO.inspect t3
+    assert  z3 == []
+
+    KVS.reader(id: tid) = :kvs.save(KVS.reader(t3, dir: 1, pos: []))
+    n1 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p))
+    nz1 = KVS.reader(n1, :args)
+    KVS.reader(id: tid) = :kvs.save(KVS.reader(n1, dir: 1))
+    IO.inspect "b1:"
+    IO.inspect n1
+
+    n2 = :kvs.take(KVS.reader(:kvs.load_reader(tid), args: p))
+    nz2 = KVS.reader(n2, :args)
+    :kvs.save n2
+    IO.inspect "b2:"
+    IO.inspect n2
+
+    assert z2 ++ z1 == nz1 ++ nz2
   end
+
 end