async-nats

https://docs.rs/async-nats/latest/
Admin
A Rust asynchronous client for the NATS.io messaging system, providing publish-subscribe,...
Tokens:1,227,453
Snippets:11,063
Trust Score:9.7
Update:1 week ago
Show doc for...
Tokens:
Raw
### service_builder example

Source: https://docs.rs/async-nats/latest/async_nats/service/trait.ServiceExt.html?search=

Connects to NATS, uses the service builder to set a description and stats handler, starts a service named 'products' with version '1.0.0', then creates an endpoint 'get' and responds to a request with 'hello'.

```rust
use async_nats::service::ServiceExt;
use futures_util::StreamExt;
let client = async_nats::connect("demo.nats.io").await?;
let mut service = client
    .service_builder()
    .description("some service")
    .stats_handler(|endpoint, stats| serde_json::json!({ "endpoint": endpoint }))
    .start("products", "1.0.0")
    .await?;

let mut endpoint = service.endpoint("get").await?;

if let Some(request) = endpoint.next().await {
    request.respond(Ok("hello".into())).await?;
}
```

--------------------------------

### Start a service and respond to a request

Source: https://docs.rs/async-nats/latest/async_nats/service/struct.Service.html?search=

Example from the struct documentation showing how to connect, start a service, create an endpoint, and respond to a request.

```rust
use async_nats::service::ServiceExt;
use futures_util::StreamExt;
let client = async_nats::connect("demo.nats.io").await?;
let mut service = client.service_builder().start("generator", "1.0.0").await?;
let mut endpoint = service.endpoint("get").await?;

if let Some(request) = endpoint.next().await {
    request.respond(Ok("hello".into())).await?;
}
```

--------------------------------

### HeaderMap get example

Source: https://docs.rs/async-nats/latest/src/async_nats/header.rs.html

Example for the get method. Returns the first value for a given key, or None if the key is not found.

```rust
# use async_nats::HeaderMap;

let mut headers = HeaderMap::new();
headers.append("Key", "Value");
let values = headers.get("Key").unwrap();
```

--------------------------------

### Example: Upload a file to an ObjectStore using put

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/object_store/mod.rs.html?search=u32+-%3E+bool

Shows how to connect to NATS, get an object store bucket, and upload a file using the put method. The example is marked as no_run and uses tokio::main.

```rust
```no_run
# #[tokio::main]
# async fn main() -> Result<(), async_nats::Error> {
let client = async_nats::connect("demo.nats.io").await?;
let jetstream = async_nats::jetstream::new(client);

let bucket = jetstream.get_object_store("store").await?;
let mut file = tokio::fs::File::open("foo.txt").await?;
bucket.put("file", &mut file).await.unwrap();
# Ok(())
# }
```
```

--------------------------------

### Start a service and respond to a request using async_nats::service::ServiceExt

Source: https://docs.rs/async-nats/latest/src/async_nats/service/mod.rs.html

Shows how to connect to a NATS server, start a service named 'generator' with version '1.0.0', get an endpoint, and respond to an incoming request. Uses the ServiceExt trait and futures_util::StreamExt. The example is marked as no_run and includes a tokio::main wrapper.

```rust
no_run
# #[tokio::main]
# async fn main() -> Result<(), async_nats::Error> {
use async_nats::service::ServiceExt;
use futures_util::StreamExt;
let client = async_nats::connect("demo.nats.io").await?;
let mut service = client.service_builder().start("generator", "1.0.0").await?;
let mut endpoint = service.endpoint("get").await?;

if let Some(request) = endpoint.next().await {
    request.respond(Ok("hello".into())).await?;
}

# Ok(())
# }

```

--------------------------------

### async_nats service endpoint_builder example

Source: https://docs.rs/async-nats/latest/src/async_nats/service/mod.rs.html

Shows how to use the service builder to start a service and add an endpoint. The example connects to a NATS server, starts a service named "service" with version "1.0.0", creates a group "v1", and adds an endpoint named "api" with subject "products".

```rust
/// ```no_run
/// # #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error> {
/// use async_nats::service::ServiceExt;
/// let client = async_nats::connect("demo.nats.io").await?;
/// let mut service = client.service_builder().start("service", "1.0.0").await?;
/// let v1 = service.group("v1");
///
/// let products = v1.endpoint_builder().name("api").add("products").await?;
/// # Ok(()) 
/// # }
/// ```
```

--------------------------------

### Send response with Request::respond

Source: https://docs.rs/async-nats/latest/src/async_nats/service/mod.rs.html?search=

Demonstrates sending a response to a request using the respond method. The example starts a service named "serviceA" with version "1.0.0.1", gets an endpoint, waits for the next request, and responds with the payload "hello".

```rust
/// ```no_run
/// # #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error> {
/// use async_nats::service::ServiceExt;
/// use futures_util::StreamExt;
/// # let client = async_nats::connect("demo.nats.io").await?;
/// # let mut service = client
/// #    .service_builder().start("serviceA", "1.0.0.1").await?;
/// let mut endpoint = service.endpoint("endpoint").await?;
/// let request = endpoint.next().await.unwrap();
/// request.respond(Ok("hello".into())).await?;
/// # Ok(())
/// # }
/// ```
```

--------------------------------

### Start a service and respond to a request

Source: https://docs.rs/async-nats/latest/src/async_nats/service/mod.rs.html?search=std%3A%3Avec

This example demonstrates how to connect to a NATS server, start a service with a name and version, create an endpoint, and respond to an incoming request. It uses the ServiceExt trait and futures_util::StreamExt. The code is marked as no_run, meaning it is not executed during documentation tests.

```rust
```no_run
# #[tokio::main]
# async fn main() -> Result<(), async_nats::Error> {
use async_nats::service::ServiceExt;
use futures_util::StreamExt;
let client = async_nats::connect("demo.nats.io").await?;
let mut service = client.service_builder().start("generator", "1.0.0").await?;
let mut endpoint = service.endpoint("get").await?;

if let Some(request) = endpoint.next().await {
    request.respond(Ok("hello".into())).await?;
}

# Ok(())
# }
```
```

--------------------------------

### FetchBuilder example: fetch with max_messages and max_bytes

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/consumer/pull/struct.FetchBuilder.html?search=std%3A%3Avec

Shows how to create a FetchBuilder using fetch() and configure it with max_messages and max_bytes before calling messages() to get a stream of messages. The example processes each message and acknowledges it.

```rust
use async_nats::jetstream::consumer::PullConsumer;
use futures_util::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer
    .fetch()
    .max_messages(100)
    .max_bytes(1024)
    .messages()
    .await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}
```

--------------------------------

### min_pending example

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/consumer/pull/struct.StreamBuilder.html

Sets the overflow threshold for minimum pending messages before the stream starts getting messages for a Consumer. Requires the Consumer to have Config::priority_groups and PriorityPolicy::Overflow enabled. The example connects to a NATS server, gets a stream and consumer, then consumes messages with a 30-second expiry, group 

```rust
use async_nats::jetstream::consumer::PullConsumer;
use futures_util::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer
    .stream()
    .expires(std::time::Duration::from_secs(30))
    .group("A")
    .min_pending(100)
    .messages()
    .await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}

```

--------------------------------

### start

Source: https://docs.rs/async-nats/latest/async_nats/service/struct.ServiceBuilder.html?search=

Starts the service with configured options.

```APIDOC
## start

### Description
Starts the service with configured options.

### Method
pub async fn start<N: ToString, V: ToString>(self, name: N, version: V) -> Result<Service, Error>

### Parameters
- **name** (N) - Required - Service name.
- **version** (V) - Required - Service version.

### Returns
- `Result<Service, Error>` - The started service or an error.
```

--------------------------------

### Complete example: connect, publish, and subscribe

Source: https://docs.rs/async-nats/latest/async_nats/index.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Complete example that connects to a NATS server, subscribes to the "messages" subject, publishes ten messages, and processes incoming messages. Requires the bytes and futures-util crates and uses the tokio runtime.

```rust
use bytes::Bytes;
use futures_util::StreamExt;

#[tokio::main]
async fn main() -> Result<(), async_nats::Error> {
    // Connect to the NATS server
    let client = async_nats::connect("demo.nats.io").await?;

    // Subscribe to the "messages" subject
    let mut subscriber = client.subscribe("messages").await?;

    // Publish messages to the "messages" subject
    for _ in 0..10 {
        client.publish("messages", "data".into()).await?;
    }

    // Receive and process messages
    while let Some(message) = subscriber.next().await {
        println!("Received message {:?}", message);
    }

    Ok(())
}
```

--------------------------------

### Get header value from HeaderMap

Source: https://docs.rs/async-nats/latest/async_nats/header/struct.HeaderMap.html?search=u32+-%3E+bool

Example of getting a header value using get, unwrapping the Option.

```rust

let mut headers = HeaderMap::new();
headers.append("Key", "Value");
let values = headers.get("Key").unwrap();
```

--------------------------------

### Manually create Box from scratch

Source: https://docs.rs/async-nats/latest/async_nats/type.Error.html

Example for `Box::from_raw` using the global allocator to manually allocate and initialize memory.

```rust
use std::alloc::{alloc, Layout};

unsafe {
    let ptr = alloc(Layout::new::<i32>()) as *mut i32;
    // In general .write is required to avoid attempting to destruct
    // the (uninitialized) previous contents of `ptr`, though for this
    // simple example `*ptr = 5` would have worked as well.
    ptr.write(5);
    let x = Box::from_raw(ptr);
}
```

--------------------------------

### Get subslice safely with get

Source: https://docs.rs/async-nats/latest/async_nats/subject/struct.Subject.html?search=

Example for `get` that returns `Some` for a valid range and `None` for invalid ranges (non-boundary or out-of-bounds).

```rust
let v = String::from("🗻∈🌏");

assert_eq!(Some("🗻"), v.get(0..4));

// indices not on UTF-8 sequence boundaries
assert!(v.get(1..).is_none());
assert!(v.get(..8).is_none());

// out of bounds
assert!(v.get(..42).is_none());
```

--------------------------------

### Service usage example

Source: https://docs.rs/async-nats/latest/async_nats/service/struct.Service.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Example from the Service documentation showing how to create a service, add an endpoint, and respond to a request. It uses `ServiceExt` and `StreamExt` traits.

```rust
use async_nats::service::ServiceExt;
use futures_util::StreamExt;
let client = async_nats::connect("demo.nats.io").await?;
let mut service = client.service_builder().start("generator", "1.0.0").await?;
let mut endpoint = service.endpoint("get").await?;

if let Some(request) = endpoint.next().await {
    request.respond(Ok("hello".into())).await?;
}

```

--------------------------------

### Get last header value from HeaderMap

Source: https://docs.rs/async-nats/latest/async_nats/header/struct.HeaderMap.html?search=u32+-%3E+bool

Example of getting the last header value using get_last.

```rust

let mut headers = HeaderMap::new();
headers.append("Key", "Value");
let values = headers.get_last("Key").unwrap();
```

--------------------------------

### Store create() example

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/kv/mod.rs.html

Connects to a NATS server, creates a key-value bucket, and demonstrates that create() succeeds on a new key and fails on an existing key. Uses async_nats::connect and jetstream::create_key_value.

```rust
/// # #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error> {
/// let client = async_nats::connect("demo.nats.io:4222").await?;
/// let jetstream = async_nats::jetstream::new(client);
/// let kv = jetstream
///     .create_key_value(async_nats::jetstream::kv::Config {
///         bucket: "kv".to_string(),
///         history: 10,
///         ..Default::default()
///     })
///     .await?;
///
/// let status = kv.create("key", "value".into()).await;
/// assert!(status.is_ok());
///
/// let status = kv.create("key", "value".into()).await;
/// assert!(status.is_err());
///
/// # Ok(())
/// # }

```

--------------------------------

### Create Key-Value Bucket Example

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/context.rs.html

Shows how to create a key-value bucket using the async_nats JetStream context. The example connects to a NATS server, creates a jetstream context, and calls create_key_value with a Config specifying the bucket name and history.

```rust
# #[tokio::main]
# async fn main() -> Result<(), async_nats::Error> {
let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let kv = jetstream
    .create_key_value(async_nats::jetstream::kv::Config {
        bucket: "kv".to_string(),
        history: 10,
        ..Default::default()
    })
    .await?;
# Ok(())
# }
```

--------------------------------

### Consumer::info() usage example

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/consumer/type.OrderedPushConsumer.html

Connects to a local NATS server and creates a pull consumer. The example shows the typical setup for using the info() method.

```rust
use async_nats::jetstream::consumer::PullConsumer;
let client = async_nats::connect("localhost:4222").await?;
```

--------------------------------

### Connect to NATS and publish example

Source: https://docs.rs/async-nats/latest/async_nats/fn.connect_with_options.html?search=u32+-%3E+bool

Example usage of connect_with_options to connect to demo.nats.io and publish a message to the 'test' subject.

```rust
let mut nc =
    async_nats::connect_with_options("demo.nats.io", async_nats::ConnectOptions::new()).await?;
nc.publish("test", "data".into()).await?;
```

--------------------------------

### substr_range example

Source: https://docs.rs/async-nats/latest/async_nats/subject/struct.Subject.html?search=

Uses `substr_range` to get ranges of substrings from a split iterator.

```rust
use core::range::Range;

let data = "a, b, b, a";
let mut iter = data.split(", ").map(|s| data.substr_range(s).unwrap());

assert_eq!(iter.next(), Some(Range { start: 0, end: 1 }));
assert_eq!(iter.next(), Some(Range { start: 3, end: 4 }));
assert_eq!(iter.next(), Some(Range { start: 6, end: 7 }));
assert_eq!(iter.next(), Some(Range { start: 9, end: 10 }));
```

--------------------------------

### new

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/consumer/pull.rs.html?search=

Constructor for `BatchBuilder` with default settings.

```APIDOC
## `new`

### Description
Creates a new `BatchBuilder` with default configuration values.

### Method
`new`

### Signature
`pub fn new(consumer: &'a Consumer<Config>) -> Self`

### Parameters
- **consumer** (`&'a Consumer<Config>`) - Required - The consumer to build the batch for.

### Returns
- `Self` - A new `BatchBuilder` instance.
```

--------------------------------

### Object Store Struct and Get Method

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/object_store/mod.rs.html?search=u32+-%3E+bool

The ObjectStore struct and its get method, which retrieves an object as an async stream. Includes a usage example with tokio::io::AsyncReadExt.

```rust
/// A blob store capable of storing large objects efficiently in streams.
#[derive(Clone)]
pub struct ObjectStore {
    pub(crate) name: String,
    pub(crate) stream: crate::jetstream::stream::Stream,
}

impl ObjectStore {
    /// Gets an [Object] from the [ObjectStore].
    ///
    /// [Object] implements [tokio::io::AsyncRead] that allows
    /// to read the data from Object Store.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use tokio::io::AsyncReadExt;
    /// let client = async_nats::connect("demo.nats.io").await?;
    /// let jetstream = async_nats::jetstream::new(client);
    ///
    /// let bucket = jetstream.get_object_store("store").await?;
    /// let mut object = bucket.get("FOO").await?;
    ///
    /// // Object implements `tokio::io::AsyncRead`.
    /// let mut bytes = vec![];
    /// object.read_to_end(&mut bytes).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn get<T: AsRef<str> + Send>(&self, object_name: T) -> Result<Object, GetError> {
        self.get_impl(object_name).await
    }

    fn get_impl<'bucket, 'future, T>(
        &'bucket self,
        object_name: T,
    ) -> BoxFuture<'future, Result<Object, GetError>>
    where
        T: AsRef<str> + Send + 'future,
        'bucket: 'future,
```

--------------------------------

### Get object info from the ObjectStore

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/object_store/mod.rs.html?search=std%3A%3Avec

Retrieves ObjectInfo for an object. The example connects to a NATS server, gets the object store, and retrieves info for the object named 'FOO'.

```rust
```no_run
# #[tokio::main]
# async fn main() -> Result<(), async_nats::Error> {
let client = async_nats::connect("demo.nats.io").await?;
let jetstream = async_nats::jetstream::new(client);

let bucket = jetstream.get_object_store("store").await?;
let info = bucket.info("FOO").await?;
# Ok(())
# }
```
```

--------------------------------

### cached_info() usage example

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/consumer/mod.rs.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Shows how to retrieve a consumer and call cached_info() to get its cached Info. The example is part of the method's documentation and assumes an existing stream and consumer.

```rust
///     .get_stream("events")
///     .await?
///     .get_consumer("pull")
///     .await?;
///
/// let info = consumer.cached_info();
/// # Ok(())
/// # }
/// ```
```

--------------------------------

### group method example

Source: https://docs.rs/async-nats/latest/async_nats/service/struct.Service.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Example for the `group` method, showing how to create a group with a common prefix and then add an endpoint to that group.

```rust
use async_nats::service::ServiceExt;
let client = async_nats::connect("demo.nats.io").await?;
let mut service = client.service_builder().start("service", "1.0.0").await?;

let v1 = service.group("v1");
let products = v1.endpoint("products").await?;

```

--------------------------------

### truncate_to_day method example

Source: https://docs.rs/async-nats/latest/async_nats/datetime/type.DateTime.html

Truncates the datetime to the start of the day, setting the time to midnight.

```rust
assert_eq!(
    datetime!(2022-02-18 15:30:45.123 +01).truncate_to_day(),
    datetime!(2022-02-18 0:00 +01)
);
```

--------------------------------

### endpoint method example

Source: https://docs.rs/async-nats/latest/async_nats/service/struct.Service.html?search=Option%3CT%3E%2C+%28T+-%3E+U%29+-%3E+Option%3CU%3E

Example for the `endpoint` method, showing how to add a new endpoint to the service.

```rust
use async_nats::service::ServiceExt;
let client = async_nats::connect("demo.nats.io").await?;
let mut service = client.service_builder().start("service", "1.0.0").await?;

let products = service.endpoint("products").await?;

```

--------------------------------

### ObjectStore put() usage example

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/object_store/mod.rs.html?search=

Example of using the `put` method to upload a file to an object store. It connects to a NATS server, gets an object store, opens a file, and puts it into the store.

```rust
no_run
# #[tokio::main]
# async fn main() -> Result<(), async_nats::Error> {
let client = async_nats::connect("demo.nats.io").await?;
let jetstream = async_nats::jetstream::new(client);

let bucket = jetstream.get_object_store("store").await?;
let mut file = tokio::fs::File::open("foo.txt").await?;
bucket.put("file", &mut file).await.unwrap();
# Ok(())
# }

```

--------------------------------

### messages example

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/consumer/pull/struct.FetchBuilder.html

Creates the actual stream with the provided configuration. This is the final step to obtain a Batch of messages.

```rust
use async_nats::jetstream::consumer::PullConsumer;
use futures_util::StreamExt;
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer.fetch().max_messages(100).messages().await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}
```

--------------------------------

### Respond to a request with respond method

Source: https://docs.rs/async-nats/latest/src/async_nats/service/mod.rs.html?search=std%3A%3Avec

Demonstrates how to respond to a request using the respond method. It connects to demo.nats.io, starts a service named "serviceA" with version "1.0.0.1", gets an endpoint, waits for a request, and responds with the payload "hello".

```rust
use async_nats::service::ServiceExt;
use futures_util::StreamExt;
let client = async_nats::connect("demo.nats.io").await?;
let mut service = client
   .service_builder().start("serviceA", "1.0.0.1").await?;
let mut endpoint = service.endpoint("endpoint").await?;
let request = endpoint.next().await.unwrap();
request.respond(Ok("hello".into())).await?;
```

--------------------------------

### Ordered Pull Consumer example with get_or_create_stream

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/consumer/pull.rs.html?search=u32+-%3E+bool

Example usage of an ordered pull consumer in async_nats. It connects to a local NATS server, creates or gets a stream named 'events', and then uses the consumer to fetch messages. The example is marked as no_run and uses tokio main.

```rust
    /// Returns a stream of messages for Ordered Pull Consumer.
    ///
    /// Ordered consumers uses single replica ephemeral consumer, no matter the replication factor of the
    /// Stream. It does not use acks, instead it tracks sequences and recreate itself whenever it
    /// sees mismatch.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn mains() -> Result<(), async_nats::Error> {
    /// use futures_util::StreamExt;
    /// use futures_util::TryStreamExt;
    ///
    /// let client = async_nats::connect("localhost:4222").await?;
    /// let jetstream = async_nats::jetstream::new(client);
    ///
    /// let stream = jetstream
    ///     .get_or_create_stream(async_nats::jetstream::stream::Config {
    ///         name: "events".to_string(),
```

--------------------------------

### BatchBuilder::min_pending

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/consumer/pull/struct.BatchBuilder.html

Sets the overflow threshold for minimum pending messages before the stream starts getting messages.

```APIDOC
## BatchBuilder::min_pending

### Description
Sets overflow threshold for minimum pending messages before this stream will start getting messages. To use overflow, Consumer needs to have enabled Config::priority_groups and PriorityPolicy::Overflow set.

### Method
min_pending

### Signature
pub fn min_pending(self, min_pending: usize) -> Self

### Parameters
- **min_pending** (usize) - Required - The minimum pending messages threshold.

### Returns
- **Self** - The builder instance for chaining.

### Example
```rust
use async_nats::jetstream::consumer::PullConsumer;
use futures_util::StreamExt;

let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let consumer: PullConsumer = jetstream
    .get_stream("events")
    .await?
    .get_consumer("pull")
    .await?;

let mut messages = consumer
    .batch()
    .expires(std::time::Duration::from_secs(30))
    .group("A")
    .min_pending(100)
    .messages()
    .await?;

while let Some(message) = messages.next().await {
    let message = message?;
    println!("message: {:?}", message);
    message.ack().await?;
}
```

```

--------------------------------

### create() example

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/kv/struct.Store.html

Adds a key/value pair only if the key does not exist; returns an error if it does. The example creates twice to show the error.

```Rust
let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let kv = jetstream
    .create_key_value(async_nats::jetstream::kv::Config {
        bucket: "kv".to_string(),
        history: 10,
        ..Default::default()
    })
    .await?;

let status = kv.create("key", "value".into()).await;
assert!(status.is_ok());

let status = kv.create("key", "value".into()).await;
assert!(status.is_err());
```

--------------------------------

### get_object_store example

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/context/struct.Context.html

Gets an existing object store bucket by name. Requires the `object-store` crate feature.

```rust
let client = async_nats::connect("demo.nats.io").await?;
let jetstream = async_nats::jetstream::new(client);
let bucket = jetstream.get_object_store("bucket").await?;
```

--------------------------------

### starts_with

Source: https://docs.rs/async-nats/latest/async_nats/subject/struct.Subject.html?search=u32+-%3E+bool

Documentation for the `starts_with` method on string slices.

```APIDOC
## `starts_with<P>(&self, pat: P) -> bool`

### Description
Returns `true` if the given pattern matches a prefix of this string slice, otherwise `false`. The pattern can be a `&str`, `char`, a slice of `char`s, or a function or closure that determines if a character matches. For `char`, slice of `char`s, or closures, only the first character is checked.

### Signature
```rust
pub fn starts_with<P>(&self, pat: P) -> bool
where
    P: Pattern,
```

### Parameters
- `pat` (P) - The pattern to check as a prefix. Can be a `&str`, `char`, a slice of `char`s, or a function/closure that determines if a character matches.

### Returns
- `bool` - `true` if the pattern matches a prefix of the string, `false` otherwise.

### Examples
```rust
let bananas = "bananas";

assert!(bananas.starts_with("bana"));
assert!(!bananas.starts_with("nana"));

// Note that both of these assert successfully.
assert!(bananas.starts_with(&['b', 'a', 'n', 'a']));
assert!(bananas.starts_with(&['a', 'b', 'c', 'd']));
```
```

--------------------------------

### create_object_store example

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/context/struct.Context.html

Creates a new object store bucket. Requires the `object-store` crate feature.

```rust
let client = async_nats::connect("demo.nats.io").await?;
let jetstream = async_nats::jetstream::new(client);
let bucket = jetstream
    .create_object_store(async_nats::jetstream::object_store::Config {
        bucket: "bucket".to_string(),
        ..Default::default()
    })
    .await?;
```

--------------------------------

### status() example

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/kv/struct.Store.html

Queries the server and returns the status of the key-value bucket. Requires an existing key-value bucket created with create_key_value.

```Rust
let client = async_nats::connect("demo.nats.io:4222").await?;
let jetstream = async_nats::jetstream::new(client);
let kv = jetstream
    .create_key_value(async_nats::jetstream::kv::Config {
        bucket: "kv".to_string(),
        history: 10,
        ..Default::default()
    })
    .await?;
let status = kv.status().await?;
println!("status: {:?}", status);
```

--------------------------------

### Stream::info example

Source: https://docs.rs/async-nats/latest/async_nats/jetstream/stream/struct.Stream.html?search=

Retrieves info about a Stream from the server, updates the cached info inside Stream and returns it. The example connects to a NATS server, gets the "events" stream, and calls info.

```rust
let client = async_nats::connect("localhost:4222").await?;
let jetstream = async_nats::jetstream::new(client);

let mut stream = jetstream.get_stream("events").await?;

let info = stream.info().await?;
```

--------------------------------

### BatchBuilder example

Source: https://docs.rs/async-nats/latest/src/async_nats/jetstream/consumer/pull.rs.html?search=u32+-%3E+bool

Example for the `BatchBuilder` struct, showing how to create a batch with `max_messages` and `max_bytes` and consume messages.

```rust
/// # #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error>  {
/// use async_nats::jetstream::consumer::PullConsumer;
/// use futures_util::StreamExt;
/// let client = async_nats::connect("localhost:4222").await?;
/// let jetstream = async_nats::jetstream::new(client);
///
/// let consumer: PullConsumer = jetstream
///     .get_stream("events")
///     .await?
///     .get_consumer("pull")
///     .await?;
///
/// let mut messages = consumer
///     .batch()
///     .max_messages(100)
///     .max_bytes(1024)
///     .messages()
///     .await?;
///
/// while let Some(message) = messages.next().await {
///     let message = message?;
///     println!("message: {:?}", message);
///     message.ack().await?;
/// }
/// # Ok(())
/// # }
/// ```
```

--------------------------------

### Create service endpoint with endpoint_builder

Source: https://docs.rs/async-nats/latest/src/async_nats/service/mod.rs.html?search=u32+-%3E+bool

Shows how to use `ServiceExt` and `service_builder()` to start a service, create a group, and add an endpoint named "api" with subject "products" using `endpoint_builder()`. Requires a running NATS server at demo.nats.io and returns Result<(), async_nats::Error>.

```rust
/// ```no_run
/// # #[tokio::main]
/// # async fn main() -> Result<(), async_nats::Error> {
/// use async_nats::service::ServiceExt;
/// let client = async_nats::connect("demo.nats.io").await?;
/// let mut service = client.service_builder().start("service", "1.0.0").await?;
/// let v1 = service.group("v1");
///
/// let products = v1.endpoint_builder().name("api").add("products").await?;
/// # Ok(())
/// # }
/// ```
```
Tokens:
Raw