jetstream_extra/
lib.rs

1// Copyright 2025 Synadia Communications Inc.
2// Licensed under the Apache License, Version 2.0 (the "License");
3// you may not use this file except in compliance with the License.
4// You may obtain a copy of the License at
5//
6// http://www.apache.org/licenses/LICENSE-2.0
7//
8// Unless required by applicable law or agreed to in writing, software
9// distributed under the License is distributed on an "AS IS" BASIS,
10// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11// See the License for the specific language governing permissions and
12// limitations under the License.
13
14//! JetStream extensions for async-nats.
15//!
16//! This crate provides additional functionality for NATS JetStream that extends
17//! the capabilities of the async-nats client.
18//!
19//! # Features
20//!
21//! ## Batch Publishing
22//!
23//! The [batch_publish] module provides atomic batch publishing capabilities:
24//!
25//! ```no_run
26//! # use jetstream_extra::batch_publish::BatchPublishExt;
27//! # async fn example(client: impl BatchPublishExt) -> Result<(), Box<dyn std::error::Error>> {
28//! // Publish multiple messages as an atomic batch
29//! let mut batch = client.batch_publish().build();
30//! batch.add("events.1", "data1".into()).await?;
31//! batch.add("events.2", "data2".into()).await?;
32//! let ack = batch.commit("events.3", "final".into()).await?;
33//! # Ok(())
34//! # }
35//! ```
36//!
37//! ## Fast Ingest Batch Publishing
38//!
39//! The [batch_publish_fast] module provides high-throughput, non-atomic batch publishing
40//! using JetStream's fast-ingest feature (nats-server 2.14+):
41//!
42//! ```no_run
43//! # use jetstream_extra::batch_publish_fast::FastPublishExt;
44//! # async fn example(client: impl FastPublishExt) -> Result<(), Box<dyn std::error::Error>> {
45//! let mut batch = client.fast_publish().build()?;
46//! for i in 0..1000 {
47//!     batch.add("events.data", format!("msg {i}").into()).await?;
48//! }
49//! let ack = batch.commit("events.done", "final".into()).await?;
50//! # Ok(())
51//! # }
52//! ```
53//!
54//! ## Batch Fetching
55//!
56//! The [batch_fetch] module provides efficient batch fetching of messages from streams
57//! using the DIRECT.GET API:
58//!
59//! ```no_run
60//! # use jetstream_extra::batch_fetch::BatchFetchExt;
61//! # use futures::StreamExt;
62//! # async fn example(context: async_nats::jetstream::Context) -> Result<(), Box<dyn std::error::Error>> {
63//! // Fetch 100 messages from a stream
64//! let mut messages = context
65//!     .get_batch("my_stream", 100)
66//!     .send()
67//!     .await?;
68//!
69//! while let Some(msg) = messages.next().await {
70//!     let msg = msg?;
71//!     println!("Message: {:?}", msg);
72//! }
73//! # Ok(())
74//! # }
75//! ```
76//!
77//! See the [batch_publish] and [batch_fetch] modules for detailed documentation.
78
79pub mod batch_fetch;
80pub mod batch_publish;
81pub mod batch_publish_fast;
82
83pub use async_nats::Subject;
84/// Re-exported type returned by Direct Get operation.
85pub use async_nats::jetstream::message::StreamMessage;
jetstream_extra/lib.rs

jetstream_extra/
lib.rs
