sqs-lambda-partial-batch-rust-sam

sqs-lambda-partial-batch-rust-sam/Cargo.toml

@@ -10,9 +10,9 @@ name = "handler"
 path = "src/bin/handler.rs"
 
 [dependencies]
-aws_lambda_events = { version = "0.7", default-features = false, features = ["sqs"] }
+aws_lambda_events = { version = "0.15", default-features = false, features = ["sqs"] }
 futures = "0.3"
-lambda_runtime = "0.11"
+lambda_runtime = "0.13"
 serde = {version = "1", features = ["derive"] }
 serde_json = "1"
 tokio = { version = "1", features = ["macros"] }

sqs-lambda-partial-batch-rust-sam/README.md

@@ -13,7 +13,7 @@ Important: this application uses various AWS services and there are costs associ
 * [Git Installed](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git)
 * [AWS Serverless Application Model](https://docs.aws.amazon.com/serverless-application-model/latest/developerguide/serverless-sam-cli-install.html) (AWS SAM) installed
 * [Rust](https://www.rust-lang.org/) 1.56.0 or higher
-* [cargo-zigbuild](https://github.com/messense/cargo-zigbuild) and [Zig](https://ziglang.org/) for cross-compilation
+* [CargoLambda](https://www.cargo-lambda.info/guide/installation.html)
 * Make sure to run "rustup target add aarch64-unknown-linux-gnu;"
 
 ## Deployment Instructions
@@ -28,11 +28,11 @@ Important: this application uses various AWS services and there are costs associ
     ```
 3. Install dependencies and build:
     ```
-    make build
+    sam build
     ```
 4. From the command line, use AWS SAM to deploy the AWS resources for the pattern as specified in the template.yml file:
     ```
-    make deploy
+    sam deploy
     ```
 5. During the prompts:
     * Enter a stack name
@@ -117,7 +117,7 @@ do_something MyStruct { name: "Daniele", surname: "Frasca" }
  
 1. Delete the stack
     ```bash
-    make delete
+    sam delete
     ```
 2. Confirm the stack has been deleted
     ```bash

sqs-lambda-partial-batch-rust-sam/src/bin/handler.rs

@@ -1,8 +1,10 @@
-use aws_lambda_events::event::sqs::SqsEvent;
+use aws_lambda_events::{
+    event::sqs::SqsEvent,
+    sqs::{BatchItemFailure, SqsBatchResponse},
+};
 use futures::future::join_all;
 use lambda_runtime::{run, service_fn, Error, LambdaEvent};
-use serde::{Deserialize, Serialize};
-use serde_json::Value;
+use serde::Deserialize;
 use std::sync::{Arc, Mutex};
 
 #[tokio::main]
@@ -19,54 +21,41 @@ async fn main() -> Result<(), Error> {
     .await
 }
 
-pub async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<Value, Error> {
+pub async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<SqsBatchResponse, Error> {
     println!("Input {:?}", event);
-    let failed_message: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
+    let failed_message = Arc::new(Mutex::new(Vec::with_capacity(event.payload.records.len())));
     let mut tasks = Vec::with_capacity(event.payload.records.len());
 
     for record in event.payload.records.into_iter() {
         let failed_message = failed_message.clone();
-
         tasks.push(tokio::spawn(async move {
             if let Some(body) = &record.body {
                 let request = serde_json::from_str::<MyStruct>(body);
                 if let Ok(request) = request {
                     do_something(&request).await.map_or_else(
                         |_e| {
-                            failed_message
-                                .lock()
-                                .unwrap()
-                                .push(record.message_id.unwrap().clone());
+                            failed_message.lock().unwrap().push(BatchItemFailure {
+                                item_identifier: record.message_id.unwrap_or_default(),
+                            });
                         },
                         |_| (),
                     );
                 } else {
-                    failed_message
-                        .lock()
-                        .unwrap()
-                        .push(record.message_id.unwrap().clone());
+                    failed_message.lock().unwrap().push(BatchItemFailure {
+                        item_identifier: record.message_id.unwrap_or_default(),
+                    });
                 }
             }
         }));
     }
 
     join_all(tasks).await;
 
-    let response = BatchItemFailures {
-        batch_item_failures: failed_message
-            .lock()
-            .unwrap()
-            .clone()
-            .into_iter()
-            .map(|message_id| {
-                ItemIdentifier {
-                    item_identifier: message_id,
-                }
-            })
-            .collect(),
-    };
+    let failed_message = failed_message.lock().unwrap().clone();
 
-    Ok(serde_json::to_value(response).unwrap())
+    Ok(SqsBatchResponse {
+        batch_item_failures: failed_message,
+    })
 }
 
 async fn do_something(request: &MyStruct) -> Result<(), Error> {
@@ -79,15 +68,3 @@ pub struct MyStruct {
     pub name: String,
     pub surname: String,
 }
-
-#[derive(Deserialize, Serialize, Debug, Clone, PartialEq, Eq)]
-pub struct BatchItemFailures {
-    #[serde(rename = "batchItemFailures")]
-    pub batch_item_failures: Vec<ItemIdentifier>,
-}
-
-#[derive(Deserialize, Serialize, Debug, Clone, PartialEq, Eq)]
-pub struct ItemIdentifier {
-    #[serde(rename = "itemIdentifier")]
-    pub item_identifier: String,
-}

sqs-lambda-partial-batch-rust-sam/template.yml

@@ -34,7 +34,7 @@ Resources:
   MyFunction:
     Type: AWS::Serverless::Function 
     Properties:
-      CodeUri: ./build/handler
+      CodeUri: .
       Events:
         MySQSEvent:
           Type: SQS
@@ -43,6 +43,8 @@ Resources:
             BatchSize: 10
             FunctionResponseTypes:
               - ReportBatchItemFailures
+    Metadata:
+      BuildMethod: rust-cargolambda
 
 Outputs:
   MySqsQueueName:
@@ -54,4 +56,3 @@ Outputs:
   MySqsQueueURL:
     Description: SQS queue URL
     Value: !Ref MySqsQueue
-