diff options
-rw-r--r-- | docs/rabbitmqctl.1.xml | 4 | ||||
-rw-r--r-- | include/rabbit.hrl | 2 | ||||
-rw-r--r-- | src/rabbit_access_control.erl | 2 | ||||
-rw-r--r-- | src/rabbit_channel.erl | 17 | ||||
-rw-r--r-- | src/rabbit_error_logger.erl | 2 | ||||
-rw-r--r-- | src/rabbit_exchange.erl | 28 | ||||
-rw-r--r-- | src/rabbit_upgrade_functions.erl | 10 |
7 files changed, 49 insertions, 16 deletions
diff --git a/docs/rabbitmqctl.1.xml b/docs/rabbitmqctl.1.xml index 6b02abe4..392a479a 100644 --- a/docs/rabbitmqctl.1.xml +++ b/docs/rabbitmqctl.1.xml @@ -847,6 +847,10 @@ <listitem><para>Whether the exchange will be deleted automatically when no longer used.</para></listitem> </varlistentry> <varlistentry> + <term>internal</term> + <listitem><para>Whether the exchange is internal, i.e. cannot be directly published to by a client.</para></listitem> + </varlistentry> + <varlistentry> <term>arguments</term> <listitem><para>Exchange arguments.</para></listitem> </varlistentry> diff --git a/include/rabbit.hrl b/include/rabbit.hrl index a1987fb2..2b4347dd 100644 --- a/include/rabbit.hrl +++ b/include/rabbit.hrl @@ -51,7 +51,7 @@ -record(resource, {virtual_host, kind, name}). --record(exchange, {name, type, durable, auto_delete, arguments}). +-record(exchange, {name, type, durable, auto_delete, internal, arguments}). -record(amqqueue, {name, durable, auto_delete, exclusive_owner = none, arguments, pid}). diff --git a/src/rabbit_access_control.erl b/src/rabbit_access_control.erl index bc588013..1826347d 100644 --- a/src/rabbit_access_control.erl +++ b/src/rabbit_access_control.erl @@ -316,7 +316,7 @@ add_vhost(VHostPath) -> write), [rabbit_exchange:declare( rabbit_misc:r(VHostPath, exchange, Name), - Type, true, false, []) || + Type, true, false, false, []) || {Name,Type} <- [{<<"">>, direct}, {<<"amq.direct">>, direct}, diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index 19613a57..b46bf1b4 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -350,6 +350,17 @@ check_write_permitted(Resource, #ch{username = Username}) -> check_read_permitted(Resource, #ch{username = Username}) -> check_resource_access(Username, Resource, read). +check_internal_exchange(#exchange{name = Name, + internal = IsInternal}) -> + case IsInternal of + true -> + rabbit_misc:protocol_error(access_refused, + "cannot publish to internal exchange: " + ++ "~p~n", + [Name]); + false -> ok + end. + expand_queue_name_shortcut(<<>>, #ch{most_recently_declared_queue = <<>>}) -> rabbit_misc:protocol_error( not_found, "no previously declared queue", []); @@ -444,6 +455,7 @@ handle_method(#'basic.publish'{exchange = ExchangeNameBin, ExchangeName = rabbit_misc:r(VHostPath, exchange, ExchangeNameBin), check_write_permitted(ExchangeName, State), Exchange = rabbit_exchange:lookup_or_die(ExchangeName), + check_internal_exchange(Exchange), %% We decode the content's properties here because we're almost %% certain to want to look at delivery-mode and priority. DecodedContent = rabbit_binary_parser:ensure_content_decoded(Content), @@ -689,7 +701,7 @@ handle_method(#'exchange.declare'{exchange = ExchangeNameBin, passive = false, durable = Durable, auto_delete = AutoDelete, - internal = false, + internal = Internal, nowait = NoWait, arguments = Args}, _, State = #ch{virtual_host = VHostPath}) -> @@ -712,10 +724,11 @@ handle_method(#'exchange.declare'{exchange = ExchangeNameBin, CheckedType, Durable, AutoDelete, + Internal, Args) end, ok = rabbit_exchange:assert_equivalence(X, CheckedType, Durable, - AutoDelete, Args), + AutoDelete, Internal, Args), return_ok(State, NoWait, #'exchange.declare_ok'{}); handle_method(#'exchange.declare'{exchange = ExchangeNameBin, diff --git a/src/rabbit_error_logger.erl b/src/rabbit_error_logger.erl index 42861f86..dd009c83 100644 --- a/src/rabbit_error_logger.erl +++ b/src/rabbit_error_logger.erl @@ -49,7 +49,7 @@ boot() -> init([DefaultVHost]) -> #exchange{} = rabbit_exchange:declare( rabbit_misc:r(DefaultVHost, exchange, ?LOG_EXCH_NAME), - topic, true, false, []), + topic, true, false, false, []), {ok, #resource{virtual_host = DefaultVHost, kind = exchange, name = ?LOG_EXCH_NAME}}. diff --git a/src/rabbit_exchange.erl b/src/rabbit_exchange.erl index 61a24388..3a7eaaad 100644 --- a/src/rabbit_exchange.erl +++ b/src/rabbit_exchange.erl @@ -33,11 +33,11 @@ -include("rabbit.hrl"). -include("rabbit_framing.hrl"). --export([recover/0, declare/5, lookup/1, lookup_or_die/1, list/1, info_keys/0, +-export([recover/0, declare/6, lookup/1, lookup_or_die/1, list/1, info_keys/0, info/1, info/2, info_all/1, info_all/2, publish/2, delete/2]). %% this must be run inside a mnesia tx -export([maybe_auto_delete/1]). --export([assert_equivalence/5, assert_args_equivalence/2, check_type/1]). +-export([assert_equivalence/6, assert_args_equivalence/2, check_type/1]). %%---------------------------------------------------------------------------- @@ -49,13 +49,14 @@ -type(type() :: atom()). -spec(recover/0 :: () -> 'ok'). --spec(declare/5 :: - (name(), type(), boolean(), boolean(), rabbit_framing:amqp_table()) +-spec(declare/6 :: + (name(), type(), boolean(), boolean(), boolean(), + rabbit_framing:amqp_table()) -> rabbit_types:exchange()). -spec(check_type/1 :: (binary()) -> atom() | rabbit_types:connection_exit()). --spec(assert_equivalence/5 :: - (rabbit_types:exchange(), atom(), boolean(), boolean(), +-spec(assert_equivalence/6 :: + (rabbit_types:exchange(), atom(), boolean(), boolean(), boolean(), rabbit_framing:amqp_table()) -> 'ok' | rabbit_types:connection_exit()). -spec(assert_args_equivalence/2 :: @@ -90,7 +91,7 @@ %%---------------------------------------------------------------------------- --define(INFO_KEYS, [name, type, durable, auto_delete, arguments]). +-define(INFO_KEYS, [name, type, durable, auto_delete, internal, arguments]). recover() -> Xs = rabbit_misc:table_fold( @@ -113,11 +114,12 @@ recover_with_bindings(Bs, [X = #exchange{type = Type} | Xs], Bindings) -> recover_with_bindings([], [], []) -> ok. -declare(XName, Type, Durable, AutoDelete, Args) -> +declare(XName, Type, Durable, AutoDelete, Internal, Args) -> X = #exchange{name = XName, type = Type, durable = Durable, auto_delete = AutoDelete, + internal = Internal, arguments = Args}, %% We want to upset things if it isn't ok; this is different from %% the other hooks invocations, where we tend to ignore the return @@ -170,14 +172,17 @@ check_type(TypeBin) -> assert_equivalence(X = #exchange{ durable = Durable, auto_delete = AutoDelete, + internal = Internal, type = Type}, - Type, Durable, AutoDelete, RequiredArgs) -> + Type, Durable, AutoDelete, Internal, RequiredArgs) -> (type_to_module(Type)):assert_args_equivalence(X, RequiredArgs); -assert_equivalence(#exchange{ name = Name }, _Type, _Durable, _AutoDelete, +assert_equivalence(#exchange{ name = Name }, + _Type, _Durable, _Internal, _AutoDelete, _Args) -> rabbit_misc:protocol_error( not_allowed, - "cannot redeclare ~s with different type, durable or autodelete value", + "cannot redeclare ~s with different type, durable, " ++ + "internal or autodelete value", [rabbit_misc:rs(Name)]). assert_args_equivalence(#exchange{ name = Name, arguments = Args }, @@ -215,6 +220,7 @@ i(name, #exchange{name = Name}) -> Name; i(type, #exchange{type = Type}) -> Type; i(durable, #exchange{durable = Durable}) -> Durable; i(auto_delete, #exchange{auto_delete = AutoDelete}) -> AutoDelete; +i(internal, #exchange{internal = Internal}) -> Internal; i(arguments, #exchange{arguments = Arguments}) -> Arguments; i(Item, _) -> throw({bad_argument, Item}). diff --git a/src/rabbit_upgrade_functions.erl b/src/rabbit_upgrade_functions.erl index 1c56d51d..2ba885d0 100644 --- a/src/rabbit_upgrade_functions.erl +++ b/src/rabbit_upgrade_functions.erl @@ -27,6 +27,7 @@ -rabbit_upgrade({remove_user_scope, []}). -rabbit_upgrade({hash_passwords, []}). -rabbit_upgrade({add_ip_to_listener, []}). +-rabbit_upgrade({add_internal_to_exchange, []}). %% ------------------------------------------------------------------- @@ -35,6 +36,7 @@ -spec(remove_user_scope/0 :: () -> 'ok'). -spec(hash_passwords/0 :: () -> 'ok'). -spec(add_ip_to_listener/0 :: () -> 'ok'). +-spec(add_internal_to_exchange/0 :: () -> 'ok'). -endif. @@ -71,6 +73,14 @@ add_ip_to_listener() -> end, [node, protocol, host, ip_address, port]). +add_internal_to_exchange() -> + mnesia( + rabbit_exchange, + fun ({exchange, Name, Type, Durable, AutoDelete, Args}) -> + {exchange, Name, Type, Durable, AutoDelete, false, Args} + end, + [name, type, durable, auto_delete, internal, arguments]). + %%-------------------------------------------------------------------- mnesia(TableName, Fun, FieldList) -> |