Message Queues

Message queue that allows for events to be transmitted between different software components.

Basic Queue

from import Queue myQueue = Queue('demo')

Integrating with a Serverless Function

You can use a Serverless Function to respond to the events from a queue.

from import Queue from import ServerlessFunction myQueue = Queue('demo') queue_event_trigger = myQueue.create_event_trigger() @ServerlessFunction("demofunction", events=[queue_event_trigger], permissions=[myQueue.available_permissions.READ_EVENTS]) def hello_world(event,context): print(event)

Message Ordering

By default, message order is not guaranteed, but you can designate a queue as First in First Out (fifo). This feature impacts performance and functionality of the queue, therefor you should be aware of the trade-offs.

from import Queue myQueue = Queue('demo', is_fifo=True)

Handling Errors

Sometimes there can be erroneous messages passed into the Queue that are not able to be properly handled. The message retry policy for a queue can be configured with the max_receive_count parameter, which determines the number of time a message should be retried. You can also provide a Dead Letter Queue (dql), which is a seperate queue to move messages into once they have exceeded the max_receive_count.

from import Queue myQueue = Queue('demo', dlq_arn="<aws_queue_arn>", max_receive_count=10)

By default, SQS sends batches of events to the handling function, and if the function throws an exception during the handling of a batch, the queue will mark the entire batch as failed. You can configure your queue to handle individual message failures by setting the batch_failure parameter on the event and return a structured response from your function when a message fails.

from import Queue from import ServerlessFunction myQueue = Queue('demo') queue_event_trigger = myQueue.create_event_trigger( batch_size=5, batch_failure=False ) @ServerlessFunction("demofunction", events=[queue_event_trigger], permissions=[myQueue.available_permissions.READ_EVENTS]) def hello_world(event,context): failed_items = [] for record in event.get('records'): try: process_message(record) except Exception: failed_items.append(record.get('message_id')) if failed_items: return {"batchItemFailures": [{"itemIdentifier": x} for x in failed_items]}