Struct BasicAggregator 

pub struct BasicAggregator<G, I> { /* private fields */ }

Decayed aggregate sum, count and average over a stream of items.

Examples

Basic Aggregation

use std::time::{Duration, Instant};
use fermentation::{ForwardDecay, g};
use fermentation::aggregate::{BasicAggregator, Aggregator};

let decay = ForwardDecay::new(Instant::now(), g::Polynomial::new(2));
let landmark = decay.landmark();
let now = landmark + Duration::from_secs(10);
let stream = vec![
    (landmark + Duration::from_secs(5), 4.0),
    (landmark + Duration::from_secs(7), 8.0),
    (landmark + Duration::from_secs(3), 3.0),
    (landmark + Duration::from_secs(8), 6.0),
    (landmark + Duration::from_secs(4), 4.0),
];

let mut aggregator = BasicAggregator::new(decay);

for item in stream {
    aggregator.update(item);
}

let epsilon = 0.01;

assert_eq!(aggregator.sum(now), 9.67);
assert_eq!(aggregator.count(now), 1.63);
assert!(aggregator.average() >= (5.93 - epsilon) && aggregator.average() <= (5.93 + epsilon));

Update Landmark

use std::time::{Duration, Instant};
use fermentation::{ForwardDecay, g};
use fermentation::aggregate::{BasicAggregator, Aggregator};

let decay = ForwardDecay::new(Instant::now(), g::Exponential::new(0.2));
let landmark = decay.landmark();
let new_landmark = landmark + Duration::from_secs(1);
let now = landmark + Duration::from_secs(10);
let stream = vec![
    (landmark + Duration::from_secs(5), 4.0),
    (landmark + Duration::from_secs(7), 8.0),
    (landmark + Duration::from_secs(3), 3.0),
    (landmark + Duration::from_secs(8), 6.0),
    (landmark + Duration::from_secs(4), 4.0),
];

let mut aggregator = BasicAggregator::new(decay);
let mut clone = aggregator.clone();

clone.reset(new_landmark);

for item in stream {
    aggregator.update(item);
    clone.update(item);
}

aggregator.update_landmark(new_landmark);

let epsilon = 0.0001;

assert!((aggregator.sum(now) - clone.sum(now)).abs() < epsilon);
assert!((aggregator.count(now) - clone.count(now)).abs() < epsilon);
assert!((aggregator.average() - clone.average()).abs() < epsilon);

Implementations

impl<I> BasicAggregator<Exponential, I>

where I: Item,

pub fn update_landmark(&mut self, landmark: Instant)

impl<G, I> BasicAggregator<G, I>

where G: Function, I: Item,

pub fn new(decay: ForwardDecay<G>) -> Self

pub fn sum(&self, timestamp: Instant) -> f64

pub fn static_sum(&self) -> f64

pub fn count(&self, timestamp: Instant) -> f64

pub fn static_count(&self) -> f64

pub fn average(&self) -> f64

pub fn decay(&mut self) -> &ForwardDecay<G>

Trait Implementations

impl<G, I> Aggregator for BasicAggregator<G, I>

where G: Function, I: Item,

type Item = I

fn update(&mut self, item: I)

Update the aggregation with the given item.

fn reset(&mut self, landmark: Instant)

Reset the aggregation to the initial state. This is equivalent to creating a new aggregator with the same decay model and the given landmark.

impl<G: Clone, I: Clone> Clone for BasicAggregator<G, I>

fn clone(&self) -> BasicAggregator<G, I>

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<G: Copy, I: Copy> Copy for BasicAggregator<G, I>