Packages

kvs

6.7.6
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,next/8,format/1]).



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,binary_to_term(Bin,[safe])} 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) -> 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(_,_,_,_,_,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 ->

     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);

                  _ -> T end;

next(_,_,_,_,_,T,_,_) -> T.



seq(_,_) -> integer_to_list(erlang:system_time(nano_seconds)).

create_table(_,_) -> [].

add_table_index(_, _) -> ok.

dump() -> ok.