yuanbiao
/
emqx
1
0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Rewrite the hooks module

This commit is contained in:
Feng Lee 2018-09-05 23:21:06 +08:00
parent 96122cf966
commit 4635921458
1 changed files with 106 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

199
src/emqx_hooks.erl
View File

@ -16,142 +16,160 @@
-behaviour(gen_server). -behaviour(gen_server).
-export([start_link/0]). -export([start_link/0, stop/0]).
%% Hooks API %% Hooks API
-export([add/3, add/4, delete/2, run/2, run/3, lookup/1]). -export([add/2, add/3, add/4, del/2, run/2, run/3, lookup/1]).
%% gen_server Function Exports %% gen_server Function Exports
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
terminate/2, code_change/3]). code_change/3]).
-record(state, {}). -type(hookpoint() :: atom()).
-type(action() :: function() | mfa()).
-type(filter() :: function() | mfa()).
-type(hooktag() :: atom() | string() | binary()). -record(callback, {action :: action(),
filter :: filter(),
priority :: integer()}).
-export_type([hooktag/0]). -record(hook, {name :: hookpoint(), callbacks :: list(#callback{})}).
-record(callback, {tag :: hooktag(), -export_type([hookpoint/0, action/0, filter/0]).
function :: function(),
init_args = [] :: list(any()),
priority = 0 :: integer()}).
-record(hook, {name :: atom(), callbacks = [] :: list(#callback{})}).
-define(TAB, ?MODULE). -define(TAB, ?MODULE).
-define(SERVER, ?MODULE).
-spec(start_link() -> emqx_types:startlink_ret()).
start_link() -> start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). gen_server:start_link({local, ?SERVER}, ?MODULE, [], [{hibernate_after, 60000}]).
%%-------------------------------------------------------------------- -spec(stop() -> ok).
stop() ->
gen_server:stop(?SERVER, normal, infinity).
%%------------------------------------------------------------------------------
%% Hooks API %% Hooks API
%%-------------------------------------------------------------------- %%------------------------------------------------------------------------------
-spec(add(atom(), function() | {hooktag(), function()}, list(any())) -> ok). %% @doc Register a callback
add(HookPoint, Function, InitArgs) when is_function(Function) -> -spec(add(hookpoint(), action() | #callback{}) -> emqx_types:ok_or_error(already_exists)).
add(HookPoint, {undefined, Function}, InitArgs, 0); add(HookPoint, Callback) when is_record(Callback, callback) ->
gen_server:call(?SERVER, {add, HookPoint, Callback}, infinity);
add(HookPoint, Action) when is_function(Action); is_tuple(Action) ->
add(HookPoint, #callback{action = Action, priority = 0}).
add(HookPoint, {Tag, Function}, InitArgs) when is_function(Function) -> -spec(add(hookpoint(), action(), filter() | integer() | list())
add(HookPoint, {Tag, Function}, InitArgs, 0). -> emqx_types:ok_or_error(already_exists)).
add(HookPoint, Action, InitArgs) when is_function(Action), is_list(InitArgs) ->
add(HookPoint, #callback{action = {Action, InitArgs}, priority = 0});
add(HookPoint, Action, Filter) when is_function(Filter); is_tuple(Filter) ->
add(HookPoint, #callback{action = Action, filter = Filter, priority = 0});
add(HookPoint, Action, Priority) when is_integer(Priority) ->
add(HookPoint, #callback{action = Action, priority = Priority}).
-spec(add(atom(), function() | {hooktag(), function()}, list(any()), integer()) -> ok). -spec(add(hookpoint(), action(), filter(), integer())
add(HookPoint, Function, InitArgs, Priority) when is_function(Function) -> -> emqx_types:ok_or_error(already_exists)).
add(HookPoint, {undefined, Function}, InitArgs, Priority); add(HookPoint, Action, Filter, Priority) ->
add(HookPoint, {Tag, Function}, InitArgs, Priority) when is_function(Function) -> add(HookPoint, #callback{action = Action, filter = Filter, priority = Priority}).
gen_server:call(?MODULE, {add, HookPoint, {Tag, Function}, InitArgs, Priority}).
-spec(delete(atom(), function() | {hooktag(), function()}) -> ok). %% @doc Unregister a callback.
delete(HookPoint, Function) when is_function(Function) -> -spec(del(hookpoint(), action()) -> ok).
delete(HookPoint, {undefined, Function}); del(HookPoint, Action) ->
delete(HookPoint, {Tag, Function}) when is_function(Function) -> gen_server:cast(?SERVER, {del, HookPoint, Action}).
gen_server:call(?MODULE, {delete, HookPoint, {Tag, Function}}).
%% @doc Run hooks without Acc. %% @doc Run hooks.
-spec(run(atom(), list(Arg :: any())) -> ok | stop). -spec(run(atom(), list(Arg :: any())) -> ok | stop).
run(HookPoint, Args) -> run(HookPoint, Args) ->
run_(lookup(HookPoint), Args). run_(lookup(HookPoint), Args).
%% @doc Run hooks with Accumulator.
-spec(run(atom(), list(Arg :: any()), any()) -> any()). -spec(run(atom(), list(Arg :: any()), any()) -> any()).
run(HookPoint, Args, Acc) -> run(HookPoint, Args, Acc) ->
run_(lookup(HookPoint), Args, Acc). run_(lookup(HookPoint), Args, Acc).
%% @private %% @private
run_([#callback{function = Fun, init_args = InitArgs} | Callbacks], Args) -> run_([#callback{action = Action, filter = Filter} | Callbacks], Args) ->
case apply(Fun, lists:append([Args, InitArgs])) of case filtered(Filter, Args) orelse execute(Action, Args) of
true -> run_(Callbacks, Args);
ok -> run_(Callbacks, Args); ok -> run_(Callbacks, Args);
stop -> stop; stop -> stop;
_Any -> run_(Callbacks, Args) _Any -> run_(Callbacks, Args)
end; end;
run_([], _Args) -> run_([], _Args) ->
ok. ok.
%% @private %% @private
run_([#callback{function = Fun, init_args = InitArgs} | Callbacks], Args, Acc) -> run_([#callback{action = Action, filter = Filter} | Callbacks], Args, Acc) ->
case apply(Fun, lists:append([Args, [Acc], InitArgs])) of Args1 = Args ++ [Acc],
case filtered(Filter, Args1) orelse execute(Action, Args1) of
true -> run_(Callbacks, Args, Acc);
ok -> run_(Callbacks, Args, Acc); ok -> run_(Callbacks, Args, Acc);
{ok, NewAcc} -> run_(Callbacks, Args, NewAcc); {ok, NewAcc} -> run_(Callbacks, Args, NewAcc);
stop -> {stop, Acc}; stop -> {stop, Acc};
{stop, NewAcc} -> {stop, NewAcc}; {stop, NewAcc} -> {stop, NewAcc};
_Any -> run_(Callbacks, Args, Acc) _Any -> run_(Callbacks, Args, Acc)
end; end;
run_([], _Args, Acc) -> run_([], _Args, Acc) ->
{ok, Acc}. {ok, Acc}.
-spec(lookup(atom()) -> [#callback{}]). filtered(undefined, _Args) ->
false;
filtered(Filter, Args) ->
execute(Filter, Args).
execute(Action, Args) when is_function(Action) ->
erlang:apply(Action, Args);
execute({Fun, InitArgs}, Args) when is_function(Fun) ->
erlang:apply(Fun, Args ++ InitArgs);
execute({M, F, A}, Args) ->
erlang:apply(M, F, Args ++ A).
%% @doc Lookup callbacks.
-spec(lookup(hookpoint()) -> [#callback{}]).
lookup(HookPoint) -> lookup(HookPoint) ->
case ets:lookup(?TAB, HookPoint) of case ets:lookup(?TAB, HookPoint) of
[#hook{callbacks = Callbacks}] -> Callbacks; [#hook{callbacks = Callbacks}] ->
Callbacks;
[] -> [] [] -> []
end. end.
%%-------------------------------------------------------------------- %%-----------------------------------------------------------------------------
%% gen_server callbacks %% gen_server callbacks
%%-------------------------------------------------------------------- %%-----------------------------------------------------------------------------
init([]) -> init([]) ->
_ = emqx_tables:new(?TAB, [set, protected, {keypos, #hook.name}, _ = emqx_tables:new(?TAB, [{keypos, #hook.name}, {read_concurrency, true}]),
{read_concurrency, true}]), {ok, #{}}.
{ok, #state{}}.
handle_call({add, HookPoint, {Tag, Function}, InitArgs, Priority}, _From, State) -> handle_call({add, HookPoint, Callback = #callback{action = Action}}, _From, State) ->
Callback = #callback{tag = Tag, function = Function, Reply = case lists:keyfind(Action, 2, Callbacks = lookup(HookPoint)) of
init_args = InitArgs, priority = Priority}, true ->
{reply, {error, already_exists};
case ets:lookup(?TAB, HookPoint) of false ->
[#hook{callbacks = Callbacks}] -> insert_hook(HookPoint, add_callback(Callback, Callbacks))
case contain_(Tag, Function, Callbacks) of end,
false -> {reply, Reply, State};
insert_hook_(HookPoint, add_callback_(Callback, Callbacks));
true ->
{error, already_hooked}
end;
[] ->
insert_hook_(HookPoint, [Callback])
end, State};
handle_call({delete, HookPoint, {Tag, Function}}, _From, State) -> handle_call({del, HookPoint, Action}, _From, State) ->
{reply, case lists:keydelete(Action, 2, lookup(HookPoint)) of
case ets:lookup(?TAB, HookPoint) of [] ->
[#hook{callbacks = Callbacks}] -> ets:delete(?TAB, HookPoint);
case contain_(Tag, Function, Callbacks) of Callbacks ->
true -> insert_hook(HookPoint, Callbacks)
insert_hook_(HookPoint, del_callback_(Tag, Function, Callbacks)); end,
false -> {reply, ok, State};
{error, not_found}
end;
[] ->
{error, not_found}
end, State};
handle_call(Req, _From, State) -> handle_call(Req, _From, State) ->
{reply, {error, {unexpected_request, Req}}, State}. emqx_logger:error("[Hooks] unexpected call: ~p", [Req]),
{reply, ignored, State}.
handle_cast(_Msg, State) -> handle_cast(Msg, State) ->
emqx_logger:error("[Hooks] unexpected msg: ~p", [Msg]),
{noreply, State}. {noreply, State}.
handle_info(_Info, State) -> handle_info(Info, State) ->
emqx_logger:error("[Hooks] unexpected info: ~p", [Info]),
{noreply, State}. {noreply, State}.
terminate(_Reason, _State) -> terminate(_Reason, _State) ->
@ -160,26 +178,21 @@ terminate(_Reason, _State) ->
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
%%-------------------------------------------------------------------- %%-----------------------------------------------------------------------------
%% Internal functions %% Internal functions
%%-------------------------------------------------------------------- %%-----------------------------------------------------------------------------
insert_hook_(HookPoint, Callbacks) -> insert_hook(HookPoint, Callbacks) ->
ets:insert(?TAB, #hook{name = HookPoint, callbacks = Callbacks}), ok. ets:insert(?TAB, #hook{name = HookPoint, callbacks = Callbacks}), ok.
add_callback_(Callback, Callbacks) -> add_callback(C, Callbacks) ->
lists:keymerge(#callback.priority, Callbacks, [Callback]). add_callback(C, Callbacks, []).
del_callback_(Tag, Function, Callbacks) -> add_callback(C, [], Acc) ->
lists:filter( lists:reverse([C|Acc]);
fun(#callback{tag = Tag1, function = Func1}) -> add_callback(C1 = #callback{priority = P1}, [C2 = #callback{priority = P2}|More], Acc)
not ((Tag =:= Tag1) andalso (Function =:= Func1)) when P1 =< P2 ->
end, Callbacks). add_callback(C1, More, [C2|Acc]);
add_callback(C1, More, Acc) ->
contain_(_Tag, _Function, []) -> lists:append(lists:reverse(Acc), [C1 | More]).
false;
contain_(Tag, Function, [#callback{tag = Tag, function = Function}|_Callbacks]) ->
true;
contain_(Tag, Function, [_Callback | Callbacks]) ->
contain_(Tag, Function, Callbacks).