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
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
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181

//! Operation dispatch for the Kinesis emulator.
//!
//! [`Operation`] enumerates all 39 supported Kinesis API operations.
//! [`dispatch`] routes a parsed operation to its handler function.
//!
//! Action handler submodules are internal implementation details; import
//! [`Operation`] and [`dispatch`] directly from this module.

#[doc(hidden)]
pub mod add_tags_to_stream;
#[doc(hidden)]
pub mod create_stream;
#[doc(hidden)]
pub mod decrease_stream_retention_period;
#[doc(hidden)]
pub mod delete_resource_policy;
#[doc(hidden)]
pub mod delete_stream;
#[doc(hidden)]
pub mod deregister_stream_consumer;
#[doc(hidden)]
pub mod describe_account_settings;
#[doc(hidden)]
pub mod describe_limits;
#[doc(hidden)]
pub mod describe_stream;
#[doc(hidden)]
pub mod describe_stream_consumer;
#[doc(hidden)]
pub mod describe_stream_summary;
#[doc(hidden)]
pub mod disable_enhanced_monitoring;
#[doc(hidden)]
pub mod enable_enhanced_monitoring;
#[doc(hidden)]
pub mod get_records;
#[doc(hidden)]
pub mod get_resource_policy;
#[doc(hidden)]
pub mod get_shard_iterator;
#[doc(hidden)]
pub mod increase_stream_retention_period;
#[doc(hidden)]
pub mod list_shards;
#[doc(hidden)]
pub mod list_stream_consumers;
#[doc(hidden)]
pub mod list_streams;
#[doc(hidden)]
pub mod list_tags_for_resource;
#[doc(hidden)]
pub mod list_tags_for_stream;
#[doc(hidden)]
pub mod merge_shards;
#[doc(hidden)]
pub mod put_record;
#[doc(hidden)]
pub mod put_records;
#[doc(hidden)]
pub mod put_resource_policy;
#[doc(hidden)]
pub mod register_stream_consumer;
#[doc(hidden)]
pub mod remove_tags_from_stream;
#[doc(hidden)]
pub mod split_shard;
#[doc(hidden)]
pub mod start_stream_encryption;
#[doc(hidden)]
pub mod stop_stream_encryption;
#[cfg(not(target_arch = "wasm32"))]
#[doc(hidden)]
pub mod subscribe_to_shard;
#[doc(hidden)]
pub mod tag_resource;
#[doc(hidden)]
pub mod untag_resource;
#[doc(hidden)]
pub mod update_account_settings;
#[doc(hidden)]
pub mod update_max_record_size;
#[doc(hidden)]
pub mod update_shard_count;
#[doc(hidden)]
pub mod update_stream_mode;
#[doc(hidden)]
pub mod update_stream_warm_throughput;

pub use ferrokinesis_core::operation::Operation;

#[cfg(target_arch = "wasm32")]
use crate::constants;
use crate::error::KinesisErrorResponse;
use crate::store::Store;
use serde_json::Value;

/// Routes a parsed [`Operation`] to its handler.
///
/// Returns `Ok(None)` for operations that produce an empty 200 response, or
/// `Ok(Some(body))` with a JSON response body. `SubscribeToShard` is handled
/// separately in `server.rs` via `execute_streaming` and should never reach
/// this function during normal operation.
///
/// # Errors
///
/// Returns [`KinesisErrorResponse`] if the underlying handler returns an error,
/// or an `InternalFailure` (HTTP 500) if `SubscribeToShard` is dispatched here
/// (which should never happen in practice).
pub async fn dispatch(
    store: &Store,
    operation: Operation,
    data: Value,
) -> Result<Option<Value>, KinesisErrorResponse> {
    store.check_available()?;
    match operation {
        Operation::AddTagsToStream => add_tags_to_stream::execute(store, data).await,
        Operation::CreateStream => create_stream::execute(store, data).await,
        Operation::DecreaseStreamRetentionPeriod => {
            decrease_stream_retention_period::execute(store, data).await
        }
        Operation::DeleteResourcePolicy => delete_resource_policy::execute(store, data).await,
        Operation::DeleteStream => delete_stream::execute(store, data).await,
        Operation::DeregisterStreamConsumer => {
            deregister_stream_consumer::execute(store, data).await
        }
        Operation::DescribeAccountSettings => describe_account_settings::execute(store, data).await,
        Operation::DescribeLimits => describe_limits::execute(store, data).await,
        Operation::DescribeStream => describe_stream::execute(store, data).await,
        Operation::DescribeStreamConsumer => describe_stream_consumer::execute(store, data).await,
        Operation::DescribeStreamSummary => describe_stream_summary::execute(store, data).await,
        Operation::DisableEnhancedMonitoring => {
            disable_enhanced_monitoring::execute(store, data).await
        }
        Operation::EnableEnhancedMonitoring => {
            enable_enhanced_monitoring::execute(store, data).await
        }
        Operation::GetRecords => get_records::execute(store, data).await,
        Operation::GetResourcePolicy => get_resource_policy::execute(store, data).await,
        Operation::GetShardIterator => get_shard_iterator::execute(store, data).await,
        Operation::IncreaseStreamRetentionPeriod => {
            increase_stream_retention_period::execute(store, data).await
        }
        Operation::ListShards => list_shards::execute(store, data).await,
        Operation::ListStreamConsumers => list_stream_consumers::execute(store, data).await,
        Operation::ListStreams => list_streams::execute(store, data).await,
        Operation::ListTagsForResource => list_tags_for_resource::execute(store, data).await,
        Operation::ListTagsForStream => list_tags_for_stream::execute(store, data).await,
        Operation::MergeShards => merge_shards::execute(store, data).await,
        Operation::PutRecord => put_record::execute(store, data).await,
        Operation::PutRecords => put_records::execute(store, data).await,
        Operation::PutResourcePolicy => put_resource_policy::execute(store, data).await,
        Operation::RegisterStreamConsumer => register_stream_consumer::execute(store, data).await,
        Operation::RemoveTagsFromStream => remove_tags_from_stream::execute(store, data).await,
        Operation::SplitShard => split_shard::execute(store, data).await,
        Operation::StartStreamEncryption => start_stream_encryption::execute(store, data).await,
        Operation::StopStreamEncryption => stop_stream_encryption::execute(store, data).await,
        Operation::SubscribeToShard => unexpected_subscribe_to_shard(),
        Operation::TagResource => tag_resource::execute(store, data).await,
        Operation::UntagResource => untag_resource::execute(store, data).await,
        Operation::UpdateAccountSettings => update_account_settings::execute(store, data).await,
        Operation::UpdateMaxRecordSize => update_max_record_size::execute(store, data).await,
        Operation::UpdateShardCount => update_shard_count::execute(store, data).await,
        Operation::UpdateStreamMode => update_stream_mode::execute(store, data).await,
        Operation::UpdateStreamWarmThroughput => {
            update_stream_warm_throughput::execute(store, data).await
        }
    }
}

#[cfg(not(target_arch = "wasm32"))]
fn unexpected_subscribe_to_shard() -> Result<Option<Value>, KinesisErrorResponse> {
    Err(KinesisErrorResponse::server_error(None, None))
}

#[cfg(target_arch = "wasm32")]
fn unexpected_subscribe_to_shard() -> Result<Option<Value>, KinesisErrorResponse> {
    Err(KinesisErrorResponse::client_error(
        constants::INVALID_ARGUMENT,
        Some("SubscribeToShard is not supported in this build."),
    ))
}