6.11.1

.formatter.exs

@@ -1,5 +1,5 @@
 # Used by "mix format"
 [
   inputs: ["{mix,.formatter}.exs", "{config,lib,test}/**/*.{ex,exs}"],
-  line_length: 133
+  line_length: 100
 ]

lib/KVS.ex

@@ -32,5 +32,4 @@ defmodule KVS do
       end
     end)
   end
-
 end

mix.exs

@@ -2,7 +2,13 @@ defmodule KVS.Mixfile do
   use Mix.Project
 
   def project do
-    [app: :kvs, version: "6.11.0", description: "KVS Abstract Chain Database", package: package(), deps: deps()]
+    [
+      app: :kvs,
+      version: "6.11.1",
+      description: "KVS Abstract Chain Database",
+      package: package(),
+      deps: deps()
+    ]
   end
 
   def application do
@@ -19,8 +25,8 @@ defmodule KVS.Mixfile do
 
   defp deps do
     [
-     {:ex_doc, "~> 0.11", only: :dev},
-    # {:rocksdb, "~> 1.3.2"}
+      {:ex_doc, "~> 0.11", only: :dev},
+#      {:rocksdb, "~> 1.3.2"}
     ]
   end
 end

src/kvs.app.src

@@ -1,6 +1,6 @@
 {application, kvs,
    [{description, "KVS Abstract Chain Database"},
-    {vsn, "6.11.0"},
+    {vsn, "6.11.1"},
     {registered, []},
     {applications, [kernel,stdlib]},
     {mod, { kvs, []}},

src/kvs.erl

@@ -9,7 +9,7 @@
 -include("cursors.hrl").
 -include("kvs.hrl").
 -include("backend.hrl").
--export([dump/0,check/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,fetch/2,fetch/3,feed/2]).
 -export(?API).
 -export(?STREAM).
 -compile(export_all).
@@ -110,7 +110,6 @@ seq_gen() ->
                 {ok, _} -> {Key,skip} end end,
     [ Init(atom_to_list(Name))  || {Name,_Fields} <- cursors() ].
 
-
 put(Records,#kvs{mod=Mod}) when is_list(Records) -> Mod:put(Records);
 put(Record,#kvs{mod=Mod}) -> Mod:put(Record).
 get(RecordName, Key, #kvs{mod=Mod}) -> Mod:get(RecordName, Key).
@@ -120,33 +119,10 @@ index(Tab, Key, Value,#kvs{mod=DBA}) -> DBA:index(Tab, Key, Value).
 seq(Tab, Incr,#kvs{mod=DBA}) -> DBA:seq(Tab, Incr).
 dump(#kvs{mod=Mod}) -> Mod:dump().
 feed(Key,#kvs{st=Mod}=KVS) -> (Mod:take((kvs:reader(Key))#reader{args=-1}))#reader.args.
-
 remove(Rec,Feed) -> remove(Rec,Feed,#kvs{mod=dba(),st=kvs_stream()}).
-
 remove(Rec,Feed, #kvs{st=Mod}=KVS) -> 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.
-
-% tests
-
-check() ->
-    Id1 = {list1,kvs:seq([],[])},
-    Id2 = {list2,kvs:seq([],[])},
-    X   = 5,
-    W1   = kvs:save(kvs:writer(Id1)),
-    W2   = kvs:save(kvs:writer(Id2)),
-    [ kvs:save(kvs:add((kvs:writer(Id1))#writer{args={'$msg',[],[],[],[],[]}})) || _ <- lists:seq(1,X) ],
-    [ kvs:append({'$msg',[],[],[],[],[]},Id2) || _ <- lists:seq(1,X) ],
-    R1  = kvs:save(kvs:reader(Id1)),
-    R2  = kvs:save(kvs:reader(Id2)),
-    R = kvs:take((kvs:load_reader(R2#reader.id))#reader{args=20}),
-    B = kvs:feed(Id1),
-    C = kvs:feed(Id2),
-    ?assertMatch(5,length(R#reader.args)),
- %   ?assertMatch(X,length(B)),
- ok.
-
 fetch(Table, Key) -> fetch(Table, Key, []).
 fetch(Table, Key, Default) -> case get(Table, Key) of
                                         {ok, Value} -> Value;

src/layers/kvs_stream.erl

@@ -54,17 +54,15 @@ w({error,X},_,_)                          -> {error,X}.
 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).
 take(#reader{cache=[]}=C) -> C#reader{args=[]};
-take(#reader{dir=D,cache=B,args=N,pos=P}=C)  -> take(acc(D),N,C,C,[],P,B).
+take(#reader{dir=D,cache=B,args=N,pos=P}=C)  -> take(acc(D),N,C,C,[],P).
 
-take(_,_,{error,_},C2,R,P,B) -> C2#reader{args=lists:flatten(R),pos=P,cache=B};
-take(_,0,_,C2,R,P,B)         -> C2#reader{args=lists:flatten(R),pos=P,cache=B};
-take(A,N,#reader{cache={T,I},pos=P}=C,C2,R,_,_) ->
-    take(A,N-1,?MODULE:A(C),C2,[element(2,kvs:get(T,I))|R],P,{T,I}).
+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,_) -> take(A,N-1,?MODULE:A(C),C2,[element(2,kvs:get(T,I))|R],P).
 
 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,_,_) ->
-    drop(A,N-1,?MODULE:A(C),C2,P,B).
+drop(A,N,#reader{cache=B,pos=P}=C,C2,_,_) -> drop(A,N-1,?MODULE:A(C),C2,P,B).
 
 % new, save, load, up, down, top, bot
 

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,format/1]).
+-export([ref/0,next/8,format/1,bt/1]).
 
+bt(X)      -> binary_to_term(X,[safe]).
 start()    -> ok.
 stop()     -> ok.
 destroy()  -> ok.
@@ -20,11 +21,12 @@ initialize() -> [ kvs:initialize(kvs_rocks,Module) || Module <- kvs:modules() ].
 ref() -> application:get_env(kvs,rocks_ref,[]).
 index(_,_,_) -> [].
 get(Tab, Key) ->
-    Address = <<(list_to_binary(lists:concat(["/",format(Tab),"/"])))/binary,(term_to_binary(Key))/binary>>,
+    Address = <<(list_to_binary(lists:concat(["/",format(Tab),"/"])))/binary,
+                (term_to_binary(Key))/binary>>,
 %    io:format("KVS.GET.Address: ~s~n",[Address]),
     case rocksdb:get(ref(), Address, []) of
          not_found -> {error,not_found};
-         {ok,Bin} -> {ok,binary_to_term(Bin,[safe])} end.
+         {ok,Bin} -> {ok,bt(Bin)} end.
 
 put(Records) when is_list(Records) -> lists:map(fun(Record) -> put(Record) end, Records);
 put(Record) -> 
@@ -53,8 +55,7 @@ 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 ->
      case binary:part(A,0,S) of Key ->
-          next(I,Key,S,rocksdb:iterator_move(I, next), [],
-                       [binary_to_term(X,[safe])|T],N,C+1);
+          next(I,Key,S,rocksdb:iterator_move(I, next), [], [bt(X)|T],N,C+1);
                   _ -> T end;
 next(_,_,_,_,_,T,_,_) -> T.
 

src/stores/kvs_st.erl

@@ -6,6 +6,7 @@
 -export(?STREAM).
 -export([prev/8,ref/0,feed_key/2]).
 
+bt(X) -> kvs_rocks:bt(X).
 ref() -> kvs_rocks:ref().
 
 % section: kvs_stream prelude
@@ -26,11 +27,11 @@ next (#reader{feed=Feed,cache=I}=C) when is_tuple(I) ->
    Key = feed_key(I,Feed),
    rocksdb:iterator_move(I, {seek,Key}),
    case rocksdb:iterator_move(I, next) of
-        {ok,_,Bin} -> C#reader{cache=binary_to_term(Bin,[safe])};
+        {ok,_,Bin} -> C#reader{cache=bt(Bin)};
             {error,Reason} -> {error,Reason} end;
 next (#reader{cache=I}=C) when is_reference(I) ->
    case rocksdb:iterator_move(I, next) of
-        {ok,_,Bin} -> C#reader{cache=binary_to_term(Bin,[safe])};
+        {ok,_,Bin} -> C#reader{cache=bt(Bin)};
             {error,Reason} -> {error,Reason} end.
 
 prev (#reader{cache=[]}) -> {error,empty};
@@ -38,11 +39,11 @@ prev (#reader{cache=I,id=Feed}=C) when is_tuple(I) ->
    Key = feed_key(I,Feed),
    rocksdb:iterator_move(I, {seek,Key}),
    case rocksdb:iterator_move(I, prev) of
-        {ok,_,Bin} -> C#reader{cache=binary_to_term(Bin,[safe])};
+        {ok,_,Bin} -> C#reader{cache=bt(Bin)};
             {error,Reason} -> {error,Reason} end;
 prev (#reader{cache=I}=C) when is_reference(I) ->
    case rocksdb:iterator_move(I, prev) of
-        {ok,_,Bin} -> C#reader{cache=binary_to_term(Bin,[safe])};
+        {ok,_,Bin} -> C#reader{cache=bt(Bin)};
             {error,Reason} -> {error,Reason} end.
 
 % section: take, drop
@@ -52,7 +53,7 @@ drop(#reader{args=N}) when N < 0 -> #reader{};
 drop(#reader{args=N,feed=Feed,cache=I}=C) when N == 0 ->
    Key = list_to_binary(lists:concat(["/",kvs_rocks:format(Feed)])),
    case rocksdb:iterator_move(I, {seek,Key}) of
-        {ok,_,Bin} -> C#reader{cache=binary_to_term(Bin,[safe])};
+        {ok,_,Bin} -> C#reader{cache=bt(Bin)};
                  _ -> C#reader{cache=[]} end;
 
 drop(#reader{args=N,feed=Feed,cache=I}=C) when N > 0 ->
@@ -68,30 +69,15 @@ drop(#reader{args=N,feed=Feed,cache=I}=C) when N > 0 ->
      end,
            {First,{<<131,106>>,1}},
            lists:seq(0,N)),
-   C#reader{cache=binary_to_term(element(1,element(2,Term)))}.
+   C#reader{cache=bt(element(1,element(2,Term)))}.
 
 take(#reader{args=N,feed=Feed,cache={T,O}}=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)}),
-   Fir = binary_to_term(BERT),
-   Res = kvs_rocks:next(I,Key,size(Key),K,BERT,[],N+1,0),
-   io:format("Fir: ~p~n",[Fir]),
-   io:format("Res: ~p~n",[Res]),
-   case {Res,length(Res) < N + 1} of
-        {[],_}    -> C#reader{args=[],cache=I};
-        {[H|X],false} -> C#reader{args=X,cache={e(1,H),e(2,H)}};
-        {[H|X],true} -> C#reader{args=Res,cache=[]} end;
-
-take(#reader{args=N,feed=Feed,cache=I}=C) when is_reference(I) ->
-   Key = list_to_binary(lists:concat(["/",kvs_rocks:format(Feed)])),
-   {ok,K,BERT} = rocksdb:iterator_move(I, {seek,Key}),
-   Fir = binary_to_term(BERT),
-   Res = kvs_rocks:next(I,Key,size(Key),K,BERT,[],N+1,0),
-   io:format("Fir: ~p~n",[Fir]),
-   io:format("Res: ~p~n",[Res]),
-   case {Res,length(Res) < N + 1} of
-        {[],_}    -> C#reader{args=[],cache=I};
+   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=[]} end.
 
@@ -99,7 +85,7 @@ take(#reader{args=N,feed=Feed,cache=I}=C) when is_reference(I) ->
 
 load_reader (Id) ->
     case kvs:get(reader,Id) of
-         {ok,#reader{}=C} -> C#reader{cache=element(2,rocksdb:iterator(ref(),[]))};
+         {ok,#reader{}=C} -> C;
               _ -> #reader{id=[]} end.
 
 writer (Id) -> case kvs:get(writer,Id) of {ok,W} -> W; {error,_} -> #writer{id=Id} end.
@@ -109,7 +95,7 @@ reader (Id) ->
              Key = list_to_binary(lists:concat(["/",kvs_rocks:format(Feed)])),
              {ok,I} = rocksdb:iterator(ref(), []),
              {ok,K,BERT} = rocksdb:iterator_move(I, {seek,Key}),
-             F = binary_to_term(BERT),
+             F = bt(BERT),
              #reader{id=kvs:seq([],[]),feed=Id,cache={e(1,F),e(2,F)}};
          {error,_} -> #reader{} end.
 save (C) -> NC = c4(C,[]), kvs:put(NC), NC.

test/test_helper.exs

@@ -1,10 +1,89 @@
-
-:kvs.join
+require KVS
+:kvs.join()
 ExUnit.start()
 
 defmodule BPE.Test do
   use ExUnit.Case, async: true
-  test "kvs" do
-    assert :kvs.check == :ok
+
+  test "basic" do
+    id1 = {:basic, :kvs.seq([], [])}
+    id2 = {:basic, :kvs.seq([], [])}
+    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(id2)
+        assert :lists.reverse(c) == KVS.reader(x2, :args)
+
+      _ ->
+        []
+    end
+
+    assert :lists.reverse(KVS.reader(x1, :args)) == :lists.reverse(b)
+    assert length(KVS.reader(x1, :args)) == length(KVS.reader(x2, :args))
+    assert x == length(b)
+  end
+
+  test "take" do
+    id = {:partial, :kvs.seq([], [])}
+    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))
+    b = :kvs.feed(id)
+    # mnesia
+    assert KVS.reader(t, :args) == b
+  end
+
+  test "partial take" do
+    id = {:partial, :kvs.seq([], [])}
+    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))
+    rid = KVS.reader(r, :id)
+    p = 2
+    IO.inspect(:kvs.feed(id))
+
+    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)
+    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)
+    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)
+    assert {:erlang.element(1, hd(a)), :erlang.element(2, hd(a))} == cache
+    IO.inspect({cache, t, a})
+    assert length(a) == 1
   end
 end