merge bug20654 into default

7 files changed, 178 insertions, 86 deletions

diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index c707112f..ffda0698 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -143,6 +143,7 @@
                 host     :: string() | atom(),
                 port     :: non_neg_integer()}).
 -type(not_found() :: {'error', 'not_found'}).
+-type(routing_result() :: 'routed' | 'unroutable' | 'not_delivered').
 
 -endif.
 
diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
new file mode 100644
index 00000000..b2e85820
--- /dev/null
+++ b/src/rabbit_basic.erl
@@ -0,0 +1,71 @@
+%%   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_basic).
+-include("rabbit.hrl").
+-include("rabbit_framing.hrl").
+
+-export([publish/4, message/4]).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(publish/4 :: (bool(), bool(), maybe(txn()), message()) ->
+             {ok, routing_result(), [pid()]} | not_found()).
+-spec(message/4 :: (exchange_name(), routing_key(), binary(), binary()) ->
+             message()).
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+publish(Mandatory, Immediate, Txn,
+        Message = #basic_message{exchange_name = ExchangeName}) ->
+    case rabbit_exchange:lookup(ExchangeName) of
+        {ok, X} ->
+            {RoutingRes, DeliveredQPids} =
+                rabbit_exchange:publish(X, Mandatory, Immediate, Txn, Message),
+            {ok, RoutingRes, DeliveredQPids};
+        Other ->
+            Other
+    end.
+
+message(ExchangeName, RoutingKeyBin, ContentTypeBin, BodyBin) ->
+    {ClassId, _MethodId} = rabbit_framing:method_id('basic.publish'),
+    Content = #content{class_id = ClassId,
+                       properties = #'P_basic'{content_type = ContentTypeBin},
+                       properties_bin = none,
+                       payload_fragments_rev = [BodyBin]},
+    #basic_message{exchange_name  = ExchangeName,
+                   routing_key    = RoutingKeyBin,
+                   content        = Content,
+                   persistent_key = none}.
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index e837c84f..738e9017 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -306,7 +306,9 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin,
                                routing_key = RoutingKey,
                                mandatory = Mandatory,
                                immediate = Immediate},
-              Content, State = #ch{ virtual_host = VHostPath}) ->
+              Content, State = #ch{ virtual_host = VHostPath,
+                                    transaction_id = TxnKey,
+                                    writer_pid = WriterPid}) ->
     ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin),
     check_write_permitted(ExchangeName, State),
     Exchange = rabbit_exchange:lookup_or_die(ExchangeName),
@@ -317,12 +319,29 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin,
                         true  -> rabbit_guid:guid();
                         false -> none
                     end,
-    {noreply, publish(Mandatory, Immediate,
-                      #basic_message{exchange_name  = ExchangeName,
-                                     routing_key    = RoutingKey,
-                                     content        = DecodedContent,
-                                     persistent_key = PersistentKey},
-                      rabbit_exchange:route(Exchange, RoutingKey, DecodedContent), State)};
+    Message = #basic_message{exchange_name  = ExchangeName,
+                             routing_key    = RoutingKey,
+                             content        = DecodedContent,
+                             persistent_key = PersistentKey},
+    {RoutingRes, DeliveredQPids} =
+        rabbit_exchange:publish(Exchange, Mandatory, Immediate, TxnKey,
+                                Message),
+    case RoutingRes of
+        routed ->
+            ok;
+        unroutable ->
+            %% FIXME: 312 should be replaced by the ?NO_ROUTE
+            %% definition, when we move to >=0-9
+            ok = basic_return(Message, WriterPid, 312, <<"unroutable">>);
+        not_delivered ->
+            %% FIXME: 313 should be replaced by the ?NO_CONSUMERS
+            %% definition, when we move to >=0-9
+            ok = basic_return(Message, WriterPid, 313, <<"not_delivered">>)
+    end,
+    {noreply, case TxnKey of
+                  none -> State;
+                  _    -> add_tx_participants(DeliveredQPids, State)
+              end};
 
 handle_method(#'basic.ack'{delivery_tag = DeliveryTag,
                            multiple = Multiple},
@@ -551,6 +570,13 @@ handle_method(#'exchange.declare'{exchange = ExchangeNameBin,
             {ok, FoundX} -> FoundX;
             {error, not_found} ->
                 check_name('exchange', ExchangeNameBin),
+                case rabbit_misc:r_arg(VHostPath, exchange, Args,
+                                       <<"alternate-exchange">>) of
+                    undefined -> ok;
+                    AName     -> check_read_permitted(ExchangeName, State),
+                                 check_write_permitted(AName, State),
+                                 ok
+                end,
                 rabbit_exchange:declare(ExchangeName,
                                         CheckedType,
                                         Durable,
@@ -764,30 +790,6 @@ binding_action(Fun, ExchangeNameBin, QueueNameBin, RoutingKey, Arguments,
         ok -> return_ok(State, NoWait, ReturnMethod)
     end.
 
-publish(Mandatory, Immediate, Message, QPids,
-        State = #ch{transaction_id = TxnKey, writer_pid = WriterPid}) ->
-    Handled = deliver(QPids, Mandatory, Immediate, TxnKey,
-                      Message, WriterPid),
-    case TxnKey of
-        none -> State;
-        _    -> add_tx_participants(Handled, State)
-    end.
-
-deliver(QPids, Mandatory, Immediate, Txn, Message, WriterPid) ->
-    case rabbit_router:deliver(QPids, Mandatory, Immediate, Txn, Message) of
-        {ok, DeliveredQPids}   -> DeliveredQPids;
-        {error, unroutable}    ->
-            %% FIXME: 312 should be replaced by the ?NO_ROUTE
-            %% definition, when we move to >=0-9
-            ok = basic_return(Message, WriterPid, 312, <<"unroutable">>),
-            [];
-        {error, not_delivered} ->
-            %% FIXME: 313 should be replaced by the ?NO_CONSUMERS
-            %% definition, when we move to >=0-9
-            ok = basic_return(Message, WriterPid, 313, <<"not_delivered">>),
-            []
-    end.
-
 basic_return(#basic_message{exchange_name = ExchangeName,
                             routing_key   = RoutingKey,
                             content       = Content},
diff --git a/src/rabbit_error_logger.erl b/src/rabbit_error_logger.erl
index dc5824f1..d73edb73 100644
--- a/src/rabbit_error_logger.erl
+++ b/src/rabbit_error_logger.erl
@@ -74,7 +74,9 @@ publish(_Other, _Format, _Data, _State) ->
     ok.
 
 publish1(RoutingKey, Format, Data, LogExch) ->
-    {ok, _QueueNames} = rabbit_exchange:simple_publish(
-                          false, false, LogExch, RoutingKey, <<"text/plain">>,
-                          list_to_binary(io_lib:format(Format, Data))),
+    {ok, _RoutingRes, _DeliveredQPids} =
+        rabbit_basic:publish(false, false, none,
+                             rabbit_basic:message(
+                               LogExch, RoutingKey, <<"text/plain">>,
+                               list_to_binary(io_lib:format(Format, Data)))),
     ok.
diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl
index d6d70cba..0a7a9bc7 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -36,8 +36,7 @@
 
 -export([recover/0, declare/5, lookup/1, lookup_or_die/1,
          list/1, info/1, info/2, info_all/1, info_all/2,
-         simple_publish/6, simple_publish/3,
-         route/3]).
+         publish/5]).
 -export([add_binding/4, delete_binding/4, list_bindings/1]).
 -export([delete/2]).
 -export([delete_queue_bindings/1, delete_transient_queue_bindings/1]).
@@ -57,8 +56,6 @@
 
 -ifdef(use_specs).
 
--type(publish_res() :: {'ok', [pid()]} |
-      not_found() | {'error', 'unroutable' | 'not_delivered'}).
 -type(bind_res() :: 'ok' | {'error',
                             'queue_not_found' |
                             'exchange_not_found' |
@@ -75,11 +72,8 @@
 -spec(info/2 :: (exchange(), [info_key()]) -> [info()]).
 -spec(info_all/1 :: (vhost()) -> [[info()]]).
 -spec(info_all/2 :: (vhost(), [info_key()]) -> [[info()]]).
--spec(simple_publish/6 ::
-      (bool(), bool(), exchange_name(), routing_key(), binary(), binary()) ->
-             publish_res()).
--spec(simple_publish/3 :: (bool(), bool(), message()) -> publish_res()).
--spec(route/3 :: (exchange(), routing_key(), decoded_content()) -> [pid()]).
+-spec(publish/5 :: (exchange(), bool(), bool(), maybe(txn()), message()) ->
+             {routing_result(), [pid()]}).
 -spec(add_binding/4 ::
       (exchange_name(), queue_name(), routing_key(), amqp_table()) ->
              bind_res() | {'error', 'durability_settings_incompatible'}).
@@ -194,36 +188,44 @@ info_all(VHostPath) -> map(VHostPath, fun (X) -> info(X) end).
 
 info_all(VHostPath, Items) -> map(VHostPath, fun (X) -> info(X, Items) end).
 
-%% Usable by Erlang code that wants to publish messages.
-simple_publish(Mandatory, Immediate, ExchangeName, RoutingKeyBin,
-               ContentTypeBin, BodyBin) ->
-    {ClassId, _MethodId} = rabbit_framing:method_id('basic.publish'),
-    Content = #content{class_id = ClassId,
-                       properties = #'P_basic'{content_type = ContentTypeBin},
-                       properties_bin = none,
-                       payload_fragments_rev = [BodyBin]},
-    Message = #basic_message{exchange_name = ExchangeName,
-                             routing_key = RoutingKeyBin,
-                             content = Content,
-                             persistent_key = none},
-    simple_publish(Mandatory, Immediate, Message).
-
-%% Usable by Erlang code that wants to publish messages.
-simple_publish(Mandatory, Immediate,
-               Message = #basic_message{exchange_name = ExchangeName,
-                                        routing_key = RoutingKey,
-					content = Content}) ->
-    case lookup(ExchangeName) of
-        {ok, Exchange} ->
-            QPids = route(Exchange, RoutingKey, Content),
-            rabbit_router:deliver(QPids, Mandatory, Immediate,
-                                  none, Message);
-        {error, Error} -> {error, Error}
+publish(X, Mandatory, Immediate, Txn, Message) ->
+    publish(X, [], Mandatory, Immediate, Txn, Message).
+
+publish(X, Seen, Mandatory, Immediate, Txn,
+        Message = #basic_message{routing_key = RK, content = C}) ->
+    case rabbit_router:deliver(route(X, RK, C),
+                               Mandatory, Immediate, Txn, Message) of
+        {_, []} = R ->
+            #exchange{name = XName, arguments = Args} = X,
+            case rabbit_misc:r_arg(XName, exchange, Args,
+                                   <<"alternate-exchange">>) of
+                undefined ->
+                    R;
+                AName ->
+                    NewSeen = [XName | Seen],
+                    case lists:member(AName, NewSeen) of
+                        true ->
+                            R;
+                        false ->
+                            case lookup(AName) of
+                                {ok, AX} ->
+                                    publish(AX, NewSeen,
+                                            Mandatory, Immediate, Txn,
+                                            Message);
+                                {error, not_found} ->
+                                    rabbit_log:warning(
+                                      "alternate exchange for ~s "
+                                      "does not exist: ~s",
+                                      [rabbit_misc:rs(XName),
+                                       rabbit_misc:rs(AName)]),
+                                    R
+                            end
+                    end
+            end;
+        R ->
+            R
     end.
 
-sort_arguments(Arguments) ->
-    lists:keysort(1, Arguments).
-
 %% return the list of qpids to which a message with a given routing
 %% key, sent to a particular exchange, should be delivered.
 %%
@@ -250,6 +252,9 @@ route(X = #exchange{type = fanout}, _RoutingKey, _Content) ->
 route(X = #exchange{type = direct}, RoutingKey, _Content) ->
     match_routing_key(X, RoutingKey).
 
+sort_arguments(Arguments) ->
+    lists:keysort(1, Arguments).
+
 %% TODO: Maybe this should be handled by a cursor instead.
 %% TODO: This causes a full scan for each entry with the same exchange
 match_bindings(#exchange{name = Name}, Match) ->
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index 7a395a6f..72e16f0f 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -39,7 +39,7 @@
 -export([not_found/1]).
 -export([get_config/1, get_config/2, set_config/2]).
 -export([dirty_read/1]).
--export([r/3, r/2, rs/1]).
+-export([r/3, r/2, r_arg/4, rs/1]).
 -export([enable_cover/0, report_cover/0]).
 -export([throw_on_error/2, with_exit_handler/2, filter_exit_map/2]).
 -export([with_user/2, with_vhost/2, with_user_and_vhost/3]).
@@ -78,12 +78,14 @@
 -spec(get_config/2 :: (atom(), A) -> A).
 -spec(set_config/2 :: (atom(), any()) -> 'ok').
 -spec(dirty_read/1 :: ({atom(), any()}) -> {'ok', any()} | not_found()).
--spec(r/3 :: (vhost() | r(atom()), K, resource_name()) -> r(K) 
-              when is_subtype(K, atom())).
+-spec(r/3 :: (vhost() | r(atom()), K, resource_name()) ->
+             r(K) when is_subtype(K, atom())).
 -spec(r/2 :: (vhost(), K) -> #resource{virtual_host :: vhost(),
                                        kind         :: K,
                                        name         :: '_'}
                                  when is_subtype(K, atom())).
+-spec(r_arg/4 :: (vhost() | r(atom()), K, amqp_table(), binary()) ->
+             undefined | r(K)  when is_subtype(K, atom())).
 -spec(rs/1 :: (r(atom())) -> string()).
 -spec(enable_cover/0 :: () -> 'ok' | {'error', any()}).
 -spec(report_cover/0 :: () -> 'ok').
@@ -173,6 +175,14 @@ r(VHostPath, Kind, Name) when is_binary(Name) andalso is_binary(VHostPath) ->
 r(VHostPath, Kind) when is_binary(VHostPath) ->
     #resource{virtual_host = VHostPath, kind = Kind, name = '_'}.
 
+r_arg(#resource{virtual_host = VHostPath}, Kind, Table, Key) ->
+    r_arg(VHostPath, Kind, Table, Key);
+r_arg(VHostPath, Kind, Table, Key) ->
+    case lists:keysearch(Key, 1, Table) of
+        {value, {_, longstr, NameBin}} -> r(VHostPath, Kind, NameBin);
+        false                          -> undefined
+    end.
+
 rs(#resource{virtual_host = VHostPath, kind = Kind, name = Name}) ->
     lists:flatten(io_lib:format("~s '~s' in vhost '~s'",
                                 [Kind, Name, VHostPath])).
diff --git a/src/rabbit_router.erl b/src/rabbit_router.erl
index 0b06a063..57166428 100644
--- a/src/rabbit_router.erl
+++ b/src/rabbit_router.erl
@@ -51,7 +51,7 @@
 
 -spec(start_link/0 :: () -> {'ok', pid()} | 'ignore' | {'error', any()}).
 -spec(deliver/5 :: ([pid()], bool(), bool(), maybe(txn()), message()) ->
-             {'ok', [pid()]} | {'error', 'unroutable' | 'not_delivered'}).
+             {routing_result(), [pid()]}).
 
 -endif.
 
@@ -98,14 +98,15 @@ deliver_per_node(NodeQPids, Mandatory = false, Immediate = false,
     %% therefore safe to use a fire-and-forget cast here and return
     %% the QPids - the semantics is preserved. This scales much better
     %% than the non-immediate case below.
-    {ok, lists:flatmap(
-           fun ({Node, QPids}) ->
-                   gen_server2:cast(
-                     {?SERVER, Node},
-                     {deliver, QPids, Mandatory, Immediate, Txn, Message}),
-                   QPids
-           end,
-           NodeQPids)};
+    {routed,
+     lists:flatmap(
+       fun ({Node, QPids}) ->
+               gen_server2:cast(
+                 {?SERVER, Node},
+                 {deliver, QPids, Mandatory, Immediate, Txn, Message}),
+               QPids
+       end,
+       NodeQPids)};
 deliver_per_node(NodeQPids, Mandatory, Immediate,
                  Txn, Message) ->
     R = rabbit_misc:upmap(
@@ -179,6 +180,6 @@ run_bindings(QPids, IsMandatory, IsImmediate, Txn, Message) ->
       QPids).
 
 %% check_delivery(Mandatory, Immediate, {WasRouted, QPids})
-check_delivery(true, _   , {false, []}) -> {error, unroutable};
-check_delivery(_   , true, {_    , []}) -> {error, not_delivered};
-check_delivery(_   , _   , {_    , Qs}) -> {ok, Qs}.
+check_delivery(true, _   , {false, []}) -> {unroutable, []};
+check_delivery(_   , true, {_    , []}) -> {not_delivered, []};
+check_delivery(_   , _   , {_    , Qs}) -> {routed, Qs}.