Class: Cosmo::API::KV

Inherits:
Object
  • Object
show all
Defined in:
lib/cosmo/api/kv.rb,
sig/cosmo/api/kv.rbs

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(name, options = nil) ⇒ KV

Returns a new instance of KV.

Parameters:

  • name (::String)
  • options (Hash[Symbol, untyped], nil) (defaults to: nil) 


8
9
10
11
12
 
# File 'lib/cosmo/api/kv.rb', line 8

def initialize(name, options = nil)
  @name = name
  @options = Hash(options)
  @kv = client.kv(@name, **@options)
end

Instance Attribute Details

#kv ⇒ Object (readonly)

Returns the value of attribute kv.

Returns:

  • (Object) 


6
7
8
 
# File 'lib/cosmo/api/kv.rb', line 6

def kv
  @kv
end

Instance Method Details

#clean ⇒ Object

Returns:

  • (Object) 


61
62
63
 
# File 'lib/cosmo/api/kv.rb', line 61

def clean
  client.purge("KV_#{@name}", ">")
end

#client ⇒ Client

Returns:



88
89
90
 
# File 'lib/cosmo/api/kv.rb', line 88

def client
  Client.instance
end

#count ⇒ ::Integer Also known as: size

Returns:

  • (::Integer) 


65
66
67
68
69
 
# File 'lib/cosmo/api/kv.rb', line 65

def count
  keys.size
rescue NATS::KeyValue::NoKeysFoundError, NATS::JetStream::Error::NotFound
  0
end

#delete(key) ⇒ Object

Writes a KV-Operation tombstone. On a ttl-bearing bucket this leaves the subject occupied, so a subsequent #set(ttl:) CAS with last_seq: 0 will keep failing -- use #erase on those buckets instead.

Parameters:

  • key (::String, ::Integer)

Returns:

  • (Object) 


31
32
33
 
# File 'lib/cosmo/api/kv.rb', line 31

def delete(key)
  kv.delete(key)
end

#erase(key) ⇒ ::Integer?

Removes key leaving no trace at all -- unlike #delete/#purge, which write a KV-Operation tombstone message. Mirrors how per-message Nats-TTL expiry removes a key, so callers never have to distinguish "deleted" from "TTL-expired" on read.

Parameters:

  • key (::String, ::Integer)

Returns:

  • (::Integer, nil) 


57
58
59
 
# File 'lib/cosmo/api/kv.rb', line 57

def erase(key)
  client.purge("KV_#{@name}", "$KV.#{@name}.#{key}")
end

#get(key) ⇒ Object

Parameters:

  • key (::String, ::Integer)

Returns:

  • (Object) 


22
23
24
25
26
 
# File 'lib/cosmo/api/kv.rb', line 22

def get(key)
  kv.get(key)
rescue NATS::KeyValue::KeyNotFoundError
  # nop
end

#keys(subject = nil, limit: 25) ⇒ Array[::String]

Parameters:

  • subject (::String, nil) (defaults to: nil)
  • limit: (::Integer) (defaults to: 25)

Returns:

  • (Array[::String]) 


35
36
37
38
39
40
41
42
43
44
45
46
 
# File 'lib/cosmo/api/kv.rb', line 35

def keys(subject = nil, limit: 25)
  results = []
  watcher = kv.watch(subject || ">", ignore_deletes: true, meta_only: true)
  watcher.each do |entry|
    break unless entry

    results << entry.key
    break if results.size >= limit
  end
  watcher.stop
  results
end

#publish_cas(key, value, ttl, last_seq:) ⇒ Object

CAS = Compare-And-Swap: publish value with a per-message Nats-TTL, but only if the subject's current last sequence matches last_seq (sent as the Nats-Expected-Last-Subject-Sequence header). Raises NATS::KeyValue::KeyWrongLastSequenceError if it doesn't match -- e.g. last_seq: 0 means "only publish if nothing exists here yet".

Parameters:

  • key (::String, ::Integer)
  • value (::String)
  • ttl (::Integer, nil)
  • last_seq: (::Integer)

Returns:

  • (Object) 


79
80
81
82
83
84
85
86
 
# File 'lib/cosmo/api/kv.rb', line 79

def publish_cas(key, value, ttl, last_seq:)
  headers = { "Nats-Expected-Last-Subject-Sequence" => last_seq.to_s, "Nats-TTL" => "#{ttl.to_i}s" }
  client.js.publish("$KV.#{@name}.#{key}", value, header: headers)
rescue NATS::JetStream::Error::APIError => e
  raise NATS::KeyValue::KeyWrongLastSequenceError, e.description if e.err_code == 10_071

  raise
end

#purge(key) ⇒ Object

Writes a KV-Operation tombstone (same issue as #delete on ttl buckets).

Parameters:

  • key (::String, ::Integer)

Returns:

  • (Object) 


49
50
51
 
# File 'lib/cosmo/api/kv.rb', line 49

def purge(key)
  kv.purge(key)
end

#set(key, value, ttl: nil) ⇒ Object

Parameters:

  • key (::String, ::Integer)
  • value (Object)
  • ttl: (::Integer, nil) (defaults to: nil)

Returns:

  • (Object) 


14
15
16
17
18
19
20
 
# File 'lib/cosmo/api/kv.rb', line 14

def set(key, value, ttl: nil)
  return kv.put(key, value.to_s) unless ttl

  # Pass ttl: (seconds) to set per-message expiry.
  # Raises `NATS::KeyValue::KeyWrongLastSequenceError` when the key is live.
  publish_cas(key, value.to_s, ttl, last_seq: 0).seq
end