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

    This class integrates with natsy's codec system to provide automatic serialization/deserialization of values and uses Kotlin's Result type for error handling.

    Example:

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

// Put and get typed values
kv.put("user:123:theme", "dark").getOrThrow()
val theme = kv.get<String>("user:123:theme").getOrThrow()

// Watch for changes
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> Result<Long> put(String key, T value) Put a typed value for a key
      final <T extends Any> Result<T> get(String key) Get a typed value for a key
      final <T extends Any> Result<T> get(String key, Long revision) Get a specific revision of a typed value
      final <T extends Any> Result<Long> create(String key, T value) Create a new key-value pair only if the key doesn't exist
      final <T extends Any> Result<Long> update(String key, T value, Long expectedRevision) Update a key only if the expected revision matches
      final Result<Unit> delete(String key) Delete a key (soft delete - places a delete marker)
      final Result<Unit> delete(String key, Long expectedRevision) Delete a key only if the expected revision matches
      final Result<Unit> purge(String key) Purge all history for a key (hard delete)
      final Result<Unit> purge(String key, Long expectedRevision) Purge a key only if the expected revision matches
      final Flow<String> keys() Get all keys in the bucket
      final Flow<String> keys(String filter) Get keys matching a filter pattern
      final <T extends Any> Flow<KeyValueEntry<T>> watch(String key) 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() 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) Get the history of a key
      final Result<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> Result<Long> put(String key, T value)

        Put a typed value for a key

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

        Result containing the revision number on success

      • get

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

        Get a typed value for a key

        Parameters:
        key - The key
        Returns:

        Result containing the value or null if not found

      • get

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

        Get a specific revision of a typed value

        Parameters:
        key - The key
        revision - The revision number
        Returns:

        Result containing the value at that revision or null if not found

      • create

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

        Create a new key-value pair only if the key doesn't exist

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

        Result containing the revision number on success

      • update

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

        Update a key only if the expected revision matches

        Parameters:
        key - The key
        value - The value to store
        expectedRevision - The expected current revision
        Returns:

        Result containing the new revision number on success

      • delete

         final Result<Unit> delete(String key)

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

        Parameters:
        key - The key to delete
        Returns:

        Result indicating success or failure

      • delete

         final Result<Unit> delete(String key, Long expectedRevision)

        Delete a key only if the expected revision matches

        Parameters:
        key - The key to delete
        expectedRevision - The expected current revision
        Returns:

        Result indicating success or failure

      • purge

         final Result<Unit> purge(String key)

        Purge all history for a key (hard delete)

        Parameters:
        key - The key to purge
        Returns:

        Result indicating success or failure

      • purge

         final Result<Unit> purge(String key, Long expectedRevision)

        Purge a key only if the expected revision matches

        Parameters:
        key - The key to purge
        expectedRevision - The expected current revision
        Returns:

        Result indicating success or failure

      • keys

         final Flow<String> keys()

        Get all keys in the bucket

        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)

        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.
        Returns:

        Flow of typed KeyValueEntry objects

      • watchAll

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

        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.

        Returns:

        Flow of typed KeyValueEntry objects for all keys

      • history

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

        Get the history of a key

        Parameters:
        key - The key
        Returns:

        Flow of typed KeyValueEntry objects representing the history

      • status

         final Result<KeyValueStatus> status()

        Get the status of the bucket

        Returns:

        Result containing the KeyValueStatus