rabbitmq-server/deps/rabbitmq_mqtt/BUILD.bazel

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

311 lines
6.7 KiB
Python
Raw Normal View History

load("@rules_erlang//:eunit2.bzl", "eunit")
load("@rules_erlang//:xref2.bzl", "xref")
load("@rules_erlang//:dialyze.bzl", "dialyze", "plt")
2021-04-19 15:50:15 +08:00
load(
"//:rabbitmq.bzl",
"BROKER_VERSION_REQUIREMENTS_ANY",
"RABBITMQ_DIALYZER_OPTS",
"assert_suites",
2021-05-11 18:03:27 +08:00
"broker_for_integration_suites",
"rabbitmq_app",
2021-05-11 18:03:27 +08:00
"rabbitmq_integration_suite",
"rabbitmq_suite",
)
load(
":app.bzl",
"all_beam_files",
"all_srcs",
"all_test_beam_files",
"test_suite_beam_files",
2021-04-19 15:50:15 +08:00
)
APP_NAME = "rabbitmq_mqtt"
APP_DESCRIPTION = "RabbitMQ MQTT Adapter"
APP_MODULE = "rabbit_mqtt"
APP_ENV = """[
{ssl_cert_login,false},
{allow_anonymous, true},
{vhost, <<"/">>},
{exchange, <<"amq.topic">>},
2023-07-13 22:38:47 +08:00
{max_session_expiry_interval_seconds, 86400}, %% 1 day
2021-04-19 15:50:15 +08:00
{retained_message_store, rabbit_mqtt_retained_msg_store_dets},
%% only used by DETS store
{retained_message_store_dets_sync_interval, 2000},
{prefetch, 10},
{ssl_listeners, []},
{tcp_listeners, [1883]},
{num_tcp_acceptors, 10},
{num_ssl_acceptors, 10},
{tcp_listen_options, [{backlog, 128},
{nodelay, true},
{send_timeout, 15000},
{send_timeout_close, true}
]},
2021-04-19 15:50:15 +08:00
{proxy_protocol, false},
{sparkplug, false},
{mailbox_soft_limit, 200},
{max_packet_size_unauthenticated, 65536},
%% 256 MB is upper limit defined by MQTT spec
%% We set 16 MB as defined in deps/rabbit/Makefile max_message_size
{max_packet_size_authenticated, 16777216},
{topic_alias_maximum, 16}
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
]
"""
2021-04-19 15:50:15 +08:00
all_beam_files(name = "all_beam_files")
all_test_beam_files(name = "all_test_beam_files")
all_srcs(name = "all_srcs")
2021-04-19 15:50:15 +08:00
test_suite_beam_files(name = "test_suite_beam_files")
2021-04-19 15:50:15 +08:00
# gazelle:erlang_app_extra_app ssl
rabbitmq_app(
name = "erlang_app",
srcs = [":all_srcs"],
hdrs = [":public_hdrs"],
2021-04-19 15:50:15 +08:00
app_description = APP_DESCRIPTION,
app_env = APP_ENV,
app_extra_keys = BROKER_VERSION_REQUIREMENTS_ANY,
2021-04-19 15:50:15 +08:00
app_module = APP_MODULE,
app_name = APP_NAME,
beam_files = [":beam_files"],
extra_apps = ["ssl"],
license_files = [":license_files"],
priv = [":priv"],
deps = [
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
"//deps/amqp10_common:erlang_app",
"//deps/rabbit:erlang_app",
"//deps/rabbit_common:erlang_app",
"@ranch//:erlang_app",
],
2021-04-19 15:50:15 +08:00
)
xref(
name = "xref",
additional_libs = [
"//deps/rabbitmq_cli:erlang_app", # keep
],
target = ":erlang_app",
)
plt(
name = "deps_plt",
for_target = ":erlang_app",
ignore_warnings = True,
libs = ["@rules_elixir//elixir"], # keep
plt = "//:base_plt",
deps = ["//deps/rabbitmq_cli:erlang_app"], # keep
)
dialyze(
name = "dialyze",
dialyzer_opts = RABBITMQ_DIALYZER_OPTS,
plt = ":deps_plt",
target = ":erlang_app",
)
eunit(
name = "eunit",
compiled_suites = [
":test_rabbit_auth_backend_mqtt_mock_beam",
":test_event_recorder_beam",
":test_util_beam",
],
target = ":test_erlang_app",
)
broker_for_integration_suites(
extra_plugins = [
"//deps/rabbitmq_management:erlang_app",
"//deps/rabbitmq_web_mqtt:erlang_app",
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
"//deps/rabbitmq_consistent_hash_exchange:erlang_app",
"//deps/rabbitmq_stomp:erlang_app",
"//deps/rabbitmq_stream:erlang_app",
],
)
2021-04-19 15:50:15 +08:00
rabbitmq_integration_suite(
name = "auth_SUITE",
additional_beam = [
"test/rabbit_auth_backend_mqtt_mock.beam",
"test/util.beam",
],
2024-08-30 18:35:28 +08:00
shard_count = 18,
runtime_deps = [
"@emqtt//:erlang_app",
"@meck//:erlang_app",
],
)
rabbitmq_integration_suite(
name = "cluster_SUITE",
size = "large",
additional_beam = [
":test_util_beam",
],
flaky = True,
shard_count = 4,
2023-06-22 12:02:27 +08:00
sharding_method = "case",
runtime_deps = [
"@emqtt//:erlang_app",
],
)
rabbitmq_integration_suite(
name = "command_SUITE",
additional_beam = [
":test_util_beam",
],
runtime_deps = [
"@emqtt//:erlang_app",
],
)
rabbitmq_integration_suite(
name = "config_SUITE",
)
rabbitmq_integration_suite(
name = "config_schema_SUITE",
)
rabbitmq_integration_suite(
name = "java_SUITE",
additional_beam = [
":test_util_beam",
],
2023-06-22 12:02:27 +08:00
shard_count = 2,
sharding_method = "group",
)
rabbitmq_suite(
name = "processor_SUITE",
size = "small",
Allow to use Khepri database to store metadata instead of Mnesia [Why] Mnesia is a very powerful and convenient tool for Erlang applications: it is a persistent disc-based database, it handles replication accross multiple Erlang nodes and it is available out-of-the-box from the Erlang/OTP distribution. RabbitMQ relies on Mnesia to manage all its metadata: * virtual hosts' properties * intenal users * queue, exchange and binding declarations (not queues data) * runtime parameters and policies * ... Unfortunately Mnesia makes it difficult to handle network partition and, as a consequence, the merge conflicts between Erlang nodes once the network partition is resolved. RabbitMQ provides several partition handling strategies but they are not bullet-proof. Users still hit situations where it is a pain to repair a cluster following a network partition. [How] @kjnilsson created Ra [1], a Raft consensus library that RabbitMQ already uses successfully to implement quorum queues and streams for instance. Those queues do not suffer from network partitions. We created Khepri [2], a new persistent and replicated database engine based on Ra and we want to use it in place of Mnesia in RabbitMQ to solve the problems with network partitions. This patch integrates Khepri as an experimental feature. When enabled, RabbitMQ will store all its metadata in Khepri instead of Mnesia. This change comes with behavior changes. While Khepri remains disabled, you should see no changes to the behavior of RabbitMQ. If there are changes, it is a bug. After Khepri is enabled, there are significant changes of behavior that you should be aware of. Because it is based on the Raft consensus algorithm, when there is a network partition, only the cluster members that are in the partition with at least `(Number of nodes in the cluster ÷ 2) + 1` number of nodes can "make progress". In other words, only those nodes may write to the Khepri database and read from the database and expect a consistent result. For instance in a cluster of 5 RabbitMQ nodes: * If there are two partitions, one with 3 nodes, one with 2 nodes, only the group of 3 nodes will be able to write to the database. * If there are three partitions, two with 2 nodes, one with 1 node, none of the group can write to the database. Because the Khepri database will be used for all kind of metadata, it means that RabbitMQ nodes that can't write to the database will be unable to perform some operations. A list of operations and what to expect is documented in the associated pull request and the RabbitMQ website. This requirement from Raft also affects the startup of RabbitMQ nodes in a cluster. Indeed, at least a quorum number of nodes must be started at once to allow nodes to become ready. To enable Khepri, you need to enable the `khepri_db` feature flag: rabbitmqctl enable_feature_flag khepri_db When the `khepri_db` feature flag is enabled, the migration code performs the following two tasks: 1. It synchronizes the Khepri cluster membership from the Mnesia cluster. It uses `mnesia_to_khepri:sync_cluster_membership/1` from the `khepri_mnesia_migration` application [3]. 2. It copies data from relevant Mnesia tables to Khepri, doing some conversion if necessary on the way. Again, it uses `mnesia_to_khepri:copy_tables/4` from `khepri_mnesia_migration` to do it. This can be performed on a running standalone RabbitMQ node or cluster. Data will be migrated from Mnesia to Khepri without any service interruption. Note that during the migration, the performance may decrease and the memory footprint may go up. Because this feature flag is considered experimental, it is not enabled by default even on a brand new RabbitMQ deployment. More about the implementation details below: In the past months, all accesses to Mnesia were isolated in a collection of `rabbit_db*` modules. This is where the integration of Khepri mostly takes place: we use a function called `rabbit_khepri:handle_fallback/1` which selects the database and perform the query or the transaction. Here is an example from `rabbit_db_vhost`: * Up until RabbitMQ 3.12.x: get(VHostName) when is_binary(VHostName) -> get_in_mnesia(VHostName). * Starting with RabbitMQ 3.13.0: get(VHostName) when is_binary(VHostName) -> rabbit_khepri:handle_fallback( #{mnesia => fun() -> get_in_mnesia(VHostName) end, khepri => fun() -> get_in_khepri(VHostName) end}). This `rabbit_khepri:handle_fallback/1` function relies on two things: 1. the fact that the `khepri_db` feature flag is enabled, in which case it always executes the Khepri-based variant. 4. the ability or not to read and write to Mnesia tables otherwise. Before the feature flag is enabled, or during the migration, the function will try to execute the Mnesia-based variant. If it succeeds, then it returns the result. If it fails because one or more Mnesia tables can't be used, it restarts from scratch: it means the feature flag is being enabled and depending on the outcome, either the Mnesia-based variant will succeed (the feature flag couldn't be enabled) or the feature flag will be marked as enabled and it will call the Khepri-based variant. The meat of this function really lives in the `khepri_mnesia_migration` application [3] and `rabbit_khepri:handle_fallback/1` is a wrapper on top of it that knows about the feature flag. However, some calls to the database do not depend on the existence of Mnesia tables, such as functions where we need to learn about the members of a cluster. For those, we can't rely on exceptions from Mnesia. Therefore, we just look at the state of the feature flag to determine which database to use. There are two situations though: * Sometimes, we need the feature flag state query to block because the function interested in it can't return a valid answer during the migration. Here is an example: case rabbit_khepri:is_enabled(RemoteNode) of true -> can_join_using_khepri(RemoteNode); false -> can_join_using_mnesia(RemoteNode) end * Sometimes, we need the feature flag state query to NOT block (for instance because it would cause a deadlock). Here is an example: case rabbit_khepri:get_feature_state() of enabled -> members_using_khepri(); _ -> members_using_mnesia() end Direct accesses to Mnesia still exists. They are limited to code that is specific to Mnesia such as classic queue mirroring or network partitions handling strategies. Now, to discover the Mnesia tables to migrate and how to migrate them, we use an Erlang module attribute called `rabbit_mnesia_tables_to_khepri_db` which indicates a list of Mnesia tables and an associated converter module. Here is an example in the `rabbitmq_recent_history_exchange` plugin: -rabbit_mnesia_tables_to_khepri_db( [{?RH_TABLE, rabbit_db_rh_exchange_m2k_converter}]). The converter module — `rabbit_db_rh_exchange_m2k_converter` in this example — is is fact a "sub" converter module called but `rabbit_db_m2k_converter`. See the documentation of a `mnesia_to_khepri` converter module to learn more about these modules. [1] https://github.com/rabbitmq/ra [2] https://github.com/rabbitmq/khepri [3] https://github.com/rabbitmq/khepri_mnesia_migration See #7206. Co-authored-by: Jean-Sébastien Pédron <jean-sebastien@rabbitmq.com> Co-authored-by: Diana Parra Corbacho <dparracorbac@vmware.com> Co-authored-by: Michael Davis <mcarsondavis@gmail.com>
2023-01-05 20:57:50 +08:00
runtime_deps = [
"@meck//:erlang_app",
],
deps = [
"//deps/amqp_client:erlang_app",
"//deps/rabbit_common:erlang_app",
],
)
rabbitmq_integration_suite(
name = "proxy_protocol_SUITE",
2023-06-14 19:09:37 +08:00
additional_beam = [
":test_util_beam",
],
)
rabbitmq_integration_suite(
name = "reader_SUITE",
additional_beam = [
":test_util_beam",
":test_event_recorder_beam",
],
runtime_deps = [
"@emqtt//:erlang_app",
],
)
rabbitmq_integration_suite(
name = "retainer_SUITE",
additional_beam = [
":test_util_beam",
],
2023-02-22 02:24:15 +08:00
shard_count = 6,
runtime_deps = [
"@emqtt//:erlang_app",
],
)
rabbitmq_integration_suite(
name = "mqtt_shared_SUITE",
size = "large",
additional_beam = [
":test_util_beam",
":test_event_recorder_beam",
],
shard_count = 5,
runtime_deps = [
"//deps/rabbitmq_management_agent:erlang_app",
"@emqtt//:erlang_app",
"@gun//:erlang_app",
"@meck//:erlang_app",
],
)
rabbitmq_integration_suite(
2023-04-23 15:53:44 +08:00
name = "v5_SUITE",
Support MQTT 5.0 features No Local, RAP, Subscription IDs Support subscription options "No Local" and "Retain As Published" as well as Subscription Identifiers. All three MQTT 5.0 features can be set on a per subscription basis. Due to wildcards in topic filters, multiple subscriptions can match a given topic. Therefore, to implement Retain As Published and Subscription Identifiers, the destination MQTT connection process needs to know what subscription(s) caused it to receive the message. There are a few ways how this could be implemented: 1. The destination MQTT connection process is aware of all its subscriptions. Whenever, it receives a message, it can match the message's routing key / topic against all its known topic filters. However, to iteratively match the routing key against all topic filters for every received message can become very expensive in the worst case when the MQTT client creates many subscriptions containing wildcards. This could be the case for an MQTT client that acts as a bridge or proxy or dispatcher: It could subscribe via a wildcard for each of its own clients. 2. Instead of interatively matching the topic of the received message against all topic filters that contain wildcards, a better approach would be for every MQTT subscriber connection process to maintain a local trie datastructure (similar to how topic exchanges are implemented) and perform matching therefore more efficiently. However, this does not sound optimal either because routing is effectively performed twice: in the topic exchange and again against a much smaller trie in each destination connection process. 3. Given that the topic exchange already perform routing, a much more sensible way would be to send the matched binding key(s) to the destination MQTT connection process. A subscription (topic filter) maps to a binding key in AMQP 0.9.1 routing. Therefore, for the first time in RabbitMQ, the routing function should not only output a list of unique destination queues, but also the binding keys (subscriptions) that caused the message to be routed to the destination queue. This commit therefore implements the 3rd approach. The downside of the 3rd approach is that it requires API changes to the routing function and topic exchange. Specifically, this commit adds a new function rabbit_exchange:route/3 that accepts a list of routing options. If that list contains version 2, the caller of the routing function knows how to handle the return value that could also contain binding keys. This commits allows an MQTT connection process, the channel process, and at-most-once dead lettering to handle binding keys. Binding keys are included as AMQP 0.9.1 headers into the basic message. Therefore, whenever a message is sent from an MQTT client or AMQP 0.9.1 client or AMQP 1.0 client or STOMP client, the MQTT receiver will know the subscription identifier that caused the message to be received. Note that due to the low number of allowed wildcard characters (# and +), the cardinality of matched binding keys shouldn't be high even if the topic contains for example 3 levels and the message is sent to for example 5 million destination queues. In other words, sending multiple distinct basic messages to the destination shouldn't hurt the delegate optimisation too much. The delegate optimisation implemented for classic queues and rabbit_mqtt_qos0_queue(s) still takes place for all basic messages that contain the same set of matched binding keys. The topic exchange returns all matched binding keys by remembering the edges walked down to the leaves. As an optimisation, only for MQTT queues are binding keys being returned. This does add a small dependency from app rabbit to app rabbitmq_mqtt which is not optimal. However, this dependency should be simple to remove when omitting this optimisation. Another important feature of this commit is persisting subscription options and subscription identifiers because they are part of the MQTT 5.0 session state. In MQTT v3 and v4, the only subscription information that were part of the session state was the topic filter and the QoS level. Both information were implicitly stored in the form of bindings: The topic filter as the binding key and the QoS level as the destination queue name of the binding. For MQTT v5 we need to persist more subscription information. From a domain perspective, it makes sense to store subscription options as part of subscriptions, i.e. bindings, even though they are currently not used in routing. Therefore, this commits stores subscription options as binding arguments. Storing subscription options as binding arguments comes in turn with new challenges: How to handle mixed version clusters and upgrading an MQTT session from v3 or v4 to v5? Imagine an MQTT client connects via v5 with Session Expiry Interval > 0 to a new node in a mixed version cluster, creates a subscription, disconnects, and subsequently connects via v3 to an old node. The client should continue to receive messages. To simplify such edge cases, this commit introduces a new feature flag called mqtt_v5. If mqtt_v5 is disabled, clients cannot connect to RabbitMQ via MQTT 5.0. This still doesn't entirely solve the problem of MQTT session upgrades (v4 to v5 client) or session downgrades (v5 to v4 client). Ideally, once mqtt_v5 is enabled, all MQTT bindings contain non-empty binding arguments. However, this will require a feature flag migration function to modify all MQTT bindings. To be more precise, all MQTT bindings need to be deleted and added because the binding argument is part of the Mnesia table key. Since feature flag migration functions are non-trivial to implement in RabbitMQ (they can run on every node multiple times and concurrently), this commit takes a simpler approach: All v3 / v4 sessions keep the empty binding argument []. All v5 sessions use the new binding argument [#mqtt_subscription_opts{}]. This requires only handling a session upgrade / downgrade by creating a binding (with the new binding arg) and deleting the old binding (with the old binding arg) when processing the CONNECT packet. Note that such session upgrades or downgrades should be rather rare in practice. Therefore these binding transactions shouldn't hurt peformance. The No Local option is implemented within the MQTT publishing connection process: The message is not sent to the MQTT destination if the destination queue name matches the current MQTT client ID and the message was routed due to a subscription that has the No Local flag set. This avoids unnecessary traffic on the MQTT queue. The alternative would have been that the "receiving side" (same process) filters the message out - which would have been more consistent in how Retain As Published and Subscription Identifiers are implemented, but would have caused unnecessary load on the MQTT queue.
2023-04-19 21:32:34 +08:00
size = "large",
2023-04-23 15:53:44 +08:00
additional_beam = [
":test_util_beam",
2023-04-23 15:53:44 +08:00
],
shard_count = 2,
2023-04-23 15:53:44 +08:00
runtime_deps = [
"@emqtt//:erlang_app",
"@gun//:erlang_app",
],
)
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
rabbitmq_integration_suite(
name = "protocol_interop_SUITE",
size = "medium",
additional_beam = [
":test_util_beam",
],
shard_count = 2,
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
runtime_deps = [
Introduce new AMQP 1.0 address format ## What? Introduce a new address format (let's call it v2) for AMQP 1.0 source and target addresses. The old format (let's call it v1) is described in https://github.com/rabbitmq/rabbitmq-server/tree/v3.13.x/deps/rabbitmq_amqp1_0#routing-and-addressing The only v2 source address format is: ``` /queue/:queue ``` The 4 possible v2 target addresses formats are: ``` /exchange/:exchange/key/:routing-key /exchange/:exchange /queue/:queue <null> ``` where the last AMQP <null> value format requires that each message’s `to` field contains one of: ``` /exchange/:exchange/key/:routing-key /exchange/:exchange /queue/:queue ``` ## Why? The AMQP address v1 format comes with the following flaws: 1. Obscure address format: Without reading the documentation, the differences for example between source addresses ``` /amq/queue/:queue /queue/:queue :queue ``` are unknown to users. Hence, the address format is obscure. 2. Implicit creation of topologies Some address formats implicitly create queues (and bindings), such as source address ``` /exchange/:exchange/:binding-key ``` or target address ``` /queue/:queue ``` These queues and bindings are never deleted (by the AMQP 1.0 plugin.) Implicit creation of such topologies is also obscure. 3. Redundant address formats ``` /queue/:queue :queue ``` have the same meaning and are therefore redundant. 4. Properties section must be parsed to determine whether a routing key is present Target address ``` /exchange/:exchange ``` requires RabbitMQ to parse the properties section in order to check whether the message `subject` is set. If `subject` is not set, the routing key will default to the empty string. 5. Using `subject` as routing key misuses the purpose of this field. According to the AMQP spec, the message `subject` field's purpose is: > A common field for summary information about the message content and purpose. 6. Exchange names, queue names and routing keys must not contain the "/" (slash) character. The current 3.13 implemenation splits by "/" disallowing these characters in exchange, and queue names, and routing keys which is unnecessary prohibitive. 7. Clients must create a separate link per target exchange While this is reasonable working assumption, there might be rare use cases where it could make sense to create many exchanges (e.g. 1 exchange per queue, see https://github.com/rabbitmq/rabbitmq-server/discussions/10708) and have a single application publish to all these exchanges. With the v1 address format, for an application to send to 500 different exchanges, it needs to create 500 links. Due to these disadvantages and thanks to #10559 which allows clients to explicitly create topologies, we can create a simpler, clearer, and better v2 address format. ## How? ### Design goals Following the 7 cons from v1, the design goals for v2 are: 1. The address format should be simple so that users have a chance to understand the meaning of the address without necessarily consulting the docs. 2. The address format should not implicitly create queues, bindings, or exchanges. Instead, topologies should be created either explicitly via the new management node prior to link attachment (see #10559), or in future, we might support the `dynamic` source or target properties so that RabbitMQ creates queues dynamically. 3. No redundant address formats. 4. The target address format should explicitly state whether the routing key is present, empty, or will be provided dynamically in each message. 5. `Subject` should not be used as routing key. Instead, a better fitting field should be used. 6. Exchange names, queue names, and routing keys should allow to contain valid UTF-8 encoded data including the "/" character. 7. Allow both target exchange and routing key to by dynamically provided within each message. Furthermore 8. v2 must co-exist with v1 for at least some time. Applications should be able to upgrade to RabbitMQ 4.0 while continuing to use v1. Examples include AMQP 1.0 shovels and plugins communicating between a 4.0 and a 3.13 cluster. Starting with 4.1, we should change the AMQP 1.0 shovel and plugin clients to use only the new v2 address format. This will allow AMQP 1.0 and plugins to communicate between a 4.1 and 4.2 cluster. We will deprecate v1 in 4.0 and remove support for v1 in a later 4.x version. ### Additional Context The address is usually a String, but can be of any type. The [AMQP Addressing extension](https://docs.oasis-open.org/amqp/addressing/v1.0/addressing-v1.0.html) suggests that addresses are URIs and are therefore hierarchical and could even contain query parameters: > An AMQP address is a URI reference as defined by RFC3986. > the path expression is a sequence of identifier segments that reflects a path through an > implementation specific relationship graph of AMQP nodes and their termini. > The path expression MUST resolve to a node’s terminus in an AMQP container. The [Using the AMQP Anonymous Terminus for Message Routing Version 1.0](https://docs.oasis-open.org/amqp/anonterm/v1.0/cs01/anonterm-v1.0-cs01.html) extension allows for the target being `null` and the `To` property to contain the node address. This corresponds to AMQP 0.9.1 where clients can send each message on the same channel to a different `{exchange, routing-key}` destination. The following v2 address formats will be used. ### v2 addresses A new deprecated feature flag `amqp_address_v1` will be introduced in 4.0 which is permitted by default. Starting with 4.1, we should change the AMQP 1.0 shovel and plugin AMQP 1.0 clients to use only the new v2 address format. However, 4.1 server code must still understand the 4.0 AMQP 1.0 shovel and plugin AMQP 1.0 clients’ v1 address format. The new deprecated feature flag will therefore be denied by default in 4.2. This allows AMQP 1.0 shovels and plugins to work between * 4.0 and 3.13 clusters using v1 * 4.1 and 4.0 clusters using v2 from 4.1 to v4.0 and v1 from 4.0 to 4.1 * 4.2 and 4.1 clusters using v2 without having to support both v1 and v2 at the same time in the AMQP 1.0 shovel and plugin clients. While supporting both v1 and v2 in these clients is feasible, it's simpler to switch the client code directly from v1 to v2. ### v2 source addresses The source address format is ``` /queue/:queue ``` If the deprecated feature flag `amqp_address_v1` is permitted and the queue does not exist, the queue will be auto-created. If the deprecated feature flag `amqp_address_v1` is denied, the queue must exist. ### v2 target addresses v1 requires attaching a new link for each destination exchange. v2 will allow dynamic `{exchange, routing-key}` combinations for a given link. v2 therefore allows for the rare use cases where a single AMQP 1.0 publisher app needs to send to many different exchanges. Setting up a link per destination exchange could be cumbersome. Hence, v2 will support the dynamic `{exchange, routing-key}` combinations of AMQP 0.9.1. To achieve this, we make use of the "Anonymous Terminus for Message Routing" extension: The target address will contain the AMQP value null. The `To` field in each message must be set and contain either address format ``` /exchange/:exchange/key/:routing-key ``` or ``` /exchange/:exchange ``` when using the empty routing key. The `to` field requires an address type and is better suited than the `subject field. Note that each message will contain this `To` value for the anonymous terminus. Hence, we should save some bytes being sent across the network and stored on disk. Using a format ``` /e/:exchange/k/:routing-key ``` saves more bytes, but is too obscure. However, we use only `/key/` instead of `/routing-key/` so save a few bytes. This also simplifies the format because users don’t have to remember whether to use spell `routing-key` or `routing_key` or `routingkey`. The other allowed target address formats are: ``` /exchange/:exchange/key/:routing-key ``` where exchange and routing key are static on the given link. ``` /exchange/:exchange ``` where exchange and routing key are static on the given link, and routing key will be the empty string (useful for example for the fanout exchange). ``` /queue/:queue ``` This provides RabbitMQ beginners the illusion of sending a message directly to a queue without having to understand what exchanges and routing keys are. If the deprecated feature flag `amqp_address_v1` is permitted and the queue does not exist, the queue will be auto-created. If the deprecated feature flag `amqp_address_v1` is denied, the queue must exist. Besides the additional queue existence check, this queue target is different from ``` /exchange//key/:queue ``` in that queue specific optimisations might be done (in future) by RabbitMQ (for example different receiving queue types could grant different amounts of link credits to the sending clients). A write permission check to the amq.default exchange will be performed nevertheless. v2 will prohibit the v1 static link & dynamic routing-key combination where the routing key is sent in the message `subject` as that’s also obscure. For this use case, v2’s new anonymous terminus can be used where both exchange and routing key are defined in the message’s `To` field. (The bare message must not be modified because it could be signed.) The alias format ``` /topic/:topic ``` will also be removed. Sending to topic exchanges is arguably an advanced feature. Users can directly use the format ``` /exchange/amq.topic/key/:topic ``` which reduces the number of redundant address formats. ### v2 address format reference To sump up (and as stated at the top of this commit message): The only v2 source address format is: ``` /queue/:queue ``` The 4 possible v2 target addresses formats are: ``` /exchange/:exchange/key/:routing-key /exchange/:exchange /queue/:queue <null> ``` where the last AMQP <null> value format requires that each message’s `to` field contains one of: ``` /exchange/:exchange/key/:routing-key /exchange/:exchange /queue/:queue ``` Hence, all 8 listed design goals are reached.
2024-04-02 00:35:06 +08:00
"//deps/rabbitmq_amqp_client:erlang_app",
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
"//deps/rabbitmq_stomp:erlang_app",
"//deps/rabbitmq_stream_common:erlang_app",
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
"@emqtt//:erlang_app",
],
)
2023-02-08 18:34:05 +08:00
rabbitmq_suite(
name = "packet_prop_SUITE",
deps = [
"//deps/rabbitmq_ct_helpers:erlang_app",
],
2023-02-08 18:34:05 +08:00
)
rabbitmq_suite(
name = "rabbit_mqtt_confirms_SUITE",
size = "small",
deps = [
"//deps/rabbit_common:erlang_app",
],
)
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
rabbitmq_suite(
name = "util_SUITE",
size = "small",
data = [
"test/rabbitmq_mqtt.app",
],
)
rabbitmq_suite(
name = "mc_mqtt_SUITE",
size = "small",
deps = [
"//deps/amqp10_common:erlang_app",
"//deps/rabbit:erlang_app",
Message Containers (#5077) This PR implements an approach for a "protocol (data format) agnostic core" where the format of the message isn't converted at point of reception. Currently all non AMQP 0.9.1 originating messages are converted into a AMQP 0.9.1 flavoured basic_message record before sent to a queue. If the messages are then consumed by the originating protocol they are converted back from AMQP 0.9.1. For some protocols such as MQTT 3.1 this isn't too expensive as MQTT is mostly a fairly easily mapped subset of AMQP 0.9.1 but for others such as AMQP 1.0 the conversions are awkward and in some cases lossy even if consuming from the originating protocol. This PR instead wraps all incoming messages in their originating form into a generic, extensible message container type (mc). The container module exposes an API to get common message details such as size and various properties (ttl, priority etc) directly from the source data type. Each protocol needs to implement the mc behaviour such that when a message originating form one protocol is consumed by another protocol we convert it to the target protocol at that point. The message container also contains annotations, dead letter records and other meta data we need to record during the lifetime of a message. The original protocol message is never modified unless it is consumed. This includes conversion modules to and from amqp, amqpl (AMQP 0.9.1) and mqtt. COMMIT HISTORY: * Refactor away from using the delivery{} record In many places including exchange types. This should make it easier to move towards using a message container type instead of basic_message. Add mc module and move direct replies outside of exchange Lots of changes incl classic queues Implement stream support incl amqp conversions simplify mc state record move mc.erl mc dlx stuff recent history exchange Make tracking work But doesn't take a protocol agnostic approach as we just convert everything into AMQP legacy and back. Might be good enough for now. Tracing as a whole may want a bit of a re-vamp at some point. tidy make quorum queue peek work by legacy conversion dead lettering fixes dead lettering fixes CMQ fixes rabbit_trace type fixes fixes fix Fix classic queue props test assertion fix feature flag and backwards compat Enable message_container feature flag in some SUITEs Dialyzer fixes fixes fix test fixes Various Manually update a gazelle generated file until a gazelle enhancement can be made https://github.com/rabbitmq/rules_erlang/issues/185 Add message_containers_SUITE to bazel and regen bazel files with gazelle from rules_erlang@main Simplify essential proprty access Such as durable, ttl and priority by extracting them into annotations at message container init time. Move type to remove dependenc on amqp10 stuff in mc.erl mostly because I don't know how to make bazel do the right thing add more stuff Refine routing header stuff wip Cosmetics Do not use "maybe" as type name as "maybe" is a keyword since OTP 25 which makes Erlang LS complain. * Dedup death queue names * Fix function clause crashes Fix failing tests in the MQTT shared_SUITE: A classic queue message ID can be undefined as set in https://github.com/rabbitmq/rabbitmq-server/blob/fbe79ff47b4edbc0fd95457e623d6593161ad198/deps/rabbit/src/rabbit_classic_queue_index_v2.erl#L1048 Fix failing tests in the MQTT shared_SUITE-mixed: When feature flag message_containers is disabled, the message is not an #mc{} record, but a #basic_message{} record. * Fix is_utf8_no_null crash Prior to this commit, the function crashed if invalid UTF-8 was provided, e.g.: ``` 1> rabbit_misc:is_valid_shortstr(<<"😇"/utf16>>). ** exception error: no function clause matching rabbit_misc:is_utf8_no_null(<<216,61,222,7>>) (rabbit_misc.erl, line 1481) ``` * Implement mqtt mc behaviour For now via amqp translation. This is still work in progress, but the following SUITEs pass: ``` make -C deps/rabbitmq_mqtt ct-shared t=[mqtt,v5,cluster_size_1] FULL=1 make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_1] FULL=1 ``` * Shorten mc file names Module name length matters because for each persistent message the #mc{} record is persisted to disk. ``` 1> iolist_size(term_to_iovec({mc, rabbit_mc_amqp_legacy})). 30 2> iolist_size(term_to_iovec({mc, mc_amqpl})). 17 ``` This commit renames the mc modules: ``` ag -l rabbit_mc_amqp_legacy | xargs sed -i 's/rabbit_mc_amqp_legacy/mc_amqpl/g' ag -l rabbit_mc_amqp | xargs sed -i 's/rabbit_mc_amqp/mc_amqp/g' ag -l rabbit_mqtt_mc | xargs sed -i 's/rabbit_mqtt_mc/mc_mqtt/g' ``` * mc: make deaths an annotation + fixes * Fix mc_mqtt protocol_state callback * Fix test will_delay_node_restart ``` make -C deps/rabbitmq_mqtt ct-v5 t=[mqtt,cluster_size_3]:will_delay_node_restart FULL=1 ``` * Bazel run gazelle * mix format rabbitmqctl.ex * Ensure ttl annotation is refelected in amqp legacy protocol state * Fix id access in message store * Fix rabbit_message_interceptor_SUITE * dializer fixes * Fix rabbit:rabbit_message_interceptor_SUITE-mixed set_annotation/3 should not result in duplicate keys * Fix MQTT shared_SUITE-mixed Up to 3.12 non-MQTT publishes were always QoS 1 regardless of delivery_mode. https://github.com/rabbitmq/rabbitmq-server/blob/75a953ce286a10aca910c098805a4f545989af38/deps/rabbitmq_mqtt/src/rabbit_mqtt_processor.erl#L2075-L2076 From now on, non-MQTT publishes are QoS 1 if durable. This makes more sense. The MQTT plugin must send a #basic_message{} to an old node that does not understand message containers. * Field content of 'v1_0.data' can be binary Fix ``` bazel test //deps/rabbitmq_mqtt:shared_SUITE-mixed \ --test_env FOCUS="-group [mqtt,v4,cluster_size_1] -case trace" \ -t- --test_sharding_strategy=disabled ``` * Remove route/2 and implement route/3 for all exchange types. This removes the route/2 callback from rabbit_exchange_type and makes route/3 mandatory instead. This is a breaking change and will require all implementations of exchange types to update their code, however this is necessary anyway for them to correctly handle the mc type. stream filtering fixes * Translate directly from MQTT to AMQP 0.9.1 * handle undecoded properties in mc_compat amqpl: put clause in right order recover death deatails from amqp data * Replace callback init_amqp with convert_from * Fix return value of lists:keyfind/3 * Translate directly from AMQP 0.9.1 to MQTT * Fix MQTT payload size MQTT payload can be a list when converted from AMQP 0.9.1 for example First conversions tests Plus some other conversion related fixes. bazel bazel translate amqp 1.0 null to undefined mc: property/2 and correlation_id/message_id return type tagged values. To ensure we can support a variety of types better. The type type tags are AMQP 1.0 flavoured. fix death recovery mc_mqtt: impl new api Add callbacks to allow protocols to compact data before storage And make readable if needing to query things repeatedly. bazel fix * more decoding * tracking mixed versions compat * mc: flip default of `durable` annotation to save some data. Assuming most messages are durable and that in memory messages suffer less from persistence overhead it makes sense for a non existent `durable` annotation to mean durable=true. * mc conversion tests and tidy up * mc make x_header unstrict again * amqpl: death record fixes * bazel * amqp -> amqpl conversion test * Fix crash in mc_amqp:size/1 Body can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0/ ct-system t=java ``` on branch native-amqp. * Fix crash in lists:flatten/1 Data can be a single amqp-value section (instead of being a list) as shown by test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp. * Fix crash in rabbit_writer Running test ``` make -C deps/rabbitmq_amqp1_0 ct-system t=dotnet:roundtrip_to_amqp_091 ``` on branch native-amqp resulted in the following crash: ``` crasher: initial call: rabbit_writer:enter_mainloop/2 pid: <0.711.0> registered_name: [] exception error: bad argument in function size/1 called as size([<<0>>,<<"Sw">>,[<<160,2>>,<<"hi">>]]) *** argument 1: not tuple or binary in call from rabbit_binary_generator:build_content_frames/7 (rabbit_binary_generator.erl, line 89) in call from rabbit_binary_generator:build_simple_content_frames/4 (rabbit_binary_generator.erl, line 61) in call from rabbit_writer:assemble_frames/5 (rabbit_writer.erl, line 334) in call from rabbit_writer:internal_send_command_async/3 (rabbit_writer.erl, line 365) in call from rabbit_writer:handle_message/2 (rabbit_writer.erl, line 265) in call from rabbit_writer:handle_message/3 (rabbit_writer.erl, line 232) in call from rabbit_writer:mainloop1/2 (rabbit_writer.erl, line 223) ``` because #content.payload_fragments_rev is currently supposed to be a flat list of binaries instead of being an iolist. This commit fixes this crash inefficiently by calling iolist_to_binary/1. A better solution would be to allow AMQP legacy's #content.payload_fragments_rev to be an iolist. * Add accidentally deleted line back * mc: optimise mc_amqp internal format By removint the outer records for message and delivery annotations as well as application properties and footers. * mc: optimis mc_amqp map_add by using upsert * mc: refactoring and bug fixes * mc_SUITE routingheader assertions * mc remove serialize/1 callback as only used by amqp * mc_amqp: avoid returning a nested list from protocol_state * test and bug fix * move infer_type to mc_util * mc fixes and additiona assertions * Support headers exchange routing for MQTT messages When a headers exchange is bound to the MQTT topic exchange, routing will be performend based on both MQTT topic (by the topic exchange) and MQTT User Property (by the headers exchange). This combines the best worlds of both MQTT 5.0 and AMQP 0.9.1 and enables powerful routing topologies. When the User Property contains the same name multiple times, only the last name (and value) will be considered by the headers exchange. * Fix crash when sending from stream to amqpl When publishing a message via the stream protocol and consuming it via AMQP 0.9.1, the following crash occurred prior to this commit: ``` crasher: initial call: rabbit_channel:init/1 pid: <0.818.0> registered_name: [] exception exit: {{badmatch,undefined}, [{rabbit_channel,handle_deliver0,4, [{file,"rabbit_channel.erl"}, {line,2728}]}, {lists,foldl,3,[{file,"lists.erl"},{line,1594}]}, {rabbit_channel,handle_cast,2, [{file,"rabbit_channel.erl"}, {line,728}]}, {gen_server2,handle_msg,2, [{file,"gen_server2.erl"},{line,1056}]}, {proc_lib,wake_up,3, [{file,"proc_lib.erl"},{line,251}]}]} ``` This commit first gives `mc:init/3` the chance to set exchange and routing_keys annotations. If not set, `rabbit_stream_queue` will set these annotations assuming the message was originally published via the stream protocol. * Support consistent hash exchange routing for MQTT 5.0 When a consistent hash exchange is bound to the MQTT topic exchange, MQTT 5.0 messages can be routed to queues consistently based on the Correlation-Data in the PUBLISH packet. * Convert MQTT 5.0 User Property * to AMQP 0.9.1 headers * from AMQP 0.9.1 headers * to AMQP 1.0 application properties and message annotations * from AMQP 1.0 application properties and message annotations * Make use of Annotations in mc_mqtt:protocol_state/2 mc_mqtt:protocol_state/2 includes Annotations as parameter. It's cleaner to make use of these Annotations when computing the protocol state instead of relying on the caller (rabbitmq_mqtt_processor) to compute the protocol state. * Enforce AMQP 0.9.1 field name length limit The AMQP 0.9.1 spec prohibits field names longer than 128 characters. Therefore, when converting AMQP 1.0 message annotations, application properties or MQTT 5.0 User Property to AMQP 0.9.1 headers, drop any names longer than 128 characters. * Fix type specs Apply feedback from Michael Davis Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * Add mc_mqtt unit test suite Implement mc_mqtt:x_header/2 * Translate indicator that payload is UTF-8 encoded when converting between MQTT 5.0 and AMQP 1.0 * Translate single amqp-value section from AMQP 1.0 to MQTT Convert to a text representation, if possible, and indicate to MQTT client that the payload is UTF-8 encoded. This way, the MQTT client will be able to parse the payload. If conversion to text representation is not possible, encode the payload using the AMQP 1.0 type system and indiate the encoding via Content-Type message/vnd.rabbitmq.amqp. This Content-Type is not registered. Type "message" makes sense since it's a message. Vendor tree "vnd.rabbitmq.amqp" makes sense since merely subtype "amqp" is not registered. * Fix payload conversion * Translate Response Topic between MQTT and AMQP Translate MQTT 5.0 Response Topic to AMQP 1.0 reply-to address and vice versa. The Response Topic must be a UTF-8 encoded string. This commit re-uses the already defined RabbitMQ target addresses: ``` "/topic/" RK Publish to amq.topic with routing key RK "/exchange/" X "/" RK Publish to exchange X with routing key RK ``` By default, the MQTT topic exchange is configure dto be amq.topic using the 1st target address. When an operator modifies the mqtt.exchange, the 2nd target address is used. * Apply PR feedback and fix formatting Co-authored-by: Michael Davis <mcarsondavis@gmail.com> * tidy up * Add MQTT message_containers test * consistent hash exchange: avoid amqp legacy conversion When hashing on a header value. * Avoid converting to amqp legacy when using exchange federation * Fix test flake * test and dialyzer fixes * dialyzer fix * Add MQTT protocol interoperability tests Test receiving from and sending to MQTT 5.0 and * AMQP 0.9.1 * AMQP 1.0 * STOMP * Streams * Regenerate portions of deps/rabbit/app.bzl with gazelle I'm not exactly sure how this happened, but gazell seems to have been run with an older version of the rules_erlang gazelle extension at some point. This caused generation of a structure that is no longer used. This commit updates the structure to the current pattern. * mc: refactoring * mc_amqpl: handle delivery annotations Just in case they are included. Also use iolist_to_iovec to create flat list of binaries when converting from amqp with amqp encoded payload. --------- Co-authored-by: David Ansari <david.ansari@gmx.de> Co-authored-by: Michael Davis <mcarsondavis@gmail.com> Co-authored-by: Rin Kuryloski <kuryloskip@vmware.com>
2023-08-31 18:27:13 +08:00
],
)
assert_suites()
2021-04-19 15:50:15 +08:00
alias(
name = "rabbitmq_mqtt",
actual = ":erlang_app",
visibility = ["//visibility:public"],
2021-04-19 15:50:15 +08:00
)