openframe-adapters-queue-nats

NATS JetStream queue adapter for the OpenFrame Microservice Development Suite.

Part of the openframe-adapters monorepo.

What it provides

Why JetStream, not core NATS

Core NATS pub/sub is fire-and-forget (at-most-once): there is no persistence, no publish acknowledgement, and no concept of redelivery — a message a subscriber isn't currently listening for is simply gone. That means there is nothing meaningful for BaseConsumer.ack()/nack() to do on top of it.

JetStream adds a persistent stream plus a durable, acknowledgeable consumer, giving ack()/nack() real at-least-once semantics: a handler failure (nack() → msg.nak()) triggers redelivery according to the durable consumer's ack_wait/max_deliver config — the same guarantee KafkaConsumer provides via manual offset commit. This adapter targets JetStream exclusively for that reason; there is no core-NATS mode.

Installation

# Via meta-package (recommended)
pip install "openframe-adapters[nats]"

# Or directly
pip install openframe-adapters-queue-nats

Quick start

from openframe.adapters.queue.nats import NatsSettings, NatsProducer, NatsConsumer

settings = NatsSettings(nats_servers="nats://localhost:4222")

# Produce
producer = NatsProducer(settings)
await producer.start()
await producer.publish({"event": "item.created", "id": "abc"})
await producer.publish_batch([{"event": "x"}, {"event": "y"}])
await producer.close()

# Consume
consumer = NatsConsumer(settings)

async def handle(event: dict) -> None:
    print(f"Received: {event}")

await consumer.subscribe(handle)   # runs until consumer.close() called

Configuration

Typed domain objects

from openframe.adapters.queue.nats import NatsProducer, NatsConsumer, NatsSettings
from dataclasses import dataclass, asdict

@dataclass
class OrderEvent:
    order_id: str
    event_type: str

class OrderProducer(NatsProducer[OrderEvent]):
    def _serialise(self, message: OrderEvent) -> bytes:
        import json
        return json.dumps(asdict(message)).encode("utf-8")

class OrderConsumer(NatsConsumer[OrderEvent]):
    def _deserialise(self, raw: bytes) -> OrderEvent:
        import json
        return OrderEvent(**json.loads(raw.decode("utf-8")))

Plugin lifecycle (optional)

NatsPlugin is a BasePort — wire it up with ApplicationBootstrap.compose() (requires openframe-core>=3.3), the recommended zero-subclass entry point:

from openframe.core.runtime import ApplicationBootstrap
from openframe.core.ports import Capability
from openframe.adapters.queue.nats import NatsPlugin, NatsSettings

nats_plugin = NatsPlugin(NatsSettings())

async with ApplicationBootstrap.compose(nats_plugin) as app:
    plugin = app.get(Capability.QUEUE)
    producer = plugin.get_producer()
    await producer.publish({"event": "item.created"})

    consumer = plugin.make_consumer()
    await consumer.subscribe(handler)

NatsPlugin doubles as both the producer and consumer port for the QUEUE capability (get_producer() / make_consumer()), so a single instance is usually enough. If a service registers a separate producer-only and consumer-only NatsPlugin (e.g. different subjects/settings for each), pass both to compose(): ApplicationBootstrap.compose(producer_plugin, consumer_plugin). Reach for a subclassed ApplicationBootstrap with configure() only when you need per-port config=/init_timeout= or conditional registration order, and use app.registry as an escape hatch for anything neither tier covers.

Consumer acknowledgement semantics

Note: unlike KafkaConsumer's ack()/nack() (which commit/skip a global offset and ignore their message argument), NatsConsumer.ack()/nack() take the raw NATS message (not the deserialised payload) — JetStream acks are per-message, so the exact delivered message object is required to target the right ack. See consumer.py's module docstring for the full rationale.

Resilience (optional — requires openframe-core>=3.4)

openframe.core.resilience ships a CircuitBreakerProxy that composes around any BasePort, including this adapter's producer:

from openframe.core.resilience import CircuitBreakerProxy
from openframe.core.tracing import TracingProxy

producer = CircuitBreakerProxy(
    TracingProxy(plugin.get_producer(), prefix="nats.producer"),
    failure_threshold=5,
    reset_timeout=30.0,
)

Compose the circuit breaker around the traced producer, not the reverse — a short-circuited call should never produce a misleading adapter span for a call that never reached NATS. No adapter code changes are needed to support this; it wraps from the outside exactly like TracingProxy.

License

MIT — © Furious Meteors Engineering