Repository: MystenLabs/sui
File: crates/sui-indexer-alt-framework/src/ingestion/streaming_client.rs (Rust)

Code before:
        }

        /// Make `connect` fail for the next N calls
        pub fn fail_connection_times(mut self, times: usize) -> Self {
            self.connection_failures_remaining = times;
            self
        }

        /// Make `connect` timeout for the next N calls
        pub fn fail_connection_with_timeout(mut self, times: usize) -> Self {
            self.connection_timeouts_remaining = times;
            self
        }

        /// Insert an error at the back of the queue.
        pub fn insert_error(&mut self) {
            self.actions.lock().unwrap().push(StreamAction::Error);
        }

        /// Insert a timeout at the back of the queue (causes poll_next to return Pending).
        /// Defaults to 6 seconds to ensure 5-second wrapper times out first.
        pub fn insert_timeout(&mut self) {
            self.actions.lock().unwrap().push(StreamAction::Timeout {
                deadline: None,
                duration: self.timeout_duration,
            });
        }

        /// Insert a timeout with custom duration.
        pub fn insert_timeout_with_duration(&mut self, duration: Duration) {
            self.actions.lock().unwrap().push(StreamAction::Timeout {
                deadline: None,
                duration,
            });
        }

        /// Insert a checkpoint at the back of the queue.
        pub fn insert_checkpoint(&mut self, sequence_number: u64) {
            self.insert_checkpoint_range([sequence_number])
        }

        pub fn insert_checkpoint_range<I>(&mut self, checkpoint_range: I)
        where
            I: IntoIterator<Item = u64>,
        {
            let mut actions = self.actions.lock().unwrap();
            for sequence_number in checkpoint_range {
                actions.push(StreamAction::Checkpoint(sequence_number));
            }
        }
    }

Review comment (suggestion):
```suggestion
            self.insert_timeout_with_duration(self.timeout_duration);
```

Code after:
        }

        /// Make `connect` fail for the next N calls
        pub fn fail_connection_times(mut self, times: usize) -> Self {
            self.connection_failures_remaining = times;
            self
        }

        /// Make `connect` timeout for the next N calls
        pub fn fail_connection_with_timeout(mut self, times: usize) -> Self {
            self.connection_timeouts_remaining = times;
            self
        }

        /// Insert an error at the back of the queue.
        pub fn insert_error(&mut self) {
            self.actions.lock().unwrap().push(StreamAction::Error);
        }

        /// Insert a timeout at the back of the queue (causes poll_next to return Pending).
        pub fn insert_timeout(&mut self) {
            self.insert_timeout_with_duration(self.timeout_duration)
        }

        /// Insert a timeout with custom duration.
        pub fn insert_timeout_with_duration(&mut self, duration: Duration) {
            self.actions.lock().unwrap().push(StreamAction::Timeout {
                deadline: None,
                duration,
            });
        }

        /// Insert a checkpoint at the back of the queue.
        pub fn insert_checkpoint(&mut self, sequence_number: u64) {
            self.insert_checkpoint_range([sequence_number])
        }

        pub fn insert_checkpoint_range<I>(&mut self, checkpoint_range: I)
        where
            I: IntoIterator<Item = u64>,
        {
            let mut actions = self.actions.lock().unwrap();
            for sequence_number in checkpoint_range {
                actions.push(StreamAction::Checkpoint(sequence_number));
            }
        }
    }

    #[async_trait]
    impl CheckpointStreamingClient for MockStreamingClient {
        async fn connect(&mut self) -> Result<CheckpointStream> {