merge default into amqp_0_9_1

13 files changed, 207 insertions, 105 deletions

diff --git a/docs/rabbitmqctl.1.pod b/docs/rabbitmqctl.1.pod
index a0232a40..42156896 100644
--- a/docs/rabbitmqctl.1.pod
+++ b/docs/rabbitmqctl.1.pod
@@ -201,7 +201,7 @@ name
     URL-encoded name of the exchange
 
 type
-    exchange type (B<direct>, B<topic> or B<fanout>)
+    exchange type (B<direct>, B<topic>, B<fanout>, or B<headers>)
 
 durable
     whether the exchange survives server restarts
diff --git a/include/rabbit.hrl b/include/rabbit.hrl
index 7515e714..0e7fb7c9 100644
--- a/include/rabbit.hrl
+++ b/include/rabbit.hrl
@@ -142,6 +142,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/packaging/RPMS/Fedora/rabbitmq-server.spec b/packaging/RPMS/Fedora/rabbitmq-server.spec
index bed1611a..3c3be609 100644
--- a/packaging/RPMS/Fedora/rabbitmq-server.spec
+++ b/packaging/RPMS/Fedora/rabbitmq-server.spec
@@ -117,6 +117,9 @@ fi
 rm -rf %{buildroot}
 
 %changelog
+* Tue May 19 2009 Matthias Radestock <matthias@lshift.net> 1.5.5-1
+- Maintenance release for the 1.5.x series
+
 * Mon Apr 6 2009 Matthias Radestock <matthias@lshift.net> 1.5.4-1
 - Maintenance release for the 1.5.x series
 
diff --git a/packaging/debs/Debian/debian/changelog b/packaging/debs/Debian/debian/changelog
index d1ccd3a0..7c5673f7 100644
--- a/packaging/debs/Debian/debian/changelog
+++ b/packaging/debs/Debian/debian/changelog
@@ -1,3 +1,9 @@
+rabbitmq-server (1.5.5-1) hardy; urgency=low
+
+  * New Upstream Release
+
+ -- Matthias Radestock <matthias@lshift.net>  Tue, 19 May 2009 09:57:54 +0100
+
 rabbitmq-server (1.5.4-1) hardy; urgency=low
 
   * New Upstream Release
diff --git a/packaging/debs/Debian/debian/control b/packaging/debs/Debian/debian/control
index b2b3ab02..21636072 100644
--- a/packaging/debs/Debian/debian/control
+++ b/packaging/debs/Debian/debian/control
@@ -7,7 +7,7 @@ Standards-Version: 3.8.0
 
 Package: rabbitmq-server
 Architecture: all
-Depends: erlang-nox, adduser, logrotate, ${misc:Depends}
+Depends: erlang-nox, erlang-os-mon | erlang-nox (<< 1:13.b-dfsg1-1), adduser, logrotate, ${misc:Depends}
 Description: An AMQP server written in Erlang
  RabbitMQ is an implementation of AMQP, the emerging standard for high
  performance enterprise messaging. The RabbitMQ server is a robust and
diff --git a/packaging/macports/net/rabbitmq-server/Portfile b/packaging/macports/net/rabbitmq-server/Portfile
index d9d16dbb..65913256 100644
--- a/packaging/macports/net/rabbitmq-server/Portfile
+++ b/packaging/macports/net/rabbitmq-server/Portfile
@@ -1,15 +1,15 @@
 # -*- coding: utf-8; mode: tcl; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- vim:fenc=utf-8:filetype=tcl:et:sw=4:ts=4:sts=4
 # $Id$
 
-PortSystem      1.0
-
-name            rabbitmq-server
-version         1.5.3
-categories      net
-maintainers     tonyg@rabbitmq.com
-platforms       darwin
-description     The RabbitMQ AMQP Server
-long_description    \
+PortSystem 1.0
+name		rabbitmq-server
+version		1.5.5
+revision	0
+categories	net 
+maintainers	tonyg@rabbitmq.com
+platforms	darwin
+description	The RabbitMQ AMQP Server
+long_description	\
     RabbitMQ is an implementation of AMQP, the emerging standard for \
     high performance enterprise messaging. The RabbitMQ server is a \
     robust and scalable implementation of an AMQP broker.
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index eb076e94..0316788f 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -201,9 +201,7 @@ with(Name, F, E) ->
 with(Name, F) ->
     with(Name, F, fun () -> {error, not_found} end).
 with_or_die(Name, F) ->
-    with(Name, F, fun () -> rabbit_misc:protocol_error(
-                              not_found, "no ~s", [rabbit_misc:rs(Name)])
-                  end).
+    with(Name, F, fun () -> rabbit_misc:not_found(Name) end).
 
 list(VHostPath) ->
     mnesia:dirty_match_object(
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 246d0460..e76b71a9 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -300,7 +300,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),
@@ -311,12 +313,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},
@@ -547,6 +566,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,
@@ -574,8 +600,7 @@ handle_method(#'exchange.delete'{exchange = ExchangeNameBin,
     check_configure_permitted(ExchangeName, State),
     case rabbit_exchange:delete(ExchangeName, IfUnused) of
         {error, not_found} ->
-            rabbit_misc:protocol_error(
-              not_found, "no ~s", [rabbit_misc:rs(ExchangeName)]);
+            rabbit_misc:not_found(ExchangeName);
         {error, in_use} ->
             die_precondition_failed(
               "~s in use", [rabbit_misc:rs(ExchangeName)]);
@@ -741,11 +766,9 @@ binding_action(Fun, ExchangeNameBin, QueueNameBin, RoutingKey, Arguments,
     check_read_permitted(ExchangeName, State),
     case Fun(ExchangeName, QueueName, ActualRoutingKey, Arguments) of
         {error, exchange_not_found} ->
-            rabbit_misc:protocol_error(
-              not_found, "no ~s", [rabbit_misc:rs(ExchangeName)]);
+            rabbit_misc:not_found(ExchangeName);
         {error, queue_not_found} ->
-            rabbit_misc:protocol_error(
-              not_found, "no ~s", [rabbit_misc:rs(QueueName)]);
+            rabbit_misc:not_found(QueueName);
         {error, exchange_and_queue_not_found} ->
             rabbit_misc:protocol_error(
               not_found, "no ~s and no ~s", [rabbit_misc:rs(ExchangeName),
@@ -762,26 +785,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}    ->
-            ok = basic_return(Message, WriterPid, ?NO_ROUTE, <<"unroutable">>),
-            [];
-        {error, no_consumers} ->
-            ok = basic_return(Message, WriterPid, ?NO_CONSUMERS, <<"no_consumers">>),
-            []
-    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 625bea90..d809a57a 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 d60725e2..2adac273 100644
--- a/src/rabbit_exchange.erl
+++ b/src/rabbit_exchange.erl
@@ -36,8 +36,7 @@
 
 -export([recover/0, declare/4, 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' |
@@ -74,11 +71,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'}).
@@ -162,9 +156,7 @@ lookup(Name) ->
 lookup_or_die(Name) ->
     case lookup(Name) of
         {ok, X} -> X;
-        {error, not_found} ->
-            rabbit_misc:protocol_error(
-              not_found, "no ~s", [rabbit_misc:rs(Name)])
+        {error, not_found} -> rabbit_misc:not_found(Name)
     end.
 
 list(VHostPath) ->
@@ -193,36 +185,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.
 %%
@@ -249,6 +249,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 eced0b3c..72e16f0f 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -36,9 +36,10 @@
 
 -export([method_record_type/1, polite_pause/0, polite_pause/1]).
 -export([die/1, frame_error/2, protocol_error/3, protocol_error/4]).
+-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]).
@@ -72,16 +73,19 @@
       (atom() | amqp_error(), string(), [any()]) -> no_return()).
 -spec(protocol_error/4 ::
       (atom() | amqp_error(), string(), [any()], atom()) -> no_return()).
+-spec(not_found/1 :: (r(atom())) -> no_return()).
 -spec(get_config/1 :: (atom()) -> {'ok', any()} | not_found()).
 -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').
@@ -139,6 +143,8 @@ protocol_error(Error, Explanation, Params, Method) ->
     CompleteExplanation = lists:flatten(io_lib:format(Explanation, Params)),
     exit({amqp, Error, CompleteExplanation, Method}).
 
+not_found(R) -> protocol_error(not_found, "no ~s", [rs(R)]).
+
 get_config(Key) ->
     case dirty_read({rabbit_config, Key}) of
         {ok, {rabbit_config, Key, V}} -> {ok, V};
@@ -169,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 afc534f9..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' | 'no_consumers'}).
+             {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, no_consumers};
-check_delivery(_   , _   , {_    , Qs}) -> {ok, Qs}.
+check_delivery(true, _   , {false, []}) -> {unroutable, []};
+check_delivery(_   , true, {_    , []}) -> {not_delivered, []};
+check_delivery(_   , _   , {_    , Qs}) -> {routed, Qs}.