%%%------------------------------------------------------------------- %%% @author Heinz Nikolaus Gies %%% @copyright (C) 2013, Heinz Nikolaus Gies %%% @doc %%% %%% @end %%% Created : 10 Jan 2013 by Heinz Nikolaus Gies %%%------------------------------------------------------------------- -module(fifo_db). -behaviour(gen_server). %% API -export([start_link/3, start/1, start/2, start/3, sync_get/3, get/3, transact/2, delete/3, destroy/1, put/4, fold/4, fold_keys/4, list_keys/2]). -export([encode_key/1, encode_key/2, decode_key/1]). %% gen_server callbacks -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -type calback_reply() :: calback_reply(any()). -type calback_reply(ReplyType) :: {noreply, State::term()} | {reply, Reply::ReplyType, State::term()}. -type fold_fn() :: fun((Key::binary(), Value::term(), Acc) -> Acc). -type fold_key_fn() :: fun((Key::binary(), Acc) -> Acc). -type transaction_op() :: {delete, Key::binary()} | {put, Key::binary(), Value::term()}. -callback init(DBLock::string(), Name::atom(), Opts::[term()]) -> {ok, State::term()}. -callback ensure_running(State::term()) -> {ok, State::term()}. -callback put(Bucket::binary(), Keyt::binary(), Value::term(), _From::pid(), State::term()) -> calback_reply(). -callback transact(Transaction::[transaction_op()], From::pid(), State::term()) -> calback_reply(). -callback get(Buckett::binary(), Key::binary(), From::pid(), State::term()) -> calback_reply(). -callback destroy(State::term()) -> {ok, State::term()}. -callback delete(Bucket::binary(), Key::binary(), _From::pid(), State::term()) -> calback_reply(). -callback fold(Bucket::binary(), FoldFn::fold_fn(), Acc0::term(), From::pid(), State::term()) -> calback_reply(). -callback fold_keys(Bucket::binary(), FoldFn::fold_key_fn(), Acc0::term(), From::pid(), State::term()) -> calback_reply(). -callback list_keys(Bucket::binary(), _From::pid(), State::term()) -> calback_reply([binary()]). -callback terminate(Reason::atom(), State::term()) -> ok | {error, term()}. -callback code_change(OldVsn::term(), State::term(), Extra::term()) -> {ok, term()}. -ignore_xref([start_link/1, fold_keys/4]). %%%=================================================================== %%% API %%%=================================================================== -spec encode_key({Bucket::binary(), Key::binary()}) -> binary(). encode_key({Bucket, Key}) -> encode_key(Bucket, Key). -spec encode_key(Bucket::binary(), Key::binary()) -> binary(). encode_key(Bucket, Key) -> Len = byte_size(Bucket), <>. -spec decode_key(BucketKey::binary()) -> {Bucket::binary(), Key::binary()}. decode_key(<>) -> {Bucket, Key}. %%-------------------------------------------------------------------- %% @doc %% Starts the server %% %% @spec start_link() -> {ok, Pid} | ignore | {error, Error} %% @end %%-------------------------------------------------------------------- start_link(Name, Backend, Opts) -> gen_server:start_link({local, Name}, ?MODULE, [Name, Backend, Opts], []). start(Name) -> start(Name, []). start(Name, Opts) -> start(Name, application:get_env(fifo_db, backend, fifo_db_leveldb), Opts). start(Name, Backend, Opts) -> case erlang:whereis(Name) of undefined -> fifo_db_sup:start_child(Name, Backend, Opts); Pid -> gen_server:call(Pid, ensure_running) end. transact(Name, Transaction) -> gen_server:call(Name, {transact, Transaction}). put(Name, Bucket, Key, Value) -> gen_server:call(Name, {put, Bucket, Key, Value}). get(Name, Bucket, Key) -> gen_server:call(Name, {get, Bucket, Key}). sync_get(Name, Bucket, Key) -> gen_server:call(Name, {sync_get, Bucket, Key}). delete(Name, Bucket, Key) -> gen_server:call(Name, {delete, Bucket, Key}). destroy(Name) -> gen_server:call(Name, destroy). fold(Name, Bucket, FoldFn, Acc0) -> gen_server:call(Name, {fold, Bucket, FoldFn, Acc0}). fold_keys(Name, Bucket, FoldFn, Acc0) -> gen_server:call(Name, {fold_keys, Bucket, FoldFn, Acc0}). list_keys(Name, Bucket) -> gen_server:call(Name, {list_keys, Bucket}). %%%=================================================================== %%% Internal functions %%%=================================================================== init([Name, Backend, Opts]) when is_atom(Name), is_atom(Backend) -> {ok, DBLoc} = application:get_env(fifo_db, db_path), lager:info("Opening ~s in ~s as backend ~p with options: ~p", [Name, DBLoc, Backend, Opts]), {ok, State} = Backend:init(DBLoc, Name, Opts), {ok, {Backend, State}}; init(P) -> lager:error("Invalid init parameters: ~p", [P]), {stop, error, undefined}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling call messages %% %% @spec handle_call(Request, From, State) -> %% {reply, Reply, State} | %% {reply, Reply, State, Timeout} | %% {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, Reply, State} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_call(ensure_running, _From, {Backend, State}) -> {reply, ok, {Backend, Backend:ensure_running(State)}}; handle_call({put, Bucket, Key, Value}, From, {Backend, State}) -> case Backend:put(Bucket, Key, Value, From, State) of {reply, Reply, State1} -> {reply, Reply, {Backend, State1}}; {noreply, State1} -> {noreply, {Backend, State1}} end; handle_call({transact, Transaction}, From, {Backend, State}) -> case Backend:transact(Transaction, From, State) of {reply, Reply, State1} -> {reply, Reply, {Backend, State1}}; {noreply, State1} -> {noreply, {Backend, State1}} end; handle_call({get, Bucket, Key}, From, {Backend, State}) -> case Backend:get(Bucket, Key, From, State) of {reply, Reply, State1} -> {reply, Reply, {Backend, State1}}; {noreply, State1} -> {noreply, {Backend, State1}} end; handle_call({delete, Bucket, Key}, From, {Backend, State}) -> case Backend:delete(Bucket, Key, From, State) of {reply, Reply, State1} -> {reply, Reply, {Backend, State1}}; {noreply, State1} -> {noreply, {Backend, State1}} end; handle_call(destroy, _From, {Backend, State}) -> {ok, State1} = Backend:destroy(State), {reply, ok, {Backend, State1}}; handle_call({list_keys, Bucket}, From, {Backend, State}) -> case Backend:list_keys(Bucket, From, State) of {reply, Reply, State1} -> {reply, Reply, {Backend, State1}}; {noreply, State1} -> {noreply, {Backend, State1}} end; handle_call({fold, Bucket, FoldFn, Acc0}, From, {Backend, State}) -> case Backend:fold(Bucket, FoldFn, Acc0, From, State) of {reply, Reply, State1} -> {reply, Reply, {Backend, State1}}; {noreply, State1} -> {noreply, {Backend, State1}} end; handle_call({fold_keys, Bucket, FoldFn, Acc0}, From, {Backend, State}) -> case Backend:fold_keys(Bucket, FoldFn, Acc0, From, State) of {reply, Reply, State1} -> {reply, Reply, {Backend, State1}}; {noreply, State1} -> {noreply, {Backend, State1}} end. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling cast messages %% %% @spec handle_cast(Msg, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_cast(_Msg, State) -> {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% Handling all non call/cast messages %% %% @spec handle_info(Info, State) -> {noreply, State} | %% {noreply, State, Timeout} | %% {stop, Reason, State} %% @end %%-------------------------------------------------------------------- handle_info(_Info, State) -> {noreply, State}. %%-------------------------------------------------------------------- %% @private %% @doc %% This function is called by a gen_server when it is about to %% terminate. It should be the opposite of Module:init/1 and do any %% necessary cleaning up. When it returns, the gen_server terminates %% with Reason. The return value is ignored. %% %% @spec terminate(Reason, From, State) -> void() %% @end %%-------------------------------------------------------------------- terminate(Reason, {Backend, State}) -> try Backend:terminate(Reason, State) catch E:E1 -> lager:error("[fifodb] Error terminating database: ~p:~p", [E, E1]) end. %%-------------------------------------------------------------------- %% @private %% @doc %% Convert process state when code is changed %% %% @spec code_change(OldVsn, State, Extra) -> {ok, NewState} %% @end %%-------------------------------------------------------------------- code_change(OldVsn, {Backend, State}, Extra) -> {ok, State1} = Backend:code_change(OldVsn, State, Extra), {ok, {Backend, State1}}.