obzenflow_core 0.2.1

Core domain layer for ObzenFlow - pure abstractions with minimal dependencies
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
57
58
59
60
61
62
63
64
65
66

// SPDX-License-Identifier: MIT OR Apache-2.0
// SPDX-FileCopyrightText: 2025-2026 ObzenFlow Contributors
// https://obzenflow.dev

use super::{ChainEvent, ChainEventContent};
use crate::event::journal_event::{JournalEvent, Sealed};
use crate::event::payloads::flow_control_payload::FlowControlPayload;
use crate::event::payloads::observability_payload::{MetricsLifecycle, ObservabilityPayload};
use crate::event::types::{AdmissionSeq, EventId, WriterId};

impl Sealed for ChainEvent {}

impl JournalEvent for ChainEvent {
    fn id(&self) -> &EventId {
        &self.id
    }

    fn writer_id(&self) -> &WriterId {
        &self.writer_id
    }

    fn admission_seq(&self) -> Option<AdmissionSeq> {
        self.admission_seq
    }

    fn set_admission_seq(&mut self, seq: AdmissionSeq) {
        self.admission_seq = Some(seq);
    }

    /// Zero-alloc category string for metrics & fast logs.
    /// Falls back to generic labels when the name is dynamic (e.g. custom metrics).
    fn event_type_name(&self) -> &'static str {
        match &self.content {
            ChainEventContent::Data { .. } => "data",
            ChainEventContent::FlowControl(sig) => match sig {
                FlowControlPayload::Eof { .. } => "control.eof",
                FlowControlPayload::Watermark { .. } => "control.watermark",
                FlowControlPayload::CatchUpComplete { .. } => "control.catch_up_complete",
                FlowControlPayload::Checkpoint { .. } => "control.checkpoint",
                FlowControlPayload::Drain => "control.drain",
                FlowControlPayload::PipelineAbort { .. } => "control.pipeline_abort",
                FlowControlPayload::SourceContract { .. } => "control.source_contract",
                FlowControlPayload::ConsumptionProgress { .. } => "control.consumption_progress",
                FlowControlPayload::ConsumptionGap { .. } => "control.consumption_gap",
                FlowControlPayload::ConsumptionFinal { .. } => "control.consumption_final",
                FlowControlPayload::ReaderStalled { .. } => "control.reader_stalled",
                FlowControlPayload::AtLeastOnceViolation { .. } => {
                    "control.at_least_once_violation"
                }
            },
            ChainEventContent::Delivery(_) => "sink.delivery",
            ChainEventContent::Observability(obs) => match obs {
                ObservabilityPayload::Stage(_) => "lifecycle.stage",
                ObservabilityPayload::Metrics(m) => match m {
                    MetricsLifecycle::Custom { .. } => "lifecycle.metrics.custom",
                    MetricsLifecycle::HttpPullSnapshot { .. } => {
                        "lifecycle.metrics.http_pull_snapshot"
                    }
                    _ => "lifecycle.metrics",
                },
                ObservabilityPayload::Middleware(_) => "lifecycle.middleware",
                ObservabilityPayload::Backpressure(_) => "lifecycle.backpressure",
            },
        }
    }
}