Rust – 6 месяцев и подсчеты

AWS_NODEJS_CONNECTION_REUSE_ENABLED: 1

JAVA_TOOL_OPTIONS: -XX:+TieredCompilation -XX:TieredStopAtLevel=1

    let mut tasks = Vec::with_capacity(event.records.len());
....
    for record in event.records.into_iter() {
      tasks.push(tokio::spawn(async move {
         .....
         do_something(&shared_client, &request)
            .await
            .map_or_else(|e| log::error!("Error {:?}", e), |_| ());
      }))
    }

    join_all(tasks).await;
....
  }

// convert array in streams to use async
let mut records = stream::iter(event.records); 
// processing one element at a time
while let Some(record) = records.next().await {

lambda_runtime = "0.5.1"

[dependencies]
aws-config = "0.12.0" // AWS SDK config and credential provider implementations.
lambda_runtime = "0.5.1" // Lambda runtime
serde = {version = "1.0", features = ["derive"] } //  framework for serializing and deserializing
serde_json = "1.0.68" //  framework for serializing and deserializing
tokio = "1.13.0" // A runtime for writing asynchronous required for the lambda_runtime
tracing-subscriber = "0.3" // required to enable CloudWatch error logging by the runtime

// imports my libraries
use lambda_runtime::{service_fn, Error, LambdaEvent};
use serde_json::Value;

// expected by lambda_runtime to make the main function asynchronous
#[tokio::main]
// Result<(), Error> is the idiomatic way to handle errors in Rust instead of try catch them
// https://doc.rust-lang.org/rust-by-example/error/result.html#using-result-in-main
async fn main() -> Result<(), Error> { 
    // required to enable CloudWatch error logging by the runtime
    tracing_subscriber::fmt()
        // this needs to be set to false, otherwise ANSI color codes will
        // show up in a confusing manner in CloudWatch logs.
        .with_ansi(false)
        // disabling time is handy because CloudWatch will add the ingestion time.
        .without_time()
        .init();

   // some custom code to init all my classes, SDK etc outside of the handler
   let app_client = MyAppClient::builder()
        .init_something(xxx)
        .init_something_else()
        .build();

    // this is necessary because rust is not supported as a provided runtime
    lambda_runtime::run(service_fn(|event: LambdaEvent<Value>| {
        execute(&app_client, event)
    }))
    .await?; //block until the results come back. The ? is to handle the propagation of the error.
    Ok(()) // tell the caller that things were successful 
}

// our handler maybe in the future will be the only thing that we will write
// app_client is my custom code that contains all the initialisation outside of the handler
// _event is the payload and it is going to be deserialised into Value 
pub async fn execute(app_client: &dyn MyAppInitialisation, _event: LambdaEvent<Value>) -> Result<(), Error> {
    //do something

    Ok(())
}

fn prints_and_returns_10(a: i32) -> i32 {
    println!("I got the value {}", a);
    10
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn this_test_will_pass() {
        let value = prints_and_returns_10(4);
        assert_eq!(10, value);
    }

    #[test]
    fn this_test_will_fail() {
        let value = prints_and_returns_10(8);
        assert_eq!(5, value);
    }
}

use aws_lambda_events::event::sqs::SqsEvent;
use aws_sdk_cloudfront::model::{invalidation_batch, paths};
use lambda_runtime::{service_fn, Error, LambdaEvent};
utils::injections::invalidation_di::{InvalidationAppClient, InvalidationAppInitialisation}};
use my_app::{dtos::invalidation_request::InvalidationRequest, utils::injections::invalidation_di::{InvalidationAppClient, InvalidationAppInitialisation}};

#[tokio::main]
async fn main() -> Result<(), Error> {
    tracing_subscriber::fmt()
        .with_ansi(false)
        .without_time()
        .init();

    let config = aws_config::load_from_env().await;
    let cloudfront_client = aws_sdk_cloudfront::Client::new(&config);
    let distribution_name = std::env::var("DISTRIBUTION_NAME").expect("DISTRIBUTION_NAME must be set");

    let app_client = InvalidationAppClient::builder()
        .cloudfront_client(cloudfront_client)
        .distribution_id(distribution_name)
        .build();

    lambda_runtime::run(service_fn(|event: LambdaEvent<SqsEvent>| {
        execute(&app_client, event)
    }))
    .await?;
    Ok(())
}

pub async fn execute(app_client: &dyn InvalidationAppInitialisation, event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
    println!("{:?}", event);

    let mut items: Vec<String> = vec![];
    for record in event.payload.records.into_iter() {
        if let Some(body) = &record.body {
            let request = serde_json::from_str::<InvalidationRequest>(&body)?;
            items.push(format!("/?key1={}&key2={}", request.my_key1, request.my_key2));
        }
    }

    let paths = paths::Builder::default()
        .quantity(items.len().try_into().unwrap())
        .set_items(Some(items))
        .build();
    let invalidation_batch = invalidation_batch::Builder::default()
        .paths(paths)
        .set_caller_reference(Some(format!("{}", app_client.get_timestamp())))
        .build();

    app_client.get_cloudfront_client()
        .create_invalidation()
        .distribution_id(app_client.get_distribution_id())
        .invalidation_batch(invalidation_batch)
        .send()
        .await?;

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use aws_sdk_cloudfront::{Client, Config, Credentials, Region};
    use aws_smithy_client::{test_connection::TestConnection, erase::DynConnector};
    use aws_smithy_http::body::SdkBody;
    use lambda_http::Context;
    use mockall::mock;
    use my_app::error::ApplicationError;

    async fn get_mock_config() -> aws_sdk_cloudfront::Config {
        let cfg = aws_config::from_env()
            .region(Region::new("eu-central-1"))
            .credentials_provider(Credentials::new(
                "accesskey",
                "privatekey",
                None,
                None,
                "dummy",
            ))
            .load()
            .await;

        Config::new(&cfg)
    }

    fn get_request_builder() -> http::request::Builder {
        http::Request::builder()
            .header("content-type", "application/xml")
            .uri(http::uri::Uri::from_static(
                "https://cloudfront.amazonaws.com/2020-05-31/distribution/some_distribution/invalidation",
            ))
    }

    #[tokio::test]
    async fn invalidate_key() -> Result<(), ApplicationError> {
        // ARRANGE
        let conn = TestConnection::new(vec![(
            get_request_builder()
                .body(SdkBody::from(
                    r#"<InvalidationBatch xmlns="http://cloudfront.amazonaws.com/doc/2020-05-31/"><Paths><Quantity>1</Quantity><Items><Path>/?key1=something&amp;key2=something</Path></Items></Paths><CallerReference>1336468263</CallerReference></InvalidationBatch>"#,
                ))
                .unwrap(),
            http::Response::builder()
                .status(200)
                .body(SdkBody::from(r#"
<Invalidation>
   <CreateTime>2019-12-05T18:40:49.413Z</CreateTime>
   <Id>I2J0I21PCUYOIK</Id>
   <InvalidationBatch>
      <CallerReference>1336468263</CallerReference>
      <Paths>
         <Items>
            <Path>/?key1=something&amp;key2=something</Path>
         </Items>
         <Quantity>1</Quantity>
      </Paths>
   </InvalidationBatch>
   <Status>InProgress</Status>
</Invalidation>
                "#))
                .unwrap(),
        )]);
        let client = Client::from_conf_conn(get_mock_config().await, DynConnector::new(conn.clone()));

        let data = r#"{
            "Records": [
              {
                "messageId": "059f34b4-87a3-46ab-83d2-661975830a7d",
                "receiptHandle": "AQEBwJnKyrHDfsdgvbUMZj6rYigCgxlaS3SLy0a",
                "body": "{"key1":"something","key2":"something"}",
                "attributes": {
                  "ApproximateReceiveCount": "1",
                  "SentTimestamp": "1545082649183",
                  "SenderId": "AIDAIENHSNOLO23YVJ4VO",
                  "ApproximateFirstReceiveTimestamp": "1545082649185"
                },
                "eventSource": "aws:sqs",
                "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
                "aws_region": "us-east-2"
              }
            ]
          }"#;

        let sqs = serde_json::from_str::<SqsEvent>(&data).unwrap();
        let context = Context::default();
        let event = LambdaEvent::new(sqs, context);

        mock! {
            pub InvalidationAppClient {}
            impl InvalidationAppInitialisation for InvalidationAppClient {
                fn get_timestamp(&self) -> i64 {
                    1336468263
                }

                fn get_cloudfront_client(&self) -> aws_sdk_cloudfront::Client {
                    client
                }

                fn get_distribution_id(&self) -> String {
                    "something".clone()
                }
            }
        }

        let mut mock = MockInvalidationAppClient::new();
        mock.expect_get_timestamp().times(1).return_const(1336468263);
        mock.expect_get_cloudfront_client().times(1).returning(move || client.clone());
        mock.expect_get_distribution_id().times(1).return_const("some_distribution");

        // ACT
        execute(&mock, event).await?;

        // ASSERT
        assert_eq!(conn.requests().len(), 1);
        conn.assert_requests_match(&vec![]);

        Ok(())
    }
}

async fn get_mock_config() -> aws_sdk_cloudfront::Config {}

fn get_request_builder() -> http::request::Builder {}

lambda_runtime::run(service_fn(|event: LambdaEvent<SqsEvent>| {
        execute(&app_client, event)
    })).await?;
}

pub async fn execute(app_client: &dyn InvalidationAppInitialisation, event: LambdaEvent<SqsEvent>) -> Result<(), Error> { }

.....
#[tokio::test]
async fn my_test() -> Result<(), ApplicationError> {
    let data = r#"{
            "Records": [
              {
                "messageId": "059f34b4-87a3-46ab-83d2-661975830a7d",
                "receiptHandle": "AQEBwJnKyrHDfsdgvbUMZj6rYigCgxlaS3SLy0a",
                "body": "{"key1":"something","key2":"something"}",
                "attributes": {
                  "ApproximateReceiveCount": "1",
                  "SentTimestamp": "1545082649183",
                  "SenderId": "AIDAIENHSNOLO23YVJ4VO",
                  "ApproximateFirstReceiveTimestamp": "1545082649185"
                },
                "eventSource": "aws:sqs",
                "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:my-queue",
                "aws_region": "us-east-2"
              }
            ]
          }"#;

        let sqs = serde_json::from_str::<SqsEvent>(&data).unwrap();
        let context = Context::default();
        let event = LambdaEvent::new(sqs, context);
...
}

lambda_http::run(service_fn(|event: Request| execute(&app_client, event))).await?;

...
pub async fn execute(app_client: &dyn InvalidationAppInitialisation, event: Request) > Result<impl IntoResponse, Error> { }

.....
#[tokio::test]
async fn my_test() -> Result<(), ApplicationError> {
        let mut item = HashMap::new();
        item.insert("xxxx".into(), vec!["key1".into()]);
        let request = Request::default().with_query_string_parameters(item.clone());

      //or in case of a post

      let request = http::Request::builder()
            .header("Content-Type", "application/json")
            .body(Body::from(
                r#"{"key1":"something", "key2": "something"}"#,
            ))
            .unwrap();
...
}

let app_client = ListAppClient::builder()
        .list_by_brand_query(query)
        .dynamo_db_client(dynamodb_client.clone())
        .build();

lambda_runtime::run(service_fn(|event: LambdaEvent<SqsEvent>| {
        execute(&app_client, event)
    })).await?;

pub async fn execute(app_client: &dyn InvalidationAppInitialisation, event: LambdaEvent<SqsEvent>) -> Result<(), Error> {

mock! {
            pub InvalidationAppClient {}
            impl InvalidationAppInitialisation for InvalidationAppClient {
                fn get_timestamp(&self) -> i64 {
                    1336468263
                }

                fn get_cloudfront_client(&self) -> aws_sdk_cloudfront::Client {
                    client
                }

                fn get_distribution_id(&self) -> String {
                    "something".clone()
                }
            }
        }

        let mut mock = MockInvalidationAppClient::new();
        mock.expect_get_timestamp().times(1).return_const(1336468263);
        mock.expect_get_cloudfront_client().times(1).returning(move || client.clone());
        mock.expect_get_distribution_id().times(1).return_const("some_distribution");

pub async fn execute(app_client: &dyn InvalidationAppInitialisation, event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
    let mut tasks = Vec::with_capacity(event.records.len());
....
    for record in event.records.into_iter() {
      // This will be the concrete init and it will not be possible to mock do_something()
      // let shared_app_client = app_client.clone(); 
      let shared_app_client = app_client.clone_to_arc();
      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 {
                     shared_app_client.do_something(&request).await
                       .map_or_else(|e| {
                         println!("ERROR {:?}", e);
                       }, |_| ());
                 } 
             }
        }));
    }

    join_all(tasks).await;
....
  }

    fn clone_to_arc(&self) -> Arc<dyn InvalidationAppInitialisation> {
        Arc::new(self.clone())
    }

mock! {
            pub InvalidationAppClient {}
            impl InvalidationAppInitialisation for InvalidationAppClient {
                ....
                fn clone_to_arc(&self) -> Arc<dyn InvalidationAppInitialisation> {
                    Arc::new(self.clone())
                }
            }
        }

        let mut mock = MockInvalidationAppClient::new();
        mock.expect_clone_to_arc().times(1).returning(move || {
             let mut m = MockInvalidationAppClient::new();
             m.expect_do_something().times(1).returning(|_| Ok(()));
             Arc::new(m)
        });