Add NATS Transport Unit Tests, related documentation and prefix support for stream names.

[maxyloon]

closes #2103

Implement support for NATS using its built-in persistence layer JetStream.

What's in this PR

Original implementation by @joeriddles. (PR: #2299) This PR builds on that work to bring it up to date with main and add the items needed for merge:

• Unit tests (t/unit/transport/test_nats.py) — 72 tests covering Message, QoS, Channel, and Transport with no live NATS server required
• Configurable stream/consumer name prefixes via stream_name_prefix and consumer_name_prefix transport options (default STREAM_ / CONSUMER_)
• Setup extras — kombu[nats] extra pointing to requirements/extras/nats.txt
• Docs — NATS row in the README transport comparison table, NATS URL examples and transport options documented in the userguide connections page, kombu.transport.nats added to the API reference index
• Dependency updates — rebased onto current main; nats-py[nkeys]>=2.9.0,<3.0.0 confirmed compatible with nats-py 2.14.0
• Lint — all flake8 warnings resolved (F841 unused variables removed)

Testing

Install the extra and run unit tests (no server needed):

pip install "kombu[nats]"
python -m pytest t/unit/transport/test_nats.py -v

Integration tests (requires a running NATS server with JetStream):

nats-server --jetstream
python -m pytest t/integration/test_nats.py -v -E nats

Example

You can test using either:

• Installing the NATS server locally and running it with JetStream enabled: nats-server --js
• Using the demo NATS server at demo.nats.io

The transport can be tested using examples/nats_receive.py and examples/nats_send.py:

1. Start the local NATS server (optional)
2. Run python -m examples.nats_receive (append --demo if using the demo server)
3. In another window, run python -m examples.nats_send (append --demo if using the demo server)

In the receive window, you should see something like:

Received message: 'hello world'
  properties:
{   'body_encoding': 'base64',
    'delivery_info': {'exchange': 'exchange', 'routing_key': 'messages'},
    'delivery_mode': 2,
    'delivery_tag': 'fa01f1a9-f666-46a0-af71-35e5164e1ad9',
    'priority': 0}
  delivery_info:
{'exchange': 'exchange', 'routing_key': 'messages'}

[auvipy]

thanks max

[codecov]

Codecov Report

❌ Patch coverage is 92.12121% with 26 lines in your changes missing coverage. Please review.
✅ Project coverage is 83.19%. Comparing base (9ee8595) to head (f41b4c0).
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
kombu/transport/nats.py	92.12%	22 Missing and 4 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #2541      +/-   ##
==========================================
+ Coverage   82.90%   83.19%   +0.28%     
==========================================
  Files          79       80       +1     
  Lines       10402    10732     +330     
  Branches     1197     1231      +34     
==========================================
+ Hits         8624     8928     +304     
- Misses       1577     1599      +22     
- Partials      201      205       +4     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

[maxyloon]

No problem. Let me know if there is anything else I can do to get it merged.

[auvipy]

@maxyloon please cross check the suggestions locally and fix which are relevant

[maxyloon]

So I realized that the implementation uses embedds metadata in the body (like most virtual transports), but nats supports binary payloads and I wanted to make sure protocol binding for use with things like cloudevents would work with some configuration. Therefore I added a clean-body mode. So it uses headers instead for metadata and should avoid nested serialization problems

• Add optional clean-body mode for the NATS transport:
  • New transport options:
    • nats_clean_body (bool, default False) — if True, the transport publishes the application payload directly as the NATS message body (bytes) and places Kombu metadata into configurable NATS headers.
    • nats_metadata_header_prefix (str, default "Kombu-") — prefix for Kombu metadata headers (e.g. Kombu-Content-Type).
    • nats_metadata_header_names (dict, optional) — override per-field header suffixes (defaults: Content-Type, Content-Encoding, Headers, Properties, Delivery-Info).
  • Behavior:
    • Backwards-compatible by default (nats_clean_body=False).
    • In clean-body mode the transport does NOT JSON-wrap or otherwise alter the payload bytes — serialization remains the caller/serializer's responsibility.
    • Kombu metadata is carried as NATS headers (JSON-encoded when structured). Nats-TTL header handling is unchanged.
    • Consumer auto-detects mode by presence of the configured metadata headers; legacy envelope parsing is preserved.
  • Safety notes:
    • The nats_metadata_header_prefix must not start with Nats- (reserved); code emits a warning for such prefixes.
  • Tests & verification:
    • New unit tests covering helper functions, round-trip behavior, custom prefix/names, TTL interactions (total n tests added).
    • Manual integration confirmed: raw bytes published and received identical when nats_clean_body=True.

[maxyloon]

LGTM

[auvipy]

isn't it a complete new backend? why should we think about backward compatibility here then?

[Copilot]

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 3 comments.

[auvipy]

see

[maxyloon]

Good catch. The transport should not install a global asyncio event loop.

I see Kombu’s virtual transports are synchronous, so the NATS transport should behave as a sync adapter over nats-py’s async API. I’ll change this to keep a private event loop owned by each Channel, avoid asyncio.set_event_loop().

I think this keeps asyncio isolated from the rest of Celery/Kombu and avoids replacing an application’s existing thread-default loop.

[auvipy]

will asyncio work well with current celery and kombu code base properly?

[maxyloon]

Have a local branch that works now, will push it shortly. Taking some time verify some setups and sceanarios to make it fully celery compatible. Here is the docs from my working branch. Once I find it stable enough i'll open a PR for celery too.

[auvipy]

no problem take your time

[maxyloon]

isn't it a complete new backend? why should we think about backward compatibility here then?

Not sure exactly what you mean here, but I think the “backward compatibility” point here is less about compatibility with an existing released NATS backend, since this is new, and more about compatibility between the two wire formats introduced by this PR. And no I don't think its a complete new backend, just a question of wireformat which should be transient to the user.

My thinking is that; The main thing we should preserve is consumer-side interoperability: a consumer should be able to read both formats regardless of whether the publisher used envelope-in-body mode or clean-body mode (might refactor it to raw for better abstraction). That matters if a queue already contains messages when nats_clean_body is changed, or if different producers are temporarily configured differently.

So I agree the publish path can be strict and explicit, but the receive path should remain tolerant: detect Kombu metadata headers → clean-body mode; otherwise treat the payload as the full Kombu JSON envelope.

External semantics outside the module will remain the same, but raw/clean mode with skip serializing and deserializing the wireformat,

[auvipy]

isn't it a complete new backend? why should we think about backward compatibility here then?

Not sure exactly what you mean here, but I think the “backward compatibility” point here is less about compatibility with an existing released NATS backend, since this is new, and more about compatibility between the two wire formats introduced by this PR. And no I don't think its a complete new backend, just a question of wireformat which should be transient to the user.

My thinking is that; The main thing we should preserve is consumer-side interoperability: a consumer should be able to read both formats regardless of whether the publisher used envelope-in-body mode or clean-body mode (might refactor it to raw for better abstraction). That matters if a queue already contains messages when nats_clean_body is changed, or if different producers are temporarily configured differently.

So I agree the publish path can be strict and explicit, but the receive path should remain tolerant: detect Kombu metadata headers → clean-body mode; otherwise treat the payload as the full Kombu JSON envelope.

External semantics outside the module will remain the same, but raw/clean mode with skip serializing and deserializing the wireformat,

that is more clear now and make sense.

[va1ker]

Hi! Any updates on it?

[auvipy]

we need manual tester for this

[emattiza]

Just got done with manual testing on my side, i've got some feedback

test_get_returns_message pulls a message from the previous run because test_queue is shared. I fixed the flaky integration test by putting unique queue names per test + teardown cleanup. https://github.com/emattiza/kombu/commits/fix/nats-tests

Additionally, theres some overhead reduction and nats roundtrip for every _get because of the call to pull_subscribe. i wrote a helper registry dict for subscriptions so you can re-use the existing consumers. https://github.com/emattiza/kombu/commits/perf/nats-caching

i've got some other fixes related to shared thread per connection/transport instead of thread per channel and error messaging for retry-ability when possible. lmk if interested as well in those.

[auvipy]

yes please

[emattiza]

yes please

https://github.com/emattiza/kombu/tree/nats-manual-testing has got all the changes bundled for review.

 nats-manual-testing
 ├── 744dcc79  NATSError + cause chaining
 ├── e5e9683b  NotBoundError on closed channel
 ├── ad56fbed  integration test isolation
 ├── 266d717d  shared loop + client refactor
 └── e601853b  pull-subscription caching