Struct async_nats::jetstream::stream::Stream

pub struct Stream { /* private fields */ }

Handle to operations that can be performed on a Stream.

Implementations

impl Stream

pub async fn info(&mut self) -> Result<&Info, Error>

Retrieves info about Stream from the server, updates the cached info inside Stream and returns it.

Examples

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?;

Examples found in repository

src/jetstream/object_store/mod.rs (line 413)

    pub async fn seal(&mut self) -> Result<(), Error> {
        let mut stream_config = self.stream.info().await?.to_owned();
        stream_config.config.sealed = true;

        self.stream
            .context
            .update_stream(&stream_config.config)
            .await?;
        Ok(())
    }

pub fn cached_info(&self) -> &Info

Returns cached Info for the Stream. Cache is either from initial creation/retrieval of the Stream or last call to Stream::info.

Examples

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

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

let info = stream.cached_info();

pub async fn direct_get_next_for_subject<T: AsRef<str>>(
    &self,
    subject: T,
    sequence: Option<u64>
) -> Result<Message, Error>

Gets next message for a Stream.

Requires a Stream with allow_direct set to true. This is different from Stream::get_raw_message, as it can fetch Message from any replica member. This means read after write is possible, as that given replica might not yet catch up with the leader.

Examples

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

let stream = jetstream.create_stream(async_nats::jetstream::stream::Config {
    name: "events".to_string(),
    subjects: vec!["events.>".to_string()],
    allow_direct: true,
    ..Default::default()
}).await?;

jetstream.publish("events.data".into(), "data".into()).await?;
let pub_ack = jetstream.publish("events.data".into(), "data".into()).await?;

let message =  stream
    .direct_get_next_for_subject("events.data", Some(pub_ack.await?.sequence)).await?;

pub async fn direct_get_first_for_subject<T: AsRef<str>>(
    &self,
    subject: T
) -> Result<Message, Error>

Gets first message from Stream.

Requires a Stream with allow_direct set to true. This is different from Stream::get_raw_message, as it can fetch Message from any replica member. This means read after write is possible, as that given replica might not yet catch up with the leader.

Examples

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

let stream = jetstream.create_stream(async_nats::jetstream::stream::Config {
    name: "events".to_string(),
    subjects: vec!["events.>".to_string()],
    allow_direct: true,
    ..Default::default()
}).await?;

let pub_ack = jetstream.publish("events.data".into(), "data".into()).await?;

let message =  stream.direct_get_first_for_subject("events.data").await?;

pub async fn direct_get(&self, sequence: u64) -> Result<Message, Error>

Gets message from Stream with given sequence id.

Requires a Stream with allow_direct set to true. This is different from Stream::get_raw_message, as it can fetch Message from any replica member. This means read after write is possible, as that given replica might not yet catch up with the leader.

Examples

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

let stream = jetstream.create_stream(async_nats::jetstream::stream::Config {
    name: "events".to_string(),
    subjects: vec!["events.>".to_string()],
    allow_direct: true,
    ..Default::default()
}).await?;

let pub_ack = jetstream.publish("events.data".into(), "data".into()).await?;

let message =  stream.direct_get(pub_ack.await?.sequence).await?;

pub async fn direct_get_last_for_subject<T: AsRef<str>>(
    &self,
    subject: T
) -> Result<Message, Error>

Gets last message for a given subject.

Requires a Stream with allow_direct set to true. This is different from Stream::get_raw_message, as it can fetch Message from any replica member. This means read after write is possible, as that given replica might not yet catch up with the leader.

Examples

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

let stream = jetstream.create_stream(async_nats::jetstream::stream::Config {
    name: "events".to_string(),
    subjects: vec!["events.>".to_string()],
    allow_direct: true,
    ..Default::default()
}).await?;

jetstream.publish("events.data".into(), "data".into()).await?;

let message =  stream.direct_get_last_for_subject("events.data").await?;

Examples found in repository

src/jetstream/kv/mod.rs (line 239)

    pub async fn entry<T: Into<String>>(&self, key: T) -> Result<Option<Entry>, Error> {
        let key: String = key.into();
        if !is_valid_key(key.as_ref()) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid key",
            )));
        }

        let subject = format!("{}{}", self.prefix.as_str(), &key);

        let result: Option<(Message, Operation, u64, OffsetDateTime)> = {
            if self.stream.info.config.allow_direct {
                let message = self
                    .stream
                    .direct_get_last_for_subject(subject.as_str())
                    .await;

                match message {
                    Ok(message) => {
                        let headers = message.headers.as_ref().ok_or_else(|| {
                            std::io::Error::new(io::ErrorKind::Other, "did not found headers")
                        })?;
                        let operation = headers.get(KV_OPERATION).map_or_else(
                            || Operation::Put,
                            |operation| match operation
                                .iter()
                                .next()
                                .cloned()
                                .unwrap_or_else(|| KV_OPERATION_PUT.to_string())
                                .as_ref()
                            {
                                KV_OPERATION_PURGE => Operation::Purge,
                                KV_OPERATION_DELETE => Operation::Delete,
                                _ => Operation::Put,
                            },
                        );
                        let sequence = headers
                            .get(header::NATS_SEQUENCE)
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::NotFound,
                                    "did not found sequence header",
                                )
                            })?
                            .iter()
                            .next()
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::NotFound,
                                    "did not found sequence header value",
                                )
                            })?
                            .parse()?;
                        let created = headers
                            .get(header::NATS_TIME_STAMP)
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::NotFound,
                                    "did not found timestamp header",
                                )
                            })?
                            .iter()
                            .next()
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::NotFound,
                                    "did not found timestamp header value",
                                )
                            })
                            .and_then(|created| {
                                OffsetDateTime::parse(created, &Rfc3339).map_err(|err| {
                                    std::io::Error::new(
                                        io::ErrorKind::Other,
                                        format!("failed to parse Nats-Time-Stamp: {err}"),
                                    )
                                })
                            })?;

                        Some((message.message, operation, sequence, created))
                    }
                    Err(err) => {
                        let e: std::io::Error = *err.downcast().unwrap();
                        if e.kind() == ErrorKind::NotFound {
                            None
                        } else {
                            return Err(Box::new(e));
                        }
                    }
                }
            } else {
                let raw_message = self
                    .stream
                    .get_last_raw_message_by_subject(subject.as_str())
                    .await;
                match raw_message {
                    Ok(raw_message) => {
                        let operation = kv_operation_from_stream_message(&raw_message);
                        // TODO: unnecessary expensive, cloning whole Message.
                        let nats_message = Message::try_from(raw_message.clone())?;
                        Some((
                            nats_message,
                            operation,
                            raw_message.sequence,
                            raw_message.time,
                        ))
                    }
                    Err(err) => {
                        let e: std::io::Error = *err.downcast().unwrap();
                        let d = e.get_ref().unwrap();
                        let de = d.downcast_ref::<response::Error>().unwrap();
                        // 10037 is returned when there are no messages found.
                        if de.code == 10037 {
                            None
                        } else {
                            return Err(Box::new(e));
                        }
                    }
                }
            }
        };

        match result {
            Some((message, operation, revision, created)) => {
                if message.status == Some(StatusCode::NO_RESPONDERS) {
                    return Ok(None);
                }

                let entry = Entry {
                    bucket: self.name.clone(),
                    key,
                    value: message.payload.to_vec(),
                    revision,
                    created,
                    operation,
                    delta: 0,
                };
                Ok(Some(entry))
            }
            // TODO: remember to touch this when Errors are in place.
            None => Ok(None),
        }
    }

pub async fn get_raw_message(&self, sequence: u64) -> Result<RawMessage, Error>

Get a raw message from the stream.

Examples

#[tokio::main]
use futures::StreamExt;
use futures::TryStreamExt;

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

let stream = context.get_or_create_stream(async_nats::jetstream::stream::Config {
    name: "events".to_string(),
    max_messages: 10_000,
    ..Default::default()
}).await?;

let publish_ack = context.publish("events".to_string(), "data".into()).await?;
let raw_message = stream.get_raw_message(publish_ack.await?.sequence).await?;
println!("Retrieved raw message {:?}", raw_message);

pub async fn get_last_raw_message_by_subject(
    &self,
    stream_subject: &str
) -> Result<RawMessage, Error>

Get the last raw message from the stream by subject.

Examples

#[tokio::main]
use futures::StreamExt;
use futures::TryStreamExt;

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

let stream = context.get_or_create_stream(async_nats::jetstream::stream::Config {
    name: "events".to_string(),
    max_messages: 10_000,
    ..Default::default()
}).await?;

let publish_ack = context.publish("events".to_string(), "data".into()).await?;
let raw_message = stream.get_last_raw_message_by_subject("events").await?;
println!("Retrieved raw message {:?}", raw_message);

Examples found in repository

src/jetstream/object_store/mod.rs (line 207)

    pub async fn info<T: AsRef<str>>(&self, object_name: T) -> Result<ObjectInfo, Error> {
        let object_name = object_name.as_ref();
        let object_name = encode_object_name(object_name);
        if !is_valid_object_name(&object_name) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid object name",
            )));
        }

        // Grab last meta value we have.
        let subject = format!("$O.{}.M.{}", &self.name, &object_name);

        let message = self
            .stream
            .get_last_raw_message_by_subject(subject.as_str())
            .await?;
        let decoded_payload = base64::decode(message.payload)
            .map_err(|err| Box::new(std::io::Error::new(ErrorKind::Other, err)))?;
        let object_info = serde_json::from_slice::<ObjectInfo>(&decoded_payload)?;

        Ok(object_info)
    }

src/jetstream/kv/mod.rs (line 317)

    pub async fn entry<T: Into<String>>(&self, key: T) -> Result<Option<Entry>, Error> {
        let key: String = key.into();
        if !is_valid_key(key.as_ref()) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid key",
            )));
        }

        let subject = format!("{}{}", self.prefix.as_str(), &key);

        let result: Option<(Message, Operation, u64, OffsetDateTime)> = {
            if self.stream.info.config.allow_direct {
                let message = self
                    .stream
                    .direct_get_last_for_subject(subject.as_str())
                    .await;

                match message {
                    Ok(message) => {
                        let headers = message.headers.as_ref().ok_or_else(|| {
                            std::io::Error::new(io::ErrorKind::Other, "did not found headers")
                        })?;
                        let operation = headers.get(KV_OPERATION).map_or_else(
                            || Operation::Put,
                            |operation| match operation
                                .iter()
                                .next()
                                .cloned()
                                .unwrap_or_else(|| KV_OPERATION_PUT.to_string())
                                .as_ref()
                            {
                                KV_OPERATION_PURGE => Operation::Purge,
                                KV_OPERATION_DELETE => Operation::Delete,
                                _ => Operation::Put,
                            },
                        );
                        let sequence = headers
                            .get(header::NATS_SEQUENCE)
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::NotFound,
                                    "did not found sequence header",
                                )
                            })?
                            .iter()
                            .next()
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::NotFound,
                                    "did not found sequence header value",
                                )
                            })?
                            .parse()?;
                        let created = headers
                            .get(header::NATS_TIME_STAMP)
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::NotFound,
                                    "did not found timestamp header",
                                )
                            })?
                            .iter()
                            .next()
                            .ok_or_else(|| {
                                io::Error::new(
                                    io::ErrorKind::NotFound,
                                    "did not found timestamp header value",
                                )
                            })
                            .and_then(|created| {
                                OffsetDateTime::parse(created, &Rfc3339).map_err(|err| {
                                    std::io::Error::new(
                                        io::ErrorKind::Other,
                                        format!("failed to parse Nats-Time-Stamp: {err}"),
                                    )
                                })
                            })?;

                        Some((message.message, operation, sequence, created))
                    }
                    Err(err) => {
                        let e: std::io::Error = *err.downcast().unwrap();
                        if e.kind() == ErrorKind::NotFound {
                            None
                        } else {
                            return Err(Box::new(e));
                        }
                    }
                }
            } else {
                let raw_message = self
                    .stream
                    .get_last_raw_message_by_subject(subject.as_str())
                    .await;
                match raw_message {
                    Ok(raw_message) => {
                        let operation = kv_operation_from_stream_message(&raw_message);
                        // TODO: unnecessary expensive, cloning whole Message.
                        let nats_message = Message::try_from(raw_message.clone())?;
                        Some((
                            nats_message,
                            operation,
                            raw_message.sequence,
                            raw_message.time,
                        ))
                    }
                    Err(err) => {
                        let e: std::io::Error = *err.downcast().unwrap();
                        let d = e.get_ref().unwrap();
                        let de = d.downcast_ref::<response::Error>().unwrap();
                        // 10037 is returned when there are no messages found.
                        if de.code == 10037 {
                            None
                        } else {
                            return Err(Box::new(e));
                        }
                    }
                }
            }
        };

        match result {
            Some((message, operation, revision, created)) => {
                if message.status == Some(StatusCode::NO_RESPONDERS) {
                    return Ok(None);
                }

                let entry = Entry {
                    bucket: self.name.clone(),
                    key,
                    value: message.payload.to_vec(),
                    revision,
                    created,
                    operation,
                    delta: 0,
                };
                Ok(Some(entry))
            }
            // TODO: remember to touch this when Errors are in place.
            None => Ok(None),
        }
    }

pub async fn delete_message(&self, sequence: u64) -> Result<bool, Error>

Delete a message from the stream.

Examples

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

let stream = context.get_or_create_stream(async_nats::jetstream::stream::Config {
    name: "events".to_string(),
    max_messages: 10_000,
    ..Default::default()
}).await?;

let publish_ack = context.publish("events".to_string(), "data".into()).await?;
stream.delete_message(publish_ack.await?.sequence).await?;

pub fn purge(&self) -> Purge<No, No>

Purge Stream messages.

Examples

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

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

Examples found in repository

src/jetstream/stream.rs (line 544)

    pub async fn purge_subject<T>(&self, subject: T) -> Result<PurgeResponse, Error>
    where
        T: Into<String>,
    {
        self.purge().filter(subject).await
    }

pub async fn purge_subject<T>(&self, subject: T) -> Result<PurgeResponse, Error>where
    T: Into<String>,

Purge Stream messages for a matching subject.

Examples

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

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

Examples found in repository

src/jetstream/object_store/mod.rs (line 172)

    pub async fn delete<T: AsRef<str>>(&self, object_name: T) -> Result<(), Error> {
        let object_name = object_name.as_ref();
        let mut object_info = self.info(object_name).await?;
        object_info.chunks = 0;
        object_info.size = 0;
        object_info.deleted = true;

        let data = serde_json::to_vec(&object_info)?;

        let mut headers = HeaderMap::default();
        headers.insert(NATS_ROLLUP, HeaderValue::from_str(ROLLUP_SUBJECT)?);

        let subject = format!("$O.{}.M.{}", &self.name, encode_object_name(object_name));

        self.stream
            .context
            .publish_with_headers(subject, headers, data.into())
            .await?;

        let chunk_subject = format!("$O.{}.C.{}", self.name, object_info.nuid);

        self.stream.purge_subject(&chunk_subject).await?;

        Ok(())
    }

    /// Retrieves [Object] [ObjectInfo].
    ///
    /// # Examples
    ///
    /// ```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(())
    /// # }
    /// ```
    pub async fn info<T: AsRef<str>>(&self, object_name: T) -> Result<ObjectInfo, Error> {
        let object_name = object_name.as_ref();
        let object_name = encode_object_name(object_name);
        if !is_valid_object_name(&object_name) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid object name",
            )));
        }

        // Grab last meta value we have.
        let subject = format!("$O.{}.M.{}", &self.name, &object_name);

        let message = self
            .stream
            .get_last_raw_message_by_subject(subject.as_str())
            .await?;
        let decoded_payload = base64::decode(message.payload)
            .map_err(|err| Box::new(std::io::Error::new(ErrorKind::Other, err)))?;
        let object_info = serde_json::from_slice::<ObjectInfo>(&decoded_payload)?;

        Ok(object_info)
    }

    /// Puts an [Object] into the [ObjectStore].
    /// This method implements `tokio::io::AsyncRead`.
    ///
    /// # Examples
    ///
    /// ```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(())
    /// # }
    /// ```
    pub async fn put<T>(
        &self,
        meta: T,
        data: &mut (impl tokio::io::AsyncRead + std::marker::Unpin),
    ) -> Result<ObjectInfo, Error>
    where
        ObjectMeta: From<T>,
    {
        let object_meta: ObjectMeta = meta.into();

        let encoded_object_name = encode_object_name(&object_meta.name);
        if !is_valid_object_name(&encoded_object_name) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid object name",
            )));
        }
        // Fetch any existing object info, if there is any for later use.
        let maybe_existing_object_info = match self.info(&encoded_object_name).await {
            Ok(object_info) => Some(object_info),
            Err(_) => None,
        };

        let object_nuid = nuid::next();
        let chunk_subject = format!("$O.{}.C.{}", &self.name, &object_nuid);

        let mut object_chunks = 0;
        let mut object_size = 0;

        let mut buffer = Box::new([0; DEFAULT_CHUNK_SIZE]);
        let mut context = ring::digest::Context::new(&SHA256);

        loop {
            let n = data.read(&mut *buffer).await?;

            if n == 0 {
                break;
            }
            context.update(&buffer[..n]);

            object_size += n;
            object_chunks += 1;

            // FIXME: this is ugly
            let payload = bytes::Bytes::from(buffer[..n].to_vec());

            self.stream
                .context
                .publish(chunk_subject.clone(), payload)
                .await?;
        }
        let digest = context.finish();
        let subject = format!("$O.{}.M.{}", &self.name, &encoded_object_name);
        let object_info = ObjectInfo {
            name: object_meta.name,
            description: object_meta.description,
            link: object_meta.link,
            bucket: self.name.clone(),
            nuid: object_nuid,
            chunks: object_chunks,
            size: object_size,
            digest: format!(
                "SHA-256={}",
                base64::encode_config(digest, base64::URL_SAFE)
            ),
            modified: OffsetDateTime::now_utc(),
            deleted: false,
        };

        let mut headers = HeaderMap::new();
        headers.insert(NATS_ROLLUP, ROLLUP_SUBJECT.parse::<HeaderValue>()?);
        let data = serde_json::to_vec(&object_info)?;

        // publish meta.
        self.stream
            .context
            .publish_with_headers(subject, headers, data.into())
            .await?;

        // Purge any old chunks.
        if let Some(existing_object_info) = maybe_existing_object_info {
            let chunk_subject = format!("$O.{}.C.{}", &self.name, &existing_object_info.nuid);

            self.stream.purge_subject(&chunk_subject).await?;
        }

        Ok(object_info)
    }

pub async fn create_consumer<C: IntoConsumerConfig + FromConsumer>(
    &self,
    config: C
) -> Result<Consumer<C>, Error>

Create a new Durable or Ephemeral Consumer (if durable_name was not provided) and returns the info from the server about created Consumer

Examples

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

let stream = jetstream.get_stream("events").await?;
let info = stream.create_consumer(consumer::pull::Config {
    durable_name: Some("pull".to_string()),
    ..Default::default()
}).await?;

Examples found in repository

src/jetstream/consumer/push.rs (lines 647-650)

async fn recreate_ephemeral_consumer(
    context: Context,
    config: OrderedConfig,
    stream_name: String,
    sequence: u64,
) -> Result<(), Error> {
    let stream = context.get_stream(stream_name.clone()).await?;

    let deliver_policy = {
        if sequence == 0 {
            DeliverPolicy::All
        } else {
            DeliverPolicy::ByStartSequence {
                start_sequence: sequence + 1,
            }
        }
    };
    tokio::time::timeout(
        Duration::from_secs(5),
        stream.create_consumer(jetstream::consumer::push::OrderedConfig {
            deliver_policy,
            ..config
        }),
    )
    .await
    .map_err(|_| io::Error::new(ErrorKind::TimedOut, "timed out"))??;
    Ok(())
}

src/jetstream/object_store/mod.rs (lines 121-125)

    pub async fn get<T: AsRef<str>>(&self, object_name: T) -> Result<Object<'_>, Error> {
        let object_info = self.info(object_name).await?;
        // if let Some(link) = object_info.link {
        //     return self.get(link.name).await;
        // }

        let chunk_subject = format!("$O.{}.C.{}", self.name, object_info.nuid);

        let subscription = self
            .stream
            .create_consumer(crate::jetstream::consumer::push::OrderedConfig {
                filter_subject: chunk_subject,
                deliver_subject: self.stream.context.client.new_inbox(),
                ..Default::default()
            })
            .await?
            .messages()
            .await?;

        Ok(Object::new(subscription, object_info))
    }

    /// 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> {
    /// let client = async_nats::connect("demo.nats.io").await?;
    /// let jetstream = async_nats::jetstream::new(client);
    ///
    /// let bucket = jetstream.get_object_store("store").await?;
    /// bucket.delete("FOO").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn delete<T: AsRef<str>>(&self, object_name: T) -> Result<(), Error> {
        let object_name = object_name.as_ref();
        let mut object_info = self.info(object_name).await?;
        object_info.chunks = 0;
        object_info.size = 0;
        object_info.deleted = true;

        let data = serde_json::to_vec(&object_info)?;

        let mut headers = HeaderMap::default();
        headers.insert(NATS_ROLLUP, HeaderValue::from_str(ROLLUP_SUBJECT)?);

        let subject = format!("$O.{}.M.{}", &self.name, encode_object_name(object_name));

        self.stream
            .context
            .publish_with_headers(subject, headers, data.into())
            .await?;

        let chunk_subject = format!("$O.{}.C.{}", self.name, object_info.nuid);

        self.stream.purge_subject(&chunk_subject).await?;

        Ok(())
    }

    /// Retrieves [Object] [ObjectInfo].
    ///
    /// # Examples
    ///
    /// ```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(())
    /// # }
    /// ```
    pub async fn info<T: AsRef<str>>(&self, object_name: T) -> Result<ObjectInfo, Error> {
        let object_name = object_name.as_ref();
        let object_name = encode_object_name(object_name);
        if !is_valid_object_name(&object_name) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid object name",
            )));
        }

        // Grab last meta value we have.
        let subject = format!("$O.{}.M.{}", &self.name, &object_name);

        let message = self
            .stream
            .get_last_raw_message_by_subject(subject.as_str())
            .await?;
        let decoded_payload = base64::decode(message.payload)
            .map_err(|err| Box::new(std::io::Error::new(ErrorKind::Other, err)))?;
        let object_info = serde_json::from_slice::<ObjectInfo>(&decoded_payload)?;

        Ok(object_info)
    }

    /// Puts an [Object] into the [ObjectStore].
    /// This method implements `tokio::io::AsyncRead`.
    ///
    /// # Examples
    ///
    /// ```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(())
    /// # }
    /// ```
    pub async fn put<T>(
        &self,
        meta: T,
        data: &mut (impl tokio::io::AsyncRead + std::marker::Unpin),
    ) -> Result<ObjectInfo, Error>
    where
        ObjectMeta: From<T>,
    {
        let object_meta: ObjectMeta = meta.into();

        let encoded_object_name = encode_object_name(&object_meta.name);
        if !is_valid_object_name(&encoded_object_name) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid object name",
            )));
        }
        // Fetch any existing object info, if there is any for later use.
        let maybe_existing_object_info = match self.info(&encoded_object_name).await {
            Ok(object_info) => Some(object_info),
            Err(_) => None,
        };

        let object_nuid = nuid::next();
        let chunk_subject = format!("$O.{}.C.{}", &self.name, &object_nuid);

        let mut object_chunks = 0;
        let mut object_size = 0;

        let mut buffer = Box::new([0; DEFAULT_CHUNK_SIZE]);
        let mut context = ring::digest::Context::new(&SHA256);

        loop {
            let n = data.read(&mut *buffer).await?;

            if n == 0 {
                break;
            }
            context.update(&buffer[..n]);

            object_size += n;
            object_chunks += 1;

            // FIXME: this is ugly
            let payload = bytes::Bytes::from(buffer[..n].to_vec());

            self.stream
                .context
                .publish(chunk_subject.clone(), payload)
                .await?;
        }
        let digest = context.finish();
        let subject = format!("$O.{}.M.{}", &self.name, &encoded_object_name);
        let object_info = ObjectInfo {
            name: object_meta.name,
            description: object_meta.description,
            link: object_meta.link,
            bucket: self.name.clone(),
            nuid: object_nuid,
            chunks: object_chunks,
            size: object_size,
            digest: format!(
                "SHA-256={}",
                base64::encode_config(digest, base64::URL_SAFE)
            ),
            modified: OffsetDateTime::now_utc(),
            deleted: false,
        };

        let mut headers = HeaderMap::new();
        headers.insert(NATS_ROLLUP, ROLLUP_SUBJECT.parse::<HeaderValue>()?);
        let data = serde_json::to_vec(&object_info)?;

        // publish meta.
        self.stream
            .context
            .publish_with_headers(subject, headers, data.into())
            .await?;

        // Purge any old chunks.
        if let Some(existing_object_info) = maybe_existing_object_info {
            let chunk_subject = format!("$O.{}.C.{}", &self.name, &existing_object_info.nuid);

            self.stream.purge_subject(&chunk_subject).await?;
        }

        Ok(object_info)
    }

    /// Creates a [Watch] stream over changes in the [ObjectStore].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use futures::StreamExt;
    /// 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 watcher = bucket.watch().await.unwrap();
    /// while let Some(object) = watcher.next().await {
    ///     println!("detected changes in {:?}", object?);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn watch(&self) -> Result<Watch<'_>, Error> {
        let subject = format!("$O.{}.M.>", self.name);
        let ordered = self
            .stream
            .create_consumer(crate::jetstream::consumer::push::OrderedConfig {
                deliver_policy: super::consumer::DeliverPolicy::New,
                deliver_subject: self.stream.context.client.new_inbox(),
                description: Some("object store watcher".to_string()),
                filter_subject: subject,
                ..Default::default()
            })
            .await?;
        Ok(Watch {
            subscription: ordered.messages().await?,
        })
    }

    /// Returns a [List] stream with all not deleted [Objects][Object] in the [ObjectStore].
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use futures::StreamExt;
    /// 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 list = bucket.list().await.unwrap();
    /// while let Some(object) = list.next().await {
    ///     println!("object {:?}", object?);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn list(&self) -> Result<List<'_>, Error> {
        trace!("starting Object List");
        let subject = format!("$O.{}.M.>", self.name);
        let ordered = self
            .stream
            .create_consumer(crate::jetstream::consumer::push::OrderedConfig {
                deliver_policy: super::consumer::DeliverPolicy::All,
                deliver_subject: self.stream.context.client.new_inbox(),
                description: Some("object store list".to_string()),
                filter_subject: subject,
                ..Default::default()
            })
            .await?;
        Ok(List {
            done: ordered.info.num_pending == 0,
            subscription: ordered.messages().await?,
        })
    }

src/jetstream/kv/mod.rs (lines 396-403)

    pub async fn watch<T: AsRef<str>>(&self, key: T) -> Result<Watch<'_>, Error> {
        let subject = format!("{}{}", self.prefix.as_str(), key.as_ref());

        let consumer = self
            .stream
            .create_consumer(super::consumer::push::OrderedConfig {
                deliver_subject: self.stream.context.client.new_inbox(),
                description: Some("kv watch consumer".to_string()),
                filter_subject: subject,
                replay_policy: super::consumer::ReplayPolicy::Instant,
                deliver_policy: DeliverPolicy::New,
                ..Default::default()
            })
            .await?;

        Ok(Watch {
            subscription: consumer.messages().await?,
            prefix: self.prefix.clone(),
            bucket: self.name.clone(),
        })
    }

    /// Creates a [futures::Stream] over [Entries][Entry] for all keys, which yields
    /// values whenever there are changes in the bucket.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use futures::StreamExt;
    /// 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 mut entries = kv.watch_all().await?;
    /// while let Some(entry) = entries.next().await {
    ///     println!("entry: {:?}", entry);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn watch_all(&self) -> Result<Watch<'_>, Error> {
        self.watch(ALL_KEYS).await
    }

    pub async fn get<T: Into<String>>(&self, key: T) -> Result<Option<Vec<u8>>, Error> {
        match self.entry(key).await {
            Ok(Some(entry)) => match entry.operation {
                Operation::Put => Ok(Some(entry.value)),
                _ => Ok(None),
            },
            Ok(None) => Ok(None),
            Err(err) => Err(err),
        }
    }

    /// Updates a value for a given key, but only if passed `revision` is the last `revision` in
    /// the bucket.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use futures::StreamExt;
    /// 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 revision = kv.put("key", "value".into()).await?;
    /// kv.update("key", "updated".into(), revision).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn update<T: AsRef<str>>(
        &self,
        key: T,
        value: Bytes,
        revision: u64,
    ) -> Result<u64, Error> {
        if !is_valid_key(key.as_ref()) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid key",
            )));
        }
        let subject = format!("{}{}", self.prefix.as_str(), key.as_ref());

        let mut headers = crate::HeaderMap::default();
        headers.insert(
            header::NATS_EXPECTED_LAST_SUBJECT_SEQUENCE,
            HeaderValue::from(revision),
        );

        self.stream
            .context
            .publish_with_headers(subject, headers, value)
            .await?
            .await
            .map(|publish_ack| publish_ack.sequence)
    }

    /// Deletes a given key. This is a non-destructive operation, which sets a `DELETE` marker.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use futures::StreamExt;
    /// 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?;
    /// kv.put("key", "value".into()).await?;
    /// kv.delete("key").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn delete<T: AsRef<str>>(&self, key: T) -> Result<(), Error> {
        if !is_valid_key(key.as_ref()) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid key",
            )));
        }
        let mut subject = String::new();
        if self.use_jetstream_prefix {
            subject.push_str(&self.stream.context.prefix);
            subject.push('.');
        }
        subject.push_str(self.put_prefix.as_ref().unwrap_or(&self.prefix));
        subject.push_str(key.as_ref());

        let mut headers = crate::HeaderMap::default();
        // TODO: figure out which headers k/v should be where.
        headers.insert(KV_OPERATION, KV_OPERATION_DELETE.parse::<HeaderValue>()?);

        self.stream
            .context
            .publish_with_headers(subject, headers, "".into())
            .await?;
        Ok(())
    }

    /// Purges all the revisions of a entry destructively, leaving behind a single purge entry in-place.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use futures::StreamExt;
    /// 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?;
    /// kv.put("key", "value".into()).await?;
    /// kv.put("key", "another".into()).await?;
    /// kv.purge("key").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn purge<T: AsRef<str>>(&self, key: T) -> Result<(), Error> {
        if !is_valid_key(key.as_ref()) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid key",
            )));
        }

        let subject = format!("{}{}", self.prefix.as_str(), key.as_ref());

        let mut headers = crate::HeaderMap::default();
        headers.insert(KV_OPERATION, HeaderValue::from(KV_OPERATION_PURGE));
        headers.insert(NATS_ROLLUP, HeaderValue::from(ROLLUP_SUBJECT));

        self.stream
            .context
            .publish_with_headers(subject, headers, "".into())
            .await?;
        Ok(())
    }

    /// Returns a [futures::Stream] that allows iterating over all [Operations][Operation] that
    /// happen for given key.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use futures::StreamExt;
    /// 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 mut entries = kv.history("kv").await?;
    /// while let Some(entry) = entries.next().await {
    ///     println!("entry: {:?}", entry);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn history<T: AsRef<str>>(&self, key: T) -> Result<History<'_>, Error> {
        if !is_valid_key(key.as_ref()) {
            return Err(Box::new(io::Error::new(
                io::ErrorKind::InvalidInput,
                "invalid key",
            )));
        }
        let subject = format!("{}{}", self.prefix.as_str(), key.as_ref());

        let consumer = self
            .stream
            .create_consumer(super::consumer::push::OrderedConfig {
                deliver_subject: self.stream.context.client.new_inbox(),
                description: Some("kv history consumer".to_string()),
                filter_subject: subject,
                replay_policy: super::consumer::ReplayPolicy::Instant,
                ..Default::default()
            })
            .await?;

        Ok(History {
            subscription: consumer.messages().await?,
            done: false,
            prefix: self.prefix.clone(),
            bucket: self.name.clone(),
        })
    }

    /// Returns a [futures::Stream] that allows iterating over all keys in the bucket.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// # #[tokio::main]
    /// # async fn main() -> Result<(), async_nats::Error> {
    /// use futures::StreamExt;
    /// 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 mut entries = kv.keys().await?;
    /// while let Some(key) = entries.next() {
    ///     println!("key: {:?}", key);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn keys(&self) -> Result<collections::hash_set::IntoIter<String>, Error> {
        let subject = format!("{}>", self.prefix.as_str());

        let consumer = self
            .stream
            .create_consumer(super::consumer::push::OrderedConfig {
                deliver_subject: self.stream.context.client.new_inbox(),
                description: Some("kv history consumer".to_string()),
                filter_subject: subject,
                headers_only: true,
                replay_policy: super::consumer::ReplayPolicy::Instant,
                ..Default::default()
            })
            .await?;

        let mut entries = History {
            done: consumer.info.num_pending == 0,
            subscription: consumer.messages().await?,
            prefix: self.prefix.clone(),
            bucket: self.name.clone(),
        };

        let mut keys = HashSet::new();
        while let Some(entry) = entries.try_next().await? {
            keys.insert(entry.key);
        }
        Ok(keys.into_iter())
    }

src/jetstream/stream.rs (line 723)

    pub async fn get_or_create_consumer<T: FromConsumer + IntoConsumerConfig>(
        &self,
        name: &str,
        config: T,
    ) -> Result<Consumer<T>, Error> {
        let subject = format!("CONSUMER.INFO.{}.{}", self.info.config.name, name);

        match self.context.request(subject, &json!({})).await? {
            Response::Err { error } if error.status == 404 => self.create_consumer(config).await,
            Response::Err { error } => Err(Box::new(io::Error::new(
                ErrorKind::Other,
                format!(
                    "nats: error while getting or creating stream: {}, {}, {}",
                    error.code, error.status, error.description
                ),
            ))),
            Response::Ok::<consumer::Info>(info) => Ok(Consumer::new(
                T::try_from_consumer_config(info.config.clone())?,
                info,
                self.context.clone(),
            )),
        }
    }

pub async fn consumer_info<T: AsRef<str>>(&self, name: T) -> Result<Info, Error>

Retrieve Info about Consumer from the server.

Examples

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

let stream = jetstream.get_stream("events").await?;
let info = stream.consumer_info("pull").await?;

Examples found in repository

src/jetstream/stream.rs (line 684)

    pub async fn get_consumer<T: FromConsumer + IntoConsumerConfig>(
        &self,
        name: &str,
    ) -> Result<Consumer<T>, Error> {
        let info = self.consumer_info(name).await?;

        Ok(Consumer::new(
            T::try_from_consumer_config(info.config.clone())?,
            info,
            self.context.clone(),
        ))
    }

pub async fn get_consumer<T: FromConsumer + IntoConsumerConfig>(
    &self,
    name: &str
) -> Result<Consumer<T>, Error>

Get Consumer from the the server. Consumer iterators can be used to retrieve Messages for a given Consumer.

Examples

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

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

pub async fn get_or_create_consumer<T: FromConsumer + IntoConsumerConfig>(
    &self,
    name: &str,
    config: T
) -> Result<Consumer<T>, Error>

Create a Consumer with the given configuration if it is not present on the server. Returns a handle to the Consumer.

Note: This does not validate if the Consumer on the server is compatible with the configuration passed in except Push/Pull compatibility.

Examples

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

let stream = jetstream.get_stream("events").await?;
let consumer = stream.get_or_create_consumer("pull", consumer::pull::Config {
    durable_name: Some("pull".to_string()),
    ..Default::default()
}).await?;

pub async fn delete_consumer(&self, name: &str) -> Result<DeleteStatus, Error>

Delete a Consumer from the server.

Examples

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

jetstream.get_stream("events").await?
    .delete_consumer("pull").await?;

Trait Implementations

impl Clone for Stream

fn clone(&self) -> Stream

Returns a copy of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Stream

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.