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.instance.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) 


73
74
75
 
# File 'lib/cosmo/api/kv.rb', line 73

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

#count ⇒ ::Integer Also known as: size

Returns:

  • (::Integer) 


77
78
79
80
81
 
# File 'lib/cosmo/api/kv.rb', line 77

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

#delete(key) ⇒ Object

Parameters:

  • key (::String, ::Integer)

Returns:

  • (Object) 


52
53
54
 
# File 'lib/cosmo/api/kv.rb', line 52

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

#get(key) ⇒ Object

Parameters:

  • key (::String, ::Integer)

Returns:

  • (Object) 


46
47
48
49
50
 
# File 'lib/cosmo/api/kv.rb', line 46

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]) 


56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/cosmo/api/kv.rb', line 56

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

#purge(key) ⇒ Object

Parameters:

  • key (::String, ::Integer)

Returns:

  • (Object) 


69
70
71
 
# File 'lib/cosmo/api/kv.rb', line 69

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

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

rubocop:disable Metrics/AbcSize, Metrics/MethodLength

Parameters:

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

Returns:

  • (Object) 


14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/cosmo/api/kv.rb', line 14

def set(key, value, ttl: nil) # rubocop:disable Metrics/AbcSize, Metrics/MethodLength
  return kv.put(key, value.to_s) unless ttl

  # Pass ttl: (seconds) to set a per-message expiry.
  # Raises `NATS::KeyValue::KeyWrongLastSequenceError` when the key is live.
  begin
    value = value.to_s
    put = lambda do |last_seq:|
      headers = { "Nats-Expected-Last-Subject-Sequence" => last_seq.to_s, "Nats-TTL" => "#{ttl.to_i}s" }
      Client.instance.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

    put.call(last_seq: 0)
    kv.send(:_get, key) # fetch the created entry to get its revision
  rescue NATS::KeyValue::KeyWrongLastSequenceError
    # `kv.get` converts KeyDeletedError → KeyNotFoundError, hiding tombstone info.
    # Use private _get instead — it raises KeyDeletedError with the entry's revision
    begin
      kv.send(:_get, key)
    rescue NATS::KeyValue::KeyDeletedError => e
      put.call(last_seq: e.entry.revision)
      return kv.send(:_get, key)
    end

    raise
  end
end