Enum AnyReceiver 

pub enum AnyReceiver {
    Grpc(GrpcTransport),
    Memory(MemoryTransport),
    Pipe(PipeTransport),
    File(FileTransport),
    Http(HttpTransport),
    Redis(RedisTransport),
}

Available on crate feature transport only.

Type-erased transport receiver.

Wraps any concrete transport receiver behind an enum for runtime dispatch. Created by the transport factory from config, mirroring AnySender.

Uses enum dispatch (not trait objects) because TransportReceiver has impl Future return types and an associated Token type that prevent dyn dispatch.

The AnyReceiver::recv method wraps each backend token in the corresponding AnyToken variant. AnyReceiver::commit extracts the inner tokens for the active backend and forwards to that backend’s own commit – tokens from a different variant are silently skipped (they cannot legitimately appear but the code stays defensive).

Variants

Grpc(GrpcTransport)

Available on crate feature transport-grpc only.

Memory(MemoryTransport)

Available on crate feature transport-memory only.

Pipe(PipeTransport)

Available on crate feature transport-pipe only.

File(FileTransport)

Available on crate feature transport-file only.

Http(HttpTransport)

Available on crate feature transport-http only.

Redis(RedisTransport)

Available on crate feature transport-redis only.

Implementations

impl AnyReceiver

pub async fn from_config(key: &str) -> TransportResult<Self>

Create a receiver from the config cascade.

Reads the transport config from the given key in the config cascade and creates the appropriate receiver.

Example config

transport:
  input:
    type: kafka
    kafka:
      brokers: ["kafka:9092"]
      group_id: "my-consumer"

let receiver = AnyReceiver::from_config("transport.input").await?;

pub async fn from_transport_config( config: &TransportConfig, ) -> TransportResult<Self>

Create a receiver from an explicit TransportConfig.

pub async fn from_config_with_governor( key: &str, governor: &SelfRegulationGovernor, ) -> TransportResult<Self>

Available on crate feature governor only.

Create a governed receiver from the config cascade (governor feature).

Identical to from_config but threads the supplied SelfRegulationGovernor’s pressure into the inbound brake of every backend that can honour it – the Kafka pause-partitions gate and the HTTP/gRPC 503/unavailable shed – so a factory-built receiver actually engages the default-on governor instead of silently dropping the inbound brake.

Construction order: the governor (and its pressure) is built by the runtime BEFORE this call, so the pressure latch already exists and is merely cloned (cheap Arc bump) into each transport here.

Errors

Same as from_config.

pub async fn from_transport_config_with_governor( config: &TransportConfig, governor: &SelfRegulationGovernor, ) -> TransportResult<Self>

Available on crate feature governor only.

Create a governed receiver from an explicit TransportConfig (governor feature).

The governor-aware sibling of from_transport_config. Backends that own an inbound brake are wired to the governor’s shared pressure:

• Kafka: the consumer’s assigned partitions are paused/resumed via SelfRegulationGovernor::attach_kafka_gate (the full gate_actuator -> InboundGate -> with_inbound_gate dance).
• HTTP / gRPC: the embedded receive server is built with with_pressure(Some(governor.pressure())), so it sheds with 503 / Status::unavailable while the pressure latch holds.

Backends with no inbound brake (memory, pipe, file, redis) construct exactly as in from_transport_config – the byte-budget lever already reaches them through the governed driver.

Errors

Same as from_transport_config.

Trait Implementations

impl TransportBase for AnyReceiver

async fn close(&self) -> TransportResult<()>

Shutdown the transport gracefully.

fn is_healthy(&self) -> bool

Check if the transport is healthy and connected.

fn name(&self) -> &'static str

Get transport name for logging/metrics.

impl TransportReceiver for AnyReceiver

async fn recv_limited( &self, limits: RecvLimits, ) -> TransportResult<WorkBatch<AnyToken>>

Forward the byte-aware recv to each inner transport so the governed driver’s byte budget reaches the transport that can honour it (Kafka’s recv-arena). Transports without a byte-aware override fall back to the trait default (record-bounded recv), which is correct for the one-record-at-a-time channel/stream transports.

type Token = AnyToken

The token type for this transport.

async fn recv(&self, max: usize) -> TransportResult<WorkBatch<AnyToken>>

Receive up to max records as one WorkBatch.

async fn commit(&self, tokens: &[AnyToken]) -> TransportResult<()>

Commit/acknowledge processed messages.