store_kai

samples/rebar.config

@@ -3,5 +3,6 @@
 {deps_dir,["deps"]}.
 {deps, [
     {active,  ".*", {git, "git://github.com/synrc/active", {tag,"0.9"}}},
+    {kai,     ".*", {git, "git://github.com/synrc/kai", "HEAD"}},
     {kvs,     ".*", {git, "git://github.com/synrc/kvs", "HEAD"}}
 ]}.

src/kvs.erl

@@ -67,6 +67,8 @@ add(Record) when is_tuple(Record) ->
             Cid = table_type(case element(#iterator.feed_id, Record) of
                 undefined -> element(1,Record);
                 Fid -> Fid end),
+                
+            io:format("CName:~p Cid:~p",[CName, Cid]),
 
             Container = case kvs:get(CName, Cid) of
                 {ok,C} -> C;

src/store_kai.erl

@@ -0,0 +1,72 @@
+-module(store_kai).
+-author('Maxim Sokhatsky').
+-copyright('Synrc Research Center s.r.o.').
+-include_lib("kai/include/kai.hrl").
+-include_lib("kvs/include/config.hrl").
+-include_lib("kvs/include/metainfo.hrl").
+-include_lib("stdlib/include/qlc.hrl").
+-compile(export_all).
+
+start() -> kai:start(), ok.
+stop() -> kai_store:stop(), ok.
+version() -> {version,"KVS KAI"}.
+join() -> initialize(), ok.
+join(Node) -> initialize(), ok.
+initialize() -> ok.
+dir() -> [{table,T}||T<-kvs:modules()].
+
+put(Records) when is_list(Records) -> lists:foreach(fun kai_put/1, Records);
+put(Record) -> kai_put(Record).
+
+kai_put(Record) ->
+    Data = #data{key = element(2,Record), bucket = table_to_num(element(1,Record)),
+        last_modified = now(), checksum = erlang:md5(term_to_binary(Record)),
+        vector_clocks = vclock:fresh(), value = Record },
+    kai_store:put(Data).
+
+update(Record, Object) -> ok.
+
+get(Tab, Key) ->
+    Data = #data{key=Key,bucket=table_to_num(Tab)},
+    kai_get(Data).
+
+kai_get(Data) ->
+    case kai_store:get(Data) of
+         #data{value=Value} -> {ok,Value};
+         undefined -> {error,not_found};
+         E -> {error,E} end.
+
+delete(Tab, Key) ->
+    Data = #data{key=Key,bucket=table_to_num(Tab)},
+    kai_delete(Data).
+
+kai_delete(Data) ->
+    case kai_store:delete(Data) of
+         ok -> ok;
+         E -> {error,E} end.
+
+key_to_bin(Key) ->
+    if is_integer(Key) -> erlang:list_to_binary(integer_to_list(Key));
+       is_list(Key) -> erlang:list_to_binary(Key);
+       is_atom(Key) -> erlang:list_to_binary(erlang:atom_to_list(Key));
+       is_binary(Key) -> Key;
+       true ->  [ListKey] = io_lib:format("~p", [Key]), erlang:list_to_binary(ListKey) end.
+
+all(RecordName) ->
+    {list_of_data,List} = kai_store:list(table_to_num(RecordName)),
+    [ begin {ok,Val}=kai_get(Data),Val end || Data <- List ].
+
+all_by_index(Tab, IndexId, IndexVal) -> [].
+
+table_to_num(user)         -> 10;
+table_to_num(subscription) -> 11;
+table_to_num(group)        -> 12;
+table_to_num(id_seq)       -> 13;
+table_to_num(product)      -> 14;
+table_to_num(acl)          -> 15;
+table_to_num(feed)         -> 16;
+table_to_num(entry)        -> 17;
+table_to_num(comment)      -> 18;
+table_to_num(Name) -> case kvs:config(kvs,table_to_num) of
+                        [] -> unknown_table;
+                        Module -> Module:table_to_num(Name) end.

src/store_mnesia.erl

@@ -9,7 +9,7 @@ start()    -> mnesia:start().
 stop()     -> mnesia:stop().
 destroy()  -> [mnesia:delete_table(list_to_atom(T))||{_,T}<-kvs:dir()], mnesia:delete_schema([node()]), ok.
 version()  -> {version,"KVS MNESIA"}.
-dir()      -> [{table,atom_to_list(T)}||T<-mnesia:system_info(local_tables)].
+dir()      -> [{table,T}||T<-mnesia:system_info(local_tables)].
 join()     -> mnesia:change_table_copy_type(schema, node(), disc_copies), initialize().
 join(Node) ->
     mnesia:change_config(extra_db_nodes, [Node]),