Support publishing AMQP 1.0 to Event Exchange

 ## What?

Prior to this commit, the `rabbitmq_event_exchange` internally published
always AMQP 0.9.1 messages to the `amq.rabbitmq.event` topic exchange.
This commit allows users to configure the plugin to publish AMQP 1.0
messages instead.

 ## Why?

Prior to this commit, when an AMQP 1.0 client consumed events,
event properties that are lists were omitted, for example property
`client_properties` of event `connection.created` or property
`arguments` of event `queue.created` because of the following sequence:
1. The event exchange plugins listens for all kind of internal events.
2. The event exchange plugin re-publishes all events as AMQP 0.9.1 message to the event exchange.
3. Later, when an AMQP 1.0 client consumes this message, the broker must translate the message from AMQP 0.9.1 to AMQP 1.0.
4. This translation follows the rules outlined in https://www.rabbitmq.com/docs/conversions#amqpl-amqp
5. Specifically, in this table the row before the last one describes the rule we're hitting here. It says that if the AMQP 0.9.1
header value is not an `x-` prefixed header and its value is an array or table, then this header is not converted.
That's because AMQP 1.0 application-properties must be simple types as mandated in https://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-messaging-v1.0-os.html#type-application-properties

 ## How?

The user can configure the plugin as follows to have the plugin
internally publish AMQP 1.0 messages:
```
event_exchange.protocol = amqp_1_0
```

To support complex types such as lists, the plugin sets all event
properties as message-annotations. The plugin prefixes all message
annotation keys with `x-opt-` to comply with the AMQP 1.0 spec.

 ## Alternative Design

An alternative design would have been to format all event properties
e.g. as JSON within the message body. However, this breaks routing on
specific event property values via a headers exchange.

 ## Documentation
rabbitmq/rabbitmq-website#2129

Author: ansd

deps/rabbit/src/mc_amqpl.erl

@@ -43,7 +43,6 @@
 -define(AMQP10_FOOTER, <<"x-amqp-1.0-footer">>).
 -define(PROTOMOD, rabbit_framing_amqp_0_9_1).
 -define(CLASS_ID, 60).
--define(LONGSTR_UTF8_LIMIT, 4096).
 
 -opaque state() :: #content{}.
 
@@ -682,19 +681,13 @@ wrap(_Type, undefined) ->
 wrap(Type, Val) ->
     {Type, Val}.
 
-from_091(longstr, V)
-  when is_binary(V) andalso
-       byte_size(V) =< ?LONGSTR_UTF8_LIMIT ->
-    %% if a longstr is longer than 4096 bytes we just assume it is binary
-    %% it _may_ still be valid utf8 but checking this for every longstr header
-    %% value is going to be excessively slow
-    case mc_util:is_utf8_no_null(V) of
+from_091(longstr, V) ->
+    case mc_util:is_utf8_no_null_limited(V) of
         true ->
             {utf8, V};
         false ->
             {binary, V}
     end;
-from_091(longstr, V) -> {binary, V};
 from_091(long, V) -> {long, V};
 from_091(unsignedbyte, V) -> {ubyte, V};
 from_091(short, V) -> {short, V};

deps/rabbit/src/mc_util.erl

@@ -3,6 +3,7 @@
 -include("mc.hrl").
 
 -export([is_valid_shortstr/1,
+         is_utf8_no_null_limited/1,
          is_utf8_no_null/1,
          uuid_to_urn_string/1,
          urn_string_to_uuid/1,
@@ -12,12 +13,24 @@
          is_x_header/1
         ]).
 
+-define(UTF8_SCAN_LIMIT, 4096).
+
 -spec is_valid_shortstr(term()) -> boolean().
 is_valid_shortstr(Bin) when ?IS_SHORTSTR_LEN(Bin) ->
     is_utf8_no_null(Bin);
 is_valid_shortstr(_) ->
     false.
 
+-spec is_utf8_no_null_limited(term()) -> boolean().
+is_utf8_no_null_limited(Bin)
+  when byte_size(Bin) =< ?UTF8_SCAN_LIMIT ->
+    is_utf8_no_null(Bin);
+is_utf8_no_null_limited(_Term) ->
+    %% If longer than 4096 bytes, just assume it's not UTF-8.
+    %% It _may_ still be valid UTF-8 but checking this
+    %% on the hot path is going to be excessively slow.
+    false.
+
 -spec is_utf8_no_null(term()) -> boolean().
 is_utf8_no_null(Term) ->
     utf8_scan(Term, fun (C) -> C > 0 end).

deps/rabbit/test/mc_unit_SUITE.erl

@@ -313,34 +313,37 @@ amqpl_amqp_bin_amqpl(_Config) ->
     %% incoming amqpl converted to amqp, serialized / deserialized then converted
     %% back to amqpl.
     %% simulates a legacy message published then consumed to a stream
-    Props = #'P_basic'{content_type = <<"text/plain">>,
-                       content_encoding = <<"gzip">>,
-                       headers = [{<<"a-stream-offset">>, long, 99},
-                                  {<<"a-string">>, longstr, <<"a string">>},
-                                  {<<"a-bool">>, bool, false},
-                                  {<<"a-unsignedbyte">>, unsignedbyte, 1},
-                                  {<<"a-unsignedshort">>, unsignedshort, 1},
-                                  {<<"a-unsignedint">>, unsignedint, 1},
-                                  {<<"a-signedint">>, signedint, 1},
-                                  {<<"a-timestamp">>, timestamp, 1},
-                                  {<<"a-double">>, double, 1.0},
-                                  {<<"a-float">>, float, 1.0},
-                                  {<<"a-void">>, void, undefined},
-                                  {<<"a-binary">>, binary, <<"data">>},
-                                  {<<"a-array">>, array, [{long, 1}, {long, 2}]},
-                                  {<<"x-stream-filter">>, longstr, <<"apple">>}
-                                 ],
-                       delivery_mode = 2,
-                       priority = 98,
-                       correlation_id = <<"corr">> ,
-                       reply_to = <<"reply-to">>,
-                       expiration = <<"1">>,
-                       message_id = <<"msg-id">>,
-                       timestamp = 99,
-                       type = <<"45">>,
-                       user_id = <<"banana">>,
-                       app_id = <<"rmq">>
-                      },
+    String5k = binary:copy(<<"x">>, 5000),
+    Props = #'P_basic'{
+               content_type = <<"text/plain">>,
+               content_encoding = <<"gzip">>,
+               headers = [{<<"a-stream-offset">>, long, 99},
+                          {<<"a-string">>, longstr, <<"a string">>},
+                          {<<"a-very-long-string">>, longstr, String5k},
+                          {<<"a-bool">>, bool, false},
+                          {<<"a-unsignedbyte">>, unsignedbyte, 1},
+                          {<<"a-unsignedshort">>, unsignedshort, 1},
+                          {<<"a-unsignedint">>, unsignedint, 1},
+                          {<<"a-signedint">>, signedint, 1},
+                          {<<"a-timestamp">>, timestamp, 1},
+                          {<<"a-double">>, double, 1.0},
+                          {<<"a-float">>, float, 1.0},
+                          {<<"a-void">>, void, undefined},
+                          {<<"a-binary">>, binary, <<"data">>},
+                          {<<"a-array">>, array, [{long, 1}, {long, 2}]},
+                          {<<"x-stream-filter">>, longstr, <<"apple">>}
+                         ],
+               delivery_mode = 2,
+               priority = 98,
+               correlation_id = <<"corr">> ,
+               reply_to = <<"reply-to">>,
+               expiration = <<"1">>,
+               message_id = <<"msg-id">>,
+               timestamp = 99,
+               type = <<"45">>,
+               user_id = <<"banana">>,
+               app_id = <<"rmq">>
+              },
     Content = #content{properties = Props,
                        payload_fragments_rev = [<<"data">>]},
     Msg = mc:init(mc_amqpl, Content, annotations()),
@@ -404,6 +407,9 @@ amqpl_amqp_bin_amqpl(_Config) ->
 
     ?assertEqual({long, 99}, Get(<<"a-stream-offset">>, AP10)),
     ?assertEqual({utf8, <<"a string">>}, Get(<<"a-string">>, AP10)),
+    %% We expect that a very long string is not scanned for valid UTF-8
+    %% and instead directly turned into a binary.
+    ?assertEqual({binary, String5k}, Get(<<"a-very-long-string">>, AP10)),
     ?assertEqual(false, Get(<<"a-bool">>, AP10)),
     ?assertEqual({ubyte, 1}, Get(<<"a-unsignedbyte">>, AP10)),
     ?assertEqual({ushort, 1}, Get(<<"a-unsignedshort">>, AP10)),

deps/rabbitmq_event_exchange/Makefile

@@ -1,6 +1,12 @@
 PROJECT = rabbitmq_event_exchange
 PROJECT_DESCRIPTION = Event Exchange Type
 
+define PROJECT_ENV
+	  [
+		{protocol, amqp_0_9_1}
+	  ]
+endef
+
 define PROJECT_APP_EXTRA_KEYS
 	{broker_version_requirements, []}
 endef

deps/rabbitmq_event_exchange/README.md

@@ -1,154 +1,7 @@
 # RabbitMQ Event Exchange
 
-## Overview
-
-This plugin exposes the internal RabbitMQ event mechanism as messages that clients
-can consume. It's useful
-if you want to keep track of certain events, e.g. when queues, exchanges, bindings, users,
-connections, channels are created and deleted. This plugin filters out stats
-events, so you are almost certainly going to get better results using
-the management plugin for stats.
-
-## How it Works
-
-It declares a topic exchange called `amq.rabbitmq.event` **in the default
-virtual host**. All events are published to this exchange with routing
-keys like 'exchange.created', 'binding.deleted' etc, so you can
-subscribe to only the events you're interested in.
-
-The exchange behaves similarly to 'amq.rabbitmq.log': everything gets
-published there; if you don't trust a user with the information that
-gets published, don't allow them access.
-
-
-## Installation
-
-This plugin ships with RabbitMQ. Like with all other plugins, it must be
-enabled before it can be used:
-
-```bash
-[sudo] rabbitmq-plugins enable rabbitmq_event_exchange
-```
-
-## Event format
-
-Each event has various properties associated with it. These are
-translated into AMQP 0-9-1 data encoding and inserted in the message headers. The
-**message body is always blank**.
-
-## Events
-
-So far RabbitMQ and related plugins emit events with the following routing keys:
-
-### RabbitMQ Broker
-
-Queue, Exchange and Binding events:
-
- * `queue.deleted`
- * `queue.created`
- * `exchange.created`
- * `exchange.deleted`
- * `binding.created`
- * `binding.deleted`
-
-Connection and Channel events:
-
- * `connection.created`
- * `connection.closed`
- * `channel.created`
- * `channel.closed`
-
-Consumer events:
-
- * `consumer.created`
- * `consumer.deleted`
-
-Policy and Parameter events:
-
- * `policy.set`
- * `policy.cleared`
- * `parameter.set`
- * `parameter.cleared`
-
-Virtual host events:
-
- * `vhost.created`
- * `vhost.deleted`
- * `vhost.limits.set`
- * `vhost.limits.cleared`
-
-User related events:
-
- * `user.authentication.success`
- * `user.authentication.failure`
- * `user.created`
- * `user.deleted`
- * `user.password.changed`
- * `user.password.cleared`
- * `user.tags.set`
-
-Permission events:
-
- * `permission.created`
- * `permission.deleted`
- * `topic.permission.created`
- * `topic.permission.deleted`
-
-Alarm events:
-
- * `alarm.set`
- * `alarm.cleared`
-
-### Shovel Plugin
-
-Worker events:
-
- * `shovel.worker.status`
- * `shovel.worker.removed`
-
-### Federation Plugin
-
-Link events:
-
- * `federation.link.status`
- * `federation.link.removed`
-
-## Example
-
-There is a usage example using the Java client in `examples/java`.
-
-
-## Configuration
-
- * `rabbitmq_event_exchange.vhost`: what vhost should the `amq.rabbitmq.event` exchange be declared in. Default: `rabbit.default_vhost` (`<<"/">>`).
-
-
-## Uninstalling
-
-If you want to remove the exchange which this plugin creates, first
-disable the plugin and restart the broker. Then you can delete the exchange,
-e.g. with :
-
-    rabbitmqctl eval 'rabbit_exchange:delete(rabbit_misc:r(<<"/">>, exchange, <<"amq.rabbitmq.event">>), false, <<"username">>).'
-
-
-## Building from Source
-
-Building is no different from [building other RabbitMQ plugins](https://www.rabbitmq.com/plugin-development.html).
-
-TL;DR:
-
-    git clone https://github.com.com/rabbitmq/rabbitmq-public-umbrella.git umbrella
-    cd umbrella
-    make co
-    make up BRANCH=stable
-    cd deps
-    git clone https://github.com/rabbitmq/rabbitmq-event-exchange.git rabbitmq_event_exchange
-    cd rabbitmq_event_exchange
-    make dist
-
+See the [website](https://www.rabbitmq.com/docs/event-exchange) for documentation.
 
 ## License
 
-Released under the Mozilla Public License 2.0,
-the same as RabbitMQ.
+Released under the Mozilla Public License 2.0, the same as RabbitMQ.

deps/rabbitmq_event_exchange/priv/schema/rabbitmq_event_exchange.schema

@@ -5,3 +5,7 @@
 fun(Conf) ->
     list_to_binary(cuttlefish:conf_get("event_exchange.vhost", Conf))
 end}.
+
+{mapping, "event_exchange.protocol", "rabbitmq_event_exchange.protocol", [
+    {datatype, {enum, [amqp_0_9_1, amqp_1_0]}}
+]}.