Add Qpid JMS client message selector tests

Author: ansd

deps/rabbit/src/rabbit_amqp_util.erl

@@ -14,6 +14,14 @@
          jms_header_to_amqp_field_name/1
         ]).
 
+-type field_name() :: durable | priority |
+                      message_id | user_id | to | subject | reply_to |
+                      correlation_id | content_type | content_encoding |
+                      absolute_expiry_time | creation_time | group_id |
+                      group_sequence | reply_to_group_id.
+
+-export_type([field_name/0]).
+
 -spec protocol_error(term(), io:format(), [term()]) ->
     no_return().
 protocol_error(Condition, Msg, Args) ->
@@ -30,12 +38,13 @@ capabilities(Capabilities) ->
     Caps = [{symbol, C} || C <- Capabilities],
     {array, symbol, Caps}.
 
+-spec section_field_name_to_atom(binary()) -> field_name().
 %% header section
 section_field_name_to_atom(<<"durable">>) -> durable;
 section_field_name_to_atom(<<"priority">>) -> priority;
-section_field_name_to_atom(<<"ttl">>) -> ttl;
-section_field_name_to_atom(<<"first-acquirer">>) -> first_acquirer;
-section_field_name_to_atom(<<"delivery-count">>) -> delivery_count;
+%% ttl, first-acquirer, and delivery-count are unsupported
+%% because setting a JMS message selector on these fields is invalid.
+
 %% properties section
 section_field_name_to_atom(<<"message-id">>) -> message_id;
 section_field_name_to_atom(<<"user-id">>) -> user_id;
@@ -49,9 +58,10 @@ section_field_name_to_atom(<<"absolute-expiry-time">>) -> absolute_expiry_time;
 section_field_name_to_atom(<<"creation-time">>) -> creation_time;
 section_field_name_to_atom(<<"group-id">>) -> group_id;
 section_field_name_to_atom(<<"group-sequence">>) -> group_sequence;
-section_field_name_to_atom(<<"reply-to-group-id">>) -> reply_to_group_id.
-
+section_field_name_to_atom(<<"reply-to-group-id">>) -> reply_to_group_id;
+section_field_name_to_atom(Other) -> erlang:error({unsupported_field_name, Other}).
 
+-spec jms_header_to_amqp_field_name(binary()) -> field_name() | binary().
 %% "Message header field references are restricted to
 %% JMSDeliveryMode, JMSPriority, JMSMessageID, JMSTimestamp, JMSCorrelationID, and JMSType."
 %% https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#message-selector-syntax
@@ -64,7 +74,6 @@ jms_header_to_amqp_field_name(<<"JMSCorrelationID">>) -> correlation_id;
 jms_header_to_amqp_field_name(<<"JMSType">>) -> subject;
 %% amqp-bindmap-jms-v1.0-wd10 § 3.2.2 JMS-defined ’JMSX’ Properties
 jms_header_to_amqp_field_name(<<"JMSXUserID">>) -> user_id;
-jms_header_to_amqp_field_name(<<"JMSXDeliveryCount">>) -> delivery_count;
 jms_header_to_amqp_field_name(<<"JMSXGroupID">>) -> group_id;
 jms_header_to_amqp_field_name(<<"JMSXGroupSeq">>) -> group_sequence;
 jms_header_to_amqp_field_name(Other) -> Other.

deps/rabbit/src/rabbit_fifo_client.erl

@@ -44,10 +44,11 @@
 -define(UNLIMITED_PREFETCH_COUNT, 2000). %% something large for ra
 %% controls the timer for closing cached segments
 -define(CACHE_SEG_TIMEOUT, 5000).
+-define(DEFAULT_MSG_PRIORITY, 4). %% defined in both AMQP and JMS
 
 -type seq() :: non_neg_integer().
 -type milliseconds() :: non_neg_integer().
--type filter() :: none | [atom()].
+-type filter() :: none | [rabbit_amqp_util:field_name()].
 
 -record(consumer, {key :: rabbit_fifo:consumer_key(),
                    % status = up :: up | cancelled,
@@ -1133,12 +1134,31 @@ msg_meta(_Msg, none) ->
     #{};
 msg_meta(Msg, FieldNames) ->
     Fields = lists:filtermap(fun(Name) ->
-                                     case mc:property(Name, Msg) of
-                                         {_Type, Val} ->
-                                             {true, {Name, Val}};
+                                     case get_field_value(Name, Msg) of
                                          undefined ->
-                                             false
+                                             false;
+                                         Val ->
+                                             {true, {Name, Val}}
                                      end
                              end, FieldNames),
     maps:merge(mc:routing_headers(Msg, []),
                maps:from_list(Fields)).
+
+get_field_value(durable, Msg) ->
+    mc:is_persistent(Msg);
+get_field_value(priority, Msg) ->
+    case mc:priority(Msg) of
+        undefined ->
+            ?DEFAULT_MSG_PRIORITY;
+        P ->
+            P
+    end;
+get_field_value(creation_time, Msg) ->
+    mc:timestamp(Msg);
+get_field_value(Name, Msg) ->
+    case mc:property(Name, Msg) of
+        {_Type, Val} ->
+            Val;
+        undefined ->
+            undefined
+    end.

deps/rabbit/src/rabbit_fifo_filter_jms.erl

@@ -9,6 +9,24 @@
 
 -export([eval/2]).
 
+%% "When used in a message selector JMSDeliveryMode is treated as having
+%% the values 'PERSISTENT' and 'NON_PERSISTENT'."
+%% https://jakarta.ee/specifications/messaging/3.1/jakarta-messaging-spec-3.1#special-notes
+-define(DELIVERY_MODE_PERSISTENT, <<"PERSISTENT">>).
+-define(DELIVERY_MODE_NON_PERSISTENT, <<"NON_PERSISTENT">>).
+-define(IS_DELIVERY_MODE(Val),
+        Val =:= ?DELIVERY_MODE_PERSISTENT orelse
+        Val =:= ?DELIVERY_MODE_NON_PERSISTENT).
+
+%% "For PERSISTENT messages, the durable field of header MUST be set to true.
+%% For NON PERSISTENT messages, the durable field of header MUST be either
+%% set to false or omitted."
+%% amqp-bindmap-jms-v1.0-wd10
+is_durable(?DELIVERY_MODE_PERSISTENT) ->
+    true;
+is_durable(?DELIVERY_MODE_NON_PERSISTENT) ->
+    false.
+
 %% Evaluates a parsed JMS message selector expression against message metadata.
 -spec eval(term(), #{atom() | binary() => atom() | binary() | number()}) ->
     boolean().
@@ -81,9 +99,9 @@ eval0({'not', Expr}, Headers) ->
 
 %% Comparison operators
 eval0({'=' = Op, Expr1, Expr2}, Headers) ->
-    compare(Op, eval0(Expr1, Headers), eval0(Expr2, Headers));
+    compare_eq(Op, eval0(Expr1, Headers), eval0(Expr2, Headers));
 eval0({'<>' = Op, Expr1, Expr2}, Headers) ->
-    compare(Op, eval0(Expr1, Headers), eval0(Expr2, Headers));
+    compare_eq(Op, eval0(Expr1, Headers), eval0(Expr2, Headers));
 eval0({'>' = Op, Expr1, Expr2}, Headers) ->
     compare(Op, eval0(Expr1, Headers), eval0(Expr2, Headers));
 eval0({'<' = Op, Expr1, Expr2}, Headers) ->
@@ -199,6 +217,13 @@ compare(_, _, _) ->
     %% the value of the operation is false."
     false.
 
+compare_eq(Op, Left, Right) when is_boolean(Left) andalso ?IS_DELIVERY_MODE(Right) ->
+    compare(Op, Left, is_durable(Right));
+compare_eq(Op, Left, Right) when is_boolean(Right) andalso ?IS_DELIVERY_MODE(Left) ->
+    compare(Op, is_durable(Left), Right);
+compare_eq(Op, Left, Right) ->
+    compare(Op, Left, Right).
+
 arithmetic(_, undefined, _) ->
     undefined;
 arithmetic(_, _, undefined) ->

deps/rabbit/src/rabbit_jms_selector_parser.erl

@@ -6,6 +6,14 @@
 
 extract_value({_Token, _Line, Value}) -> Value.
 
+process_identifier({_Token, Line, <<"JMSXDeliveryCount">>}) ->
+    %% "A clarification has been added to state that the effect of setting a
+    %% message selector on a property (such as JMSXDeliveryCount) which is set
+    %% by the provider on receive is undefined."
+    return_error(Line, "setting message selector on JMSXDeliveryCount is disallowed");
+process_identifier({_Token, _Line, Value}) ->
+    rabbit_amqp_util:jms_header_to_amqp_field_name(Value).
+
 process_like_pattern({string, Line, Value}) ->
     case unicode:characters_to_list(Value) of
         L when is_list(L) ->
@@ -210,7 +218,7 @@ yecctoken2string1(Other) ->
 
 
 
--file("rabbit_jms_selector_parser.erl", 213).
+-file("rabbit_jms_selector_parser.erl", 221).
 
 -dialyzer({nowarn_function, yeccpars2/7}).
 -compile({nowarn_unused_function,  yeccpars2/7}).
@@ -1499,7 +1507,7 @@ yeccpars2_21_(__Stack0) ->
  [___1 | __Stack] = __Stack0,
  [begin
 
-    {identifier, rabbit_amqp_util:jms_header_to_amqp_field_name(extract_value(___1))}
+    {identifier, process_identifier(___1)}
   end | __Stack].
 
 -compile({inline,yeccpars2_22_/1}).
@@ -1827,4 +1835,4 @@ yeccpars2_86_(__Stack0) ->
   end | __Stack].
 
 
--file("rabbit_jms_selector_parser.yrl", 141).
+-file("rabbit_jms_selector_parser.yrl", 149).

deps/rabbit/src/rabbit_jms_selector_parser.yrl

@@ -111,7 +111,7 @@ primary -> identifier_expr : '$1'.
 
 %% Identifiers (header fields or property references)
 identifier_expr -> identifier :
-    {identifier, rabbit_amqp_util:jms_header_to_amqp_field_name(extract_value('$1'))}.
+    {identifier, process_identifier('$1')}.
 
 %% Literals
 literal -> integer : {integer, extract_value('$1')}.
@@ -123,6 +123,14 @@ Erlang code.
 
 extract_value({_Token, _Line, Value}) -> Value.
 
+process_identifier({_Token, Line, <<"JMSXDeliveryCount">>}) ->
+    %% "A clarification has been added to state that the effect of setting a
+    %% message selector on a property (such as JMSXDeliveryCount) which is set
+    %% by the provider on receive is undefined."
+    return_error(Line, "setting message selector on JMSXDeliveryCount is disallowed");
+process_identifier({_Token, _Line, Value}) ->
+    rabbit_amqp_util:jms_header_to_amqp_field_name(Value).
+
 process_like_pattern({string, Line, Value}) ->
     case unicode:characters_to_list(Value) of
         L when is_list(L) ->

deps/rabbit/src/rabbit_quorum_queue.erl

@@ -450,7 +450,6 @@ consumer_filter(Spec, Args, Q) ->
             error
     end.
 
-%%TODO map JMS fields to AMQP fields
 parse_jms_selector(JmsSelector) ->
     String = unicode:characters_to_list(JmsSelector),
     case rabbit_jms_selector_lexer:string(String) of
@@ -461,13 +460,13 @@ parse_jms_selector(JmsSelector) ->
                     %                  [?MODULE, ?FUNCTION_NAME, JmsSelector, Tokens, _Expr]),
                     Ok;
                 {error, Reason} ->
-                    rabbit_log:warning("failed to parse JMS Selector '~s': ~p",
+                    rabbit_log:warning("failed to parse JMS message selector '~s': ~p",
                                        [JmsSelector, Reason]),
                     error
             end;
         {error, {_Line, _Mod, ErrDescriptor}, _Locaction} ->
             Reason = lists:flatten(leex:format_error(ErrDescriptor)),
-            rabbit_log:warning("failed to scan JMS selector '~s': ~p",
+            rabbit_log:warning("failed to scan JMS message selector '~s': ~p",
                                [JmsSelector, Reason]),
             error
     end.

deps/rabbit/test/amqp_jms_SUITE.erl

@@ -42,7 +42,9 @@ groups() ->
        message_types_jms_to_jms,
        message_types_jms_to_amqp,
        temporary_queue_rpc,
-       temporary_queue_delete
+       temporary_queue_delete,
+       message_selector_application_properties,
+       message_selector_header_fields
       ]
      }].
 
@@ -134,6 +136,12 @@ temporary_queue_rpc(Config) ->
 temporary_queue_delete(Config) ->
     ok = run_jms_test(?FUNCTION_NAME, Config).
 
+message_selector_application_properties(Config) ->
+    ok = run_jms_test(?FUNCTION_NAME, Config).
+
+message_selector_header_fields(Config) ->
+    ok = run_jms_test(?FUNCTION_NAME, Config).
+
 %% -------------------------------------------------------------------
 %% Helpers
 %% -------------------------------------------------------------------