AWS Lambda is a serverless computing service provided by Amazon Web Services.
The Lambda functions can perform any kind of computing task, from serving web pages and processing streams of data to calling APIs and integrating with other AWS services, like SQS, Kinesis etc.
In case of this article, we will create SQS consumer, running on AWS lambda.
SQS stands for Simple Queue Service, which means a web service that gives you access to a message queue that can be used to store messages while waiting for a computer to process them.
SQS can be used as a classic pub/sub queue solution, and it can support the First In First Out principle if needed.
More about SQS you can find on official amazon pages.
How to create a lambda code
Cargo gives us a beautiful helper command to deal with Rust and AWS Lambda, called cargo-lamda.
We need to install it by running cargo install cargo-lambda.
After that let's create our first lambda project with cargo lambda new rust-lambda-example.
After running this command, you need to answer a few questions to help cargo-lambda create the needed template.
As I mentioned earlier, our goal is to create an SQS consumer (a rust lambda that is triggered by SQS events), so your answers should be like:
cargo lambda new rust-lambda-example
? Is this function an HTTP function? No
? AWS Event type that this function receives sqs::SqsEvent
Let's take a look at new template!
Your new template should look something like the one below.
I've removed all comments and annotations and some blocks of code to make the most basic example, so don't worry if your code might look a bit different.
use aws_lambda_events::event::sqs::SqsEvent;
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
run(service_fn(function_handler)).await
}
function_handler is a function that will handle every event, that's a "heart" of our future lambda.
You can easily pass additional parameters to your function_handler function by changing the code like this.
async fn function_handler(
event: LambdaEvent<SqsEvent>,
conn: &SomeConnection,
) -> Result<(), Error> {
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Error> {
let conn = SomeConnection {}; // an abstract example, replace it with the real one if needed
run(service_fn(|event: LambdaEvent<SqsEvent>| {
function_handler(event, &conn)
}))
.await
}
Let's have a quick look at our dependencies in Cargo.toml
[dependencies]
aws_lambda_events = { version = "0.6.3", default-features = false, features = ["sqs"] }
lambda_runtime = "0.6.0"
tokio = { version = "1", features = ["macros"] }
Here we see the awslabs lambda runtime, built on tokio runtime and aws_lambda_events crate, which provides a collection of common AWS events structs.
Ok, let's teach our handler to print all SQS messages bodies, by adding simple code to our handler
async fn function_handler(event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
event
.payload
.records
.into_iter()
.filter_map(|m| m.body) // since body is optional, we want to skip all None values
.for_each(|m| println!("{m}"));
Ok(())
}
How to test it locally
Write a simple test
First, let's add a new dependency, a well-known json library called serde-json.
Add this line to our Cargo.toml dependencies block: serde_json = "~1.0.59".
Next. let's create some test dataset:
- Create a "fixtures" folder
- Create "example-sqs-event.json" file into this folder
- Fill "example-sqs-event.json" file with the data below
{ "Records": [ { "messageId" : "MessageID_1", "receiptHandle" : "MessageReceiptHandle", "body" : "Hello from message 1!", "md5OfBody" : "fce0ea8dd236ccb3ed9b37dae260836f", "md5OfMessageAttributes" : "582c92c5c5b6ac403040a4f3ab3115c9", "eventSourceARN": "arn:aws:sqs:us-west-2:123456789012:SQSQueue", "eventSource": "aws:sqs", "awsRegion": "us-west-2", "attributes" : { "ApproximateReceiveCount" : "2", "SentTimestamp" : "1520621625029", "SenderId" : "AROAIWPX5BD2BHG722MW4:sender", "ApproximateFirstReceiveTimestamp" : "1520621634884" }, "messageAttributes" : { "Attribute3" : { "binaryValue" : "MTEwMA==", "stringListValues" : ["abc", "123"], "binaryListValues" : ["MA==", "MQ==", "MA=="], "dataType" : "Binary" }, "Attribute2" : { "stringValue" : "123", "stringListValues" : [ ], "binaryListValues" : ["MQ==", "MA=="], "dataType" : "Number" }, "Attribute1" : { "stringValue" : "AttributeValue1", "stringListValues" : [ ], "binaryListValues" : [ ], "dataType" : "String" } } }, { "messageId" : "MessageID_2", "receiptHandle" : "MessageReceiptHandle", "body" : "Hello from message 2!", "md5OfBody" : "fce0ea8dd236ccb3ed9b37dae260836f", "md5OfMessageAttributes" : "582c92c5c5b6ac403040a4f3ab3115c9", "eventSourceARN": "arn:aws:sqs:us-west-2:123456789012:SQSQueue", "eventSource": "aws:sqs", "awsRegion": "us-west-2", "attributes" : { "ApproximateReceiveCount" : "2", "SentTimestamp" : "1520621625029", "SenderId" : "AROAIWPX5BD2BHG722MW4:sender", "ApproximateFirstReceiveTimestamp" : "1520621634884" }, "messageAttributes" : { "Attribute3" : { "binaryValue" : "MTEwMA==", "stringListValues" : ["abc", "123"], "binaryListValues" : ["MA==", "MQ==", "MA=="], "dataType" : "Binary" }, "Attribute2" : { "stringValue" : "123", "stringListValues" : [ ], "binaryListValues" : ["MQ==", "MA=="], "dataType" : "Number" }, "Attribute1" : { "stringValue" : "AttributeValue1", "stringListValues" : [ ], "binaryListValues" : [ ], "dataType" : "String" } } } ] }
Now let's write a test! Add this block of code to the end of your main file.
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_function_handler() {
let data = include_bytes!("fixtures/example-sqs-event.json");
let parsed: SqsEvent = serde_json::from_slice(data).unwrap();
let context = lambda_runtime::Context::default();
let event = LambdaEvent::new(parsed, context);
function_handler(event).await.expect("failed to handle event");
}
}
After running the test, output should be:
running 1 test
Hello from message 1!
Hello from message 2!
test tests::test_function_handler ... ok
Great! Our handler is working as expected!
Emulate
cargo lambda start command emulates the AWS Lambda control plane API.
Run this command at the root of the project and you will see such output
INFO invoke server listening on 127.0.0.1:9000
Now while our lambda is working, let's try to send an event to it by running
cargo lambda invoke rust-lambda-example --data-file src/fixtures/example-sqs-event.json in a separate window.
After invoking lambda, the output should be updated by something like this
INFO starting lambda function function="rust-lambda-example"
[Running 'cargo run --bin rust-lambda-example']
Compiling rust-lambda-example v0.1.0 ({your_path}/rust-lambda-example)
Finished dev [unoptimized + debuginfo] target(s) in 2.37s
Running `target/debug/rust-lambda-example`
Hello from message 1!
Hello from message 2!
Cool! That works.
AWS lambda Environment variables
Let's imagine that we need to use AWS lambda environment variables. How should we test it locally?
Easy! You just need to work with them like with classic Rust environment variables!
Let's update our main file with
async fn main() -> Result<(), Error> {
let env_var = std::env::var("MODE").expect("Environment error");
println!("{env_var}");
run(service_fn(function_handler)).await
}
and then rerun our API emulation with MODE=test cargo lambda start.
Let's try to send an event again with lambda invoke rust-lambda-example --data-file src/fixtures/example-sqs-event.json.
The output will be
[Running 'cargo run --bin rust-lambda-example']
Finished dev [unoptimized + debuginfo] target(s) in 0.02s
Running `target/debug/rust-lambda-example`
test
Hello from message 1!
Hello from message 2!
Not surprising, but it works!
The next step is to configure and run our code on AWS lambda, so let's move to the next article.
P.S. You can find a working example on my GitHub repository :)