Message Containers

deps/amqp10_common/test/binary_generator_SUITE.erl

@@ -1,11 +1,11 @@
 -module(binary_generator_SUITE).
 
+-compile(nowarn_export_all).
 -compile(export_all).
 
 -export([
          ]).
 
--include_lib("common_test/include/ct.hrl").
 -include_lib("eunit/include/eunit.hrl").
 
 %%%===================================================================

deps/amqp10_common/test/binary_parser_SUITE.erl

@@ -55,16 +55,16 @@ end_per_testcase(_TestCase, _Config) ->
 roundtrip(_Config) ->
     Terms = [null,
              {described,
-              {utf8, <<"URL">>},
+              {symbol, <<"URL">>},
               {utf8, <<"http://example.org/hello-world">>}},
              {described,
-              {utf8, <<"URL">>},
-              {utf8, <<"https://rabbitmq.com">>}},
+              {symbol, <<"URL">>},
+              {binary, <<"https://rabbitmq.com">>}},
              {array, ubyte, [{ubyte, 1}, {ubyte, 255}]},
              {boolean, false},
              {list, [{utf8, <<"hi">>},
                      {described,
-                      {utf8, <<"URL">>},
+                      {symbol, <<"URL">>},
                       {utf8, <<"http://example.org/hello-world">>}}
                     ]},
              {list, [{int, 123},

deps/rabbit/BUILD.bazel

@@ -535,6 +535,11 @@ rabbitmq_integration_suite(
     size = "medium",
 )
 
+rabbitmq_integration_suite(
+    name = "message_containers_SUITE",
+    size = "medium",
+)
+
 rabbitmq_integration_suite(
     name = "metrics_SUITE",
     size = "medium",
@@ -778,6 +783,15 @@ rabbitmq_suite(
     ],
 )
 
+rabbitmq_suite(
+    name = "mc_SUITE",
+    size = "small",
+    deps = [
+        "//deps/amqp10_common:erlang_app",
+        "//deps/rabbit_common:erlang_app",
+    ],
+)
+
 rabbitmq_suite(
     name = "rabbit_stream_coordinator_SUITE",
     deps = [

deps/rabbit/app.bzl

@@ -10,6 +10,7 @@ def all_beam_files(name = "all_beam_files"):
         name = "behaviours",
         srcs = [
             "src/gm.erl",
+            "src/mc.erl",
             "src/rabbit_backing_queue.erl",
             "src/rabbit_credential_validator.erl",
             "src/rabbit_exchange_type.erl",
@@ -34,6 +35,10 @@ def all_beam_files(name = "all_beam_files"):
             "src/gatherer.erl",
             "src/internal_user.erl",
             "src/lqueue.erl",
+            "src/mc_amqp.erl",
+            "src/mc_amqpl.erl",
+            "src/mc_compat.erl",
+            "src/mc_util.erl",
             "src/mirrored_supervisor.erl",
             "src/mirrored_supervisor_sups.erl",
             "src/pg_local.erl",
@@ -254,6 +259,7 @@ def all_test_beam_files(name = "all_test_beam_files"):
         testonly = True,
         srcs = [
             "src/gm.erl",
+            "src/mc.erl",
             "src/rabbit_backing_queue.erl",
             "src/rabbit_credential_validator.erl",
             "src/rabbit_exchange_type.erl",
@@ -279,6 +285,10 @@ def all_test_beam_files(name = "all_test_beam_files"):
             "src/gatherer.erl",
             "src/internal_user.erl",
             "src/lqueue.erl",
+            "src/mc_amqp.erl",
+            "src/mc_amqpl.erl",
+            "src/mc_compat.erl",
+            "src/mc_util.erl",
             "src/mirrored_supervisor.erl",
             "src/mirrored_supervisor_sups.erl",
             "src/pg_local.erl",
@@ -503,6 +513,7 @@ def all_srcs(name = "all_srcs"):
             "include/amqqueue.hrl",
             "include/amqqueue_v2.hrl",
             "include/gm_specs.hrl",
+            "include/mc.hrl",
             "include/rabbit_global_counters.hrl",
             "include/vhost.hrl",
             "include/vhost_v2.hrl",
@@ -535,6 +546,11 @@ def all_srcs(name = "all_srcs"):
             "src/gm.erl",
             "src/internal_user.erl",
             "src/lqueue.erl",
+            "src/mc.erl",
+            "src/mc_amqp.erl",
+            "src/mc_amqpl.erl",
+            "src/mc_compat.erl",
+            "src/mc_util.erl",
             "src/mirrored_supervisor.erl",
             "src/mirrored_supervisor_sups.erl",
             "src/pg_local.erl",
@@ -1985,3 +2001,22 @@ def test_suite_beam_files(name = "test_suite_beam_files"):
         erlc_opts = "//:test_erlc_opts",
         deps = ["//deps/amqp_client:erlang_app", "//deps/rabbitmq_ct_helpers:erlang_app"],
     )
+    erlang_bytecode(
+        name = "message_containers_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/message_containers_SUITE.erl"],
+        outs = ["test/message_containers_SUITE.beam"],
+        app_name = "rabbit",
+        erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp_client:erlang_app"],
+    )
+    erlang_bytecode(
+        name = "mc_SUITE_beam_files",
+        testonly = True,
+        srcs = ["test/mc_SUITE.erl"],
+        outs = ["test/mc_SUITE.beam"],
+        hdrs = ["include/mc.hrl"],
+        app_name = "rabbit",
+        erlc_opts = "//:test_erlc_opts",
+        deps = ["//deps/amqp10_common:erlang_app", "//deps/rabbit_common:erlang_app"],
+    )

deps/rabbit/include/mc.hrl

@@ -0,0 +1,23 @@
+-type death_key() :: {Queue :: rabbit_misc:resource_name(), rabbit_dead_letter:reason()}.
+-type death_anns() :: #{first_time := non_neg_integer(), %% the timestamp of the first
+                        last_time := non_neg_integer(), %% the timestamp of the last
+                        ttl => non_neg_integer()}.
+-record(death, {
+                exchange :: rabbit_misc:resource_name(),
+                routing_keys = [] :: [rabbit_types:routing_key()],
+                count = 0 :: non_neg_integer(),
+                anns :: death_anns()
+               }).
+
+-record(deaths, {first :: death_key(),
+                 last :: death_key(),
+                 records = #{} :: #{death_key() := #death{}}}).
+
+
+%% good enough for most use cases
+-define(IS_MC(Msg), element(1, Msg) == mc andalso tuple_size(Msg) == 5).
+
+%% "Field names MUST start with a letter, '$' or '#' and may continue with letters, '$' or '#', digits, or
+%% underlines, to a maximum length of 128 characters." [AMQP 0.9.1 4.2.5.5 Field Tables]
+%% Given that the valid chars are ASCII chars, 1 char is encoded as 1 byte.
+-define(AMQP_LEGACY_FIELD_NAME_MAX_LEN, 128).