kvs/src/stores/kvs_st.erl

-module(kvs_st).
-description('KVS STREAM NATIVE ROCKS').
-include("kvs.hrl").
-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]).

% section: kvs_stream prelude

se(X,Y,Z) -> setelement(X,Y,Z).
e(X,Y) -> element(X,Y).
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.

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,_) -> 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

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. Если курсор установлен в конец списка и уже вернул его последний элемент
%     то результат вычитки будет равным пустому списку


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

load_reader(Id) ->
    case kvs:get(reader,Id) of
         {ok,#reader{}=C} -> C;
              _ -> #reader{id=[]} end.

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)}};
  {error,_} -> save(#writer{id=Id}), reader(Id) end.
save(C) -> NC = c4(C,[]), kvs:put(NC), NC.

% add

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}.

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
   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}]).

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;
        {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.