Add OSS changes for rabbitmq_jms plugin

This commit adds OSS changes required for the new VMware Tanzu RabbitMQ
`rabbitmq_jms` plugin.

The deleted tests in this commit moved to `~/tanzu-rabbitmq/deps/rabbitmq_jms/test/`.

Author: ansd

deps/amqp10_client/src/amqp10_client_session.erl

@@ -83,7 +83,8 @@
 -type target_def() :: #{address => terminus_address(),
                         durable => terminus_durability()}.
 -type source_def() :: #{address => terminus_address(),
-                        durable => terminus_durability()}.
+                        durable => terminus_durability(),
+                        distribution_mode => distribution_mode()}.
 
 -type attach_role() :: {sender, target_def()} | {receiver, source_def(), pid()}.
 
@@ -818,12 +819,19 @@ make_source(#{role := {receiver, Source, _Pid}} = AttachArgs) ->
                        _ ->
                            undefined
                    end,
+    DistMode = case Source of
+                   #{distribution_mode := Mode} ->
+                       amqp10_util:dist_mode_from_atom(Mode);
+                   _ ->
+                       undefined
+               end,
     Dynamic = maps:get(dynamic, Source, false),
     TranslatedFilter = translate_filters(maps:get(filter, AttachArgs, #{})),
     #'v1_0.source'{address = make_address(Source),
                    durable = {uint, Durable},
                    expiry_policy = ExpiryPolicy,
                    dynamic = Dynamic,
+                   distribution_mode = DistMode,
                    filter = TranslatedFilter,
                    capabilities = make_capabilities(Source)}.
 

deps/amqp10_common/include/amqp10_filter.hrl

@@ -5,14 +5,27 @@
 %% Copyright (c) 2007-2026 Broadcom. All Rights Reserved. The term “Broadcom” refers to Broadcom Inc. and/or its subsidiaries. All rights reserved.
 
 %% AMQP Filter Expressions Version 1.0 Committee Specification Draft 01
+%% https://docs.oasis-open.org/amqp/filtex/v1.0/csd01/filtex-v1.0-csd01.html#_Toc67929253
+-define(CAP_FILTEX_SQL, <<"AMQP_FILTEX_SQL_V1_0">>).
+-define(CAP_FILTEX_PROP, <<"AMQP_FILTEX_PROP_V1_0">>).
 %% https://docs.oasis-open.org/amqp/filtex/v1.0/csd01/filtex-v1.0-csd01.html#_Toc67929266
 -define(DESCRIPTOR_NAME_PROPERTIES_FILTER, <<"amqp:properties-filter">>).
 -define(DESCRIPTOR_CODE_PROPERTIES_FILTER, 16#173).
 -define(DESCRIPTOR_NAME_APPLICATION_PROPERTIES_FILTER, <<"amqp:application-properties-filter">>).
 -define(DESCRIPTOR_CODE_APPLICATION_PROPERTIES_FILTER, 16#174).
-
 %% A filter with this name contains an AMQP SQL expression.
 -define(FILTER_NAME_SQL, <<"sql-filter">>).
 %% https://docs.oasis-open.org/amqp/filtex/v1.0/csd01/filtex-v1.0-csd01.html#_Toc67929276
 -define(DESCRIPTOR_NAME_SQL_FILTER, <<"amqp:sql-filter">>).
 -define(DESCRIPTOR_CODE_SQL_FILTER, 16#120).
+
+%% A filter with this name contains a JMS message selector.
+%% We use the same name as Qpid JMS in
+%% https://github.com/apache/qpid-jms/blob/2.10.0/qpid-jms-client/src/main/java/org/apache/qpid/jms/provider/amqp/AmqpSupport.java#L75
+-define(FILTER_NAME_SELECTOR, <<"jms-selector">>).
+%% SQL-based filtering syntax
+%% This capability and these descriptors are defined in
+%% https://www.amqp.org/specification/1.0/filters
+-define(CAP_SELECTOR, <<"APACHE.ORG:SELECTOR">>).
+-define(DESCRIPTOR_NAME_SELECTOR_FILTER, <<"apache.org:selector-filter:string">>).
+-define(DESCRIPTOR_CODE_SELECTOR_FILTER, 16#0000468C00000004).

deps/amqp10_common/include/amqp10_types.hrl

@@ -26,3 +26,6 @@
 
 % [3.2.16]
 -define(MESSAGE_FORMAT, 0).
+
+% [3.5.7]
+-type distribution_mode() :: move | copy.

deps/amqp10_common/src/amqp10_util.erl

@@ -7,7 +7,11 @@
 
 -module(amqp10_util).
 -include_lib("amqp10_common/include/amqp10_types.hrl").
--export([link_credit_snd/3]).
+-include_lib("amqp10_common/include/amqp10_framing.hrl").
+
+-export([link_credit_snd/3,
+         dist_mode_to_atom/1,
+         dist_mode_from_atom/1]).
 
 %% AMQP 1.0 §2.6.7
 -spec link_credit_snd(sequence_no(), uint(), sequence_no()) -> uint().
@@ -18,3 +22,15 @@ link_credit_snd(DeliveryCountRcv, LinkCreditRcv, DeliveryCountSnd) ->
     %% LinkCreditSnd can be negative when receiver decreases credits
     %% while messages are in flight. Maintain a floor of zero.
     max(0, LinkCreditSnd).
+
+-spec dist_mode_to_atom({symbol, binary()}) -> distribution_mode().
+dist_mode_to_atom(?V_1_0_STD_DIST_MODE_MOVE) ->
+    move;
+dist_mode_to_atom(?V_1_0_STD_DIST_MODE_COPY) ->
+    copy.
+
+-spec dist_mode_from_atom(distribution_mode()) -> {symbol, binary()}.
+dist_mode_from_atom(move) ->
+    ?V_1_0_STD_DIST_MODE_MOVE;
+dist_mode_from_atom(copy) ->
+    ?V_1_0_STD_DIST_MODE_COPY.

deps/rabbit/Makefile

@@ -83,6 +83,7 @@ define PROJECT_ENV
 	    %% See https://www.rabbitmq.com/docs/consumers#acknowledgement-timeout
 	    %% 30 minutes
 	    {consumer_timeout, 1800000},
+	    {consumer_disconnected_timeout, 60000},
 
 	    %% used by rabbit_peer_discovery_classic_config
 	    {cluster_nodes, {[], disc}},
@@ -248,7 +249,7 @@ define ct_master.erl
 endef
 
 PARALLEL_CT_SET_1_A = unit_rabbit_ssl unit_cluster_formation_locking_mocks unit_cluster_formation_sort_nodes unit_collections unit_config_value_encryption unit_connection_tracking
-PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_filter_prop amqp_filter_sql amqp_filter_sql_unit amqp_dotnet amqp_jms signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management
+PARALLEL_CT_SET_1_B = amqp_address amqp_auth amqp_filter_prop amqp_filter_sql amqp_filter_sql_unit amqp_dotnet signal_handling single_active_consumer unit_access_control_authn_authz_context_propagation unit_access_control_credential_validation unit_amqp091_content_framing unit_amqp091_server_properties unit_app_management
 PARALLEL_CT_SET_1_C = amqp_proxy_protocol amqpl_consumer_ack bindings rabbit_db_maintenance rabbit_db_msup rabbit_db_policy rabbit_db_queue rabbit_db_topic_exchange cluster_limit cluster_minority default_queue_type_prop term_to_binary_compat_prop topic_permission unicode unit_access_control
 PARALLEL_CT_SET_1_D = amqqueue_backward_compatibility channel_interceptor channel_operation_timeout classic_queue_prop config_schema peer_discovery_dns peer_discovery_tmp_hidden_node per_node_limit per_user_connection_channel_limit
 
@@ -352,8 +353,8 @@ ifdef TRACE_SUPERVISOR2
 RMQ_ERLC_OPTS += -DTRACE_SUPERVISOR2=true
 endif
 
-# https://www.erlang.org/doc/apps/parsetools/leex.html#file/2
 ifndef NON_DETERMINISTIC
+# https://www.erlang.org/doc/apps/parsetools/leex.html#file/2
 YRL_ERLC_OPTS += +deterministic
 endif
 

deps/rabbit/include/rabbit_global_counters.hrl

@@ -3,7 +3,7 @@
 %% Dead Letter counters:
 %%
 %% The following two counters are mutually exclusive because
-%% quorum queue dead-letter-strategy at-least-once is incompatible with overflow drop-head.
+%% dead-letter-strategy at-least-once is incompatible with overflow drop-head.
 -define(MESSAGES_DEAD_LETTERED_MAXLEN, 1).
 -define(MESSAGES_DEAD_LETTERED_CONFIRMED, 1).
 -define(MESSAGES_DEAD_LETTERED_EXPIRED, 2).

deps/rabbit/src/amqqueue.erl

@@ -52,6 +52,7 @@
          % type_state
          get_type_state/1,
          set_type_state/2,
+         update_type_state/2,
          % state
          get_state/1,
          set_state/2,
@@ -465,11 +466,10 @@ set_pid(#amqqueue{} = Queue, Pid) ->
 % policy
 
 -spec get_policy(amqqueue()) -> proplists:proplist() | none | undefined.
-
 get_policy(#amqqueue{policy = Policy}) -> Policy.
 
--spec set_policy(amqqueue(), binary() | none | undefined) -> amqqueue().
-
+-spec set_policy(amqqueue(), proplists:proplist() | none | undefined) ->
+    amqqueue().
 set_policy(#amqqueue{} = Queue, Policy) ->
     Queue#amqqueue{policy = Policy}.
 
@@ -499,6 +499,11 @@ set_type_state(#amqqueue{} = Queue, TState) ->
 set_type_state(Queue, _TState) ->
     Queue.
 
+-spec update_type_state(amqqueue(), fun((map()) -> map())) -> amqqueue().
+update_type_state(#amqqueue{} = Queue, Fun) ->
+    TState = get_type_state(Queue),
+    set_type_state(Queue, Fun(TState)).
+
 % state
 
 -spec get_state(amqqueue()) -> atom() | none.

deps/rabbit/src/mc.erl

@@ -23,6 +23,7 @@
          message_id/1,
          property/2,
          timestamp/1,
+         received_timestamp/1,
          priority/1,
          set_ttl/2,
          x_header/2,
@@ -287,6 +288,11 @@ timestamp(#?MODULE{annotations = Anns}) ->
 timestamp(BasicMsg) ->
     mc_compat:timestamp(BasicMsg).
 
+-spec received_timestamp(state()) -> non_neg_integer().
+received_timestamp(#?MODULE{annotations = #{?ANN_RECEIVED_AT_TIMESTAMP := Ts}})
+  when is_integer(Ts) ->
+    Ts.
+
 -spec priority(state()) -> undefined | non_neg_integer().
 priority(#?MODULE{annotations = Anns}) ->
     maps:get(?ANN_PRIORITY, Anns, undefined);