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
}

// Keys are subject-shaped: letters, digits and - . / = _ only, dot-separated. A key with a
// character outside that set - a colon, say - is rejected by an IllegalArgumentException.
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}")
}

    • 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, was deleted or purged. A live key whose payload is empty is present: put(key, "") reads back as "", not as null

      • 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. An empty payload is a value, the same way it is on get

      • 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, Set<KvWatchOption> options, Long fromRevision, DecodeErrorStrategy onDecodeError)

        Watch for changes to a specific key

        The watch opens by replaying the latest revision of every matching key - not the whole history (io/nats/client/impl/NatsKeyValueWatchSubscription.java:41) - and then emits a KeyValueEntry on every change. KvWatchOption.INCLUDE_HISTORY and KvWatchOption.UPDATES_ONLY are the two ways to start somewhere else; watchEvents is what says where that opening replay ends.

        The watch buffer is unbounded: a missed revision means a permanently stale view, since the server never redelivers it. On a watch of one concrete key .conflate() bounds that memory without stalling the watcher - it fuses into the watch buffer, and where there is one key the latest element and the latest value are the same entry. It is the wrong operator as soon as a second key can arrive, including under the wildcards this overload accepts.

        Parameters:
        key - The key to watch (can include wildcards, e.g., "users.
        options - Where the watch starts and how much of each revision it carries, see KvWatchOption
        fromRevision - Start at this revision instead of at the default snapshot, replaying from it inclusive
        onDecodeError - What to do with a revision whose value cannot be decoded, see DecodeErrorStrategy.
        Returns:

        Flow of typed KeyValueEntry objects

      • watch

         final <T extends Any> Flow<KeyValueEntry<T>> watch(List<String> keys, Set<KvWatchOption> options, Long fromRevision, DecodeErrorStrategy onDecodeError)

        Watch several keys, or key patterns, through one subscription

        One consumer covers them all, so the entries arrive interleaved in server order rather than as several independent flows. More than one key needs a server at 2.10.0 or later, which is where filtering a consumer on several subjects arrived (io/nats/client/impl/NatsJetStreamImpl.java:99-103).

        .conflate() is not the operator for this flow: it keeps the latest element whatever key it belongs to, so another key's revision is what it drops.

        Parameters:
        keys - The keys to watch; each may include wildcards, and the list may not be empty
        options - Where the watch starts and how much of each revision it carries, see KvWatchOption
        fromRevision - Start at this revision instead of at the default snapshot, replaying from it inclusive
        onDecodeError - What to do with a revision whose value cannot be decoded, see watch
        Returns:

        Flow of typed KeyValueEntry objects for every key named

      • watchAll

         final <T extends Any> Flow<KeyValueEntry<T>> watchAll(Set<KvWatchOption> options, Long fromRevision, 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. Only a .buffer(n, DROP_OLDEST) - or another non-SUSPEND policy - bounds it, because a plain .buffer(n) fuses by taking the larger capacity and so cannot shrink an unbounded one. .conflate() is the wrong operator for a whole bucket either way: it keeps the latest element whatever key it belongs to.

        Parameters:
        options - Where the watch starts and how much of each revision it carries, see KvWatchOption
        fromRevision - Start at this revision instead of at the default snapshot, replaying from it inclusive
        onDecodeError - What to do with a revision whose value cannot be decoded, see watch
        Returns:

        Flow of typed KeyValueEntry objects for all keys

      • watchEvents

         final <T extends Any> Flow<KvWatchEvent<T>> watchEvents(List<String> keys, Set<KvWatchOption> options, Long fromRevision, DecodeErrorStrategy onDecodeError)

        Watch, and be told when the opening replay is over

        The same subscription watch and watchAll open, with one more element in it: every revision arrives as KvWatchEvent.Entry, and KvWatchEvent.InitialSyncComplete marks the point where the bucket's existing contents end and live changes begin. That is what makes the warm-a-cache-then-serve pattern expressible - fill from the entries, flip the readiness probe on the signal, apply what follows as updates - and it is the only place natsy surfaces jnats' endOfData (io/nats/client/api/Watcher.java:30-34).

        Do not .conflate() this flow: the signal is an element like any other, and conflation can drop the very one the flow exists for.

        Parameters:
        keys - The keys to watch, each of which may include wildcards; null, the default, watches the whole bucket as watchAll does
        options - Where the watch starts and how much of each revision it carries, see KvWatchOption
        fromRevision - Start at this revision instead of at the default snapshot, replaying from it inclusive
        onDecodeError - What to do with a revision whose value cannot be decoded, see watch
        Returns:

        Flow of KvWatchEvent, entries and the one initial-sync signal

      • 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