async fn create_state(
    base_metrics: &KinesisMetrics,
    c: KinesisSourceConnector,
) -> Result<
    (
        KinesisClient,
        String,
        HashMap<String, ShardMetrics>,
        VecDeque<(String, Option<String>)>,
    ),
    anyhow::Error,
> {
    let config = c.aws.load().await?;
    let kinesis_client = kinesis::client(&config)?;

    let shard_set = kinesis::get_shard_ids(&kinesis_client, &c.stream_name).await?;
    let mut shard_queue: VecDeque<(String, Option<String>)> = VecDeque::new();
    let mut shard_map = HashMap::new();
    for shard_id in shard_set {
        shard_queue.push_back((
            shard_id.clone(),
            kinesis::get_shard_iterator(&kinesis_client, &c.stream_name, &shard_id).await?,
        ));
        shard_map.insert(
            shard_id.clone(),
            ShardMetrics::new(base_metrics, &c.stream_name, &shard_id),
        );
    }

    Ok((
        kinesis_client,
        c.stream_name.clone(),
        shard_map,
        shard_queue,
    ))
}
/// Creates the necessary data-structures for shard management
// todo: Better error handling here! Not all errors mean we're done/can't progress.
Creates the necessary data-structures for shard management
todo: Better error handling here. Not all errors mean we're done/can't progress.
Creates the necessary data - structures for shard management todo : Better error handling here . Not all errors mean we ' re done / can ' t progress .
async fn create_state(
    base_metrics: &KinesisMetrics,
    c: KinesisSourceConnector,
) -> Result<
    (
        KinesisClient,
        String,
        HashMap<String, ShardMetrics>,
        VecDeque<(String, Option<String>)>,
    ),
    anyhow::Error,
> {
    let config = c.aws.load().await?;
    let kinesis_client = kinesis::client(&config)?;
    let shard_set = kinesis::get_shard_ids(&kinesis_client, &c.stream_name).await?;
    let mut shard_queue: VecDeque<(String, Option<String>)> = VecDeque::new();
    let mut shard_map = HashMap::new();
    for shard_id in shard_set {
        shard_queue.push_back((
            shard_id.clone(),
            kinesis::get_shard_iterator(&kinesis_client, &c.stream_name, &shard_id).await?,
        ));
        shard_map.insert(
            shard_id.clone(),
            ShardMetrics::new(base_metrics, &c.stream_name, &shard_id),
        );
    }
    Ok((
        kinesis_client,
        c.stream_name.clone(),
        shard_map,
        shard_queue,
    ))
}
async fn create_state ( base_metrics : & KinesisMetrics , c : KinesisSourceConnector , ) -> Result < ( KinesisClient , String , HashMap < String , ShardMetrics > , VecDeque < ( String , Option < String > ) > , ) , anyhow :: Error , > { let config = c . aws . load ( ) . await ? ; let kinesis_client = kinesis :: client ( & config ) ? ; let shard_set = kinesis :: get_shard_ids ( & kinesis_client , & c . stream_name ) . await ? ; let mut shard_queue : VecDeque < ( String , Option < String > ) > = VecDeque :: new ( ) ; let mut shard_map = HashMap :: new ( ) ; for shard_id in shard_set { shard_queue . push_back ( ( shard_id . clone ( ) , kinesis :: get_shard_iterator ( & kinesis_client , & c . stream_name , & shard_id ) . await ? , ) ) ; shard_map . insert ( shard_id . clone ( ) , ShardMetrics :: new ( base_metrics , & c . stream_name , & shard_id ) , ) ; } Ok ( ( kinesis_client , c . stream_name . clone ( ) , shard_map , shard_queue , ) ) }
Creates the necessary data-structures for shard management
todo: Better error handling here!
Creates the necessary data - structures for shard management todo : Better error handling here!
base_metrics
&KinesisMetrics
c
KinesisSourceConnector
base_metrics
&KinesisMetrics
c
KinesisSourceConnector