ferrokinesis 0.7.0

A local AWS Kinesis mock server for testing, written in Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

use crate::constants;
use crate::error::KinesisErrorResponse;
use crate::store::Store;
use serde_json::Value;

pub async fn execute(store: &Store, data: Value) -> Result<Option<Value>, KinesisErrorResponse> {
    let resource_arn = data[constants::RESOURCE_ARN].as_str().unwrap_or("");

    if resource_arn.is_empty() {
        return Err(KinesisErrorResponse::client_error(
            constants::INVALID_ARGUMENT,
            Some("ResourceARN is required."),
        ));
    }

    let tag_keys: Vec<String> = data[constants::TAG_KEYS]
        .as_array()
        .ok_or_else(|| {
            KinesisErrorResponse::client_error(
                constants::INVALID_ARGUMENT,
                Some("TagKeys is required."),
            )
        })?
        .iter()
        .filter_map(|v| v.as_str().map(|s| s.to_string()))
        .collect();

    // Check if this is a stream ARN
    if resource_arn.contains(":stream/")
        && !resource_arn.contains("/consumer/")
        && let Some(stream_name) = store.stream_name_from_arn(resource_arn)
    {
        store
            .update_stream(&stream_name, |stream| {
                for key in &tag_keys {
                    stream.tags.remove(key);
                }
                Ok(())
            })
            .await?;
        tracing::trace!(resource_arn, tags = tag_keys.len(), "resource untagged");
        return Ok(None);
    }

    // For non-stream resources
    let mut existing = store.get_resource_tags(resource_arn).await;
    for key in &tag_keys {
        existing.remove(key);
    }
    store.put_resource_tags(resource_arn, &existing).await?;

    tracing::trace!(resource_arn, tags = tag_keys.len(), "resource untagged");
    Ok(None)
}