Add 'include/types.hrl' and introduce some common types

This commit is contained in:
Feng Lee 2019-02-19 11:07:44 +08:00 committed by Gilbert
parent 44d3eff094
commit 7a645dd9cc
24 changed files with 75 additions and 44 deletions

View File

@ -15,6 +15,7 @@
-module(emqx).
-include("emqx.hrl").
-include("types.hrl").
%% Start/Stop the application
-export([start/0, restart/1, is_running/1, stop/0]).
@ -167,8 +168,10 @@ reboot() ->
%%------------------------------------------------------------------------------
%% Internal functions
%%------------------------------------------------------------------------------
reload_config(ConfFile) ->
{ok, [Conf]} = file:consult(ConfFile),
lists:foreach(fun({App, Vals}) ->
[application:set_env(App, Par, Val) || {Par, Val} <- Vals]
end, Conf).

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
%% Mnesia bootstrap
-export([mnesia/1]).
@ -50,7 +51,7 @@ mnesia(copy) ->
ok = ekka_mnesia:copy_table(?TAB).
%% @doc Start the banned server.
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
-export([start_link/2]).
-export([subscribe/1, subscribe/2, subscribe/3]).
@ -53,7 +54,7 @@
%% Guards
-define(is_subid(Id), (is_binary(Id) orelse is_atom(Id))).
-spec(start_link(atom(), pos_integer()) -> emqx_types:startlink_ret()).
-spec(start_link(atom(), pos_integer()) -> startlink_ret()).
start_link(Pool, Id) ->
ok = create_tabs(),
gen_server:start_link({local, emqx_misc:proc_name(?BROKER, Id)},
@ -322,7 +323,7 @@ subscribed(SubId, Topic) when ?is_subid(SubId) ->
SubPid = emqx_broker_helper:lookup_subpid(SubId),
ets:member(?SUBOPTION, {SubPid, Topic}).
-spec(get_subopts(pid(), emqx_topic:topic()) -> emqx_types:subopts() | undefined).
-spec(get_subopts(pid(), emqx_topic:topic()) -> maybe(emqx_types:subopts())).
get_subopts(SubPid, Topic) when is_pid(SubPid), is_binary(Topic) ->
lookup_value(?SUBOPTION, {SubPid, Topic});
get_subopts(SubId, Topic) when ?is_subid(SubId) ->

View File

@ -17,6 +17,7 @@
-behaviour(gen_server).
-include("logger.hrl").
-include("types.hrl").
-export([start_link/0]).
-export([register_sub/2]).
@ -35,7 +36,7 @@
-define(BATCH_SIZE, 100000).
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?HELPER}, ?MODULE, [], []).
@ -50,7 +51,7 @@ register_sub(SubPid, SubId) when is_pid(SubPid) ->
error(subid_conflict)
end.
-spec(lookup_subid(pid()) -> emqx_types:subid() | undefined).
-spec(lookup_subid(pid()) -> maybe(emqx_types:subid())).
lookup_subid(SubPid) when is_pid(SubPid) ->
emqx_tables:lookup_value(?SUBMON, SubPid).

View File

@ -16,6 +16,7 @@
-behaviour(gen_statem).
-include("types.hrl").
-include("emqx_mqtt.hrl").
-export([start_link/0, start_link/1]).
@ -112,12 +113,12 @@
bridge_mode :: boolean(),
client_id :: binary(),
clean_start :: boolean(),
username :: binary() | undefined,
password :: binary() | undefined,
username :: maybe(binary()),
password :: maybe(binary()),
proto_ver :: emqx_mqtt_types:version(),
proto_name :: iodata(),
keepalive :: non_neg_integer(),
keepalive_timer :: reference() | undefined,
keepalive_timer :: maybe(reference()),
force_ping :: boolean(),
paused :: boolean(),
will_flag :: boolean(),

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
-export([start_link/0]).
@ -46,7 +47,7 @@
-define(BATCH_SIZE, 100000).
%% @doc Start the connection manager.
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?CM}, ?MODULE, [], []).
@ -121,7 +122,7 @@ set_conn_stats(ClientId, ConnPid, Stats) when is_binary(ClientId), is_pid(ConnPi
ets:insert(?CONN_STATS_TAB, {Conn, Stats}).
%% @doc Lookup connection pid.
-spec(lookup_conn_pid(emqx_types:client_id()) -> pid() | undefined).
-spec(lookup_conn_pid(emqx_types:client_id()) -> maybe(pid())).
lookup_conn_pid(ClientId) when is_binary(ClientId) ->
emqx_tables:lookup_value(?CONN_TAB, ClientId).

View File

@ -21,6 +21,8 @@
-module(emqx_gc).
-include("types.hrl").
-export([init/1, run/3, info/1, reset/1]).
-type(opts() :: #{count => integer(),
@ -35,7 +37,7 @@
-define(ENABLED(X), (is_integer(X) andalso X > 0)).
%% @doc Initialize force GC state.
-spec(init(opts() | false) -> gc_state() | undefined).
-spec(init(opts() | false) -> maybe(gc_state())).
init(#{count := Count, bytes := Bytes}) ->
Cnt = [{cnt, {Count, Count}} || ?ENABLED(Count)],
Oct = [{oct, {Bytes, Bytes}} || ?ENABLED(Bytes)],
@ -61,7 +63,7 @@ run([{K, N}|T], St) ->
end.
%% @doc Info of GC state.
-spec(info(gc_state()) -> map() | undefined).
-spec(info(gc_state()) -> maybe(map())).
info({?MODULE, St}) ->
St;
info(undefined) ->

View File

@ -17,6 +17,7 @@
-behaviour(gen_server).
-include("logger.hrl").
-include("types.hrl").
-export([start_link/0, stop/0]).
@ -42,7 +43,7 @@
-define(TAB, ?MODULE).
-define(SERVER, ?MODULE).
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], [{hibernate_after, 1000}]).

View File

@ -15,6 +15,7 @@
-module(emqx_metrics).
-include("logger.hrl").
-include("types.hrl").
-include("emqx_mqtt.hrl").
-export([start_link/0]).
@ -87,7 +88,7 @@
-define(SERVER, ?MODULE).
%% @doc Start the metrics server.
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

View File

@ -46,6 +46,7 @@
-module(emqx_mqueue).
-include("emqx.hrl").
-include("types.hrl").
-include("emqx_mqtt.hrl").
-export([init/1]).
@ -117,7 +118,7 @@ stats(#mqueue{max_len = MaxLen, dropped = Dropped} = MQ) ->
[{len, len(MQ)}, {max_len, MaxLen}, {dropped, Dropped}].
%% @doc Enqueue a message.
-spec(in(message(), mqueue()) -> {undefined | message(), mqueue()}).
-spec(in(message(), mqueue()) -> {maybe(message()), mqueue()}).
in(#message{qos = ?QOS_0}, MQ = #mqueue{store_qos0 = false}) ->
{_Dropped = undefined, MQ};
in(Msg = #message{topic = Topic}, MQ = #mqueue{default_p = Dp,

View File

@ -15,11 +15,13 @@
%% @doc The utility functions for erlang process dictionary.
-module(emqx_pd).
-include("types.hrl").
-export([update_counter/2, get_counter/1, reset_counter/1]).
-type(key() :: term()).
-spec(update_counter(key(), number()) -> undefined | number()).
-spec(update_counter(key(), number()) -> maybe(number())).
update_counter(Key, Inc) ->
put(Key, get_counter(Key) + Inc).

View File

@ -17,6 +17,7 @@
-behaviour(gen_server).
-include("logger.hrl").
-include("types.hrl").
-export([start_link/2]).
-export([submit/1, submit/2]).
@ -34,7 +35,7 @@
-type(task() :: fun() | mfa() | {fun(), Args :: list(any())}).
%% @doc Start pool.
-spec(start_link(atom(), pos_integer()) -> emqx_types:startlink_ret()).
-spec(start_link(atom(), pos_integer()) -> startlink_ret()).
start_link(Pool, Id) ->
gen_server:start_link({local, emqx_misc:proc_name(?MODULE, Id)},
?MODULE, [Pool, Id], [{hibernate_after, 1000}]).

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
-include_lib("ekka/include/ekka.hrl").
%% Mnesia bootstrap
@ -65,7 +66,7 @@ mnesia(copy) ->
%% Start a router
%%------------------------------------------------------------------------------
-spec(start_link(atom(), pos_integer()) -> emqx_types:startlink_ret()).
-spec(start_link(atom(), pos_integer()) -> startlink_ret()).
start_link(Pool, Id) ->
gen_server:start_link({local, emqx_misc:proc_name(?MODULE, Id)},
?MODULE, [Pool, Id], [{hibernate_after, 1000}]).

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
%% Mnesia bootstrap
-export([mnesia/1]).
@ -61,7 +62,7 @@ mnesia(copy) ->
%%------------------------------------------------------------------------------
%% @doc Starts the router helper
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

View File

@ -43,6 +43,7 @@
-include("emqx.hrl").
-include("emqx_mqtt.hrl").
-include("logger.hrl").
-include("types.hrl").
-export([start_link/1]).
-export([info/1, attrs/1]).
@ -77,7 +78,7 @@
client_id :: binary(),
%% Username
username :: binary() | undefined,
username :: maybe(binary()),
%% Connection pid binding with session
conn_pid :: pid(),
@ -107,7 +108,7 @@
retry_interval = 20000 :: timeout(),
%% Retry Timer
retry_timer :: reference() | undefined,
retry_timer :: maybe(reference()),
%% All QoS1, QoS2 messages published to when client is disconnected.
%% QoS 1 and QoS 2 messages pending transmission to the Client.
@ -125,19 +126,19 @@
await_rel_timeout = 20000 :: timeout(),
%% Awaiting PUBREL Timer
await_rel_timer :: reference() | undefined,
await_rel_timer :: maybe(reference()),
%% Session Expiry Interval
expiry_interval = 7200 :: timeout(),
%% Expired Timer
expiry_timer :: reference() | undefined,
expiry_timer :: maybe(reference()),
%% Enable Stats
enable_stats :: boolean(),
%% Stats timer
stats_timer :: reference() | undefined,
stats_timer :: maybe(reference()),
%% GC State
gc_state,
@ -147,7 +148,7 @@
will_msg :: emqx:message(),
will_delay_timer :: reference() | undefined
will_delay_timer :: maybe(reference())
}).

View File

@ -17,6 +17,7 @@
-behaviour(gen_server).
-include("logger.hrl").
-include("types.hrl").
-export([start_link/1]).
-export([start_session/1, count_sessions/0]).
@ -38,7 +39,7 @@
-define(BATCH_EXIT, 100000).
%% @doc Start session supervisor.
-spec(start_link(map()) -> emqx_types:startlink_ret()).
-spec(start_link(map()) -> startlink_ret()).
start_link(SessSpec) when is_map(SessSpec) ->
gen_server:start_link({local, ?SUP}, ?MODULE, [SessSpec], []).
@ -47,7 +48,7 @@ start_link(SessSpec) when is_map(SessSpec) ->
%%------------------------------------------------------------------------------
%% @doc Start a session.
-spec(start_session(map()) -> emqx_types:startlink_ret()).
-spec(start_session(map()) -> startlink_ret()).
start_session(SessAttrs) ->
gen_server:call(?SUP, {start_session, SessAttrs}, infinity).

View File

@ -19,6 +19,7 @@
-include("emqx.hrl").
-include("emqx_mqtt.hrl").
-include("logger.hrl").
-include("types.hrl").
%% Mnesia bootstrap
-export([mnesia/1]).
@ -70,7 +71,7 @@ mnesia(copy) ->
%% API
%%------------------------------------------------------------------------------
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
-export([start_link/0]).
@ -55,7 +56,7 @@
-define(BATCH_SIZE, 100000).
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?SM}, ?MODULE, [], []).

View File

@ -15,13 +15,14 @@
-module(emqx_sm_locker).
-include("emqx.hrl").
-include("types.hrl").
-export([start_link/0]).
-export([trans/2, trans/3]).
-export([lock/1, lock/2, unlock/1]).
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
ekka_locker:start_link(?MODULE).
@ -29,7 +30,7 @@ start_link() ->
trans(ClientId, Fun) ->
trans(ClientId, Fun, undefined).
-spec(trans(emqx_types:client_id() | undefined,
-spec(trans(maybe(emqx_types:client_id()),
fun(([node()])-> any()), ekka_locker:piggyback()) -> any()).
trans(undefined, Fun, _Piggyback) ->
Fun([]);

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
-export([start_link/0]).
-export([is_enabled/0]).
@ -36,7 +37,7 @@
-type(session_pid() :: pid()).
%% @doc Start the global session manager.
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?REGISTRY}, ?MODULE, [], []).

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
-export([start_link/0, start_link/1, stop/0]).
@ -82,11 +83,11 @@
-type opts() :: #{tick_ms := timeout()}.
%% @doc Start stats server
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
start_link(#{tick_ms => timer:seconds(1)}).
-spec(start_link(opts()) -> emqx_types:startlink_ret()).
-spec(start_link(opts()) -> startlink_ret()).
start_link(Opts) ->
gen_server:start_link({local, ?SERVER}, ?MODULE, Opts, []).
@ -112,7 +113,7 @@ getstats() ->
end.
%% @doc Get stats by name
-spec(getstat(atom()) -> non_neg_integer() | undefined).
-spec(getstat(atom()) -> maybe(non_neg_integer())).
getstat(Name) ->
case ets:lookup(?TAB, Name) of
[{Name, Val}] -> Val;

View File

@ -17,6 +17,7 @@
-behavior(gen_server).
-include("logger.hrl").
-include("types.hrl").
-export([start_link/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
@ -24,10 +25,16 @@
%% compress unused warning
-export([procinfo/1]).
-type(option() :: {long_gc, false | pos_integer()}
| {long_schedule, false | pos_integer()}
| {large_heap, pos_integer()}
| {busy_port, boolean()}
| {busy_dist_port, boolean()}).
-define(SYSMON, ?MODULE).
%% @doc Start system monitor
-spec(start_link(Opts :: list(tuple())) -> emqx_types:startlink_ret()).
-spec(start_link(list(option())) -> startlink_ret()).
start_link(Opts) ->
gen_server:start_link({local, ?SYSMON}, ?MODULE, [Opts], []).

View File

@ -15,9 +15,9 @@
-module(emqx_types).
-include("emqx.hrl").
-include("types.hrl").
-export_type([zone/0]).
-export_type([startlink_ret/0, ok_or_error/1]).
-export_type([pubsub/0, topic/0, subid/0, subopts/0]).
-export_type([client_id/0, username/0, password/0, peername/0, protocol/0]).
-export_type([credentials/0, session/0]).
@ -28,8 +28,6 @@
-export_type([alarm/0, plugin/0, command/0]).
-type(zone() :: atom()).
-type(startlink_ret() :: {ok, pid()} | ignore | {error, term()}).
-type(ok_or_error(Reason) :: ok | {error, Reason}).
-type(pubsub() :: publish | subscribe).
-type(topic() :: binary()).
-type(subid() :: binary() | atom()).
@ -39,8 +37,8 @@
}).
-type(session() :: #session{}).
-type(client_id() :: binary() | atom()).
-type(username() :: binary() | undefined).
-type(password() :: binary() | undefined).
-type(username() :: maybe(binary())).
-type(password() :: maybe(binary())).
-type(peername() :: {inet:ip_address(), inet:port_number()}).
-type(protocol() :: mqtt | 'mqtt-sn' | coap | stomp | none | atom()).
-type(credentials() :: #{client_id := client_id(),

View File

@ -18,6 +18,7 @@
-include("emqx.hrl").
-include("logger.hrl").
-include("types.hrl").
-export([start_link/0]).
-export([get_env/2, get_env/3]).
@ -33,17 +34,17 @@
-define(TAB, ?MODULE).
-define(SERVER, ?MODULE).
-spec(start_link() -> emqx_types:startlink_ret()).
-spec(start_link() -> startlink_ret()).
start_link() ->
gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
-spec(get_env(emqx_types:zone() | undefined, atom()) -> undefined | term()).
-spec(get_env(maybe(emqx_types:zone()), atom()) -> maybe(term())).
get_env(undefined, Key) ->
emqx_config:get_env(Key);
get_env(Zone, Key) ->
get_env(Zone, Key, undefined).
-spec(get_env(emqx_types:zone() | undefined, atom(), term()) -> undefined | term()).
-spec(get_env(maybe(emqx_types:zone()), atom(), term()) -> maybe(term())).
get_env(undefined, Key, Def) ->
emqx_config:get_env(Key, Def);
get_env(Zone, Key, Def) ->