Corrections to rabbit_basic:message/4 and refactoring of is_message_persistent such that message/4 doesn't permit disagreement between is_persistent and the properties

2 files changed, 31 insertions, 15 deletions

diff --git a/src/rabbit_basic.erl b/src/rabbit_basic.erl
index 7595d53b..69bc44cc 100644
--- a/src/rabbit_basic.erl
+++ b/src/rabbit_basic.erl
@@ -36,6 +36,7 @@
 -export([publish/1, message/4, properties/1, delivery/4]).
 -export([publish/4, publish/7]).
 -export([build_content/2, from_content/1]).
+-export([is_message_persistent/1]).
 
 %%----------------------------------------------------------------------------
 
@@ -57,6 +58,8 @@
              publish_result()).
 -spec(build_content/2 :: (amqp_properties(), binary()) -> content()).
 -spec(from_content/1 :: (content()) -> {amqp_properties(), binary()}).
+-spec(is_message_persistent/1 :: (decoded_content()) ->
+                                      (boolean() | {'error', any()})).
 
 -endif.
 
@@ -93,11 +96,17 @@ from_content(Content) ->
 
 message(ExchangeName, RoutingKeyBin, RawProperties, BodyBin) ->
     Properties = properties(RawProperties),
-    #basic_message{exchange_name  = ExchangeName,
-                   routing_key    = RoutingKeyBin,
-                   content        = build_content(Properties, BodyBin),
-                   guid           = rabbit_guid:guid(),
-                   is_persistent  = false}.
+    Content = build_content(Properties, BodyBin),
+    case is_message_persistent(Content) of
+        {error, Other} ->
+            {error, {invalid_delivery_mode, Other}};
+        Boolean when is_boolean(Boolean) ->
+            #basic_message{exchange_name  = ExchangeName,
+                           routing_key    = RoutingKeyBin,
+                           content        = Content,
+                           guid           = rabbit_guid:guid(),
+                           is_persistent  = Boolean}
+    end.
 
 properties(P = #'P_basic'{}) ->
     P;
@@ -131,3 +140,12 @@ publish(ExchangeName, RoutingKeyBin, Mandatory, Immediate, Txn, Properties,
     publish(delivery(Mandatory, Immediate, Txn,
                      message(ExchangeName, RoutingKeyBin,
                              properties(Properties), BodyBin))).
+
+is_message_persistent(#content{properties = #'P_basic'{
+                                 delivery_mode = Mode}}) ->
+    case Mode of
+        1         -> false;
+        2         -> true;
+        undefined -> false;
+        Other     -> {error, Other}
+    end.
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 51eb93c5..4ab9f4cb 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -1011,16 +1011,14 @@ notify_limiter(LimiterPid, Acked) ->
         Count -> rabbit_limiter:ack(LimiterPid, Count)
     end.
 
-is_message_persistent(#content{properties = #'P_basic'{
-                                 delivery_mode = Mode}}) ->
-    case Mode of
-        1         -> false;
-        2         -> true;
-        undefined -> false;
-        Other     -> rabbit_log:warning("Unknown delivery mode ~p - "
-                                        "treating as 1, non-persistent~n",
-                                        [Other]),
-                     false
+is_message_persistent(Content) ->
+    case rabbit_basic:is_message_persistent(Content) of
+        {error, Other} ->
+            rabbit_log:warning("Unknown delivery mode ~p - "
+                               "treating as 1, non-persistent~n",
+                               [Other]);
+        Boolean when is_boolean(Boolean) ->
+            Boolean
     end.
 
 lock_message(true, MsgStruct, State = #ch{unacked_message_q = UAMQ}) ->