Merge pull request #55 from synrc/np

switch to o-move, prev/next

include/stream.hrl

@@ -11,7 +11,7 @@
 -spec prev(#reader{}) -> #reader{}.
 -spec drop(#reader{}) -> #reader{}.
 -spec take(#reader{}) -> #reader{}.
--spec feed (term())   -> #reader{}.
+-spec feed (term())   -> list().
 -spec load_reader (term()) -> #reader{}.
 -spec writer (term()) -> #writer{}.
 -spec reader (term()) -> #reader{}.

src/layers/kvs_stream.erl

@@ -50,7 +50,7 @@ w({ok,#writer{cache=B,count=Size}},top,C) -> C#reader{cache={tab(B),id(B)},pos=S
 w({error,X},_,_)                          -> {error,X}.
 
 % section: take, drop
-feed(Feed) -> take((reader(Feed))#reader{args=-1}).
+feed(Feed) -> #reader{args=Args} = take((reader(Feed))#reader{args=-1}), Args.
 
 drop(#reader{cache=[]}=C) -> C#reader{args=[]};
 drop(#reader{dir=D,cache=B,args=N,pos=P}=C)  -> drop(acc(D),N,C,C,P,B).

src/stores/kvs_rocks.erl

@@ -5,7 +5,7 @@
 -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]).
--export([seek_it/1]).
+-export([seek_it/1, move_it/3]).
 
 e(X,Y)     -> element(X,Y).
 bt([])     -> [];
@@ -62,6 +62,7 @@ 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]).
 
 get(Tab, Key) ->
     case rocksdb:get(ref(), key(Tab,Key), []) of

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, seek_it/1]).
+-import(kvs_rocks, [key/2, key/1, bt/1, ref/0, seek_it/1, move_it/3]).
 
 % section: kvs_stream prelude
 
@@ -15,38 +15,17 @@ si(M,T) -> se(#it.id, M, T).
 id(T) -> e(#it.id, T).
 
 % section: next, prev
-feed(Feed) -> take((reader(Feed))#reader{args=-1}).
+feed(Feed) -> #reader{args=Args} = take((reader(Feed))#reader{args=-1}), Args.
 
 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}.
 
-% handle -> seek -> move
-move_it(Key,Dir) -> 
-  Seek = fun(F,{ok,H})            -> {F(H,{seek,Key}),H};
-            (F,{{ok,_,_},H})      -> F(H,Dir);
-            (F,{{ok,_}, H})       -> F(H,Dir);
-            (_,{error,Error})     -> {error,Error};
-            (_,{{error,Error},_}) -> {error,Error};
-            (F,{R,O})             -> F(R,O) end,
-
-  case lists:foldl(Seek, {ref(),[]},
-    [fun rocksdb:iterator/2, fun rocksdb:iterator_move/2, fun rocksdb:iterator_move/2]) of
-    {ok,_,Bin} -> {ok,bt(Bin)};
-    {error, Error} -> {error,Error}
-  end.
-
 % iterator -> specific feed reader
-read_it(C,{ok,F,V,H}) -> C#reader{cache={e(1,V),id(V)}, args=lists:reverse(H)}; % real cache {F,e(1,V),id(V)}
+read_it(C,{ok,_F,V,H}) -> C#reader{cache={e(1,V),id(V)}, args=lists:reverse(H)}; % real cache {F,e(1,V),id(V)}
 read_it(C,_) -> C.
-read_it(C, Feed, Move) ->
-  case Move of 
-    {ok, Bin} when element(1,Bin) =:= Feed -> C#reader{cache=Bin};
-    {ok,_} -> C;
-    {error,_} -> C
-  end.
-
-next(#reader{feed=Feed,cache=I}=C) -> read_it(C,Feed,move_it(key(Feed,I),next)).
-prev(#reader{cache=I,feed=Feed}=C) -> read_it(C,Feed,move_it(key(Feed,I),prev)).
+
+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
 

test/old_test.exs

@@ -25,7 +25,7 @@ defmodule OLD.Test do
     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.reader(:kvs.feed(id1), :args)
+    b = :kvs.feed(id1)
 
     case :application.get_env(:kvs, :dba_st, :kvs_st) do
       :kvs_st ->
@@ -69,7 +69,7 @@ defmodule OLD.Test do
     :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.reader(:kvs.feed(feed), :args)
+    b = :kvs.feed(feed)
     #: mnesia
     assert KVS.reader(t, :args) == b
   end
@@ -85,7 +85,7 @@ defmodule OLD.Test do
     :kvs.save(KVS.reader(t, dir: 1))
     log("t:", t)
     n = :kvs.take(KVS.reader(:kvs.load_reader(rid), args: 5))
-    b = KVS.reader(:kvs.feed(feed), :args)
+    b = :kvs.feed(feed)
     log("n:", n)
     assert KVS.reader(n, :args) == KVS.reader(t, :args)
     assert KVS.reader(t, :args) == b

test/st_test.exs

@@ -15,11 +15,21 @@ defmodule ST.Test do
 
     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)}, dir: 0) == :kvs.top(r0))
     test "bot",  kvs, do: (r0=:kvs.reader(:feed); assert KVS.reader(r0, cache: {:msg, Enum.at(kvs[:ids],9)}, dir: 1) == :kvs.bot(r0))
     
+    test "next", kvs, do: kvs[:ids] |> Enum.each(&assert(KVS.reader(cache: {:msg,&1}) = :kvs.next(:kvs.reader(:feed))))
+
+    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}) = :kvs.prev(:kvs.load_reader(rid)))
+    end
+   
 
     test "take-ø", kvs do
         r = KVS.reader() = :kvs.reader("/empty-feed")