obs-otel 0.2.1

OpenTelemetry Protocol (OTLP) sinks for the obs SDK.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

//! Per-sink batching helper. Spec 20 ยง 4.

use std::time::{Duration, Instant};

use parking_lot::Mutex;

/// Bounded batch with size + age trigger. Calls `flush_fn` when one of
/// the triggers fires.
#[derive(Debug)]
pub struct Batch<T> {
    inner: Mutex<BatchInner<T>>,
    max_records: usize,
    max_age: Duration,
}

#[derive(Debug)]
struct BatchInner<T> {
    items: Vec<T>,
    opened_at: Instant,
}

impl<T> Batch<T> {
    /// New batch.
    #[must_use]
    pub fn new(max_records: usize, max_age: Duration) -> Self {
        Self {
            inner: Mutex::new(BatchInner {
                items: Vec::with_capacity(max_records),
                opened_at: Instant::now(),
            }),
            max_records,
            max_age,
        }
    }

    /// Push one item. Returns `Some(items)` when the batch is ready to
    /// flush; otherwise `None`.
    pub fn push(&self, item: T) -> Option<Vec<T>> {
        let mut inner = self.inner.lock();
        inner.items.push(item);
        if inner.items.len() >= self.max_records || inner.opened_at.elapsed() >= self.max_age {
            let drained = std::mem::replace(&mut inner.items, Vec::with_capacity(self.max_records));
            inner.opened_at = Instant::now();
            return Some(drained);
        }
        None
    }

    /// Drain whatever is currently buffered.
    pub fn drain(&self) -> Vec<T> {
        let mut inner = self.inner.lock();
        let drained = std::mem::replace(&mut inner.items, Vec::with_capacity(self.max_records));
        inner.opened_at = Instant::now();
        drained
    }
}