Honor the delivery_mode argument in Exchange.Message

Exchange.Message accepts a delivery_mode argument and documents it as
'Set custom delivery mode. Defaults to :attr:delivery_mode', but the
body always used self.delivery_mode, so the caller's value was
discarded. Asking for a transient message on a persistent exchange
still produced delivery_mode=2, and vice versa.

Fall back to the exchange default only when no delivery mode is given,
matching Producer._delivery_details, which already does
'delivery_mode or exchange.delivery_mode'. Exchange.publish() passes no
delivery mode, so its behaviour is unchanged.

Author: uttam12331

kombu/entity.py

@@ -249,7 +249,8 @@ def Message(self, body, delivery_mode=None, properties=None, **kwargs):
             headers (Dict): Message headers.
         """
         properties = {} if properties is None else properties
-        properties['delivery_mode'] = maybe_delivery_mode(self.delivery_mode)
+        properties['delivery_mode'] = maybe_delivery_mode(
+            delivery_mode or self.delivery_mode)
         if (isinstance(body, str) and
                 properties.get('content_encoding', None)) is None:
             kwargs['content_encoding'] = 'utf-8'

t/unit/test_entity.py

@@ -148,6 +148,20 @@ def test_create_message(self) -> None:
         Exchange('foo', channel=chan).Message({'foo': 'bar'})
         assert 'prepare_message' in chan
 
+    def test_create_message_delivery_mode(self) -> None:
+        chan = get_conn().channel()
+        exchange = Exchange('foo', channel=chan, delivery_mode='persistent')
+
+        # an explicit delivery_mode overrides the exchange default
+        message = exchange.Message({'foo': 'bar'}, delivery_mode='transient')
+        assert message['properties']['delivery_mode'] == \
+            Exchange.TRANSIENT_DELIVERY_MODE
+
+        # without one, the exchange default is used
+        message = exchange.Message({'foo': 'bar'})
+        assert message['properties']['delivery_mode'] == \
+            Exchange.PERSISTENT_DELIVERY_MODE
+
     def test_publish(self) -> None:
         chan = get_conn().channel()
         Exchange('foo', channel=chan).publish('the quick brown fox')