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

fix issue #206

This commit is contained in:
Feng Lee 2015-07-25 14:16:36 +08:00
parent c6c3926962
commit 98250313ab
1 changed files with 9 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
src/emqttd_bridge.erl
View File

@ -24,6 +24,7 @@
%%% %%%
%%% @end %%% @end
%%%----------------------------------------------------------------------------- %%%-----------------------------------------------------------------------------
-module(emqttd_bridge). -module(emqttd_bridge).
-author("Feng Lee <feng@emqtt.io>"). -author("Feng Lee <feng@emqtt.io>").
@ -48,14 +49,14 @@
topic_suffix = <<>>, topic_suffix = <<>>,
topic_prefix = <<>>, topic_prefix = <<>>,
mqueue :: emqttd_mqueue:mqueue(), mqueue :: emqttd_mqueue:mqueue(),
max_queue_len = 0, max_queue_len = 10000,
ping_down_interval = ?PING_DOWN_INTERVAL, ping_down_interval = ?PING_DOWN_INTERVAL,
status = up}). status = up}).
-type option() :: {max_queue_len, pos_integer()} | -type option() :: {qos, mqtt_qos()} |
{qos, mqtt_qos()} |
{topic_suffix, binary()} | {topic_suffix, binary()} |
{topic_prefix, binary()} | {topic_prefix, binary()} |
{max_queue_len, pos_integer()} |
{ping_down_interval, pos_integer()}. {ping_down_interval, pos_integer()}.
-export_type([option/0]). -export_type([option/0]).
@ -85,7 +86,7 @@ init([Node, SubTopic, Options]) ->
MQueue = emqttd_mqueue:new(qname(Node, SubTopic), MQueue = emqttd_mqueue:new(qname(Node, SubTopic),
[{max_len, State#state.max_queue_len}], [{max_len, State#state.max_queue_len}],
emqttd_alarm:alarm_fun()), emqttd_alarm:alarm_fun()),
emqttd_pubsub:subscribe({SubTopic, State#state.qos}), emqttd_pubsub:subscribe(SubTopic, State#state.qos),
{ok, State#state{mqueue = MQueue}}; {ok, State#state{mqueue = MQueue}};
false -> false ->
{stop, {cannot_connect, Node}} {stop, {cannot_connect, Node}}
@ -102,7 +103,9 @@ parse_opts([{topic_prefix, Prefix} | Opts], State) ->
parse_opts([{max_queue_len, Len} | Opts], State) -> parse_opts([{max_queue_len, Len} | Opts], State) ->
parse_opts(Opts, State#state{max_queue_len = Len}); parse_opts(Opts, State#state{max_queue_len = Len});
parse_opts([{ping_down_interval, Interval} | Opts], State) -> parse_opts([{ping_down_interval, Interval} | Opts], State) ->
parse_opts(Opts, State#state{ping_down_interval = Interval*1000}). parse_opts(Opts, State#state{ping_down_interval = Interval*1000});
parse_opts([_Opt | Opts], State) ->
parse_opts(Opts, State).
qname(Node, SubTopic) when is_atom(Node) -> qname(Node, SubTopic) when is_atom(Node) ->
qname(atom_to_list(Node), SubTopic); qname(atom_to_list(Node), SubTopic);