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

Merge pull request #1574 from emqtt/emq24 

Version 2.3.7
This commit is contained in:
huangdan 2018-04-21 23:19:53 +08:00 committed by GitHub
parent 1f842e4a19 f5b77f2e7b
commit 1da9e4397e
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 77 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
Makefile
View File

@ -9,7 +9,7 @@ dep_gproc = git https://github.com/uwiger/gproc
dep_getopt = git https://github.com/jcomellas/getopt v0.8.2
dep_lager = git https://github.com/basho/lager master
dep_esockd = git https://github.com/emqtt/esockd v5.2.1
dep_ekka = git https://github.com/emqtt/ekka v0.2.2
dep_ekka = git https://github.com/emqtt/ekka v0.2.3
dep_mochiweb = git https://github.com/emqtt/mochiweb v4.2.2
dep_pbkdf2 = git https://github.com/emqtt/pbkdf2 2.0.1
dep_lager_syslog = git https://github.com/basho/lager_syslog

6
etc/emq.conf
View File

@ -135,6 +135,12 @@ cluster.autoclean = 5m
## Value: String
## cluster.k8s.app_name = emq
## Kubernates Namespace
##
## Value: String
## cluster.k8s.namespace = default
##--------------------------------------------------------------------
## Node Args
##--------------------------------------------------------------------

9
priv/emq.schema
View File

@ -124,6 +124,10 @@
{datatype, string}
]}.
{mapping, "cluster.k8s.namespace", "ekka.cluster_discovery", [
{datatype, string}
]}.
{translation, "ekka.cluster_discovery", fun(Conf) ->
Strategy = cuttlefish:conf_get("cluster.discovery", Conf),
Filter = fun(Opts) -> [{K, V} || {K, V} <- Opts, V =/= undefined] end,
@ -152,7 +156,8 @@
[{apiserver, cuttlefish:conf_get("cluster.k8s.apiserver", Conf)},
{service_name, cuttlefish:conf_get("cluster.k8s.service_name", Conf)},
{address_type, cuttlefish:conf_get("cluster.k8s.address_type", Conf, ip)},
{app_name, cuttlefish:conf_get("cluster.k8s.app_name", Conf)}];
{app_name, cuttlefish:conf_get("cluster.k8s.app_name", Conf)},
{namespace, cuttlefish:conf_get("cluster.k8s.namespace", Conf)}];
(manual) ->
[ ]
end,
@ -434,7 +439,7 @@ end}.
ConsoleLogLevel = cuttlefish:conf_get("log.console.level", Conf),
ConsoleLogFile = cuttlefish:conf_get("log.console.file", Conf),
ConsoleHandler = {lager_console_backend, ConsoleLogLevel},
ConsoleHandler = {lager_console_backend, [ConsoleLogLevel]},
ConsoleFileHandler = {lager_file_backend, [{file, ConsoleLogFile},
{level, ConsoleLogLevel},
{size, cuttlefish:conf_get("log.console.size", Conf)},

2
src/emqttd.app.src
View File

@ -1,6 +1,6 @@
{application,emqttd,
[{description,"Erlang MQTT Broker"},
{vsn,"2.3.6"},
{vsn,"2.3.7"},
{modules,[]},
{registered,[emqttd_sup]},
{applications,[kernel,stdlib,gproc,lager,esockd,mochiweb,

12
src/emqttd_packet.erl
View File

@ -56,7 +56,7 @@ format_header(#mqtt_packet_header{type = Type,
dup = Dup,
qos = QoS,
retain = Retain}, S) ->
S1 = if
S1 = if
S == undefined -> <<>>;
true -> [", ", S]
end,
@ -78,13 +78,13 @@ format_variable(#mqtt_packet_connect{
clean_sess = CleanSess,
keep_alive = KeepAlive,
client_id = ClientId,
will_topic = WillTopic,
will_msg = WillMsg,
username = Username,
will_topic = WillTopic,
will_msg = WillMsg,
username = Username,
password = Password}) ->
Format = "ClientId=~s, ProtoName=~s, ProtoVsn=~p, CleanSess=~s, KeepAlive=~p, Username=~s, Password=~s",
Args = [ClientId, ProtoName, ProtoVer, CleanSess, KeepAlive, Username, format_password(Password)],
{Format1, Args1} = if
{Format1, Args1} = if
WillFlag -> { Format ++ ", Will(Q~p, R~p, Topic=~s, Msg=~s)",
Args ++ [WillQoS, i(WillRetain), WillTopic, WillMsg] };
true -> {Format, Args}
@ -93,7 +93,7 @@ format_variable(#mqtt_packet_connect{
format_variable(#mqtt_packet_connack{ack_flags = AckFlags,
return_code = ReturnCode}) ->
io_lib:format("AckFlags=~p, RetainCode=~p", [AckFlags, ReturnCode]);
io_lib:format("AckFlags=~p, ReturnCode=~p", [AckFlags, ReturnCode]);
format_variable(#mqtt_packet_publish{topic_name = TopicName,
packet_id = PacketId}) ->

26
src/emqttd_protocol.erl
View File

@ -137,7 +137,7 @@ session(#proto_state{session = Session}) ->
%% CONNECT – Client requests a connection to a Server
%% A Client can only send the CONNECT Packet once over a Network Connection.
%% A Client can only send the CONNECT Packet once over a Network Connection.
-spec(received(mqtt_packet(), proto_state()) -> {ok, proto_state()} | {error, term()}).
received(Packet = ?PACKET(?CONNECT),
State = #proto_state{connected = false, stats_data = Stats}) ->
@ -162,11 +162,12 @@ received(Packet = ?PACKET(Type), State = #proto_state{stats_data = Stats}) ->
subscribe(RawTopicTable, ProtoState = #proto_state{client_id = ClientId,
username = Username,
session = Session}) ->
session = Session,
mountpoint = MountPoint}) ->
TopicTable = parse_topic_table(RawTopicTable),
case emqttd_hooks:run('client.subscribe', [ClientId, Username], TopicTable) of
{ok, TopicTable1} ->
emqttd_session:subscribe(Session, TopicTable1);
emqttd_session:subscribe(Session, mount(MountPoint, TopicTable1));
{stop, _} ->
ok
end,
@ -174,10 +175,11 @@ subscribe(RawTopicTable, ProtoState = #proto_state{client_id = ClientId,
unsubscribe(RawTopics, ProtoState = #proto_state{client_id = ClientId,
username = Username,
session = Session}) ->
session = Session,
mountpoint = MountPoint}) ->
case emqttd_hooks:run('client.unsubscribe', [ClientId, Username], parse_topics(RawTopics)) of
{ok, TopicTable} ->
emqttd_session:unsubscribe(Session, TopicTable);
emqttd_session:unsubscribe(Session, mount(MountPoint, TopicTable));
{stop, _} ->
ok
end,
@ -228,7 +230,8 @@ process(?CONNECT_PACKET(Var), State0) ->
%% ACCEPT
{?CONNACK_ACCEPT, SP, State2#proto_state{session = Session, is_superuser = IsSuperuser}};
{error, Error} ->
{stop, {shutdown, Error}, State2}
?LOG(error, "Username '~s' login failed for ~p", [Username, Error], State2),
{?CONNACK_SERVER, false, State2}
end;
{error, Reason}->
?LOG(error, "Username '~s' login failed for ~p", [Username, Reason], State1),
@ -449,14 +452,14 @@ start_keepalive(Sec, #proto_state{keepalive_backoff = Backoff}) when Sec > 0 ->
validate_connect(Connect = #mqtt_packet_connect{}, ProtoState) ->
case validate_protocol(Connect) of
true ->
true ->
case validate_clientid(Connect, ProtoState) of
true ->
true ->
?CONNACK_ACCEPT;
false ->
false ->
?CONNACK_INVALID_ID
end;
false ->
false ->
?CONNACK_PROTO_VER
end.
@ -498,7 +501,7 @@ validate_packet(?SUBSCRIBE_PACKET(_PacketId, TopicTable)) ->
validate_packet(?UNSUBSCRIBE_PACKET(_PacketId, Topics)) ->
validate_topics(filter, Topics);
validate_packet(_Packet) ->
validate_packet(_Packet) ->
ok.
validate_topics(_Type, []) ->
@ -593,4 +596,3 @@ unmount(MountPoint, Msg = #mqtt_message{topic = Topic}) ->
{MountPoint, Topic0} -> Msg#mqtt_message{topic = Topic0};
_ -> Msg
end.

7
src/emqttd_session.erl
View File

@ -32,7 +32,7 @@
%%
%% QoS 1 and QoS 2 messages pending transmission to the Client.
%%
%% QoS 2 messages which have been received from the Client, but have not
%% QoS 2 messages which have been received from the Client, but have not
%% been completely acknowledged.
%%
%% Optionally, QoS 0 messages pending transmission to the Client.
@ -720,7 +720,7 @@ enqueue_msg(Msg, State = #state{mqueue = Q}) ->
%%--------------------------------------------------------------------
redeliver(Msg = #mqtt_message{qos = QoS}, State) ->
deliver(Msg#mqtt_message{dup = if QoS =:= ?QOS2 -> false; true -> true end}, State);
deliver(Msg#mqtt_message{dup = if QoS =:= ?QOS0 -> false; true -> true end}, State);
redeliver({pubrel, PacketId}, #state{client_pid = Pid}) ->
Pid ! {redeliver, {?PUBREL, PacketId}}.
@ -759,7 +759,7 @@ acked(pubrec, PacketId, State = #state{client_id = ClientId,
case Inflight:lookup(PacketId) of
{publish, Msg, _Ts} ->
emqttd_hooks:run('message.acked', [ClientId, Username], Msg),
State#state{inflight = Inflight:update(PacketId, {pubrel, PacketId, os:timestamp()})};
State#state{inflight = Inflight:update(PacketId, {pubrel, PacketId, os:timestamp()})};
{pubrel, PacketId, _Ts} ->
?LOG(warning, "Duplicated PUBREC Packet: ~p", [PacketId], State),
State
@ -853,4 +853,3 @@ shutdown(Reason, State) ->
gc(State) ->
emqttd_gc:maybe_force_gc(#state.force_gc_count, State).

21
src/emqttd_sm.erl
View File

@ -183,15 +183,16 @@ handle_cast(Msg, State) ->
handle_info({'DOWN', MRef, process, DownPid, _Reason}, State) ->
case dict:find(MRef, State#state.monitors) of
{ok, ClientId} ->
case mnesia:dirty_read({mqtt_session, ClientId}) of
[] ->
ok;
[Sess = #mqtt_session{sess_pid = DownPid}] ->
mnesia:dirty_delete_object(Sess);
[_Sess] ->
ok
end,
{noreply, erase_monitor(MRef, State), hibernate};
NewState =
case mnesia:dirty_read({mqtt_session, ClientId}) of
[] -> State;
[Sess = #mqtt_session{sess_pid = DownPid}] ->
mnesia:dirty_delete_object(Sess),
erase_monitor(MRef, State);
[_Sess] ->
State
end,
{noreply, NewState, hibernate};
error ->
lager:error("MRef of session ~p not found", [DownPid]),
{noreply, State}
@ -256,6 +257,7 @@ resume_session(Session = #mqtt_session{client_id = ClientId, sess_pid = SessPid}
{ok, SessPid};
false ->
?LOG(error, "Cannot resume ~p which seems already dead!", [SessPid], Session),
remove_session(Session),
{error, session_died}
end;
@ -305,4 +307,3 @@ monitor_session(ClientId, SessPid, State = #state{monitors = Monitors}) ->
erase_monitor(MRef, State = #state{monitors = Monitors}) ->
erlang:demonitor(MRef, [flush]),
State#state{monitors = dict:erase(MRef, Monitors)}.

2
src/emqttd_stats.erl
View File

@ -143,7 +143,7 @@ setstat(Stat, Val) ->
ets:update_element(?STATS_TAB, Stat, {2, Val}).
%% @doc Set stats with max
-spec(setstats(Stat :: atom(), MaxStat :: atom(), Val :: pos_integer()) -> boolean()).
-spec(setstats(Stat :: atom(), MaxStat :: atom(), Val :: pos_integer()) -> ok).
setstats(Stat, MaxStat, Val) ->
gen_server:cast(?MODULE, {setstats, Stat, MaxStat, Val}).

30
src/emqttd_vm.erl
View File

@ -156,9 +156,9 @@ microsecs() ->
(Mega * 1000000 + Sec) * 1000000 + Micro.
loads() ->
[{load1, ftos(cpu_sup:avg1()/256)},
{load5, ftos(cpu_sup:avg5()/256)},
{load15, ftos(cpu_sup:avg15()/256)}].
[{load1, ftos(avg1()/256)},
{load5, ftos(avg5()/256)},
{load15, ftos(avg15()/256)}].
get_system_info() ->
[{Key, format_system_info(Key, get_system_info(Key))} || Key <- ?SYSTEM_INFO].
@ -427,3 +427,27 @@ mapping([{owner, V}|Entries], Acc) when is_pid(V) ->
mapping([{Key, Value}|Entries], Acc) ->
mapping(Entries, [{Key, Value}|Acc]).
avg1() ->
case cpu_sup:avg1() of
SystemLoad when is_integer(SystemLoad) ->
SystemLoad;
{error, Reason} ->
lager:error("Get the average system load in the last minute fail for ~p~n", [Reason]),
0.00
end.
avg5() ->
case cpu_sup:avg5() of
SystemLoad when is_integer(SystemLoad) ->
SystemLoad;
{error, Reason} ->
lager:error("Get the average system load in the last 5 minutes fail for ~p~n", [Reason]),
0.00
end.
avg15() ->
case cpu_sup:avg15() of
SystemLoad when is_integer(SystemLoad) ->
SystemLoad;
{error, Reason} ->
lager:error("Get the average system load in the last 15 minutes fail for ~p~n", [Reason]),
0.00
end.