|
| 1 | +//! # AimDB SingleLatest Sync Demo |
| 2 | +//! This example demonstrates the `SingleLatest` buffer primitive using AimDB's synchronous API. |
| 3 | +//! |
| 4 | +//! ## What This Demo Shows |
| 5 | +//! |
| 6 | +//! 1. Building an AimDB instance with a typed record and 2 tap observers (slow and fast) |
| 7 | +//! 2. Attaching the database to get a sync handle |
| 8 | +//! 3. Creating a producer in sync context |
| 9 | +//! 4. Setting values using blocking operations |
| 10 | +//! 5. Clean shutdown with detach() |
| 11 | +
|
| 12 | +use aimdb_core::{buffer::BufferCfg, AimDbBuilder, Consumer, RuntimeContext}; |
| 13 | +use aimdb_sync::AimDbBuilderSyncExt; // Extension trait for .attach() |
| 14 | +use aimdb_tokio_adapter::{TokioAdapter, TokioRecordRegistrarExt}; // Extension for .buffer() |
| 15 | +use std::{sync::Arc, thread, time::Duration}; |
| 16 | + |
| 17 | +#[derive(Debug, Clone)] |
| 18 | +struct FeatureFlag { |
| 19 | + rollout_percent: f32, |
| 20 | +} |
| 21 | + |
| 22 | +const KEY: &str = "config.checkout_rollout"; |
| 23 | + |
| 24 | +fn main() -> Result<(), Box<dyn std::error::Error>> { |
| 25 | + println!("=== AimDB SingleLatest Sync Demo ===\n"); |
| 26 | + |
| 27 | + // Step 1: Build the database and attach it for sync API usage |
| 28 | + println!("1. Building database and attaching for sync API..."); |
| 29 | + |
| 30 | + let adapter = Arc::new(TokioAdapter); |
| 31 | + let mut builder = AimDbBuilder::new().runtime(adapter); |
| 32 | + |
| 33 | + builder.configure::<FeatureFlag>(KEY, |reg| { |
| 34 | + // Configure a SingleLatest buffer |
| 35 | + reg.buffer(BufferCfg::SingleLatest) |
| 36 | + // Register 2 taps that observe rollout percentage values |
| 37 | + .tap(rollout_observer_fast) |
| 38 | + .tap(rollout_observer_slow); |
| 39 | + }); |
| 40 | + |
| 41 | + // Build happens inside the runtime thread where Tokio context exists |
| 42 | + let handle = builder.attach()?; |
| 43 | + |
| 44 | + // Step 2: Create a synchronous producer |
| 45 | + println!("2. Creating producer and producing values..."); |
| 46 | + |
| 47 | + let producer = handle.producer::<FeatureFlag>(KEY)?; |
| 48 | + |
| 49 | + // Step 3: Produce values |
| 50 | + println!("3. Producing rollout percentage values"); |
| 51 | + |
| 52 | + for i in 0..5 { |
| 53 | + let feature_flag = FeatureFlag { |
| 54 | + rollout_percent: i as f32 * 25.0, |
| 55 | + }; |
| 56 | + println!( |
| 57 | + " Main: Setting FeatureFlag {}%", |
| 58 | + feature_flag.rollout_percent |
| 59 | + ); |
| 60 | + |
| 61 | + // Use blocking send |
| 62 | + producer.set(feature_flag)?; |
| 63 | + |
| 64 | + // Producer publishes new value every second |
| 65 | + if i < 4 { |
| 66 | + thread::sleep(Duration::from_millis(1000)); |
| 67 | + } |
| 68 | + } |
| 69 | + |
| 70 | + // Give the slow tap time to observe the final update before shutdown. Its read cycle is 2000ms, therefore wait a little longer than that. |
| 71 | + thread::sleep(Duration::from_millis(2500)); |
| 72 | + |
| 73 | + // Step 4: Clean shutdown |
| 74 | + // Detach the handle to gracefully shut down the runtime thread |
| 75 | + |
| 76 | + println!("4. Shutting down..."); |
| 77 | + |
| 78 | + handle.detach()?; |
| 79 | + |
| 80 | + println!("Done. The sync producer published rollout updates observed by fast and slow SingleLatest taps"); |
| 81 | + |
| 82 | + Ok(()) |
| 83 | +} |
| 84 | + |
| 85 | +// A fast tap observer |
| 86 | +async fn rollout_observer_fast(ctx: RuntimeContext, consumer: Consumer<FeatureFlag>) { |
| 87 | + let mut reader = consumer.subscribe(); |
| 88 | + |
| 89 | + while let Ok(feature_flag) = reader.recv().await { |
| 90 | + ctx.log().info(&format!( |
| 91 | + "[fast] rollout: {}%", |
| 92 | + feature_flag.rollout_percent |
| 93 | + )); |
| 94 | + } |
| 95 | +} |
| 96 | + |
| 97 | +// A slow tap observer |
| 98 | +async fn rollout_observer_slow(ctx: RuntimeContext, consumer: Consumer<FeatureFlag>) { |
| 99 | + let mut reader = consumer.subscribe(); |
| 100 | + let time = ctx.time(); |
| 101 | + |
| 102 | + while let Ok(feature_flag) = reader.recv().await { |
| 103 | + ctx.log().info(&format!( |
| 104 | + "[slow] rollout: {}%", |
| 105 | + feature_flag.rollout_percent |
| 106 | + )); |
| 107 | + |
| 108 | + time.sleep_millis(2000).await; |
| 109 | + } |
| 110 | +} |
0 commit comments