connector-aws-sqs/com.river.connector.aws.sqs/onMessage

onMessage

fun <Error class: unknown class>.onMessage( queue: SqsQueue, concurrency: Int = 1, receiveConfiguration: ReceiveConfiguration.() -> Unit = {}, commitConfiguration: CommitConfiguration.() -> Unit = {}, onError: OnError = OnError.Retry(250.milliseconds), onMessage: suspend (<Error class: unknown class>) -> Acknowledgment): Job

This function is a high-level abstraction that combines receiving messages, processing them, and sending acknowledgments. It is built on top of the receiveMessagesAsFlow and acknowledgmentMessageFlow functions.

Creates a flow that continuously receives messages from an Amazon Simple Queue Service (SQS) queue, processes them using a provided function, and sends acknowledgments for processed messages.

This function is a simplified version of onMessages that processes messages one by one. It is executed in the specified CoroutineScope and returns a Job that represents its execution.

Return

A Job representing the execution of the flow.

Example usage:

val sqsClient: SqsClient = ...

coroutineScope {
    val myQueueJob = sqsClient.onMessage(SqsQueue.name("myqueue")) { message ->
        Delete
    }

    // You may cancel myQueueJob at any time.
}

Parameters

queue

The reference of the queue from which messages will be received.

concurrency

The level of concurrency for processing messages. Defaults to 1.

receiveConfiguration

A lambda with receiver for configuring the ReceiveConfiguration for the underlying receive operation.

commitConfiguration

A lambda with receiver for configuring the CommitConfiguration for the underlying commit operation.

onError

The strategy to be used when the message processing encounters an error. Defaults to OnError.Retry.

onMessage

A function to process a received message. It receives a single message and returns a MessageAcknowledgment.