Struct Aggregator 

pub struct Aggregator { /* private fields */ }

Async aggregator handle: snapshots at a fixed interval into a shared cell that the REST server reads from.

Examples

use std::sync::Arc;
use std::time::Duration;
use dynomite::stats::{Aggregator, PoolStats, ServerStats, ServiceInfo, Snapshot, Stats};
use parking_lot::Mutex;
use tokio_util::sync::CancellationToken;

let stats = Arc::new(Stats::new(
    ServiceInfo::default(),
    PoolStats::new("dyn_o_mite"),
    ServerStats::new("redis"),
));
let sink = Arc::new(Mutex::new(Snapshot::default()));
let token = CancellationToken::new();
let agg = Aggregator::new(stats, sink, Duration::from_secs(1), Duration::from_secs(300));
let _ = tokio::spawn({ let token = token.clone(); async move { agg.run(token).await } });
token.cancel();

Implementations

impl Aggregator

pub fn new( stats: Arc<Stats>, sink: Arc<Mutex<Snapshot>>, interval: Duration, histogram_reset: Duration, ) -> Self

Create a new aggregator. The aggregation loop reads from stats and publishes to sink once every interval. Histograms are reset every histogram_reset elapsed time, the same five-minute cadence the C reference uses by default.

Examples

use std::sync::Arc;
use std::time::Duration;
use dynomite::stats::{Aggregator, PoolStats, ServerStats, ServiceInfo, Snapshot, Stats};
use parking_lot::Mutex;

let stats = Arc::new(Stats::new(
    ServiceInfo::default(),
    PoolStats::new("dyn_o_mite"),
    ServerStats::new("redis"),
));
let sink = Arc::new(Mutex::new(Snapshot::default()));
let _agg = Aggregator::new(stats, sink, Duration::from_secs(1), Duration::from_secs(300));

pub async fn run(self, cancel: CancellationToken)

Run the aggregation loop until cancel is triggered. The future returns () after observing cancellation; callers that want a clean shutdown should clone the token and call CancellationToken::cancel on it.

Examples

use std::sync::Arc;
use std::time::Duration;
use dynomite::stats::{Aggregator, PoolStats, ServerStats, ServiceInfo, Snapshot, Stats};
use parking_lot::Mutex;
use tokio_util::sync::CancellationToken;

let stats = Arc::new(Stats::new(
    ServiceInfo::default(),
    PoolStats::new("dyn_o_mite"),
    ServerStats::new("redis"),
));
let sink = Arc::new(Mutex::new(Snapshot::default()));
let token = CancellationToken::new();
let agg = Aggregator::new(stats, sink, Duration::from_secs(1), Duration::from_secs(300));
let cancel = token.clone();
let handle = tokio::spawn(async move { agg.run(cancel).await });
token.cancel();
let _ = handle.await;