Removed cloned shard_id

grumlimited · Oct 22, 2023 · 09a249b · 09a249b
1 parent 9a9880b
commit 09a249b
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 18 deletions.
diff --git a/src/kinesis/helpers.rs b/src/kinesis/helpers.rs
@@ -7,6 +7,7 @@ use aws_sdk_kinesis::operation::get_shard_iterator::{
     GetShardIteratorError, GetShardIteratorOutput,
 };
 use aws_sdk_kinesis::operation::list_shards::ListShardsOutput;
+use aws_sdk_kinesis::types::Shard;
 use chrono::Utc;
 use log::{debug, info};
 use tokio::sync::mpsc::Sender;
@@ -165,13 +166,10 @@ pub async fn get_shards(client: &AwsKinesisClient, stream: &str) -> io::Result<V
         Ok(_) => {
             let shards: Vec<String> = results
                 .iter()
-                .flat_map(|r| {
-                    r.shards()
-                        .unwrap()
-                        .iter()
-                        .map(|s| s.shard_id().unwrap().to_string())
-                        .collect::<Vec<String>>()
-                })
+                .filter_map(ListShardsOutput::shards)
+                .flat_map(|s| s.iter())
+                .filter_map(Shard::shard_id)
+                .map(str::to_string)
                 .collect::<Vec<String>>();
 
             info!("Found {} shards", shards.len());

diff --git a/src/main.rs b/src/main.rs
@@ -123,18 +123,18 @@ async fn main() -> Result<()> {
                 let shard_id = shard_id.clone();
                 let semaphore = semaphore.clone();
 
-                tokio::spawn(async move {
-                    let shard_processor = kinesis::helpers::new(
-                        client.clone(),
-                        stream_name,
-                        shard_id,
-                        from_datetime,
-                        to_datetime,
-                        semaphore,
-                        tx_records.clone(),
-                        tx_ticker_updates.clone(),
-                    );
+                let shard_processor = kinesis::helpers::new(
+                    client.clone(),
+                    stream_name,
+                    shard_id,
+                    from_datetime,
+                    to_datetime,
+                    semaphore,
+                    tx_records.clone(),
+                    tx_ticker_updates.clone(),
+                );
 
+                tokio::spawn(async move {
                     shard_processor.run().await.unwrap();
                 })
             })