Skip to content

Commit 0fcba16

Browse files
authored
Improve Kinesis event recort type (#886)
Make the encryption type an enum. Make sequence number and partition key non-optional. Signed-off-by: David Calavera <[email protected]>
1 parent 31250f3 commit 0fcba16

File tree

2 files changed

+63
-3
lines changed

2 files changed

+63
-3
lines changed

lambda-events/src/event/kinesis/event.rs

Lines changed: 26 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -62,15 +62,24 @@ pub struct KinesisEventRecord {
6262
pub struct KinesisRecord {
6363
pub approximate_arrival_timestamp: SecondTimestamp,
6464
pub data: Base64Data,
65-
pub encryption_type: Option<String>,
6665
#[serde(default)]
67-
pub partition_key: Option<String>,
66+
pub encryption_type: KinesisEncryptionType,
6867
#[serde(default)]
69-
pub sequence_number: Option<String>,
68+
pub partition_key: String,
69+
#[serde(default)]
70+
pub sequence_number: String,
7071
#[serde(default)]
7172
pub kinesis_schema_version: Option<String>,
7273
}
7374

75+
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
76+
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
77+
pub enum KinesisEncryptionType {
78+
#[default]
79+
None,
80+
Kms,
81+
}
82+
7483
#[cfg(test)]
7584
mod test {
7685
use super::*;
@@ -80,6 +89,20 @@ mod test {
8089
fn example_kinesis_event() {
8190
let data = include_bytes!("../../fixtures/example-kinesis-event.json");
8291
let parsed: KinesisEvent = serde_json::from_slice(data).unwrap();
92+
assert_eq!(KinesisEncryptionType::None, parsed.records[0].kinesis.encryption_type);
93+
94+
let output: String = serde_json::to_string(&parsed).unwrap();
95+
let reparsed: KinesisEvent = serde_json::from_slice(output.as_bytes()).unwrap();
96+
assert_eq!(parsed, reparsed);
97+
}
98+
99+
#[test]
100+
#[cfg(feature = "kinesis")]
101+
fn example_kinesis_event_encrypted() {
102+
let data = include_bytes!("../../fixtures/example-kinesis-event-encrypted.json");
103+
let parsed: KinesisEvent = serde_json::from_slice(data).unwrap();
104+
assert_eq!(KinesisEncryptionType::Kms, parsed.records[0].kinesis.encryption_type);
105+
83106
let output: String = serde_json::to_string(&parsed).unwrap();
84107
let reparsed: KinesisEvent = serde_json::from_slice(output.as_bytes()).unwrap();
85108
assert_eq!(parsed, reparsed);
Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
{
2+
"Records": [
3+
{
4+
"kinesis": {
5+
"kinesisSchemaVersion": "1.0",
6+
"partitionKey": "s1",
7+
"sequenceNumber": "49568167373333333333333333333333333333333333333333333333",
8+
"data": "SGVsbG8gV29ybGQ=",
9+
"approximateArrivalTimestamp": 1480641523.477,
10+
"encryptionType": "KMS"
11+
},
12+
"eventSource": "aws:kinesis",
13+
"eventVersion": "1.0",
14+
"eventID": "shardId-000000000000:49568167373333333333333333333333333333333333333333333333",
15+
"eventName": "aws:kinesis:record",
16+
"invokeIdentityArn": "arn:aws:iam::123456789012:role/LambdaRole",
17+
"awsRegion": "us-east-1",
18+
"eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/simple-stream"
19+
},
20+
{
21+
"kinesis": {
22+
"kinesisSchemaVersion": "1.0",
23+
"partitionKey": "s1",
24+
"sequenceNumber": "49568167373333333334444444444444444444444444444444444444",
25+
"data": "SGVsbG8gV29ybGQ=",
26+
"approximateArrivalTimestamp": 1480841523.477
27+
},
28+
"eventSource": "aws:kinesis",
29+
"eventVersion": "1.0",
30+
"eventID": "shardId-000000000000:49568167373333333334444444444444444444444444444444444444",
31+
"eventName": "aws:kinesis:record",
32+
"invokeIdentityArn": "arn:aws:iam::123456789012:role/LambdaRole",
33+
"awsRegion": "us-east-1",
34+
"eventSourceARN": "arn:aws:kinesis:us-east-1:123456789012:stream/simple-stream"
35+
}
36+
]
37+
}

0 commit comments

Comments
 (0)