Packages

kvs

7.1.3
13.5.22-aleph 13.4.16 13.4.15 13.4.14 13.4.13 13.3.1 13.2.28 11.9.1 10.8.3 10.8.2 10.3.0 9.9.2 9.9.1 9.9.0 9.8.0 9.7.0 9.4.8 9.4.7 9.4.6 9.4.5 9.4.4 9.4.3 9.4.2 9.4.1 9.4.0 8.12.0 8.11.2 8.11.1 8.10.4 8.10.3 8.10.2 8.10.1 8.10.0 8.5.2 8.5.1 8.5.0 8.4.1 8.4.0 8.3.1 8.3.0 7.11.5 7.9.1 7.7.0 7.1.3 7.1.2 7.1.1 6.12.11 6.12.10 6.12.9 6.12.8 6.12.7 6.12.6 6.12.5 6.12.4 6.12.3 6.12.2 6.12.1 6.12.0 6.11.2 6.11.1 6.11.0 6.10.2 6.10.1 6.10.0 6.9.2 6.9.1 6.9.0 6.7.7 6.7.6 6.7.5 6.7.4 6.7.3 6.7.2 6.7.1 6.7.0 6.6.0 2.1.0 0.12.1 retired

KVS Key-Value Store Abstraction Layer

HexDocs

Current section

Files

Jump to
Readme
78 Versions
0 Dependencies
Dependants
Files
Activity
kvs src stores kvs_rocks.erl
Raw

src/stores/kvs_rocks.erl 

-module(kvs_rocks).

-include("backend.hrl").

-include("kvs.hrl").

-include("metainfo.hrl").

-include_lib("stdlib/include/qlc.hrl").

-export(?BACKEND).

-export([ref/0,cut/8,next/8,prev/8,prev2/8,next2/8,format/1,bt/1]).



bt([])     -> [];

bt(X)      -> binary_to_term(X,[safe]).

start()    -> ok.

stop()     -> ok.

destroy()  -> ok.

version()  -> {version,"KVS ROCKSDB"}.

dir()      -> [].

leave() -> case ref() of [] -> skip; X -> rocksdb:close(X) end.

join(_) -> application:start(rocksdb),

           leave(), {ok, Ref} = rocksdb:open(application:get_env(kvs,rocks_name,"rocksdb"), [{create_if_missing, true}]),

           initialize(),

           application:set_env(kvs,rocks_ref,Ref).

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

%    io:format("KVS.GET.Address: ~s~n",[Address]),

    case rocksdb:get(ref(), Address, []) of

         not_found -> {error,not_found};

         {ok,Bin} -> {ok,bt(Bin)} end.



put(Records) when is_list(Records) -> lists:map(fun(Record) -> put(Record) end, Records);

put(Record) -> 

    Address = <<(list_to_binary(lists:concat(["/",format(element(1,Record)),"/"])))/binary,

                         (term_to_binary(element(2,Record)))/binary>>,

%    io:format("KVS.PUT.Address: ~s~n",[Address]),

    rocksdb:put(ref(), Address, term_to_binary(Record), [{sync,true}]).



format(X) when is_list(X) -> X;

format(X) when is_atom(X) -> atom_to_list(X);

format(X) when is_binary(X) -> binary_to_list(X);

format(X) -> io_lib:format("~p",[X]).



delete(Feed, Id) ->

    Key    = list_to_binary(lists:concat(["/",format(Feed),"/"])),

    A      = <<Key/binary,(term_to_binary(Id))/binary>>,

    rocksdb:delete(ref(), A, []).



count(_) -> 0.

all(R) -> {ok,I} = rocksdb:iterator(ref(), []),

           Key = list_to_binary(lists:concat(["/",format(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.



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:now(), integer_to_list((Mega*1000000+Sec)*1000000+Micro);

                _ -> erlang:integer_to_list(element(2,hd(lists:reverse(erlang:system_info(os_monotonic_time_source)))))

  end.



create_table(_,_) -> [].

add_table_index(_, _) -> ok.

dump() -> ok.