pub async fn handle_sqs_messages<'a, T, F, Fut>(
    sqs: SqsClient,
    queue_name: &str,
    handler: F,
) -> Result<()>
where
    T: DeserializeOwned,
    F: Fn(T) -> Fut,
    Fut: Future<Output = Result<()>> + Send + 'a,
{
    let queue_url = get_queue_url(&sqs, queue_name).await?;

    loop {
        let res = sqs
            .receive_message(ReceiveMessageRequest {
                queue_url: queue_url.clone(),
                wait_time_seconds: Some(20),
                ..Default::default()
            })
            .await?;
        if let Some(messages) = res.messages {
            for message in messages.into_iter() {
                debug!("Handling message {}", message.message_id.as_ref().unwrap());
                let body = serde_json::from_str::<T>(message.body.as_ref().unwrap())?;
                handler(body).await?;
                sqs.delete_message(DeleteMessageRequest {
                    queue_url: queue_url.clone(),
                    receipt_handle: message.receipt_handle.unwrap(),
                })
                .await?;
            }
        }
    }
}
/// Listen for messages from an SQS queue, invoking the given
/// function for every message. If the function returns a successful
/// Result, the message is deleted from the queue.
Listen for messages from an SQS queue, invoking the given
function for every message. If the function returns a successful
Result, the message is deleted from the queue.
Listen for messages from an SQS queue invoking the given function for every message . If the function returns a successful Result the message is deleted from the queue .
pub async fn handle_sqs_messages<'a, T, F, Fut>(
    sqs: SqsClient,
    queue_name: &str,
    handler: F,
) -> Result<()>
where
    T: DeserializeOwned,
    F: Fn(T) -> Fut,
    Fut: Future<Output = Result<()>> + Send + 'a,
{
    let queue_url = get_queue_url(&sqs, queue_name).await?;
    loop {
        let res = sqs
            .receive_message(ReceiveMessageRequest {
                queue_url: queue_url.clone(),
                wait_time_seconds: Some(20),
                ..Default::default()
            })
            .await?;
        if let Some(messages) = res.messages {
            for message in messages.into_iter() {
                debug!("Handling message {}", message.message_id.as_ref().unwrap());
                let body = serde_json::from_str::<T>(message.body.as_ref().unwrap())?;
                handler(body).await?;
                sqs.delete_message(DeleteMessageRequest {
                    queue_url: queue_url.clone(),
                    receipt_handle: message.receipt_handle.unwrap(),
                })
                .await?;
            }
        }
    }
}
pub async fn handle_sqs_messages < ' a , T , F , Fut > ( sqs : SqsClient , queue_name : & str , handler : F , ) -> Result < ( ) > where T : DeserializeOwned , F : Fn ( T ) -> Fut , Fut : Future < Output = Result < ( ) > > + Send + ' a , { let queue_url = get_queue_url ( & sqs , queue_name ) . await ? ; loop { let res = sqs . receive_message ( ReceiveMessageRequest { queue_url : queue_url . clone ( ) , wait_time_seconds : Some ( 20 ) , .. Default :: default ( ) } ) . await ? ; if let Some ( messages ) = res . messages { for message in messages . into_iter ( ) { debug ! ( " " message message_id as_ref ( ) unwrap ( ) ) ; let body = serde_json :: from_str :: < T > ( message . body . as_ref ( ) . unwrap ( ) ) ? ; handler ( body ) . await ? ; sqs . delete_message ( DeleteMessageRequest { queue_url : queue_url . clone ( ) , receipt_handle : message . receipt_handle . unwrap ( ) , } ) . await ? ; } } } }
Listen for messages from an SQS queue, invoking the given
function for every message.
Listen for messages from an SQS queue invoking the given function for every message .
sqs
SqsClient
queue_name
&str
handler
F
sqs
SqsClient
queue_name
&str
handler
F