simultaniously using of multiple KVS backends

include/kvs.hrl

@@ -11,5 +11,8 @@
 -record(interval,  {left,right,name}).
 -record(log,       {?CONTAINER, name, acc}).
 -record(operation, {?ITERATOR(log), body, name, status}).
+-record(kvs,       {mod,cx}).
+
+-compile({no_auto_import,[put/2]}).
 
 -endif.

src/kvs.erl

@@ -7,6 +7,13 @@
 -include("metainfo.hrl").
 -include("state.hrl").
 -include("kvs.hrl").
+-include("user.hrl").
+-include("subscription.hrl").
+-include("feed.hrl").
+-include("entry.hrl").
+-include("comment.hrl").
+-include("group.hrl").
+-include("acl.hrl").
 -include_lib("stdlib/include/qlc.hrl").
 
 % NOTE: API Documentation
@@ -14,49 +21,70 @@
 -export([start/0,stop/0]).                                        % service
 -export([destroy/0,join/0,join/1,init/2]).                        % schema change
 -export([modules/0,containers/0,tables/0,table/1,version/0]).     % meta info
--export([create/1,add/1,link/1,remove/2,remove/1]).               % chain ops
+-export([create/1,add/1,link/1,remove/2]).                        % chain ops
 -export([put/1,delete/2,next_id/2]).                              % raw ops
 -export([get/2,get/3,index/3]).                                   % read ops
 -export([load_db/1,save_db/1]).                                   % import/export
 
-start() -> DBA = ?DBA, DBA:start().
-stop() -> DBA = ?DBA, DBA:stop().
+% Public Main Backend is given in sys.config and
+% could be obtained with application:get_env(kvs,dba,store_mnesia).
+
+delete(Table,Key)  -> delete  (Table, Key, #kvs{mod=?DBA}).
+remove(Table,Key)  -> remove  (Table, Key, #kvs{mod=?DBA}).
+get(Table,Key)     -> get     (Table, Key, #kvs{mod=?DBA}).
+index(Table,K,V)   -> index   (Table, K,V, #kvs{mod=?DBA}).
+next_id(Table,DX)  -> next_id (Table, DX,  #kvs{mod=?DBA}).
+join()             -> join    ([],    #kvs{mod=?DBA}).
+join(Node)         -> join    (Node,  #kvs{mod=?DBA}).
+count(Table)       -> count   (Table, #kvs{mod=?DBA}).
+add(Table)         -> add     (Table, #kvs{mod=?DBA}).
+all(Table)         -> all     (Table, #kvs{mod=?DBA}).
+put(Table)         -> put     (Table,#kvs{mod=?DBA}).
+link(Table)        -> link    (Table,#kvs{mod=?DBA}).
+change_storage(Table,Type) -> change_storage(Table,Type,#kvs{mod=?DBA}).
+start()            -> start   (#kvs{mod=?DBA}).
+stop()             -> stop    (#kvs{mod=?DBA}).
+destroy()          -> destroy (#kvs{mod=?DBA}).
+version()          -> version (#kvs{mod=?DBA}).
+dir()              -> dir     (#kvs{mod=?DBA}).
+entries(A,B,C)     -> entries (A,B,C, #kvs{mod=?DBA}).
+% Implementation
 
-change_storage(Type) -> [ change_storage(Name,Type) || #table{name=Name} <- kvs:tables() ].
-change_storage(Table,Type) -> DBA = ?DBA, DBA:change_storage(Table,Type).
-destroy() -> DBA = ?DBA, DBA:destroy().
-join() -> DBA = ?DBA, DBA:join().
-join(Node) -> DBA = ?DBA, DBA:join(Node).
 init(Backend, Module) ->
     [ begin
-        %io:format("Creating table: ~p~n",[T]),
         Backend:create_table(T#table.name, [{attributes,T#table.fields},{T#table.copy_type, [node()]}]),
         [ Backend:add_table_index(T#table.name, Key) || Key <- T#table.keys ],
         T
     end || T <- (Module:metainfo())#schema.tables ].
 
-version() -> DBA=?DBA, DBA:version().
+start(#kvs{mod=DBA}) -> DBA:start().
+stop(#kvs{mod=DBA}) -> DBA:stop().
+change_storage(Type) -> [ change_storage(Name,Type) || #table{name=Name} <- kvs:tables() ].
+change_storage(Table,Type,#kvs{mod=DBA}) -> DBA:change_storage(Table,Type).
+destroy(#kvs{mod=DBA}) -> DBA:destroy().
+join(Node,#kvs{mod=DBA}) -> DBA:join(Node).
+version(#kvs{mod=DBA}) -> DBA:version().
 tables() -> lists:flatten([ (M:metainfo())#schema.tables || M <- modules() ]).
 table(Name) -> lists:keyfind(Name,#table.name,tables()).
-dir() -> DBA = ?DBA, DBA:dir().
+dir(#kvs{mod=DBA}) -> DBA:dir().
 modules() -> kvs:config(schema).
 containers() ->
     lists:flatten([ [ {T#table.name,T#table.fields}
         || T=#table{container=true} <- (M:metainfo())#schema.tables ]
     || M <- modules() ]).
 
-create(ContainerName) -> create(ContainerName, kvs:next_id(atom_to_list(ContainerName), 1)).
+create(ContainerName) -> create(ContainerName, kvs:next_id(atom_to_list(ContainerName), 1), #kvs{mod=?DBA}).
 
-create(ContainerName, Id) ->
+create(ContainerName, Id, Driver) ->
     kvs:info(?MODULE,"Create: ~p",[ContainerName]),
     Instance = list_to_tuple([ContainerName|proplists:get_value(ContainerName, kvs:containers())]),
     Top  = setelement(#container.id,Instance,Id),
     Top2 = setelement(#container.top,Top,undefined),
     Top3 = setelement(#container.count,Top2,0),
-    ok = kvs:put(Top3),
+    ok = kvs:put(Top3, Driver),
     Id.
 
-ensure_link(Record) ->
+ensure_link(Record, #kvs{mod=Store}=Driver) ->
 
     Id    = element(2,Record),
     Type  = table_type(element(1,Record)),
@@ -65,16 +93,15 @@ ensure_link(Record) ->
                undefined -> element(1,Record);
                      Fid -> Fid end),
 
-    Container = case kvs:get(CName, Cid) of
+    Container = case kvs:get(CName, Cid, Driver) of
         {ok,Res} -> Res;
-        {error, not_found} when Cid /= undefined ->
+        {error, _} when Cid /= undefined ->
                 NC = setelement(#container.id,
                       list_to_tuple([CName|
                             proplists:get_value(CName, kvs:containers())]), Cid),
                 NC1 = setelement(#container.count, NC, 0),
-                kvs:put(NC1),
                 NC1;
-        _ -> error end,
+        _Error -> error end,
 
     case Container of
               error -> {error, no_container};
@@ -83,99 +110,100 @@ ensure_link(Record) ->
                        Next = undefined,
                        Prev = case element(#container.top, Container) of
                                    undefined -> undefined;
-                                   Tid -> case kvs:get(Type, Tid) of
-                                              {error, not_found} -> undefined;
+                                   Tid -> case kvs:get(Type, Tid, Driver) of
+                                              {error, _} -> undefined;
                                                        {ok, Top} ->
                                         NewTop = setelement(#iterator.next, Top, Id),
-                                        kvs:put(NewTop),
+                                        kvs:put(NewTop, Driver),
                                         element(#iterator.id, NewTop) end end,
 
                        C1 = setelement(#container.top, Container, Id),
                        C2 = setelement(#container.count, C1,
                                 element(#container.count, Container)+1),
 
-                       kvs:put(C2), % Container
+                       kvs:put(C2, Driver), % Container
 
                        R  = setelement(#iterator.feeds, Record,
                             [ case F1 of
                                 {FN, Fd} -> {FN, Fd};
-                                _-> {F1, kvs:create(CName,{F1,element(#iterator.id,Record)})}
+                                _-> {F1, kvs:create(CName,{F1,element(#iterator.id,Record)},Driver)}
                               end || F1 <- element(#iterator.feeds, Record)]),
 
                        R1 = setelement(#iterator.next,    R,  Next),
                        R2 = setelement(#iterator.prev,    R1, Prev),
                        R3 = setelement(#iterator.feed_id, R2, element(#container.id, Container)),
 
-                       kvs:put(R3), % Iterator
+                       kvs:put(R3, Driver), % Iterator
 
                        kvs:info(?MODULE,"Put: ~p~n", [element(#container.id,R3)]),
 
                     {ok, R3}
             end.
 
-link(Record) ->
+link(Record,#kvs{mod=Store}=Driver) ->
     Id = element(#iterator.id, Record),
-    case kvs:get(element(1,Record), Id) of
-              {ok, Exists} -> ensure_link(Exists);
+    case kvs:get(element(1,Record), Id, Driver) of
+              {ok, Exists} -> ensure_link(Exists, Driver);
         {error, not_found} -> {error, not_found} end.
 
-add(Record) when is_tuple(Record) ->
+add(Record, #kvs{mod=Store}=Driver) when is_tuple(Record) ->
     Id = element(#iterator.id, Record),
-    case kvs:get(element(1,Record), Id) of
-        {error, not_found} -> ensure_link(Record);
+    case kvs:get(element(1,Record), Id, Driver) of
+                {error, _} -> ensure_link(Record, Driver);
          {aborted, Reason} -> {aborted, Reason};
                    {ok, _} -> {error, exist} end.
 
 reverse(#iterator.prev) -> #iterator.next;
 reverse(#iterator.next) -> #iterator.prev.
 
-relink(Container, E) ->
+relink(Container, E, Driver) ->
     Id   = element(#iterator.id, E),
     Next = element(#iterator.next, E),
     Prev = element(#iterator.prev, E),
     Top  = element(#container.top, Container),
-    case kvs:get(element(1,E), Prev) of
-         {ok, PE} -> kvs:put(setelement(#iterator.next, PE, Next));
+    case kvs:get(element(1,E), Prev, Driver) of
+         {ok, PE} -> kvs:put(setelement(#iterator.next, PE, Next), Driver);
          _ -> ok end,
-    case kvs:get(element(1,E), Next) of
-         {ok, NE} -> kvs:put(setelement(#iterator.prev, NE, Prev));
+    case kvs:get(element(1,E), Next, Driver) of
+         {ok, NE} -> kvs:put(setelement(#iterator.prev, NE, Prev), Driver);
                 _ -> ok end,
     C  = case Top of
                Id -> setelement(#container.top, Container, Prev);
                 _ -> Container end,
-    kvs:put(setelement(#container.count,C,element(#container.count,C)-1)).
+    kvs:put(setelement(#container.count,C,element(#container.count,C)-1), Driver).
+
 
-remove(Record,Id) ->
-    case kvs:get(Record,Id) of
+delete(Tab, Key, #kvs{mod=Mod}) -> Mod:delete(Tab, Key).
+
+remove(Record,Id,#kvs{mod=Mod}=Driver) ->
+    case Mod:get(Record,Id) of
          {error, not_found} -> kvs:error("Can't remove ~p~n",[{Record,Id}]);
-                     {ok,R} -> remove(R) end.
+                     {ok,R} -> do_remove(R,Driver) end.
 
-remove(E) ->
-    case kvs:get(element(#iterator.container,E),element(#iterator.feed_id,E)) of
-         {ok, Container} -> relink(Container,E);
+do_remove(E,#kvs{mod=Mod}=Driver) ->
+    case Mod:get(element(#iterator.container,E),element(#iterator.feed_id,E)) of
+         {ok, Container} -> relink(Container,E,Driver);
                        _ -> skip end,
     kvs:info(?MODULE,"Delete: ~p", [E]),
-    kvs:delete(element(1,E),element(2,E)).
+    kvs:delete(element(1,E),element(2,E), Driver).
 
-traversal( _,undefined,_,_) -> [];
-traversal(_,_,0,_) -> [];
-traversal(RecordType2, Start, Count, Direction)->
+traversal( _,undefined,_,_,Driver) -> [];
+traversal(_,_,0,_,Driver) -> [];
+traversal(RecordType2, Start, Count, Direction, Driver)->
     RecordType = table_type(RecordType2),
-    case kvs:get(RecordType, Start) of
+    case kvs:get(RecordType, Start, Driver) of
     {ok, R} ->  Prev = element(Direction, R),
                 Count1 = case Count of C when is_integer(C) -> C - 1; _-> Count end,
-                [R | traversal(RecordType2, Prev, Count1, Direction)];
+                [R | traversal(RecordType2, Prev, Count1, Direction, Driver)];
     Error ->
      io:format("Error: ~p~n",[Error]),
       [] end.
 
-entries(N)                  -> entries(N, undefined).
-entries(N,C)                -> T = kvs:table(N), entries(kvs:get(T#table.container,N), N, C).
-entries({error,_},_,_)      -> [];
-entries({ok,Container},N,C) -> entries(Container,N,C);
-entries(T,N,C)              -> traversal(N,element(#container.top,T),C,#iterator.prev).
-entries(N, Start, Count, Direction) ->
-    E = traversal(N, Start, Count, Direction),
+entries({error,_},_,_,_)      -> [];
+entries({ok,Container},N,C,Driver) -> entries(Container,N,C,Driver);
+entries(T,N,C,Driver)              -> traversal(N,element(#container.top,T),C,#iterator.prev,Driver).
+entries(N, Start, Count, Direction, Driver) ->
+    E = traversal(N, Start, Count, Direction, Driver),
     case Direction of #iterator.next -> lists:reverse(E);
                       #iterator.prev -> E end.
 
@@ -187,9 +215,8 @@ add_seq_ids() ->
     [ Init(atom_to_list(Name))  || {Name,_Fields} <- containers() ].
 
 
-put(Record) ->
-    DBA=?DBA,
-    DBA:put(Record).
+
+put(Record,#kvs{mod=DBA}) -> DBA:put(Record).
 
 table_type(user2) -> user;
 table_type(A) -> A.
@@ -205,28 +232,15 @@ find([Range|T],RecordName,Id) ->
 lookup(#interval{left=Left,right=Right,name=Name},Id) when Id =< Right, Id >= Left -> Name;
 lookup(#interval{},_Id) -> [].
 
-get(RecordName, Key) ->
-    DBA=?DBA,
+get(RecordName, Key, #kvs{mod=Mod}) ->
     case range(RecordName,Key) of
-         [] -> DBA:get(RecordName, Key);
-         Name ->  DBA:get(Name, Key) end.
-
-get(RecordName, Key, Default) ->
-    DBA=?DBA,
-    case DBA:get(RecordName, Key) of
-        {ok,{RecordName,Key,Value}} ->
-            kvs:info(?MODULE,"[kvs] get config value: ~p~n", [{RecordName, Key, Value}]),
-            {ok,Value};
-        {error, _B} ->
-            kvs:info(?MODULE,"[kvs] new config value: ~p~n", [{RecordName, Key, Default}]),
-            DBA:put({RecordName,Key,Default}),
-            {ok,Default} end.
-
-delete(Tab, Key) -> DBA=?DBA,DBA:delete(Tab, Key).
-count(RecordName) -> DBA=?DBA,DBA:count(RecordName).
-all(RecordName) -> DBA=?DBA,DBA:all(RecordName).
-index(RecordName, Key, Value) -> DBA=?DBA,DBA:index(RecordName, Key, Value).
-next_id(RecordName, Incr) -> DBA=?DBA,DBA:next_id(RecordName, Incr).
+         [] -> Mod:get(RecordName, Key);
+         Name ->  Mod:get(Name, Key) end.
+
+count(RecordName,#kvs{mod=DBA}) -> DBA:count(RecordName).
+all(RecordName,#kvs{mod=DBA}) -> DBA:all(RecordName).
+index(RecordName, Key, Value,#kvs{mod=DBA}) -> DBA:index(RecordName, Key, Value).
+next_id(RecordName, Incr,#kvs{mod=DBA}) -> DBA:next_id(RecordName, Incr).
 
 save_db(Path) ->
     Data = lists:append([all(B) || B <- [list_to_atom(Name) || {table,Name} <- kvs:dir()] ]),

src/store/store_fs.erl

@@ -15,14 +15,14 @@ join(Node) -> ok. % should be rsync or smth
 change_storage(Table,Type) -> ok.
 
 initialize() ->
-    kvs:info(?MODULE,"[store_mnesia] mnesia init.~n",[]),
+    kvs:info(?MODULE,"fs init.~n",[]),
     mnesia:create_schema([node()]),
     [ kvs:init(store_fs,Module) || Module <- kvs:modules() ],
     mnesia:wait_for_tables([ T#table.name || T <- kvs:tables()],infinity).
 
 index(Tab,Key,Value) -> ok.
 get(TableName, Key) ->
-    HashKey = wf:url_encode(base64:encode(crypto:sha(term_to_binary(Key)))),
+    HashKey = encode(base64:encode(crypto:sha(term_to_binary(Key)))),
     Dir = lists:concat(["data/",TableName,"/"]),
     case file:read_file(lists:concat([Dir,HashKey])) of
          {ok,Binary} -> {ok,binary_to_term(Binary,[safe])};
@@ -31,13 +31,13 @@ get(TableName, Key) ->
 put(Records) when is_list(Records) -> lists:map(fun(Record) -> put(Record) end, Records);
 put(Record) ->
     TableName = element(1,Record),
-    HashKey = wf:url_encode(base64:encode(crypto:sha(term_to_binary(element(2,Record))))),
+    HashKey = encode(base64:encode(crypto:sha(term_to_binary(element(2,Record))))),
     BinaryValue = term_to_binary(Record),
     Dir = lists:concat(["data/",TableName,"/"]),
     filelib:ensure_dir(Dir),
     File = lists:concat([Dir,HashKey]),
     io:format("File: ~p~n",[File]),
-    file:write_file(File,BinaryValue,[write,raw,binary,exclusive,sync]).
+    file:write_file(File,BinaryValue,[write,raw,binary,sync]).
 
 delete(Tab, Key) -> ok.
 count(RecordName) -> length(filelib:fold_files(lists:concat(["data/",RecordName]), "",true, fun(A,Acc)-> [A|Acc] end, [])).
@@ -45,3 +45,42 @@ all(R) -> filelib:fold_files(lists:concat(["data/",R]), "",true, fun(A,Acc)-> [A
 next_id(RecordName, Incr) -> mnesia:dirty_update_counter({id_seq, RecordName}, Incr).
 create_table(Name,Options) -> filelib:ensure_dir(lists:concat(["data/",Name,"/"])).
 add_table_index(Record, Field) -> ok.
+
+% URL ENCODE
+
+encode(B) when is_binary(B) -> encode(binary_to_list(B));
+encode([C | Cs]) when C >= $a, C =< $z -> [C | encode(Cs)];
+encode([C | Cs]) when C >= $A, C =< $Z -> [C | encode(Cs)];
+encode([C | Cs]) when C >= $0, C =< $9 -> [C | encode(Cs)];
+encode([C | Cs]) when C == 16#20 -> [$+ | encode(Cs)];
+
+% unreserved
+encode([C = $- | Cs]) -> [C | encode(Cs)];
+encode([C = $_ | Cs]) -> [C | encode(Cs)];
+encode([C = 46 | Cs]) -> [C | encode(Cs)];
+encode([C = $! | Cs]) -> [C | encode(Cs)];
+encode([C = $~ | Cs]) -> [C | encode(Cs)];
+encode([C = $* | Cs]) -> [C | encode(Cs)];
+encode([C = 39 | Cs]) -> [C | encode(Cs)];
+encode([C = $( | Cs]) -> [C | encode(Cs)];
+encode([C = $) | Cs]) -> [C | encode(Cs)];
+
+encode([C | Cs]) when C =< 16#7f -> escape_byte(C) ++ encode(Cs);
+encode([C | Cs]) when (C >= 16#7f) and (C =< 16#07FF) ->
+  escape_byte((C bsr 6) + 16#c0)
+  ++ escape_byte(C band 16#3f + 16#80)
+  ++ encode(Cs);
+encode([C | Cs]) when (C > 16#07FF) ->
+  escape_byte((C bsr 12) + 16#e0) % (0xe0 | C >> 12)
+  ++ escape_byte((16#3f band (C bsr 6)) + 16#80) % 0x80 | ((C >> 6) & 0x3f)
+  ++ escape_byte(C band 16#3f + 16#80) % 0x80 | (C >> 0x3f)
+  ++ encode(Cs);
+encode([C | Cs]) -> escape_byte(C) ++ encode(Cs);
+encode([]) -> [].
+
+hex_octet(N) when N =< 9 -> [$0 + N];
+hex_octet(N) when N > 15 -> hex_octet(N bsr 4) ++ hex_octet(N band 15);
+hex_octet(N) -> [N - 10 + $a].
+escape_byte(C) -> normalize(hex_octet(C)).
+normalize(H) when length(H) == 1 -> "%0" ++ H;
+normalize(H) -> "%" ++ H.