ursula-stream 0.3.0

Durable Streams state machine for Ursula: bucket and stream commands, events, and offset bookkeeping.
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

//! TTL expiry index: a min-heap of `(expires_at_ms, stream)` entries with deterministic ordering.

use super::BinaryHeap;
use super::BucketStreamId;
use super::Key;
use super::Ordering;
use super::Reverse;
use super::StreamKey;
use super::compare_stream_ids;

#[derive(Debug, Clone, Default)]
pub(super) struct TtlIndex {
    pub(super) entries: BinaryHeap<Reverse<TtlEntry>>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct TtlEntry {
    pub(super) expires_at_ms: u64,
    pub(super) stream_id: BucketStreamId,
    pub(super) key: StreamKey,
}

impl Ord for TtlEntry {
    fn cmp(&self, other: &Self) -> Ordering {
        self.expires_at_ms
            .cmp(&other.expires_at_ms)
            .then_with(|| compare_stream_ids(&self.stream_id, &other.stream_id))
            .then_with(|| self.key.data().as_ffi().cmp(&other.key.data().as_ffi()))
    }
}

impl PartialOrd for TtlEntry {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}