Package eu.vstoyanov.natsy.jetstream

Class ObjectStore

  • All Implemented Interfaces:
    
public final class ObjectStore

    Wrapper around NATS ObjectStore that provides coroutine-friendly operations

    Large objects stream: putFile reads from disk a chunk at a time, getFile writes to disk the same way, and get writes into any OutputStream the caller hands it. put and getBytes are the whole-object-in-memory pair, for payloads small enough that this is the simpler thing.

    Absence is spelled two ways, deliberately. info answers null for an object that is not there, because asking about one that may not exist is the normal thing to do. get, getBytes, getFile and delete name an object they are meant to act on, so a missing one is NatsyNotFoundException rather than a null to unpack.

    Every download is checked against the metadata it was stored with; a mismatch is NatsyIntegrityException.

    Example:

    val os = js.objectStore("file-storage") {
    description = "Application file storage"
    storageType = StorageType.FILE
    maxBucketSize = 10 * 1024 * 1024 * 1024L // 10GB
}

os.put("report.pdf", pdfData)
val data = os.getBytes("report.pdf")

os.putFile(File("/path/to/backup.zip"))
os.getFile("backup.zip", File("/path/to/restore.zip"))

    • Nested Class Summary

      Nested Classes 
      Modifier and Type Class Description

    • Field Summary

      Fields 
      Modifier and Type Field Description
      private final String bucketName

    • Constructor Summary

      Constructors 
      Constructor Description

    • Enum Constant Summary

      Enum Constants 
      Enum Constant Description

    • Constructor Detail

    • Method Detail

      • put

         final ObjectStoreInfo put(String name, ByteArray data, ObjectMetadata metadata)

        Put an object with byte array data

        Parameters:
        name - The name of the object
        data - The data to store
        metadata - Optional metadata for the object
        Returns:

        information about the stored object

      • putFile

         final ObjectStoreInfo putFile(File file, String name, ObjectMetadata metadata)

        Put an object from a file

        Streams the file content to the object store in chunks, avoiding loading the entire file into memory.

        Parameters:
        file - The file to upload
        name - The name to store it under; the file's own name by default
        metadata - Optional metadata for the object
        Returns:

        information about the stored object

      • get

         final ObjectStoreInfo get(String name, OutputStream sink)

        Stream an object into sink

        The object is written chunk by chunk as it arrives, so nothing larger than one chunk is held: this is the read a Ktor respondOutputStream route wants. It forwards to jnats' get(objectName, OutputStream) (io/nats/client/ObjectStore.java:90).

        sink is written before the object is verified. jnats writes every chunk and only then checks the chunk count, the size and the digest (io/nats/client/impl/NatsObjectStore.java:224, :246 before :261-264), so a NatsyIntegrityException arrives with part or all of the object already in sink - bytes that cannot be taken back once they are on a socket. A caller that has to be sure before it commits to them reads with getBytes or getFile instead. Only the two failures jnats raises from its own lookup - NatsyNotFoundException, and the IllegalArgumentException for a link to a whole bucket (io/nats/client/impl/NatsObjectStore.java:188, :194) - arrive with sink untouched.

        sink is never closed, and is flushed only on success, exactly as jnats leaves it (io/nats/client/impl/NatsObjectStore.java:266) - closing a response stream early is how a download gets truncated. A failure of sink itself surfaces as a eu.vstoyanov.natsy.exception.NatsyConnectionException, since natsy cannot tell a broken sink from a broken connection.

        Parameters:
        name - The name of the object
        sink - Where to write it; left open, and flushed only if the download verified
        Returns:

        the object's metadata

      • getBytes

         final ByteArray getBytes(String name)

        Read a whole object into memory

        The object is materialised: a ByteArray cannot hold more than 2 GiB, and one of any size is one the heap has to carry. For anything large, get streams into a sink of the caller's choosing and getFile streams to disk.

        Unlike get, nothing is handed to the caller until the download has been verified.

        Parameters:
        name - The name of the object
        Returns:

        the object's bytes

      • getFile

         final File getFile(String name, File file)

        Get an object and write it to a file

        Streams the object data to disk rather than loading it into memory, and puts it at file only once the store has verified it. file is left untouched if anything goes wrong - including an integrity failure, where the bytes that arrived are exactly the ones not to keep. A previous copy at that path survives a failed download.

        Parameters:
        name - The name of the object
        file - The file to write to; replaced only on success
        Returns:

        file, for chaining

      • delete

         final ObjectStoreInfo delete(String name)

        Delete an object from the store

        This marks the object as deleted. The actual data may be retained for historical purposes depending on the bucket configuration.

        Parameters:
        name - The name of the object to delete
        Returns:

        information about the deleted object

      • info

         final ObjectStoreInfo info(String name)

        Get information about an object

        Parameters:
        name - The name of the object
        Returns:

        the object's metadata, or null if there is no such object, or it has been deleted

      • list

         final Flow<ObjectStoreInfo> list()

        List the live objects in the bucket

        Returns a Flow that emits ObjectStoreInfo for each object. Deleted objects are not listed - jnats filters them out - so watch is what reports a deletion.

        A failure reaches the collector rather than completing the flow: an empty result means the bucket is empty, and nothing else.

        Returns:

        Flow of ObjectStoreInfo

      • watch

         final Flow<ObjectStoreInfo> watch(Set<ObjectWatchOption> options)

        Watch for changes to objects in the bucket

        Opens by replaying the latest revision of every object - the bucket as it stands - and then emits an ObjectStoreInfo on every change. Deletions are part of that: they arrive as an entry whose ObjectStoreInfo.deleted is true, which is what list leaves out. ObjectWatchOption is how to start somewhere else, and watchEvents is what says where the opening replay ends.

        The watch buffer is unbounded so no change is dropped on its way to the collector; the server does not redeliver watch notifications.

        Parameters:
        options - Where the watch starts and what it leaves out; see ObjectWatchOption
        Returns:

        Flow of ObjectStoreInfo for all changes

      • status

         final ObjectStoreStatus status()

        Get the status of the bucket

        Returns:

        the bucket status