Merge default into bug22039

4 files changed, 163 insertions, 31 deletions

diff --git a/src/rabbit.erl b/src/rabbit.erl
index 24c4622f7e..a8b65f4624 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -292,6 +292,7 @@ add_boot_step_dep(G, RunsSecond, RunsFirst) ->
 %%---------------------------------------------------------------------------
 
 boot_core_processes() ->
+    ok = rabbit_sup:start_child(rabbit_exchange_type),
     ok = rabbit_sup:start_child(rabbit_log),
     ok = rabbit_hooks:start(),
 
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index 09ea1e9611..be73e81847 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -134,18 +134,18 @@ declare(ExchangeName, Type, Durable, AutoDelete, Args) ->
               end
       end).
 
-typename_to_plugin_module(T) when is_binary(T) ->
-    case catch list_to_existing_atom("rabbit_exchange_type_" ++ binary_to_list(T)) of
-        {'EXIT', {badarg, _}} ->
+typename_to_plugin_module(T) ->
+    case rabbit_exchange_type:lookup_module(T) of
+        {ok, Module} ->
+            Module;
+        {error, not_found} ->
             rabbit_misc:protocol_error(
-              command_invalid, "invalid exchange type '~s'", [T]);
-        Module ->
-            Module
+              command_invalid, "invalid exchange type '~s'", [T])
     end.
 
-plugin_module_to_typename(M) when is_atom(M) ->
-    "rabbit_exchange_type_" ++ S = atom_to_list(M),
-    list_to_binary(S).
+plugin_module_to_typename(M) ->
+    {ok, TypeName} = rabbit_exchange_type:lookup_name(M),
+    TypeName.
 
 check_type(T) ->
     Module = typename_to_plugin_module(T),
diff --git a/src/rabbit_exchange_type.erl b/src/rabbit_exchange_type.erl
new file mode 100644
index 0000000000..58dcfbb6a2
--- /dev/null
+++ b/src/rabbit_exchange_type.erl
@@ -0,0 +1,107 @@
+%%   The contents of this file are subject to the Mozilla Public License
+%%   Version 1.1 (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.mozilla.org/MPL/
+%%
+%%   Software distributed under the License is distributed on an "AS IS"
+%%   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
+%%   License for the specific language governing rights and limitations
+%%   under the License.
+%%
+%%   The Original Code is RabbitMQ.
+%%
+%%   The Initial Developers of the Original Code are LShift Ltd,
+%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
+%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
+%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
+%%   Technologies LLC, and Rabbit Technologies Ltd.
+%%
+%%   Portions created by LShift Ltd are Copyright (C) 2007-2009 LShift
+%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
+%%   Copyright (C) 2007-2009 Cohesive Financial Technologies
+%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%%   (C) 2007-2009 Rabbit Technologies Ltd.
+%%
+%%   All Rights Reserved.
+%%
+%%   Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_exchange_type).
+
+-behaviour(gen_server).
+
+-export([start_link/0]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+         terminate/2, code_change/3]).
+
+-export([register/2, lookup_module/1, lookup_name/1]).
+
+-define(SERVER, ?MODULE).
+
+%%---------------------------------------------------------------------------
+
+start_link() ->
+    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+
+%%---------------------------------------------------------------------------
+
+register(TypeName, ModuleName) ->
+    gen_server:call(?SERVER, {register, TypeName, ModuleName}).
+
+lookup_module(T) when is_binary(T) ->
+    case ets:lookup(rabbit_exchange_type_modules, T) of
+        [{_, Module}] ->
+            {ok, Module};
+        [] ->
+            {error, not_found}
+    end.
+
+lookup_name(M) when is_atom(M) ->
+    [{_, TypeName}] = ets:lookup(rabbit_exchange_type_names, M),
+    {ok, TypeName}.
+
+%%---------------------------------------------------------------------------
+
+internal_register(TypeName, ModuleName)
+  when is_binary(TypeName), is_atom(ModuleName) ->
+    true = ets:insert(rabbit_exchange_type_modules, {TypeName, ModuleName}),
+    true = ets:insert(rabbit_exchange_type_names, {ModuleName, TypeName}),
+    ok.
+
+%%---------------------------------------------------------------------------
+
+init([]) ->
+    rabbit_exchange_type_modules =
+        ets:new(rabbit_exchange_type_modules, [protected, set, named_table]),
+    rabbit_exchange_type_names =
+        ets:new(rabbit_exchange_type_names, [protected, set, named_table]),
+
+    %% TODO: split out into separate boot startup steps.
+    ok = internal_register(<<"direct">>, rabbit_exchange_type_direct),
+    ok = internal_register(<<"fanout">>, rabbit_exchange_type_fanout),
+    ok = internal_register(<<"headers">>, rabbit_exchange_type_headers),
+    ok = internal_register(<<"topic">>, rabbit_exchange_type_topic),
+
+    {ok, none}.
+
+handle_call({register, TypeName, ModuleName}, _From, State) ->
+    ok = internal_register(TypeName, ModuleName),
+    {reply, ok, State};
+handle_call(Request, _From, State) ->
+    {stop, {unhandled_call, Request}, State}.
+
+handle_cast(Request, State) ->
+    {stop, {unhandled_cast, Request}, State}.
+
+handle_info(Message, State) ->
+    {stop, {unhandled_info, Message}, State}.
+
+terminate(_Reason, _State) ->
+    ok.
+
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
diff --git a/src/rabbit_multi.erl b/src/rabbit_multi.erl
index f364872eca..dc642df403 100644
--- a/src/rabbit_multi.erl
+++ b/src/rabbit_multi.erl
@@ -99,13 +99,16 @@ Available commands:
 
 action(start_all, [NodeCount], RpcTimeout) ->
     io:format("Starting all nodes...~n", []),
-    N = list_to_integer(NodeCount),
+    application:load(rabbit),
+    NodeName = rabbit_misc:nodeparts(getenv("RABBITMQ_NODENAME")),
     {NodePids, Running} =
-        start_nodes(N, N, [], true,
-                    rabbit_misc:nodeparts(
-                      getenv("RABBITMQ_NODENAME")),
-                    list_to_integer(getenv("RABBITMQ_NODE_PORT")),
-                    RpcTimeout),
+        case list_to_integer(NodeCount) of
+            1 -> {NodePid, Started} = start_node(rabbit_misc:makenode(NodeName),
+                                                 RpcTimeout),
+                 {[NodePid], Started};
+            N -> start_nodes(N, N, [], true, NodeName,
+                             get_node_tcp_listener(), RpcTimeout)
+        end,
     write_pids_file(NodePids),
     case Running of
         true  -> ok;
@@ -158,26 +161,29 @@ action(rotate_logs, [Suffix], RpcTimeout) ->
 %% Running is a boolean exhibiting success at some moment
 start_nodes(0, _, PNodePid, Running, _, _, _) -> {PNodePid, Running};
 
-start_nodes(N, Total, PNodePid, Running,
-            NodeNameBase, NodePortBase, RpcTimeout) ->
+start_nodes(N, Total, PNodePid, Running, NodeNameBase, Listener, RpcTimeout) ->
     {NodePre, NodeSuff} = NodeNameBase,
     NodeNumber = Total - N,
-    NodePre1 = if NodeNumber == 0 ->
-                       %% For compatibility with running a single node
-                       NodePre;
-                  true ->           
-                       NodePre ++ "_" ++ integer_to_list(NodeNumber)
+    NodePre1 = case NodeNumber of
+                   %% For compatibility with running a single node
+                   0 -> NodePre;
+                   _ -> NodePre ++ "_" ++ integer_to_list(NodeNumber)
                end,
-    {NodePid, Started} = start_node(rabbit_misc:makenode({NodePre1, NodeSuff}),
-                                    NodePortBase + NodeNumber,
-                                    RpcTimeout),
+    Node = rabbit_misc:makenode({NodePre1, NodeSuff}),
+    os:putenv("RABBITMQ_NODENAME", atom_to_list(Node)),
+    case Listener of
+        {NodeIpAddress, NodePortBase} ->
+            NodePort = NodePortBase + NodeNumber,
+            os:putenv("RABBITMQ_NODE_PORT", integer_to_list(NodePort)),
+            os:putenv("RABBITMQ_NODE_IP_ADDRESS", NodeIpAddress);
+        undefined ->
+            ok
+    end,
+    {NodePid, Started} = start_node(Node, RpcTimeout),
     start_nodes(N - 1, Total, [NodePid | PNodePid],
-                Started and Running,
-                NodeNameBase, NodePortBase, RpcTimeout).
+                Started and Running, NodeNameBase, Listener, RpcTimeout).
 
-start_node(Node, NodePort, RpcTimeout) ->
-    os:putenv("RABBITMQ_NODENAME", atom_to_list(Node)),
-    os:putenv("RABBITMQ_NODE_PORT", integer_to_list(NodePort)),
+start_node(Node, RpcTimeout) ->
     io:format("Starting node ~s...~n", [Node]),
     case rpc:call(Node, os, getpid, []) of
         {badrpc, _} ->
@@ -293,7 +299,7 @@ kill_wait(Pid, TimeLeft, Forceful) ->
     io:format(".", []),
     is_dead(Pid) orelse kill_wait(Pid, TimeLeft - ?RPC_SLEEP, Forceful).
 
-% Test using some OS clunkiness since we shouldn't trust 
+% Test using some OS clunkiness since we shouldn't trust
 % rpc:call(os, getpid, []) at this point
 is_dead(Pid) ->
     PidS = integer_to_list(Pid),
@@ -321,3 +327,21 @@ getenv(Var) ->
         false -> throw({missing_env_var, Var});
         Value -> Value
     end.
+
+get_node_tcp_listener() ->
+    try
+        {getenv("RABBITMQ_NODE_IP_ADDRESS"),
+         list_to_integer(getenv("RABBITMQ_NODE_PORT"))}
+    catch _ ->
+            case application:get_env(rabbit, tcp_listeners) of
+                {ok, [{_IpAddy, _Port} = Listener]} ->
+                    Listener;
+                {ok, []} ->
+                    undefined;
+                {ok, Other} ->
+                    throw({cannot_start_multiple_nodes, multiple_tcp_listeners,
+                           Other});
+                undefined ->
+                    throw({missing_configuration, tcp_listeners})
+            end
+    end.