Package eu.vstoyanov.natsy.jetstream

Class KeyValueStore

  • All Implemented Interfaces:
    
public final class KeyValueStore

    Wrapper around NATS KeyValue that provides type-safe, coroutine-friendly operations

    Values are serialized through natsy's codec registry. Every operation reports failure by throwing a eu.vstoyanov.natsy.exception.NatsyException; a null return means the key is absent, never that the read failed.

    Example:

    val kv = js.keyValue("user-prefs") {
    maxHistoryPerKey = 5
    ttl = 24.hours
}

kv.put("user:123:theme", "dark")
val theme = kv.get<String>("user:123:theme") ?: "light"

kv.watch<String>("user:123:*").collect { entry ->
    println("${entry.key} changed to ${entry.value}")
}

    • 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

    • Method Summary

      Modifier and Type Method Description
      final String getBucketName() Get the name of the bucket
      final <T extends Any> Long put(String key, T value) Put a typed value for a key
      final <T extends Any> T get(String key) Get a typed value for a key
      final <T extends Any> T get(String key, Long revision) Get a specific revision of a typed value
      final <T extends Any> Long create(String key, T value) Create a new key only if it does not already exist
      final <T extends Any> Long update(String key, T value, Long expectedRevision) Update a key only if the expected revision matchesThe compare-and-swap half of the store: expectedRevision not matching is a normal outcome of a read-modify-write loop, reported as eu.vstoyanov.natsy.exception.NatsyConflictException and not logged - the caller re-reads and retries.
      final Unit delete(String key) Delete a key (soft delete - places a delete marker)
      final Unit delete(String key, Long expectedRevision) Delete a key only if the expected revision matches
      final Unit purge(String key) Purge all history for a key (hard delete)
      final Unit purge(String key, Long expectedRevision) Purge a key only if the expected revision matches
      final Flow<String> keys() Get all keys in the bucketA failure reaches the collector rather than completing the flow: an empty result means the bucket is empty, and nothing else.
      final Flow<String> keys(String filter) Get keys matching a filter pattern
      final <T extends Any> Flow<KeyValueEntry<T>> watch(String key, DecodeErrorStrategy onDecodeError) Watch for changes to a specific keyReturns a Flow that emits KeyValueEntry objects whenever the key changes.
      final <T extends Any> Flow<KeyValueEntry<T>> watchAll(DecodeErrorStrategy onDecodeError) Watch for changes to all keys in the bucketThe watch buffer is unbounded so no revision is dropped on its way to the collector.
      final <T extends Any> Flow<KeyValueEntry<T>> history(String key, DecodeErrorStrategy onDecodeError) Get the history of a key
      final KeyValueStatus status() Get the status of the bucket

      • Methods inherited from class java.lang.Object

        clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

    • Constructor Detail

    • Method Detail

      • put

         final <T extends Any> Long put(String key, T value)

        Put a typed value for a key

        Parameters:
        key - The key
        value - The value to store
        Returns:

        the revision the value was stored at

      • get

         final <T extends Any> T get(String key)

        Get a typed value for a key

        Parameters:
        key - The key
        Returns:

        the value, or null if the key does not exist or was deleted or purged

      • get

         final <T extends Any> T get(String key, Long revision)

        Get a specific revision of a typed value

        Parameters:
        key - The key
        revision - The revision to read
        Returns:

        the value at that revision, or null if it does not exist or is not a live value

      • create

         final <T extends Any> Long create(String key, T value)

        Create a new key only if it does not already exist

        Parameters:
        key - The key
        value - The value to store
        Returns:

        the revision the value was stored at

      • update

         final <T extends Any> Long update(String key, T value, Long expectedRevision)

        Update a key only if the expected revision matches

        The compare-and-swap half of the store: expectedRevision not matching is a normal outcome of a read-modify-write loop, reported as eu.vstoyanov.natsy.exception.NatsyConflictException and not logged - the caller re-reads and retries.

        Parameters:
        key - The key
        value - The value to store
        expectedRevision - The revision the caller believes is current
        Returns:

        the new revision

      • delete

         final Unit delete(String key)

        Delete a key (soft delete - places a delete marker)

        Parameters:
        key - The key to delete

      • delete

         final Unit delete(String key, Long expectedRevision)

        Delete a key only if the expected revision matches

        Parameters:
        key - The key to delete
        expectedRevision - The revision the caller believes is current

      • purge

         final Unit purge(String key)

        Purge all history for a key (hard delete)

        Parameters:
        key - The key to purge

      • purge

         final Unit purge(String key, Long expectedRevision)

        Purge a key only if the expected revision matches

        Parameters:
        key - The key to purge
        expectedRevision - The revision the caller believes is current

      • keys

         final Flow<String> keys()

        Get all keys in the bucket

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

        Returns:

        Flow of key names

      • keys

         final Flow<String> keys(String filter)

        Get keys matching a filter pattern

        Parameters:
        filter - Subject-like filter (e.g., "users.*" or "users.>")
        Returns:

        Flow of key names matching the filter

      • watch

         final <T extends Any> Flow<KeyValueEntry<T>> watch(String key, DecodeErrorStrategy onDecodeError)

        Watch for changes to a specific key

        Returns a Flow that emits KeyValueEntry objects whenever the key changes.

        The watch buffer is unbounded: a missed revision means a permanently stale view, since the server never redelivers it. Apply .conflate() to the returned Flow when only the latest value per key matters - it fuses into the watch buffer and bounds memory without stalling the watcher.

        Parameters:
        key - The key to watch (can include wildcards, e.g., "users.
        onDecodeError - What to do with a revision whose value cannot be decoded, see DecodeErrorStrategy.
        Returns:

        Flow of typed KeyValueEntry objects

      • watchAll

         final <T extends Any> Flow<KeyValueEntry<T>> watchAll(DecodeErrorStrategy onDecodeError)

        Watch for changes to all keys in the bucket

        The watch buffer is unbounded so no revision is dropped on its way to the collector. See watch for the .conflate() alternative when only the latest value per key matters.

        Parameters:
        onDecodeError - What to do with a revision whose value cannot be decoded, see watch
        Returns:

        Flow of typed KeyValueEntry objects for all keys

      • history

         final <T extends Any> Flow<KeyValueEntry<T>> history(String key, DecodeErrorStrategy onDecodeError)

        Get the history of a key

        Parameters:
        key - The key
        onDecodeError - What to do with a revision whose value cannot be decoded, see watch
        Returns:

        Flow of typed KeyValueEntry objects representing the history

      • status

         final KeyValueStatus status()

        Get the status of the bucket

        Returns:

        the bucket status