ferrokinesis 0.1.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
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

pub mod add_tags_to_stream;
pub mod create_stream;
pub mod decrease_stream_retention_period;
pub mod delete_resource_policy;
pub mod delete_stream;
pub mod deregister_stream_consumer;
pub mod describe_account_settings;
pub mod describe_limits;
pub mod describe_stream;
pub mod describe_stream_consumer;
pub mod describe_stream_summary;
pub mod disable_enhanced_monitoring;
pub mod enable_enhanced_monitoring;
pub mod get_records;
pub mod get_resource_policy;
pub mod get_shard_iterator;
pub mod increase_stream_retention_period;
pub mod list_shards;
pub mod list_stream_consumers;
pub mod list_streams;
pub mod list_tags_for_resource;
pub mod list_tags_for_stream;
pub mod merge_shards;
pub mod put_record;
pub mod put_records;
pub mod put_resource_policy;
pub mod register_stream_consumer;
pub mod remove_tags_from_stream;
pub mod split_shard;
pub mod start_stream_encryption;
pub mod stop_stream_encryption;
pub mod subscribe_to_shard;
pub mod tag_resource;
pub mod untag_resource;
pub mod update_account_settings;
pub mod update_max_record_size;
pub mod update_shard_count;
pub mod update_stream_mode;
pub mod update_stream_warm_throughput;

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

pub async fn dispatch(
    store: &Store,
    operation: &str,
    data: Value,
) -> Result<Option<Value>, KinesisErrorResponse> {
    match operation {
        "AddTagsToStream" => add_tags_to_stream::execute(store, data).await,
        "CreateStream" => create_stream::execute(store, data).await,
        "DecreaseStreamRetentionPeriod" => {
            decrease_stream_retention_period::execute(store, data).await
        }
        "DeleteResourcePolicy" => delete_resource_policy::execute(store, data).await,
        "DeleteStream" => delete_stream::execute(store, data).await,
        "DeregisterStreamConsumer" => deregister_stream_consumer::execute(store, data).await,
        "DescribeAccountSettings" => describe_account_settings::execute(store, data).await,
        "DescribeLimits" => describe_limits::execute(store, data).await,
        "DescribeStream" => describe_stream::execute(store, data).await,
        "DescribeStreamConsumer" => describe_stream_consumer::execute(store, data).await,
        "DescribeStreamSummary" => describe_stream_summary::execute(store, data).await,
        "DisableEnhancedMonitoring" => disable_enhanced_monitoring::execute(store, data).await,
        "EnableEnhancedMonitoring" => enable_enhanced_monitoring::execute(store, data).await,
        "GetRecords" => get_records::execute(store, data).await,
        "GetResourcePolicy" => get_resource_policy::execute(store, data).await,
        "GetShardIterator" => get_shard_iterator::execute(store, data).await,
        "IncreaseStreamRetentionPeriod" => {
            increase_stream_retention_period::execute(store, data).await
        }
        "ListShards" => list_shards::execute(store, data).await,
        "ListStreamConsumers" => list_stream_consumers::execute(store, data).await,
        "ListStreams" => list_streams::execute(store, data).await,
        "ListTagsForResource" => list_tags_for_resource::execute(store, data).await,
        "ListTagsForStream" => list_tags_for_stream::execute(store, data).await,
        "MergeShards" => merge_shards::execute(store, data).await,
        "PutRecord" => put_record::execute(store, data).await,
        "PutRecords" => put_records::execute(store, data).await,
        "PutResourcePolicy" => put_resource_policy::execute(store, data).await,
        "RegisterStreamConsumer" => register_stream_consumer::execute(store, data).await,
        "RemoveTagsFromStream" => remove_tags_from_stream::execute(store, data).await,
        "SplitShard" => split_shard::execute(store, data).await,
        "StartStreamEncryption" => start_stream_encryption::execute(store, data).await,
        "StopStreamEncryption" => stop_stream_encryption::execute(store, data).await,
        "TagResource" => tag_resource::execute(store, data).await,
        "UntagResource" => untag_resource::execute(store, data).await,
        "UpdateAccountSettings" => update_account_settings::execute(store, data).await,
        "UpdateMaxRecordSize" => update_max_record_size::execute(store, data).await,
        "UpdateShardCount" => update_shard_count::execute(store, data).await,
        "UpdateStreamMode" => update_stream_mode::execute(store, data).await,
        "UpdateStreamWarmThroughput" => update_stream_warm_throughput::execute(store, data).await,
        _ => Err(KinesisErrorResponse::client_error(
            constants::UNKNOWN_OPERATION,
            None,
        )),
    }
}