This commit is contained in:
Feng Lee 2015-05-02 00:04:58 +08:00
parent 2d600711c9
commit ac3a17e0e7
8 changed files with 9 additions and 10 deletions

View File

@ -152,7 +152,7 @@ stop() ->
%%%============================================================================= %%%=============================================================================
init([]) -> init([]) ->
{ok, AcOpts} = application:get_env(access_control), {ok, AcOpts} = application:get_env(access),
ets:new(?ACCESS_CONTROL_TAB, [set, named_table, protected, {read_concurrency, true}]), ets:new(?ACCESS_CONTROL_TAB, [set, named_table, protected, {read_concurrency, true}]),
ets:insert(?ACCESS_CONTROL_TAB, {auth_modules, init_mods(auth, proplists:get_value(auth, AcOpts))}), ets:insert(?ACCESS_CONTROL_TAB, {auth_modules, init_mods(auth, proplists:get_value(auth, AcOpts))}),
ets:insert(?ACCESS_CONTROL_TAB, {acl_modules, init_mods(acl, proplists:get_value(acl, AcOpts))}), ets:insert(?ACCESS_CONTROL_TAB, {acl_modules, init_mods(acl, proplists:get_value(acl, AcOpts))}),

View File

@ -64,8 +64,7 @@ start_bridge(Node, SubTopic) when is_atom(Node) and is_binary(SubTopic) ->
-spec start_bridge(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, any()}. -spec start_bridge(atom(), binary(), [emqttd_bridge:option()]) -> {ok, pid()} | {error, any()}.
start_bridge(Node, SubTopic, Options) when is_atom(Node) and is_binary(SubTopic) -> start_bridge(Node, SubTopic, Options) when is_atom(Node) and is_binary(SubTopic) ->
{ok, Env} = application:get_env(emqttd, bridge), Options1 = emqttd_opts:merge(emqttd_broker:env(bridge), Options),
Options1 = emqttd_opts:merge(Env, Options),
supervisor:start_child(?MODULE, bridge_spec(Node, SubTopic, Options1)). supervisor:start_child(?MODULE, bridge_spec(Node, SubTopic, Options1)).
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------

View File

@ -90,7 +90,7 @@ notify(EventType, Event) ->
%% @end %% @end
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
env(Name) -> env(Name) ->
proplists:get_value(Name, application:get_env(emqttd, mqtt_broker, [])). proplists:get_value(Name, application:get_env(emqttd, broker, [])).
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
%% @doc Get broker version %% @doc Get broker version

View File

@ -92,7 +92,7 @@ limit(payload) ->
env() -> env() ->
case get({env, retained}) of case get({env, retained}) of
undefined -> undefined ->
{ok, Env} = application:get_env(emqttd, retained), Env = emqttd_broker:env(retained),
put({env, retained}, Env), Env; put({env, retained}, Env), Env;
Env -> Env ->
Env Env

View File

@ -30,10 +30,10 @@
-export([merge/2]). -export([merge/2]).
%%%----------------------------------------------------------------------------- %%------------------------------------------------------------------------------
%% @doc Merge Options %% @doc Merge Options
%% @end %% @end
%%%----------------------------------------------------------------------------- %%------------------------------------------------------------------------------
merge(Defaults, Options) -> merge(Defaults, Options) ->
lists:foldl( lists:foldl(
fun({Opt, Val}, Acc) -> fun({Opt, Val}, Acc) ->

View File

@ -42,7 +42,7 @@ start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []). supervisor:start_link({local, ?MODULE}, ?MODULE, []).
init([]) -> init([]) ->
{ok, Opts} = application:get_env(mqtt_pubsub), Opts = emqttd_broker:env(pubsub),
Schedulers = erlang:system_info(schedulers), Schedulers = erlang:system_info(schedulers),
PoolSize = proplists:get_value(pool_size, Opts, Schedulers), PoolSize = proplists:get_value(pool_size, Opts, Schedulers),
gproc_pool:new(pubsub, hash, [{size, PoolSize}]), gproc_pool:new(pubsub, hash, [{size, PoolSize}]),

View File

@ -258,7 +258,7 @@ init([ClientId, ClientPid]) ->
process_flag(trap_exit, true), process_flag(trap_exit, true),
%%TODO: Is this OK? or should monitor... %%TODO: Is this OK? or should monitor...
true = link(ClientPid), true = link(ClientPid),
{ok, SessOpts} = application:get_env(mqtt_session), SessOpts = emqttd:env(session),
State = initial_state(ClientId, ClientPid), State = initial_state(ClientId, ClientPid),
Expires = proplists:get_value(expires, SessOpts, 1) * 3600, Expires = proplists:get_value(expires, SessOpts, 1) * 3600,
MsgQueue = emqttd_queue:new(proplists:get_value(max_queue, SessOpts, 1000), MsgQueue = emqttd_queue:new(proplists:get_value(max_queue, SessOpts, 1000),

View File

@ -56,8 +56,8 @@
%% @end %% @end
%%------------------------------------------------------------------------------ %%------------------------------------------------------------------------------
start_link(Req) -> start_link(Req) ->
PktOpts = emqttd:env(packet),
{ReentryWs, ReplyChannel} = upgrade(Req), {ReentryWs, ReplyChannel} = upgrade(Req),
{ok, PktOpts} = application:get_env(emqttd, mqtt_packet),
{ok, ClientPid} = gen_server:start_link(?MODULE, [self(), Req, ReplyChannel, PktOpts], []), {ok, ClientPid} = gen_server:start_link(?MODULE, [self(), Req, ReplyChannel, PktOpts], []),
ReentryWs(#wsocket_state{request = Req, ReentryWs(#wsocket_state{request = Req,
client_pid = ClientPid, client_pid = ClientPid,