Levandor

NATS JetStream MessageSource Implementation (Task 5)

2 min read

Real NATS JetStream pull consumer implementation added to consumer.rs, completing the abstraction seam established in Task 4.

Location

/Users/levander/coding/pmv2/telegram_connector-outbound/crates/telegram_connector/src/consumer.rs

Trait Seam (Task 4, lines 1–19)

Three types form the abstraction layer that lets tests inject a fake transport:

pub enum Disposition { Ack, Nak(Option<Duration>), Term }
 
#[async_trait] pub trait IncomingMessage: Send + Sync {
    fn payload(&self) -> &[u8];
    async fn settle(&self, d: Disposition) -> anyhow::Result<()>;
}
 
#[async_trait] pub trait MessageSource: Send {
    async fn next(&mut self) -> Option<anyhow::Result<Box<dyn IncomingMessage>>>;
}

Real Implementation (Task 5, lines 21–95)

connect(url, nkey_seed)

Creates an async_nats::Client. When nkey_seed is non-empty, wraps with ConnectOptions::new().nkey(seed). Requires the nkeys feature (enabled on the workspace dep).

build_source(client, stream, durable, subject, ack_wait_secs, max_deliver)

  1. Calls jetstream::new(client) to get the JetStream context.
  2. js.get_stream(stream) — gets the named stream (local variable renamed nats_stream to avoid shadowing the stream: &str param).
  3. nats_stream.get_or_create_consumer(durable, pull::Config { ... }) — creates a durable pull consumer with explicit ack policy, configurable ack_wait, max_deliver, and filter_subject.
  4. consumer.messages() — returns the pull message stream.
  5. Wraps in NatsSource.

NatsSource / NatsMessage

TypeWrapsRole
NatsSourcejetstream::consumer::pull::StreamImplements MessageSource via StreamExt::next()
NatsMessagejetstream::MessageImplements IncomingMessage; settle() maps Disposition → AckKind

Disposition → AckKind mapping

Dispositionjetstream::AckKind
AckAck
Nak(Option<Duration>)Nak(delay)
TermTerm

async-nats 0.49 API Notes

For Agents

All type paths and variant names from async-nats 0.38 docs are compatible with 0.49. No API migration was required. Only code change driven by the upgrade: rename local stream var to nats_stream to avoid shadowing the function parameter.

  • async_nats::jetstream::consumer::pull::Stream — the pull message stream type
  • async_nats::jetstream::AckKind — ack variant enum
  • futures::StreamExt — provides .next() on the pull stream; imported via futures workspace dep

Dependencies Added

In crates/telegram_connector/Cargo.toml:

async-nats = { workspace = true }   # workspace pins 0.49, features = ["nkeys"]
futures    = { workspace = true }    # StreamExt on pull::Stream

Commit & Build

  • Commit: 95b8288 on feat/outbound-relay
  • Build: clean (no warnings)
  • Tests: 19 passing

Graph View

Backlinks