Struct DfeSource 

pub struct DfeSource { /* private fields */ }

DFE source-aware topic naming for transform services.

All DFE data flows follow the same topology:

receiver -> {source}_land -> transform -> {source}_load -> loader -> ClickHouse

DfeSource is for transform services (middleware) that sit between _land and _load. It derives input/output topic names and source-scoped consumer group IDs from a source name.

Terminal consumers (loader, archiver) do not use DfeSource — they consume from whatever topics are configured or auto-discovered, and their consumer group is simply dfe-{service} without a source component.

Examples

use hyperi_rustlib::kafka_config::{DfeSource, ServiceRole};

let source = DfeSource::new("syslog");
assert_eq!(source.input_topic(), "syslog_land");
assert_eq!(source.output_topic(), "syslog_load");

// Transform: CG includes source name
assert_eq!(
    source.consumer_group("transform-vector", ServiceRole::Transform, None, None).unwrap(),
    "dfe-transform-vector-syslog"
);

// Terminal: CG is just the service name
assert_eq!(
    source.consumer_group("loader", ServiceRole::Universal, None, None).unwrap(),
    "dfe-loader"
);

// Override always wins
assert_eq!(
    source.consumer_group("transform-vector", ServiceRole::Transform, None, Some("custom")).unwrap(),
    "custom"
);

// Transform without source is an error
let empty = DfeSource::new("");
assert!(empty.consumer_group("transform-vector", ServiceRole::Transform, None, None).is_err());

Implementations

impl DfeSource

pub fn new(name: impl Into<String>) -> Self

Create a new source with default suffixes (_land, _load).

pub fn with_suffixes( name: impl Into<String>, land_suffix: impl Into<String>, load_suffix: impl Into<String>, ) -> Self

Create a source with custom suffixes.

pub fn name(&self) -> &str

Source name (e.g. "syslog", "netflow").

pub fn input_topic(&self) -> String

Landing zone topic: {source}_land.

pub fn output_topic(&self) -> String

Load-ready topic: {source}_load.

pub fn consumer_group( &self, service: &str, role: ServiceRole, pipeline: Option<&str>, cg_override: Option<&str>, ) -> Result<String, KafkaConfigError>

Consumer group ID following DFE naming conventions.

The cg_override takes precedence when set — use it when the operator explicitly configures a consumer group in YAML/env.

When cg_override is None, the default pattern depends on the service role:

Role	Pattern	Example
Transform	dfe-{service}-{source}	dfe-transform-vector-syslog
Universal (loader, archiver)	dfe-{service}	dfe-loader

For transforms, pipeline overrides the source component in the CG (e.g. syslog-enriched instead of syslog). Either the DfeSource name or pipeline must be non-empty — a bare service name is never valid for transforms because multiple pipelines would compete.

Errors

Returns Err if the role is Transform and neither the source name nor pipeline provides a non-empty suffix.

pub fn source_from_topic(topic: &str) -> Option<&str>

Derive the source name from a topic by stripping known suffixes.

Returns None if the topic doesn’t end with a known suffix.

use hyperi_rustlib::kafka_config::DfeSource;

assert_eq!(DfeSource::source_from_topic("syslog_land"), Some("syslog"));
assert_eq!(DfeSource::source_from_topic("netflow_load"), Some("netflow"));
assert_eq!(DfeSource::source_from_topic("unknown"), None);

Trait Implementations

impl Clone for DfeSource

fn clone(&self) -> DfeSource

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for DfeSource

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl PartialEq for DfeSource

fn eq(&self, other: &DfeSource) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

impl Eq for DfeSource

impl StructuralPartialEq for DfeSource