diff options
-rw-r--r-- | include/rabbit.hrl | 1 | ||||
-rw-r--r-- | packaging/RPMS/Fedora/rabbitmq-server.spec | 3 | ||||
-rw-r--r-- | packaging/debs/Debian/debian/changelog | 6 | ||||
-rw-r--r-- | packaging/debs/Debian/debian/control | 2 | ||||
-rw-r--r-- | packaging/macports/net/rabbitmq-server/Portfile | 18 | ||||
-rw-r--r-- | src/rabbit_access_control.erl | 10 | ||||
-rw-r--r-- | src/rabbit_basic.erl | 71 | ||||
-rw-r--r-- | src/rabbit_channel.erl | 70 | ||||
-rw-r--r-- | src/rabbit_error_logger.erl | 8 | ||||
-rw-r--r-- | src/rabbit_exchange.erl | 79 | ||||
-rw-r--r-- | src/rabbit_misc.erl | 16 | ||||
-rw-r--r-- | src/rabbit_router.erl | 25 |
12 files changed, 208 insertions, 101 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/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_access_control.erl b/src/rabbit_access_control.erl index 54348d9a..99b912ec 100644 --- a/src/rabbit_access_control.erl +++ b/src/rabbit_access_control.erl @@ -45,11 +45,13 @@ -ifdef(use_specs). +-type(permission_atom() :: 'configure' | 'read' | 'write'). + -spec(check_login/2 :: (binary(), binary()) -> user()). -spec(user_pass_login/2 :: (username(), password()) -> user()). -spec(check_vhost_access/2 :: (user(), vhost()) -> 'ok'). -spec(check_resource_access/3 :: - (username(), r(atom()), non_neg_integer()) -> 'ok'). + (username(), r(atom()), permission_atom()) -> 'ok'). -spec(add_user/2 :: (username(), password()) -> 'ok'). -spec(delete_user/1 :: (username()) -> 'ok'). -spec(change_password/2 :: (username(), password()) -> 'ok'). @@ -137,6 +139,10 @@ check_vhost_access(#user{username = Username}, VHostPath) -> [VHostPath, Username]) end. +permission_index(configure) -> #permission.configure; +permission_index(write) -> #permission.write; +permission_index(read) -> #permission.read. + check_resource_access(Username, R = #resource{kind = exchange, name = <<"">>}, Permission) -> @@ -158,7 +164,7 @@ check_resource_access(Username, [#user_permission{permission = P}] -> case regexp:match( binary_to_list(Name), - binary_to_list(element(Permission, P))) of + binary_to_list(element(permission_index(Permission), P))) of {match, _, _} -> true; nomatch -> false end 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..84b414fd 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -231,13 +231,13 @@ clear_permission_cache() -> ok. check_configure_permitted(Resource, #ch{ username = Username}) -> - check_resource_access(Username, Resource, #permission.configure). + check_resource_access(Username, Resource, configure). check_write_permitted(Resource, #ch{ username = Username}) -> - check_resource_access(Username, Resource, #permission.write). + check_resource_access(Username, Resource, write). check_read_permitted(Resource, #ch{ username = Username}) -> - check_resource_access(Username, Resource, #permission.read). + check_resource_access(Username, Resource, read). expand_queue_name_shortcut(<<>>, #ch{ most_recently_declared_queue = <<>> }) -> rabbit_misc:protocol_error( @@ -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 21f219ae..ca0e337b 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}. |