Merge branch 'emqx30' of github.com:emqtt/emqttd into emqx30

src/emqx_bridge.erl

@@ -19,76 +19,77 @@
 -include("emqx.hrl").
 -include("emqx_mqtt.hrl").
 
--export([start_link/5]).
+-import(proplists, [get_value/2, get_value/3]).
+
+-export([start_link/2, start_bridge/1, stop_bridge/1, status/1]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
          code_change/3]).
 
--define(PING_DOWN_INTERVAL, 1000).
+-record(state, {client_pid, options, reconnect_time, reconnect_count,
+                def_reconnect_count, type, mountpoint, queue, store_type,
+                max_pending_messages}).
 
--record(state, {pool, id,
+-record(mqtt_msg, {qos = ?QOS0, retain = false, dup = false,
-                node, subtopic,
+                   packet_id, topic, props, payload}).
-                qos                = ?QOS_0,
-                topic_suffix       = <<>>,
-                topic_prefix       = <<>>,
-                mqueue            :: emqx_mqueue:mqueue(),
-                max_queue_len      = 10000,
-                ping_down_interval = ?PING_DOWN_INTERVAL,
-                status             = up}).
 
--type(option() :: {qos, emqx_mqtt_types:qos()} |
+start_link(Name, Options) ->
-                  {topic_suffix, binary()} |
+    gen_server:start_link({local, name(Name)}, ?MODULE, [Options], []).
-                  {topic_prefix, binary()} |
-                  {max_queue_len, pos_integer()} |
-                  {ping_down_interval, pos_integer()}).
 
--export_type([option/0]).
+start_bridge(Name) ->
+    gen_server:call(name(Name), start_bridge).
 
-%% @doc Start a bridge
+stop_bridge(Name) ->
--spec(start_link(term(), pos_integer(), atom(), binary(), [option()])
+    gen_server:call(name(Name), stop_bridge).
-      -> {ok, pid()} | ignore | {error, term()}).
+
-start_link(Pool, Id, Node, Topic, Options) ->
+status(Pid) ->
-    gen_server:start_link(?MODULE, [Pool, Id, Node, Topic, Options], [{hibernate_after, 5000}]).
+    gen_server:call(Pid, status).
 
 %%------------------------------------------------------------------------------
 %% gen_server callbacks
 %%------------------------------------------------------------------------------
 
-init([Pool, Id, Node, Topic, Options]) ->
+init([Options]) ->
     process_flag(trap_exit, true),
-    true = gproc_pool:connect_worker(Pool, {Pool, Id}),
+    case get_value(start_type, Options, manual) of
-    case net_kernel:connect_node(Node) of
+        manual -> ok;
-        true ->
+        auto -> erlang:send_after(1000, self(), start)
-            true = erlang:monitor_node(Node, true),
+    end,
-            Share = iolist_to_binary(["$bridge:", atom_to_list(Node), ":", Topic]),
+    ReconnectCount = get_value(reconnect_count, Options, 10),
-            emqx_broker:subscribe(Topic, self(), [{share, Share}, {qos, ?QOS_0}]),
+    ReconnectTime = get_value(reconnect_time, Options, 30000),
-            State = parse_opts(Options, #state{node = Node, subtopic = Topic}),
+    MaxPendingMsg = get_value(max_pending_messages, Options, 10000),
-            %%TODO: queue....
+    Mountpoint = format_mountpoint(get_value(mountpoint, Options)),
-            MQueue = emqx_mqueue:new(qname(Node, Topic), [{max_len, State#state.max_queue_len}]),
+    StoreType = get_value(store_type, Options, memory),
-            {ok, State#state{pool = Pool, id = Id, mqueue = MQueue}};
+    Type = get_value(type, Options, in),
-        false ->
+    Queue = [],
-            {stop, {cannot_connect_node, Node}}
+    {ok, #state{type                = Type,
-    end.
+                mountpoint          = Mountpoint,
+                queue               = Queue,
+                store_type          = StoreType,
+                options             = Options,
+                reconnect_count     = ReconnectCount,
+                reconnect_time      = ReconnectTime,
+                def_reconnect_count = ReconnectCount,
+                max_pending_messages = MaxPendingMsg}}.
 
-parse_opts([], State) ->
+handle_call(start_bridge, _From, State = #state{client_pid = undefined}) ->
-    State;
+    {noreply, NewState} = handle_info(start, State),
-parse_opts([{qos, QoS} | Opts], State) ->
+    {reply, <<"start bridge successfully">>, NewState};
-    parse_opts(Opts, State#state{qos = QoS});
-parse_opts([{topic_suffix, Suffix} | Opts], State) ->
-    parse_opts(Opts, State#state{topic_suffix= Suffix});
-parse_opts([{topic_prefix, Prefix} | Opts], State) ->
-    parse_opts(Opts, State#state{topic_prefix = Prefix});
-parse_opts([{max_queue_len, Len} | Opts], State) ->
-    parse_opts(Opts, State#state{max_queue_len = Len});
-parse_opts([{ping_down_interval, Interval} | Opts], State) ->
-    parse_opts(Opts, State#state{ping_down_interval = Interval});
-parse_opts([_Opt | Opts], State) ->
-    parse_opts(Opts, State).
 
-qname(Node, Topic) when is_atom(Node) ->
+handle_call(start_bridge, _From, State) ->
-    qname(atom_to_list(Node), Topic);
+    {reply, <<"bridge already started">>, State};
-qname(Node, Topic) ->
+
-    iolist_to_binary(["Bridge:", Node, ":", Topic]).
+handle_call(stop_bridge, _From, State = #state{client_pid = undefined}) ->
+    {reply, <<"bridge not started">>, State};
+
+handle_call(stop_bridge, _From, State = #state{client_pid = Pid}) ->
+    emqx_client:disconnect(Pid),
+    {reply, <<"stop bridge successfully">>, State};
+
+handle_call(status, _From, State = #state{client_pid = undefined}) ->
+    {reply, <<"Stopped">>, State};
+handle_call(status, _From, State = #state{client_pid = _Pid})->
+    {reply, <<"Running">>, State};
 
 handle_call(Req, _From, State) ->
     emqx_logger:error("[Bridge] unexpected call: ~p", [Req]),
@@ -98,65 +99,156 @@ handle_cast(Msg, State) ->
     emqx_logger:error("[Bridge] unexpected cast: ~p", [Msg]),
     {noreply, State}.
 
-handle_info({dispatch, _Topic, Msg}, State = #state{mqueue = Q, status = down}) ->
+handle_info(start, State = #state{reconnect_count = 0}) ->
-    %% TODO: how to drop???
-    {noreply, State#state{mqueue = emqx_mqueue:in(Msg, Q)}};
-
-handle_info({dispatch, _Topic, Msg}, State = #state{node = Node, status = up}) ->
-    ok = emqx_rpc:cast(Node, emqx_broker, publish, [transform(Msg, State)]),
     {noreply, State};
 
-handle_info({nodedown, Node}, State = #state{node = Node, ping_down_interval = Interval}) ->
+%%----------------------------------------------------------------
-    emqx_logger:warning("[Bridge] node down: ~s", [Node]),
+%% start in message bridge
-    erlang:send_after(Interval, self(), ping_down_node),
+%%----------------------------------------------------------------
-    {noreply, State#state{status = down}, hibernate};
+handle_info(start, State = #state{options = Options,
-
+                                  client_pid = undefined,
-handle_info({nodeup, Node}, State = #state{node = Node}) ->
+                                  reconnect_time = ReconnectTime,
-    %% TODO: Really fast??
+                                  reconnect_count = ReconnectCount,
-    case emqx:is_running(Node) of
+                                  type = in}) ->
-        true -> emqx_logger:warning("[Bridge] Node up: ~s", [Node]),
+    case emqx_client:start_link([{owner, self()}|options(Options)]) of
-                {noreply, dequeue(State#state{status = up})};
+        {ok, ClientPid, _} ->
-        false -> self() ! {nodedown, Node},
+            Subs = get_value(subscriptions, Options, []),
-                 {noreply, State#state{status = down}}
+            [emqx_client:subscribe(ClientPid, {i2b(Topic), Qos}) || {Topic, Qos} <- Subs],
+            {noreply, State#state{client_pid = ClientPid}};
+        {error,_} ->
+            erlang:send_after(ReconnectTime, self(), start),
+            {noreply, State = #state{reconnect_count = ReconnectCount-1}}
     end;
 
-handle_info(ping_down_node, State = #state{node = Node, ping_down_interval = Interval}) ->
+%%----------------------------------------------------------------
-    Self = self(),
+%% start out message bridge
-    spawn_link(fun() ->
+%%----------------------------------------------------------------
-                 case net_kernel:connect_node(Node) of
+handle_info(start, State = #state{options = Options,
-                     true -> Self ! {nodeup, Node};
+                                  client_pid = undefined,
-                     false -> erlang:send_after(Interval, Self, ping_down_node)
+                                  reconnect_time = ReconnectTime,
-                 end
+                                  reconnect_count = ReconnectCount,
-               end),
+                                  type = out}) ->
+    case emqx_client:start_link([{owner, self()}|options(Options)]) of
+        {ok, ClientPid, _} ->
+            Subs = get_value(subscriptions, Options, []),
+            [emqx_client:subscribe(ClientPid, {i2b(Topic), Qos}) || {Topic, Qos} <- Subs],
+            ForwardRules = string:tokens(get_value(forward_rule, Options, ""), ","),
+            [emqx_broker:subscribe(i2b(Topic)) || Topic <- ForwardRules, emqx_topic:validate({filter, i2b(Topic)})],
+            {noreply, State#state{client_pid = ClientPid}};
+        {error,_} ->
+            erlang:send_after(ReconnectTime, self(), start),
+            {noreply, State = #state{reconnect_count = ReconnectCount-1}}
+    end;
+
+%%----------------------------------------------------------------
+%% received local node message
+%%----------------------------------------------------------------
+handle_info({dispatch, _, #message{topic = Topic, payload = Payload, flags = #{retain := Retain}}},
+             State = #state{client_pid = Pid, mountpoint = Mountpoint, queue = Queue,
+                            store_type = StoreType, max_pending_messages = MaxPendingMsg}) ->
+    Msg = #mqtt_msg{qos     = 1,
+                    retain  = Retain,
+                    topic   = mountpoint(Mountpoint, Topic),
+                    payload = Payload},
+    case emqx_client:publish(Pid, Msg) of
+        {ok, PkgId} ->
+            {noreply, State#state{queue = store(StoreType, {PkgId, Msg}, Queue, MaxPendingMsg)}};
+        {error, Reason} ->
+            emqx_logger:error("Publish fail:~p", [Reason]),
+            {noreply, State}
+    end;
+
+%%----------------------------------------------------------------
+%% received remote node message
+%%----------------------------------------------------------------
+handle_info({publish, #{qos := QoS, dup := Dup, retain := Retain, topic := Topic,
+                        properties := Props, payload := Payload}}, State) ->
+    NewMsg0 = emqx_message:make(bridge, QoS, Topic, Payload),
+    NewMsg1 = emqx_message:set_headers(Props, emqx_message:set_flags(#{dup => Dup, retain=> Retain}, NewMsg0)),
+    emqx_broker:publish(NewMsg1),
     {noreply, State};
 
-handle_info({'EXIT', _Pid, normal}, State) ->
+%%----------------------------------------------------------------
-    {noreply, State};
+%% received remote puback message
+%%----------------------------------------------------------------
+handle_info({puback, #{packet_id := PkgId}}, State = #state{queue = Queue, store_type = StoreType}) ->
+    % lists:keydelete(PkgId, 1, Queue)
+    {noreply, State#state{queue = delete(StoreType, PkgId, Queue)}};
+
+handle_info({'EXIT', Pid, normal}, State = #state{client_pid = Pid}) ->
+    {noreply, State#state{client_pid = undefined}};
+
+handle_info({'EXIT', Pid, Reason}, State = #state{client_pid = Pid,
+                                                  reconnect_time = ReconnectTime,
+                                                  def_reconnect_count = DefReconnectCount}) ->
+    lager:warning("emqx bridge stop reason:~p", [Reason]),
+    erlang:send_after(ReconnectTime, self(), start),
+    {noreply, State#state{client_pid = undefined, reconnect_count = DefReconnectCount}};
 
 handle_info(Info, State) ->
     emqx_logger:error("[Bridge] unexpected info: ~p", [Info]),
     {noreply, State}.
 
-terminate(_Reason, #state{pool = Pool, id = Id}) ->
+terminate(_Reason, #state{}) ->
-    gproc_pool:disconnect_worker(Pool, {Pool, Id}).
+    ok.
 
 code_change(_OldVsn, State, _Extra) ->
     {ok, State}.
 
-%%--------------------------------------------------------------------
+proto_ver(mqtt3) -> v3;
-%% Internal functions
+proto_ver(mqtt4) -> v4;
-%%--------------------------------------------------------------------
+proto_ver(mqtt5) -> v5.
-
+address(Address) ->
-dequeue(State = #state{mqueue = MQ}) ->
+    case string:tokens(Address, ":") of
-    case emqx_mqueue:out(MQ) of
+        [Host] -> {Host, 1883};
-        {empty, MQ1} ->
+        [Host, Port] -> {Host, list_to_integer(Port)}
-            State#state{mqueue = MQ1};
-        {{value, Msg}, MQ1} ->
-            handle_info({dispatch, Msg#message.topic, Msg}, State),
-            dequeue(State#state{mqueue = MQ1})
     end.
+options(Options) ->
+    options(Options, []).
+options([], Acc) ->
+    Acc;
+options([{username, Username}| Options], Acc) ->
+    options(Options, [{username, Username}|Acc]);
+options([{proto_ver, ProtoVer}| Options], Acc) ->
+    options(Options, [{proto_ver, proto_ver(ProtoVer)}|Acc]);
+options([{password, Password}| Options], Acc) ->
+    options(Options, [{password, Password}|Acc]);
+options([{keepalive, Keepalive}| Options], Acc) ->
+    options(Options, [{keepalive, Keepalive}|Acc]);
+options([{client_id, ClientId}| Options], Acc) ->
+    options(Options, [{client_id, ClientId}|Acc]);
+options([{clean_start, CleanStart}| Options], Acc) ->
+    options(Options, [{clean_start, CleanStart}|Acc]);
+options([{address, Address}| Options], Acc) ->
+    {Host, Port} = address(Address),
+    options(Options, [{host, Host}, {port, Port}|Acc]);
+options([_Option | Options], Acc) ->
+    options(Options, Acc).
 
-transform(Msg = #message{topic = Topic}, #state{topic_prefix = Prefix,
+name(Id) ->
-                                                topic_suffix = Suffix}) ->
+    list_to_atom(lists:concat([?MODULE, "_", Id])).
-    Msg#message{topic = <<Prefix/binary, Topic/binary, Suffix/binary>>}.
 
+i2b(L) -> iolist_to_binary(L).
+
+mountpoint(undefined, Topic) ->
+    Topic;
+mountpoint(Prefix, Topic) ->
+    <<Prefix/binary, Topic/binary>>.
+
+format_mountpoint(undefined) ->
+    undefined;
+format_mountpoint(Prefix) ->
+    binary:replace(i2b(Prefix), <<"${node}">>, atom_to_binary(node(), utf8)).
+
+store(memory, Data, Queue, MaxPendingMsg) when length(Queue) =< MaxPendingMsg ->
+    [Data | Queue];
+store(memory, _Data, Queue, _MaxPendingMsg) ->
+    lager:error("Beyond max pending messages"),
+    Queue;
+store(disk, Data, Queue, _MaxPendingMsg)->
+    [Data | Queue].
+
+delete(memory, PkgId, Queue) ->
+    lists:keydelete(PkgId, 1, Queue);
+delete(disk, PkgId, Queue) ->
+    lists:keydelete(PkgId, 1, Queue).

src/emqx_bridge1.erl

@@ -1,254 +0,0 @@
-%% Copyright (c) 2018 EMQ Technologies Co., Ltd. All Rights Reserved.
-%%
-%% Licensed under the Apache License, Version 2.0 (the "License");
-%% you may not use this file except in compliance with the License.
-%% You may obtain a copy of the License at
-%%
-%%     http://www.apache.org/licenses/LICENSE-2.0
-%%
-%% Unless required by applicable law or agreed to in writing, software
-%% distributed under the License is distributed on an "AS IS" BASIS,
-%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-%% See the License for the specific language governing permissions and
-%% limitations under the License.
-
--module(emqx_bridge1).
-
--behaviour(gen_server).
-
--include("emqx.hrl").
--include("emqx_mqtt.hrl").
-
--import(proplists, [get_value/2, get_value/3]).
-
--export([start_link/2, start_bridge/1, stop_bridge/1, status/1]).
-
--export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
-         code_change/3]).
-
--record(state, {client_pid, options, reconnect_time, reconnect_count,
-                def_reconnect_count, type, mountpoint, queue, store_type,
-                max_pending_messages}).
-
--record(mqtt_msg, {qos = ?QOS0, retain = false, dup = false,
-                   packet_id, topic, props, payload}).
-
-start_link(Name, Options) ->
-    gen_server:start_link({local, name(Name)}, ?MODULE, [Options], []).
-
-start_bridge(Name) ->
-    gen_server:call(name(Name), start_bridge).
-
-stop_bridge(Name) ->
-    gen_server:call(name(Name), stop_bridge).
-
-status(Pid) ->
-    gen_server:call(Pid, status).
-
-%%------------------------------------------------------------------------------
-%% gen_server callbacks
-%%------------------------------------------------------------------------------
-
-init([Options]) ->
-    process_flag(trap_exit, true),
-    case get_value(start_type, Options, manual) of
-        manual -> ok;
-        auto -> erlang:send_after(1000, self(), start)
-    end,
-    ReconnectCount = get_value(reconnect_count, Options, 10),
-    ReconnectTime = get_value(reconnect_time, Options, 30000),
-    MaxPendingMsg = get_value(max_pending_messages, Options, 10000),
-    Mountpoint = format_mountpoint(get_value(mountpoint, Options)),
-    StoreType = get_value(store_type, Options, memory),
-    Type = get_value(type, Options, in),
-    Queue = [],
-    {ok, #state{type                = Type,
-                mountpoint          = Mountpoint,
-                queue               = Queue,
-                store_type          = StoreType,
-                options             = Options,
-                reconnect_count     = ReconnectCount,
-                reconnect_time      = ReconnectTime,
-                def_reconnect_count = ReconnectCount,
-                max_pending_messages = MaxPendingMsg}}.
-
-handle_call(start_bridge, _From, State = #state{client_pid = undefined}) ->
-    {noreply, NewState} = handle_info(start, State),
-    {reply, <<"start bridge successfully">>, NewState};
-
-handle_call(start_bridge, _From, State) ->
-    {reply, <<"bridge already started">>, State};
-
-handle_call(stop_bridge, _From, State = #state{client_pid = undefined}) ->
-    {reply, <<"bridge not started">>, State};
-
-handle_call(stop_bridge, _From, State = #state{client_pid = Pid}) ->
-    emqx_client:disconnect(Pid),
-    {reply, <<"stop bridge successfully">>, State};
-
-handle_call(status, _From, State = #state{client_pid = undefined}) ->
-    {reply, <<"Stopped">>, State};
-handle_call(status, _From, State = #state{client_pid = _Pid})->
-    {reply, <<"Running">>, State};
-
-handle_call(Req, _From, State) ->
-    emqx_logger:error("[Bridge] unexpected call: ~p", [Req]),
-    {reply, ignored, State}.
-
-handle_cast(Msg, State) ->
-    emqx_logger:error("[Bridge] unexpected cast: ~p", [Msg]),
-    {noreply, State}.
-
-handle_info(start, State = #state{reconnect_count = 0}) ->
-    {noreply, State};
-
-%%----------------------------------------------------------------
-%% start in message bridge
-%%----------------------------------------------------------------
-handle_info(start, State = #state{options = Options,
-                                  client_pid = undefined,
-                                  reconnect_time = ReconnectTime,
-                                  reconnect_count = ReconnectCount,
-                                  type = in}) ->
-    case emqx_client:start_link([{owner, self()}|options(Options)]) of
-        {ok, ClientPid, _} ->
-            Subs = get_value(subscriptions, Options, []),
-            [emqx_client:subscribe(ClientPid, {i2b(Topic), Qos}) || {Topic, Qos} <- Subs],
-            {noreply, State#state{client_pid = ClientPid}};
-        {error,_} ->
-            erlang:send_after(ReconnectTime, self(), start),
-            {noreply, State = #state{reconnect_count = ReconnectCount-1}}
-    end;
-
-%%----------------------------------------------------------------
-%% start out message bridge
-%%----------------------------------------------------------------
-handle_info(start, State = #state{options = Options,
-                                  client_pid = undefined,
-                                  reconnect_time = ReconnectTime,
-                                  reconnect_count = ReconnectCount,
-                                  type = out}) ->
-    case emqx_client:start_link([{owner, self()}|options(Options)]) of
-        {ok, ClientPid, _} ->
-            Subs = get_value(subscriptions, Options, []),
-            [emqx_client:subscribe(ClientPid, {i2b(Topic), Qos}) || {Topic, Qos} <- Subs],
-            ForwardRules = string:tokens(get_value(forward_rule, Options, ""), ","),
-            [emqx_broker:subscribe(i2b(Topic)) || Topic <- ForwardRules, emqx_topic:validate({filter, i2b(Topic)})],
-            {noreply, State#state{client_pid = ClientPid}};
-        {error,_} ->
-            erlang:send_after(ReconnectTime, self(), start),
-            {noreply, State = #state{reconnect_count = ReconnectCount-1}}
-    end;
-
-%%----------------------------------------------------------------
-%% received local node message
-%%----------------------------------------------------------------
-handle_info({dispatch, _, #message{topic = Topic, payload = Payload, flags = #{retain := Retain}}},
-             State = #state{client_pid = Pid, mountpoint = Mountpoint, queue = Queue,
-                            store_type = StoreType, max_pending_messages = MaxPendingMsg}) ->
-    Msg = #mqtt_msg{qos     = 1,
-                    retain  = Retain,
-                    topic   = mountpoint(Mountpoint, Topic),
-                    payload = Payload},
-    case emqx_client:publish(Pid, Msg) of
-        {ok, PkgId} ->
-            {noreply, State#state{queue = store(StoreType, {PkgId, Msg}, Queue, MaxPendingMsg)}};
-        {error, Reason} ->
-            emqx_logger:error("Publish fail:~p", [Reason]),
-            {noreply, State}
-    end;
-
-%%----------------------------------------------------------------
-%% received remote node message
-%%----------------------------------------------------------------
-handle_info({publish, #{qos := QoS, dup := Dup, retain := Retain, topic := Topic,
-                        properties := Props, payload := Payload}}, State) ->
-    NewMsg0 = emqx_message:make(bridge, QoS, Topic, Payload),
-    NewMsg1 = emqx_message:set_headers(Props, emqx_message:set_flags(#{dup => Dup, retain=> Retain}, NewMsg0)),
-    emqx_broker:publish(NewMsg1),
-    {noreply, State};
-
-%%----------------------------------------------------------------
-%% received remote puback message
-%%----------------------------------------------------------------
-handle_info({puback, #{packet_id := PkgId}}, State = #state{queue = Queue, store_type = StoreType}) ->
-    % lists:keydelete(PkgId, 1, Queue)
-    {noreply, State#state{queue = delete(StoreType, PkgId, Queue)}};
-
-handle_info({'EXIT', Pid, normal}, State = #state{client_pid = Pid}) ->
-    {noreply, State#state{client_pid = undefined}};
-
-handle_info({'EXIT', Pid, Reason}, State = #state{client_pid = Pid,
-                                                  reconnect_time = ReconnectTime,
-                                                  def_reconnect_count = DefReconnectCount}) ->
-    lager:warning("emqx bridge stop reason:~p", [Reason]),
-    erlang:send_after(ReconnectTime, self(), start),
-    {noreply, State#state{client_pid = undefined, reconnect_count = DefReconnectCount}};
-
-handle_info(Info, State) ->
-    emqx_logger:error("[Bridge] unexpected info: ~p", [Info]),
-    {noreply, State}.
-
-terminate(_Reason, #state{}) ->
-    ok.
-
-code_change(_OldVsn, State, _Extra) ->
-    {ok, State}.
-
-proto_ver(mqtt3) -> v3;
-proto_ver(mqtt4) -> v4;
-proto_ver(mqtt5) -> v5.
-address(Address) ->
-    case string:tokens(Address, ":") of
-        [Host] -> {Host, 1883};
-        [Host, Port] -> {Host, list_to_integer(Port)}
-    end.
-options(Options) ->
-    options(Options, []).
-options([], Acc) ->
-    Acc;
-options([{username, Username}| Options], Acc) ->
-    options(Options, [{username, Username}|Acc]);
-options([{proto_ver, ProtoVer}| Options], Acc) ->
-    options(Options, [{proto_ver, proto_ver(ProtoVer)}|Acc]);
-options([{password, Password}| Options], Acc) ->
-    options(Options, [{password, Password}|Acc]);
-options([{keepalive, Keepalive}| Options], Acc) ->
-    options(Options, [{keepalive, Keepalive}|Acc]);
-options([{client_id, ClientId}| Options], Acc) ->
-    options(Options, [{client_id, ClientId}|Acc]);
-options([{clean_start, CleanStart}| Options], Acc) ->
-    options(Options, [{clean_start, CleanStart}|Acc]);
-options([{address, Address}| Options], Acc) ->
-    {Host, Port} = address(Address),
-    options(Options, [{host, Host}, {port, Port}|Acc]);
-options([_Option | Options], Acc) ->
-    options(Options, Acc).
-
-name(Id) ->
-    list_to_atom(lists:concat([?MODULE, "_", Id])).
-
-i2b(L) -> iolist_to_binary(L).
-
-mountpoint(undefined, Topic) ->
-    Topic;
-mountpoint(Prefix, Topic) ->
-    <<Prefix/binary, Topic/binary>>.
-
-format_mountpoint(undefined) ->
-    undefined;
-format_mountpoint(Prefix) ->
-    binary:replace(i2b(Prefix), <<"${node}">>, atom_to_binary(node(), utf8)).
-
-store(memory, Data, Queue, MaxPendingMsg) when length(Queue) =< MaxPendingMsg ->
-    [Data | Queue];
-store(memory, _Data, Queue, _MaxPendingMsg) ->
-    lager:error("Beyond max pending messages"),
-    Queue;
-store(disk, Data, Queue, _MaxPendingMsg)->
-    [Data | Queue].
-
-delete(memory, PkgId, Queue) ->
-    lists:keydelete(PkgId, 1, Queue);
-delete(disk, PkgId, Queue) ->
-    lists:keydelete(PkgId, 1, Queue).

src/emqx_bridge1_sup.erl

@@ -1,45 +0,0 @@
-%% Copyright (c) 2018 EMQ Technologies Co., Ltd. All Rights Reserved.
-%%
-%% Licensed under the Apache License, Version 2.0 (the "License");
-%% you may not use this file except in compliance with the License.
-%% You may obtain a copy of the License at
-%%
-%%     http://www.apache.org/licenses/LICENSE-2.0
-%%
-%% Unless required by applicable law or agreed to in writing, software
-%% distributed under the License is distributed on an "AS IS" BASIS,
-%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-%% See the License for the specific language governing permissions and
-%% limitations under the License.
-
--module(emqx_bridge1_sup).
-
--behavior(supervisor).
-
--include("emqx.hrl").
-
--export([start_link/0, bridges/0]).
-
-%% Supervisor callbacks
--export([init/1]).
-
-start_link() ->
-    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
-
-%% @doc List all bridges
--spec(bridges() -> [{node(), emqx_topic:topic(), pid()}]).
-bridges() ->
-    [{Name, emqx_bridge1:status(Pid)} || {Name, Pid, _, _} <- supervisor:which_children(?MODULE)].
-
-init([]) ->
-    BridgesOpts = emqx_config:get_env(bridges, []),
-    Bridges = [spec(Opts)|| Opts <- BridgesOpts],
-    {ok, {{one_for_one, 10, 100}, Bridges}}.
-
-spec({Id, Options})->
-    #{id       => Id,
-      start    => {emqx_bridge1, start_link, [Id, Options]},
-      restart  => permanent,
-      shutdown => 5000,
-      type     => worker,
-      modules  => [emqx_bridge1]}.

src/emqx_bridge_sup.erl

@@ -14,13 +14,32 @@
 
 -module(emqx_bridge_sup).
 
+-behavior(supervisor).
+
 -include("emqx.hrl").
 
--export([start_link/3]).
+-export([start_link/0, bridges/0]).
 
--spec(start_link(node(), emqx_topic:topic(), [emqx_bridge:option()])
+%% Supervisor callbacks
-      -> {ok, pid()} | {error, term()}).
+-export([init/1]).
-start_link(Node, Topic, Options) ->
-    MFA = {emqx_bridge, start_link, [Node, Topic, Options]},
-    emqx_pool_sup:start_link({bridge, Node, Topic}, random, MFA).
 
+start_link() ->
+    supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+%% @doc List all bridges
+-spec(bridges() -> [{node(), emqx_topic:topic(), pid()}]).
+bridges() ->
+    [{Name, emqx_bridge:status(Pid)} || {Name, Pid, _, _} <- supervisor:which_children(?MODULE)].
+
+init([]) ->
+    BridgesOpts = emqx_config:get_env(bridges, []),
+    Bridges = [spec(Opts)|| Opts <- BridgesOpts],
+    {ok, {{one_for_one, 10, 100}, Bridges}}.
+
+spec({Id, Options})->
+    #{id       => Id,
+      start    => {emqx_bridge, start_link, [Id, Options]},
+      restart  => permanent,
+      shutdown => 5000,
+      type     => worker,
+      modules  => [emqx_bridge]}.

src/emqx_local_bridge.erl

@@ -0,0 +1,162 @@
+%% Copyright (c) 2018 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+
+-module(emqx_local_bridge).
+
+-behaviour(gen_server).
+
+-include("emqx.hrl").
+-include("emqx_mqtt.hrl").
+
+-export([start_link/5]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
+         code_change/3]).
+
+-define(PING_DOWN_INTERVAL, 1000).
+
+-record(state, {pool, id,
+                node, subtopic,
+                qos                = ?QOS_0,
+                topic_suffix       = <<>>,
+                topic_prefix       = <<>>,
+                mqueue            :: emqx_mqueue:mqueue(),
+                max_queue_len      = 10000,
+                ping_down_interval = ?PING_DOWN_INTERVAL,
+                status             = up}).
+
+-type(option() :: {qos, emqx_mqtt_types:qos()} |
+                  {topic_suffix, binary()} |
+                  {topic_prefix, binary()} |
+                  {max_queue_len, pos_integer()} |
+                  {ping_down_interval, pos_integer()}).
+
+-export_type([option/0]).
+
+%% @doc Start a bridge
+-spec(start_link(term(), pos_integer(), atom(), binary(), [option()])
+      -> {ok, pid()} | ignore | {error, term()}).
+start_link(Pool, Id, Node, Topic, Options) ->
+    gen_server:start_link(?MODULE, [Pool, Id, Node, Topic, Options], [{hibernate_after, 5000}]).
+
+%%------------------------------------------------------------------------------
+%% gen_server callbacks
+%%------------------------------------------------------------------------------
+
+init([Pool, Id, Node, Topic, Options]) ->
+    process_flag(trap_exit, true),
+    true = gproc_pool:connect_worker(Pool, {Pool, Id}),
+    case net_kernel:connect_node(Node) of
+        true ->
+            true = erlang:monitor_node(Node, true),
+            Share = iolist_to_binary(["$bridge:", atom_to_list(Node), ":", Topic]),
+            emqx_broker:subscribe(Topic, self(), [{share, Share}, {qos, ?QOS_0}]),
+            State = parse_opts(Options, #state{node = Node, subtopic = Topic}),
+            %%TODO: queue....
+            MQueue = emqx_mqueue:new(qname(Node, Topic), [{max_len, State#state.max_queue_len}]),
+            {ok, State#state{pool = Pool, id = Id, mqueue = MQueue}};
+        false ->
+            {stop, {cannot_connect_node, Node}}
+    end.
+
+parse_opts([], State) ->
+    State;
+parse_opts([{qos, QoS} | Opts], State) ->
+    parse_opts(Opts, State#state{qos = QoS});
+parse_opts([{topic_suffix, Suffix} | Opts], State) ->
+    parse_opts(Opts, State#state{topic_suffix= Suffix});
+parse_opts([{topic_prefix, Prefix} | Opts], State) ->
+    parse_opts(Opts, State#state{topic_prefix = Prefix});
+parse_opts([{max_queue_len, Len} | Opts], State) ->
+    parse_opts(Opts, State#state{max_queue_len = Len});
+parse_opts([{ping_down_interval, Interval} | Opts], State) ->
+    parse_opts(Opts, State#state{ping_down_interval = Interval});
+parse_opts([_Opt | Opts], State) ->
+    parse_opts(Opts, State).
+
+qname(Node, Topic) when is_atom(Node) ->
+    qname(atom_to_list(Node), Topic);
+qname(Node, Topic) ->
+    iolist_to_binary(["Bridge:", Node, ":", Topic]).
+
+handle_call(Req, _From, State) ->
+    emqx_logger:error("[Bridge] unexpected call: ~p", [Req]),
+    {reply, ignored, State}.
+
+handle_cast(Msg, State) ->
+    emqx_logger:error("[Bridge] unexpected cast: ~p", [Msg]),
+    {noreply, State}.
+
+handle_info({dispatch, _Topic, Msg}, State = #state{mqueue = Q, status = down}) ->
+    %% TODO: how to drop???
+    {noreply, State#state{mqueue = emqx_mqueue:in(Msg, Q)}};
+
+handle_info({dispatch, _Topic, Msg}, State = #state{node = Node, status = up}) ->
+    ok = emqx_rpc:cast(Node, emqx_broker, publish, [transform(Msg, State)]),
+    {noreply, State};
+
+handle_info({nodedown, Node}, State = #state{node = Node, ping_down_interval = Interval}) ->
+    emqx_logger:warning("[Bridge] node down: ~s", [Node]),
+    erlang:send_after(Interval, self(), ping_down_node),
+    {noreply, State#state{status = down}, hibernate};
+
+handle_info({nodeup, Node}, State = #state{node = Node}) ->
+    %% TODO: Really fast??
+    case emqx:is_running(Node) of
+        true -> emqx_logger:warning("[Bridge] Node up: ~s", [Node]),
+                {noreply, dequeue(State#state{status = up})};
+        false -> self() ! {nodedown, Node},
+                 {noreply, State#state{status = down}}
+    end;
+
+handle_info(ping_down_node, State = #state{node = Node, ping_down_interval = Interval}) ->
+    Self = self(),
+    spawn_link(fun() ->
+                 case net_kernel:connect_node(Node) of
+                     true -> Self ! {nodeup, Node};
+                     false -> erlang:send_after(Interval, Self, ping_down_node)
+                 end
+               end),
+    {noreply, State};
+
+handle_info({'EXIT', _Pid, normal}, State) ->
+    {noreply, State};
+
+handle_info(Info, State) ->
+    emqx_logger:error("[Bridge] unexpected info: ~p", [Info]),
+    {noreply, State}.
+
+terminate(_Reason, #state{pool = Pool, id = Id}) ->
+    gproc_pool:disconnect_worker(Pool, {Pool, Id}).
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%--------------------------------------------------------------------
+%% Internal functions
+%%--------------------------------------------------------------------
+
+dequeue(State = #state{mqueue = MQ}) ->
+    case emqx_mqueue:out(MQ) of
+        {empty, MQ1} ->
+            State#state{mqueue = MQ1};
+        {{value, Msg}, MQ1} ->
+            handle_info({dispatch, Msg#message.topic, Msg}, State),
+            dequeue(State#state{mqueue = MQ1})
+    end.
+
+transform(Msg = #message{topic = Topic}, #state{topic_prefix = Prefix,
+                                                topic_suffix = Suffix}) ->
+    Msg#message{topic = <<Prefix/binary, Topic/binary, Suffix/binary>>}.
+

src/emqx_local_bridge_sup.erl

@@ -0,0 +1,26 @@
+%% Copyright (c) 2018 EMQ Technologies Co., Ltd. All Rights Reserved.
+%%
+%% Licensed under the Apache License, Version 2.0 (the "License");
+%% you may not use this file except in compliance with the License.
+%% You may obtain a copy of the License at
+%%
+%%     http://www.apache.org/licenses/LICENSE-2.0
+%%
+%% Unless required by applicable law or agreed to in writing, software
+%% distributed under the License is distributed on an "AS IS" BASIS,
+%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+%% See the License for the specific language governing permissions and
+%% limitations under the License.
+
+-module(emqx_local_bridge_sup).
+
+-include("emqx.hrl").
+
+-export([start_link/3]).
+
+-spec(start_link(node(), emqx_topic:topic(), [emqx_local_bridge:option()])
+      -> {ok, pid()} | {error, term()}).
+start_link(Node, Topic, Options) ->
+    MFA = {emqx_local_bridge, start_link, [Node, Topic, Options]},
+    emqx_pool_sup:start_link({bridge, Node, Topic}, random, MFA).
+

src/emqx_local_bridge_sup_sup.erl

@@ -12,7 +12,7 @@
 %% See the License for the specific language governing permissions and
 %% limitations under the License.
 
--module(emqx_bridge_sup_sup).
+-module(emqx_local_bridge_sup_sup).
 
 -behavior(supervisor).
 
@@ -66,9 +66,9 @@ init([]) ->
 
 bridge_spec(Node, Topic, Options) ->
     #{id       => ?CHILD_ID(Node, Topic),
-      start    => {emqx_bridge_sup, start_link, [Node, Topic, Options]},
+      start    => {emqx_local_bridge_sup, start_link, [Node, Topic, Options]},
       restart  => permanent,
       shutdown => infinity,
       type     => supervisor,
-      modules  => [emqx_bridge_sup]}.
+      modules  => [emqx_local_bridge_sup]}.
 

src/emqx_sup.erl

@@ -62,8 +62,9 @@ init([]) ->
     %% Broker Sup
     BrokerSup = supervisor_spec(emqx_broker_sup),
     %% BridgeSup
-    BridgeSup = supervisor_spec(emqx_bridge_sup_sup),
+    LocalBridgeSup = supervisor_spec(emqx_local_bridge_sup_sup),
-    BridgeSup1 = supervisor_spec(emqx_bridge1_sup),
+
+    BridgeSup = supervisor_spec(emqx_bridge_sup),
     %% AccessControl
     AccessControl = worker_spec(emqx_access_control),
     %% Session Manager
@@ -78,8 +79,8 @@ init([]) ->
           [KernelSup,
            RouterSup,
            BrokerSup,
+           LocalBridgeSup,
            BridgeSup,
-           BridgeSup1,
            AccessControl,
            SMSup,
            SessionSup,