Packages
kvs
9.7.0
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
Current section
Files
Jump to
Current section
Files
src/kvs.erl
-module(kvs).
-behaviour(application).
-behaviour(supervisor).
-include_lib("stdlib/include/assert.hrl").
-include("api.hrl").
-include("metainfo.hrl").
-include("stream.hrl").
-include("cursors.hrl").
-include("kvs.hrl").
-include("backend.hrl").
-export([dump/0,
db/0,
metainfo/0,
ensure/1,
ensure/2,
seq_gen/0,
keys/1,
fields/1,
defined/2,
field/2,
setfield/3,
cut/2]).
-export([join/2, seq/3]).
-export(?API)
.
-export(?STREAM)
.
-export([init/1, start/2, stop/1]).
-record('$msg', {id, next, prev, user, msg}).
init([]) -> {ok, {{one_for_one, 5, 10}, []}}.
start(_, _) ->
supervisor:start_link({local, kvs}, kvs, []).
stop(_) -> ok.
dba() -> application:get_env(kvs, dba, kvs_mnesia).
seq_dba() -> application:get_env(kvs, dba_seq, kvs_mnesia).
db() -> (dba()):db().
kvs_stream() ->
application:get_env(kvs, dba_st, kvs_stream).
% kvs api
all(Table) -> all(Table, #kvs{mod = dba(), db = db()}).
delete(Table, Key) ->
delete(Table, Key, #kvs{mod = dba(), db = db()}).
get(Table, Key) ->
(?MODULE):get(Table, Key, #kvs{mod = dba(), db = db()}).
index(Table, K, V) ->
index(Table, K, V, #kvs{mod = dba()}).
keys(Feed) ->
keys(Feed, #kvs{mod = dba(), db = db()}).
match(Record) ->
match(Record, #kvs{mod = dba()}).
index_match(Record, Index) ->
index_match(Record, Index, #kvs{mod = dba()}).
join() -> join([], #kvs{mod = dba(), db = db()}).
dump() -> dump(#kvs{mod = dba()}).
join(Node) -> join(Node, #kvs{mod = dba(), db = db()}).
leave() -> leave(#kvs{mod = dba(), db = db()}).
destroy() -> destroy(#kvs{mod = dba(), db = db()}).
count(Table) -> count(Table, #kvs{mod = dba()}).
put(Record) -> (?MODULE):put(Record, #kvs{mod = dba(), db = db()}).
stop() -> stop_kvs(#kvs{mod = dba()}).
start() -> start(#kvs{mod = dba()}).
ver() -> ver(#kvs{mod = dba()}).
dir() -> dir(#kvs{mod = dba()}).
feed(Key) ->
feed(Key, #kvs{mod = dba(), st = kvs_stream(), db = db()}).
seq([], DX) -> seq([], DX, #kvs{mod = kvs_rocks});
seq(Table, DX) -> seq(Table, DX, #kvs{mod = seq_dba()}).
remove(Rec, Feed) ->
remove(Rec, Feed, #kvs{mod = dba(), st = kvs_stream(), db = db()}).
put(Records, #kvs{mod = Mod, db = Db}) when is_list(Records) ->
Mod:put(Records, Db);
put(Record, #kvs{mod = Mod, db = Db}) -> Mod:put(Record, Db).
get(RecordName, Key, #kvs{mod = Mod, db = Db}) ->
Mod:get(RecordName, Key, Db).
delete(Tab, Key, #kvs{mod = Mod, db = Db}) ->
Mod:delete(Tab, Key, Db).
count(Tab, #kvs{mod = DBA}) -> DBA:count(Tab).
index(Tab, Key, Value, #kvs{mod = DBA}) ->
DBA:index(Tab, Key, Value).
keys(Feed, #kvs{mod = DBA, db = Db}) ->
DBA:keys(Feed, Db).
match(Record, #kvs{mod = DBA}) ->
DBA:match(Record).
index_match(Record, Index, #kvs{mod = DBA}) ->
DBA:index_match(Record, Index).
seq(Tab, Incr, #kvs{mod = DBA}) -> DBA:seq(Tab, Incr).
dump(#kvs{mod = Mod}) -> Mod:dump().
feed(Tab, #kvs{st = Mod, db = Db}) -> Mod:feed(Tab, Db).
remove(Rec, Feed, #kvs{st = Mod, db = Db}) ->
Mod:remove(Rec, Feed, Db).
all(Tab, #kvs{mod = DBA, db = Db}) -> DBA:all(Tab, Db).
start(#kvs{mod = DBA}) -> DBA:start().
stop_kvs(#kvs{mod = DBA}) -> DBA:stop().
join(Node, #kvs{mod = DBA, db = Db}) -> DBA:join(Node, Db).
leave(#kvs{mod = DBA, db = Db}) -> DBA:leave(Db).
destroy(#kvs{mod = DBA, db = Db}) -> DBA:destroy(Db).
ver(#kvs{mod = DBA}) -> DBA:version().
dir(#kvs{mod = DBA}) -> DBA:dir().
% stream api
top(X) -> (kvs_stream()):top(X).
top(X,#kvs{db = Db}) -> (kvs_stream()):top(X,Db).
bot(X) -> (kvs_stream()):bot(X).
bot(X,#kvs{db = Db}) -> (kvs_stream()):bot(X,Db).
next(X) -> (kvs_stream()):next(X).
next(X,#kvs{db = Db}) -> (kvs_stream()):next(X,Db).
prev(X) -> (kvs_stream()):prev(X).
prev(X,#kvs{db = Db}) -> (kvs_stream()):prev(X,Db).
drop(X) -> (kvs_stream()):drop(X).
drop(X,#kvs{db = Db}) -> (kvs_stream()):drop(X,Db).
take(X) -> (kvs_stream()):take(X).
take(X,#kvs{db = Db}) -> (kvs_stream()):take(X,Db).
save(X) -> (kvs_stream()):save(X).
save(X,#kvs{db = Db}) -> (kvs_stream()):save(X,Db).
cut(X, Y) -> (kvs_stream()):cut(X, Y).
add(X) -> (kvs_stream()):add(X).
add(X,#kvs{db = Db}) -> (kvs_stream()):add(X,Db).
append(X, Y) -> (kvs_stream()):append(X, Y).
append(X, Y, #kvs{db = Db}) -> (kvs_stream()):append(X, Y, Db).
load_reader(X) -> (kvs_stream()):load_reader(X).
load_reader(X,#kvs{db = Db}) -> (kvs_stream()):load_reader(X,Db).
writer(X) -> (kvs_stream()):writer(X).
writer(X,#kvs{db = Db}) -> (kvs_stream()):writer(X,Db).
reader(X) -> (kvs_stream()):reader(X).
reader(X,#kvs{db = Db}) -> (kvs_stream()):reader(X,Db).
% unrevisited
ensure(#writer{} = X) -> ensure(X,#kvs{mod = dba(), db = db()}).
ensure(#writer{id = Id},#kvs{} = X) ->
case kvs:get(writer, Id, X) of
{error, _} ->
kvs:save(kvs:writer(Id, X)),
ok;
{ok, _} -> ok
end.
cursors() ->
lists:flatten([[{T#table.name, T#table.fields}
|| #table{name = Name} = T
<- (M:metainfo())#schema.tables,
Name == reader orelse Name == writer]
|| M <- modules()]).
% metainfo api
tables() ->
lists:flatten([(M:metainfo())#schema.tables
|| M <- modules()]).
modules() -> application:get_env(kvs, schema, []).
metainfo() ->
#schema{name = kvs, tables = core() ++ test_tabs()}.
core() ->
[#table{name = id_seq,
fields = record_info(fields, id_seq), keys = [thing]}].
test_tabs() ->
[#table{name = '$msg',
fields = record_info(fields, '$msg')}].
table(Name) when is_atom(Name) ->
lists:keyfind(Name, #table.name, tables());
table(_) -> false.
seq_gen() ->
Init = fun (Key) ->
case kvs:get(id_seq, Key) of
{error, _} ->
{Key, kvs:put(#id_seq{thing = Key, id = 0})};
{ok, _} -> {Key, skip}
end
end,
[Init(atom_to_list(Name))
|| {Name, _Fields} <- cursors()].
initialize(Backend, Module) ->
[begin
Backend:create_table(T#table.name,
[{attributes, T#table.fields},
{T#table.copy_type, [node()]},
{type, T#table.type}]),
[Backend:add_table_index(T#table.name, Key)
|| Key <- T#table.keys],
T
end
|| T <- (Module:metainfo())#schema.tables].
fields(Table) when is_atom(Table) ->
case table(Table) of
false -> [];
T -> T#table.fields
end.
defined(TableRecord, Field) ->
FieldsList = fields(element(1, TableRecord)),
lists:member(Field, FieldsList).
field(TableRecord, Field) ->
FieldsList = fields(element(1, TableRecord)),
Index = string:str(FieldsList, [Field]) + 1,
element(Index, TableRecord).
setfield(TableRecord, Field, Value) ->
FieldsList = fields(element(1, TableRecord)),
Index = string:str(FieldsList, [Field]) + 1,
setelement(Index, TableRecord, Value).