Struct SyncConsumer

Source

pub struct SyncConsumer<T>where
    T: Send + Sync + 'static + Debug + Clone,
{ /* private fields */ }

Expand description

Synchronous consumer for records of type T.

Thread-safe, can be cloned and shared across threads. Each clone receives data independently according to buffer semantics (SPMC, etc.).

§Thread Safety

Multiple clones of SyncConsumer<T> can be used concurrently from different threads. Each receives data independently based on the configured buffer type (SPMC, SingleLatest, etc.).

§Example

// Get value (blocks until available)
let temp = consumer.get()?;
println!("Temperature: {}°C", temp.celsius);

// Get with timeout
use std::time::Duration;
match consumer.get_with_timeout(Duration::from_millis(100)) {
    Ok(temp) => println!("Got: {}°C", temp.celsius),
    Err(_) => println!("No data available"),
}

// Try to get (non-blocking)
match consumer.try_get() {
    Ok(temp) => println!("Got: {}°C", temp.celsius),
    Err(_) => println!("No data yet"),
}

Implementations§

Source §

impl<T> SyncConsumer<T>
where T: Send + Sync + 'static + Debug + Clone,

Source

pub fn get(&self) -> DbResult<T>

Get a value, blocking until one is available.

Blocks indefinitely until a value is available from the runtime thread.

§Returns

The next available record of type T.

§Errors

DbError::RuntimeShutdown if the runtime thread has stopped

§Example

use aimdb_core::AimDbBuilder;
use aimdb_sync::AimDbBuilderSyncExt;
use aimdb_tokio_adapter::TokioAdapter;
use std::sync::Arc;

let handle = AimDbBuilder::new()
    .runtime(Arc::new(TokioAdapter))
    .attach()?;
let consumer = handle.consumer::<MyData>("my_data")?;
let data = consumer.get()?; // blocks until value available
println!("Got: {:?}", data);

Source

pub fn get_with_timeout(&self, timeout: Duration) -> DbResult<T>

Get a value with a timeout.

Blocks until a value is available or the timeout expires.

§Arguments

timeout: Maximum time to wait

§Errors

DbError::GetTimeout if the timeout expires
DbError::RuntimeShutdown if the runtime thread has stopped

§Example

use aimdb_core::AimDbBuilder;
use aimdb_sync::AimDbBuilderSyncExt;
use aimdb_tokio_adapter::TokioAdapter;
use std::sync::Arc;
use std::time::Duration;

let handle = AimDbBuilder::new()
    .runtime(Arc::new(TokioAdapter))
    .attach()?;
let consumer = handle.consumer::<MyData>("my_data")?;
match consumer.get_with_timeout(Duration::from_millis(100)) {
    Ok(data) => println!("Got: {:?}", data),
    Err(_) => println!("No data available"),
}

Source

pub fn try_get(&self) -> DbResult<T>

Try to get a value without blocking.

Returns immediately with either a value or an error if no data is available.

§Errors

DbError::GetTimeout if no data is available (non-blocking)
DbError::RuntimeShutdown if the runtime thread has stopped

§Example

use aimdb_core::AimDbBuilder;
use aimdb_sync::AimDbBuilderSyncExt;
use aimdb_tokio_adapter::TokioAdapter;
use std::sync::Arc;

let handle = AimDbBuilder::new()
    .runtime(Arc::new(TokioAdapter))
    .attach()?;
let consumer = handle.consumer::<MyData>("my_data")?;
match consumer.try_get() {
    Ok(data) => println!("Got: {:?}", data),
    Err(_) => println!("No data yet"),
}

Source

pub fn get_latest(&self) -> DbResult<T>

Get the latest value by draining all queued values.

This method drains the internal channel to get the most recent value, discarding any intermediate values. This is useful for SingleLatest-like semantics where you only care about the most recent data.

Blocks until at least one value is available, then drains all queued values and returns the last one.

§Returns

The most recent available record of type T.

§Errors

DbError::RuntimeShutdown if the runtime thread has stopped

§Example

use aimdb_core::AimDbBuilder;
use aimdb_sync::AimDbBuilderSyncExt;
use aimdb_tokio_adapter::TokioAdapter;
use std::sync::Arc;

let handle = AimDbBuilder::new()
    .runtime(Arc::new(TokioAdapter))
    .attach()?;
let consumer = handle.consumer::<MyData>("my_data")?;

// Get the latest value, skipping any queued intermediate values
let latest = consumer.get_latest()?;
println!("Latest: {:?}", latest);

Source

pub fn get_latest_with_timeout(&self, timeout: Duration) -> DbResult<T>

Get the latest value with a timeout, draining all queued values.

Like get_latest(), but with a timeout. Blocks until at least one value is available or the timeout expires, then drains all queued values and returns the last one.

§Arguments

timeout: Maximum time to wait for the first value

§Errors

DbError::GetTimeout if the timeout expires before any value arrives
DbError::RuntimeShutdown if the runtime thread has stopped

§Example

use aimdb_core::AimDbBuilder;
use aimdb_sync::AimDbBuilderSyncExt;
use aimdb_tokio_adapter::TokioAdapter;
use std::sync::Arc;
use std::time::Duration;

let handle = AimDbBuilder::new()
    .runtime(Arc::new(TokioAdapter))
    .attach()?;
let consumer = handle.consumer::<MyData>("my_data")?;

// Get the latest value within 100ms
match consumer.get_latest_with_timeout(Duration::from_millis(100)) {
    Ok(latest) => println!("Latest: {:?}", latest),
    Err(_) => println!("No data available"),
}

Trait Implementations§

Source §

impl<T> Clone for SyncConsumer<T>
where T: Send + Sync + 'static + Debug + Clone,

Source §

fn clone(&self) -> Self

Clone the consumer to share across threads.

Note: All clones share the same receiver, so only one thread will receive each value. For independent subscriptions, call handle.consumer() multiple times instead.

1.0.0 · Source§