Add support for fair queue in SQS

kombu/transport/SQS.py

@@ -463,11 +463,13 @@ def _put(self, queue, message, **kwargs):
                 # we don't want to want to have the attribute in the body
                 kwargs['MessageAttributes'] = \
                     message['properties'].pop('message_attributes')
+            # support SQS fair queue system.
+            if 'MessageGroupId' in message['properties']:
+                kwargs['MessageGroupId'] = \
+                    message['properties']['MessageGroupId']
+            # Support FIFO queues.
             if queue.endswith('.fifo'):
-                if 'MessageGroupId' in message['properties']:
-                    kwargs['MessageGroupId'] = \
-                        message['properties']['MessageGroupId']
-                else:
+                if 'MessageGroupId' not in kwargs:
                     kwargs['MessageGroupId'] = 'default'
                 if 'MessageDeduplicationId' in message['properties']:
                     kwargs['MessageDeduplicationId'] = \

t/unit/transport/test_SQS.py

@@ -1148,6 +1148,29 @@ def test_predefined_queues_put_to_fifo_queue(self):
         assert 'MessageGroupId' in sqs_queue_mock.send_message.call_args[1]
         assert 'MessageDeduplicationId' in \
             sqs_queue_mock.send_message.call_args[1]
+
+    def test_predefined_queues_put_with_message_group_id(self):
+        connection = Connection(transport=SQS.Transport, transport_options={
+            'predefined_queues': example_predefined_queues,
+        })
+        channel = connection.channel()
+
+        queue_name = 'queue-1'
+
+        exchange = Exchange('test_SQS', type='direct')
+        p = messaging.Producer(channel, exchange, routing_key=queue_name)
+
+        queue = Queue(queue_name, exchange, queue_name)
+        queue(channel).declare()
+
+        channel.sqs = Mock()
+        sqs_queue_mock = Mock()
+        channel.sqs.return_value = sqs_queue_mock
+        p.publish('message', MessageGroupId='test-group-id')
+
+        sqs_queue_mock.send_message.assert_called_once()
+        assert 'MessageGroupId' in sqs_queue_mock.send_message.call_args[1]
+        assert sqs_queue_mock.send_message.call_args[1]['MessageGroupId'] == 'test-group-id'
 
     def test_predefined_queues_put_to_queue(self):
         connection = Connection(transport=SQS.Transport, transport_options={