Struct StreamClient 

pub struct StreamClient { /* private fields */ }

Client for stream-level operations.

Implementations

impl StreamClient

pub fn new( config: ClientConfig, basin: BasinName, stream: impl Into<String>, ) -> Self

Create a new stream client.

Examples found in repository

examples/get_latest_record.rs (line 12)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let read_limit = ReadLimit::new().with_count(1);
    let read_request = ReadRequest::new(ReadStart::TailOffset(1)).with_limit(read_limit);
    let latest_record = stream_client.read(read_request).await?;

    println!("{latest_record:#?}");

    Ok(())
}

examples/consumer.rs (line 14)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let start_seq_num = 0;
    let read_session_request = ReadSessionRequest::new(ReadStart::SeqNum(start_seq_num));
    let mut read_stream = stream_client.read_session(read_session_request).await?;

    loop {
        select! {
            next_batch = read_stream.next() => {
                let Some(next_batch) = next_batch else { break };
                let next_batch = next_batch?;
                println!("{next_batch:?}");
            }
            _ = tokio::signal::ctrl_c() => break,
        }
    }

    Ok(())
}

examples/explicit_trim.rs (line 12)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let tail = stream_client.check_tail().await?;
    if tail.seq_num == 0 {
        println!("Empty stream");
        return Ok(());
    }

    let latest_seq_num = tail.seq_num - 1;
    let trim_request = CommandRecord::trim(latest_seq_num);

    let append_record_batch = AppendRecordBatch::try_from_iter([trim_request])
        .expect("valid batch with 1 command record");
    let append_input = AppendInput::new(append_record_batch);
    let _ = stream_client.append(append_input).await?;

    println!("Trim requested");

    Ok(())
}

examples/producer.rs (line 14)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let fencing_token = FencingToken::generate(32).expect("valid fencing token");

    // Set the fencing token.
    let fencing_token_record: AppendRecord = CommandRecord::fence(fencing_token.clone()).into();
    let fencing_token_batch = AppendRecordBatch::try_from_iter([fencing_token_record])
        .expect("valid batch with 1 append record");
    let fencing_token_append_input = AppendInput::new(fencing_token_batch);
    let set_fencing_token = stream_client.append(fencing_token_append_input).await?;

    let match_seq_num = set_fencing_token.tail.seq_num;

    // Stream of records
    let append_stream = futures::stream::iter([
        AppendRecord::new("record_1")?,
        AppendRecord::new("record_2")?,
    ]);

    let append_records_batching_opts = AppendRecordsBatchingOpts::new()
        .with_fencing_token(Some(fencing_token))
        .with_match_seq_num(Some(match_seq_num));

    let append_session_request =
        AppendRecordsBatchingStream::new(append_stream, append_records_batching_opts);

    let mut append_session_stream = stream_client.append_session(append_session_request).await?;

    while let Some(next) = append_session_stream.next().await {
        let next = next?;
        println!("{next:#?}");
    }

    Ok(())
}

pub fn new_with_connector<C>( config: ClientConfig, basin: BasinName, stream: impl Into<String>, connector: C, ) -> Self

where C: Service<Uri> + Send + 'static, C::Response: Read + Write + Send + Unpin, C::Future: Send, C::Error: Error + Send + Sync + 'static,

Create a new stream client using a custom connector.

pub async fn check_tail(&self) -> Result<StreamPosition, ClientError>

Check the tail of the stream.

Examples found in repository

examples/explicit_trim.rs (line 14)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let tail = stream_client.check_tail().await?;
    if tail.seq_num == 0 {
        println!("Empty stream");
        return Ok(());
    }

    let latest_seq_num = tail.seq_num - 1;
    let trim_request = CommandRecord::trim(latest_seq_num);

    let append_record_batch = AppendRecordBatch::try_from_iter([trim_request])
        .expect("valid batch with 1 command record");
    let append_input = AppendInput::new(append_record_batch);
    let _ = stream_client.append(append_input).await?;

    println!("Trim requested");

    Ok(())
}

pub async fn read(&self, req: ReadRequest) -> Result<ReadOutput, ClientError>

Retrieve a batch of records from a stream.

Examples found in repository

examples/get_latest_record.rs (line 16)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let read_limit = ReadLimit::new().with_count(1);
    let read_request = ReadRequest::new(ReadStart::TailOffset(1)).with_limit(read_limit);
    let latest_record = stream_client.read(read_request).await?;

    println!("{latest_record:#?}");

    Ok(())
}

pub async fn read_session( &self, req: ReadSessionRequest, ) -> Result<Streaming<ReadOutput>, ClientError>

Retrieve batches of records from a stream continuously.

Examples found in repository

examples/consumer.rs (line 18)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let start_seq_num = 0;
    let read_session_request = ReadSessionRequest::new(ReadStart::SeqNum(start_seq_num));
    let mut read_stream = stream_client.read_session(read_session_request).await?;

    loop {
        select! {
            next_batch = read_stream.next() => {
                let Some(next_batch) = next_batch else { break };
                let next_batch = next_batch?;
                println!("{next_batch:?}");
            }
            _ = tokio::signal::ctrl_c() => break,
        }
    }

    Ok(())
}

pub async fn append(&self, req: AppendInput) -> Result<AppendAck, ClientError>

Append a batch of records to a stream.

Examples found in repository

examples/explicit_trim.rs (line 26)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let tail = stream_client.check_tail().await?;
    if tail.seq_num == 0 {
        println!("Empty stream");
        return Ok(());
    }

    let latest_seq_num = tail.seq_num - 1;
    let trim_request = CommandRecord::trim(latest_seq_num);

    let append_record_batch = AppendRecordBatch::try_from_iter([trim_request])
        .expect("valid batch with 1 command record");
    let append_input = AppendInput::new(append_record_batch);
    let _ = stream_client.append(append_input).await?;

    println!("Trim requested");

    Ok(())
}

examples/producer.rs (line 23)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let fencing_token = FencingToken::generate(32).expect("valid fencing token");

    // Set the fencing token.
    let fencing_token_record: AppendRecord = CommandRecord::fence(fencing_token.clone()).into();
    let fencing_token_batch = AppendRecordBatch::try_from_iter([fencing_token_record])
        .expect("valid batch with 1 append record");
    let fencing_token_append_input = AppendInput::new(fencing_token_batch);
    let set_fencing_token = stream_client.append(fencing_token_append_input).await?;

    let match_seq_num = set_fencing_token.tail.seq_num;

    // Stream of records
    let append_stream = futures::stream::iter([
        AppendRecord::new("record_1")?,
        AppendRecord::new("record_2")?,
    ]);

    let append_records_batching_opts = AppendRecordsBatchingOpts::new()
        .with_fencing_token(Some(fencing_token))
        .with_match_seq_num(Some(match_seq_num));

    let append_session_request =
        AppendRecordsBatchingStream::new(append_stream, append_records_batching_opts);

    let mut append_session_stream = stream_client.append_session(append_session_request).await?;

    while let Some(next) = append_session_stream.next().await {
        let next = next?;
        println!("{next:#?}");
    }

    Ok(())
}

pub async fn append_session<S>( &self, req: S, ) -> Result<Streaming<AppendAck>, ClientError>

where S: 'static + Send + Unpin + Stream<Item = AppendInput>,

Append batches of records to a stream continuously, while guaranteeing pipelined requests are processed in order. If any request fails, the session is terminated.

Examples found in repository

examples/producer.rs (line 40)

async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let token = std::env::var("S2_ACCESS_TOKEN")?;
    let config = ClientConfig::new(token);
    let basin: BasinName = "my-favorite-basin".parse()?;
    let stream = "my-favorite-stream";
    let stream_client = StreamClient::new(config, basin, stream);

    let fencing_token = FencingToken::generate(32).expect("valid fencing token");

    // Set the fencing token.
    let fencing_token_record: AppendRecord = CommandRecord::fence(fencing_token.clone()).into();
    let fencing_token_batch = AppendRecordBatch::try_from_iter([fencing_token_record])
        .expect("valid batch with 1 append record");
    let fencing_token_append_input = AppendInput::new(fencing_token_batch);
    let set_fencing_token = stream_client.append(fencing_token_append_input).await?;

    let match_seq_num = set_fencing_token.tail.seq_num;

    // Stream of records
    let append_stream = futures::stream::iter([
        AppendRecord::new("record_1")?,
        AppendRecord::new("record_2")?,
    ]);

    let append_records_batching_opts = AppendRecordsBatchingOpts::new()
        .with_fencing_token(Some(fencing_token))
        .with_match_seq_num(Some(match_seq_num));

    let append_session_request =
        AppendRecordsBatchingStream::new(append_stream, append_records_batching_opts);

    let mut append_session_stream = stream_client.append_session(append_session_request).await?;

    while let Some(next) = append_session_stream.next().await {
        let next = next?;
        println!("{next:#?}");
    }

    Ok(())
}

Trait Implementations

impl Clone for StreamClient

fn clone(&self) -> StreamClient

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for StreamClient

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.