dynamo-subscriber 0.1.1

Subscribe DynamoDB Streams as tokio-stream
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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128

use aws_config::{retry::RetryConfig, BehaviorVersion, Region, SdkConfig};
use aws_credential_types::{provider::SharedCredentialsProvider, Credentials};
use aws_sdk_dynamodb::{
    types::{
        AttributeDefinition, AttributeValue, BillingMode, KeySchemaElement, KeyType,
        ScalarAttributeType, StreamSpecification, StreamViewType,
    },
    Client,
};
use aws_sdk_dynamodbstreams::types::Record;
use dynamo_subscriber::stream::ConsumerChannel;
use tokio::time::{sleep, Duration};
use ulid::Ulid;

const TABLE: &str = "People";
const PK: &str = "Id";

pub struct TestConfig {
    table_name: String,
    config: SdkConfig,
}

impl TestConfig {
    pub fn table_name(&self) -> &str {
        self.table_name.as_str()
    }

    pub fn aws_sdk_config(&self) -> &SdkConfig {
        &self.config
    }
}

pub async fn setup() -> TestConfig {
    let creds = Credentials::from_keys(Ulid::new(), Ulid::new(), None);
    let creds_provider = SharedCredentialsProvider::new(creds);

    let retry = RetryConfig::standard().with_max_attempts(5);

    let config = SdkConfig::builder()
        .endpoint_url("http://localhost:8000")
        .credentials_provider(creds_provider)
        .retry_config(retry)
        .behavior_version(BehaviorVersion::latest())
        .region(Some(Region::from_static("us-east-1")))
        .build();

    create_table(&config).await;

    TestConfig {
        table_name: TABLE.to_string(),
        config,
    }
}

pub async fn put_item(pk: &str, config: &SdkConfig) {
    Client::new(config)
        .put_item()
        .table_name(TABLE)
        .item(PK, AttributeValue::S(pk.into()))
        .send()
        .await
        .unwrap();
}

pub async fn teardown(config: &SdkConfig) {
    drop_table(config).await;
}

pub async fn wait_until_initialized(channel: &mut ConsumerChannel) {
    let mut res = channel.initialized();

    while !res {
        sleep(Duration::from_millis(100)).await;
        res = channel.initialized();
    }
}

pub fn pk(record: &Record) -> String {
    record
        .dynamodb()
        .and_then(|dynamodb| dynamodb.keys())
        .unwrap()
        .get(PK)
        .unwrap()
        .as_s()
        .unwrap()
        .to_string()
}

async fn create_table(config: &SdkConfig) {
    Client::new(config)
        .create_table()
        .attribute_definitions(
            AttributeDefinition::builder()
                .attribute_name(PK)
                .attribute_type(ScalarAttributeType::S)
                .build()
                .unwrap(),
        )
        .table_name(TABLE)
        .key_schema(
            KeySchemaElement::builder()
                .attribute_name(PK)
                .key_type(KeyType::Hash)
                .build()
                .unwrap(),
        )
        .billing_mode(BillingMode::PayPerRequest)
        .stream_specification(
            StreamSpecification::builder()
                .stream_enabled(true)
                .stream_view_type(StreamViewType::NewImage)
                .build()
                .unwrap(),
        )
        .send()
        .await
        .unwrap();
}

async fn drop_table(config: &SdkConfig) {
    Client::new(config)
        .delete_table()
        .table_name(TABLE)
        .send()
        .await
        .unwrap();
}