Struct async_nats::jetstream::object_store::ObjectStore

pub struct ObjectStore { /* private fields */ }

A blob store capable of storing large objects efficiently in streams.

Implementations

impl ObjectStore

pub async fn get<T: AsRef<str>>(
    &self,
    object_name: T
) -> Result<Object<'_>, Error>

Gets an Object from the ObjectStore.

Object implements tokio::io::AsyncRead that allows to read the data from Object Store.

Examples

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

pub async fn delete<T: AsRef<str>>(&self, object_name: T) -> Result<(), Error>

Gets an Object from the ObjectStore.

Object implements tokio::io::AsyncRead that allows to read the data from Object Store.

Examples

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

pub async fn info<T: AsRef<str>>(
    &self,
    object_name: T
) -> Result<ObjectInfo, Error>

Retrieves Object ObjectInfo.

Examples

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

Examples found in repository

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

    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)
    }

pub async fn put<T>(
    &self,
    meta: T,
    data: &mut impl AsyncRead + Unpin
) -> Result<ObjectInfo, Error>where
    ObjectMeta: From<T>,

Puts an Object into the ObjectStore. This method implements tokio::io::AsyncRead.

Examples

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();

pub async fn watch(&self) -> Result<Watch<'_>, Error>

Creates a Watch stream over changes in the ObjectStore.

Examples

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

pub async fn list(&self) -> Result<List<'_>, Error>

Returns a List stream with all not deleted Objects in the ObjectStore.

Examples

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

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

Seals a ObjectStore, preventing any further changes to it or its Objects.

Examples

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

let mut bucket = jetstream.get_object_store("store").await?;
bucket.seal().await.unwrap();

Trait Implementations

impl Clone for ObjectStore

fn clone(&self) -> ObjectStore

Returns a copy of the value.

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

Performs copy-assignment from source.