Class: Cosmo::Client

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

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(nats_url: ENV.fetch("NATS_URL", "nats://localhost:4222")) ⇒ Client

Returns a new instance of Client.

Parameters:

  • nats_url: (::String) (defaults to: ENV.fetch("NATS_URL", "nats://localhost:4222")) 


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

def initialize(nats_url: ENV.fetch("NATS_URL", "nats://localhost:4222"))
  Logger.debug "Connecting to NATS server at #{nats_url}..."
  @nc = NATS.connect(nats_url)
  Logger.debug "Connection established"
  @js = @nc.jetstream
end

Instance Attribute Details

#js ⇒ Object (readonly)

Returns the value of attribute js.

Returns:

  • (Object) 


12
13
14
 
# File 'lib/cosmo/client.rb', line 12

def js
  @js
end

#nc ⇒ Object (readonly)

Returns the value of attribute nc.

Returns:

  • (Object) 


12
13
14
 
# File 'lib/cosmo/client.rb', line 12

def nc
  @nc
end

Class Method Details

.instance ⇒ Client

Returns:



8
9
10
 
# File 'lib/cosmo/client.rb', line 8

def self.instance
  @instance ||= Client.new
end

Instance Method Details

#close ⇒ void

This method returns an undefined value.



130
131
132
 
# File 'lib/cosmo/client.rb', line 130

def close
  nc.close
end

#consumer_info(stream_name, consumer_name) ⇒ NATS::JetStream::API::ConsumerInfo

Parameters:

  • stream_name (::String)
  • consumer_name (::String)

Returns:

  • (NATS::JetStream::API::ConsumerInfo) 


102
103
104
 
# File 'lib/cosmo/client.rb', line 102

def consumer_info(stream_name, consumer_name)
  js.consumer_info(stream_name, consumer_name)
end

#create_kv_with_msg_ttl(name, **options) ⇒ Object

NOTE: KV manager in nats-pure hardcodes the fields it copies into StreamConfig, so allow_msg_ttl is never forwarded via create_key_value. Send the raw stream-create API request instead.

Parameters:

  • name (::String)
  • opts (Object)

Returns:

  • (Object) 


138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/cosmo/client.rb', line 138

def create_kv_with_msg_ttl(name, **options)
  payload = Utils::Json.dump({
    name: "KV_#{name}",
    subjects: ["$KV.#{name}.>"],
    storage: "file",
    allow_direct: true,
    allow_msg_ttl: true,
    allow_rollup_hdrs: true,
    max_msgs_per_subject: 1
  }.merge(options))
  resp = nc.request("$JS.API.STREAM.CREATE.KV_#{name}", payload)
  result = Utils::Json.parse(resp.data, symbolize_names: false)
  if result&.dig("error")
    msg = result.dig("error", "description").to_s
    # Two worker processes starting simultaneously can both attempt creation.
    # If another process won the race, fall back to looking up the existing bucket.
    raise NATS::JetStream::Error, msg unless msg.match?(/already in use|already exists/i)
  end
  js.key_value(name)
end

#create_stream(name, config) ⇒ Object

Parameters:

  • name (::String, Symbol)
  • config (Hash[Symbol, untyped])

Returns:

  • (Object) 


33
34
35
 
# File 'lib/cosmo/client.rb', line 33

def create_stream(name, config)
  js.add_stream(name: name, **config)
end

#cron_subjects_in_stream(stream_name, filter) ⇒ Array<String>

Return all subjects in stream_name that match filter using NATS's subjects_filter on STREAM.INFO (requires NATS ≥ 2.9).

Parameters:

  • stream_name (::String)
  • filter (::String)

Returns:

  • (Array<String>) 


57
58
59
60
61
62
63
64
 
# File 'lib/cosmo/client.rb', line 57

def cron_subjects_in_stream(stream_name, filter)
  payload = Utils::Json.dump({ subjects_filter: filter })
  resp = nc.request("$JS.API.STREAM.INFO.#{stream_name}", payload)
  data = Utils::Json.parse(resp.data, symbolize_names: false)
  (data&.dig("state", "subjects") || {}).keys
rescue StandardError
  []
end

#delete_message(name, seq) ⇒ Hash[::String, untyped]

Parameters:

  • name (::String)
  • seq (::Integer)

Returns:

  • (Hash[::String, untyped]) 


110
111
112
113
 
# File 'lib/cosmo/client.rb', line 110

def delete_message(name, seq)
  response = nc.request("$JS.API.STREAM.MSG.DELETE.#{name}", JSON.dump({ seq: seq }))
  Utils::Json.parse(response.data, symbolize_names: false)
end

#delete_stream(name, params = {}) ⇒ Object

Parameters:

  • name (::String, Symbol)
  • params (Hash[Symbol, untyped]) (defaults to: {})

Returns:

  • (Object) 


37
38
39
 
# File 'lib/cosmo/client.rb', line 37

def delete_stream(name, params = {})
  js.delete_stream(name, params)
end

#get_message(stream_name, **options) ⇒ NATS::JetStream::API::RawStreamMsg

Parameters:

  • name (::String, Symbol)
  • options (Object)

Returns:

  • (NATS::JetStream::API::RawStreamMsg) 


106
107
108
 
# File 'lib/cosmo/client.rb', line 106

def get_message(stream_name, **options)
  js.get_msg(stream_name, **options)
end

#kv(name, allow_msg_ttl: false, **options) ⇒ Object

Parameters:

  • name (::String)
  • allow_msg_ttl: (Boolean) (defaults to: false)
  • opts (Object)

Returns:

  • (Object) 


124
125
126
127
128
 
# File 'lib/cosmo/client.rb', line 124

def kv(name, allow_msg_ttl: false, **options)
  js.key_value(name)
rescue NATS::KeyValue::BucketNotFoundError
  allow_msg_ttl ? create_kv_with_msg_ttl(name, **options) : js.create_key_value({ bucket: name }.merge(options))
end

#list_consumers(stream_name) ⇒ Array[Hash[::String, untyped]]

Parameters:

  • stream_name (::String)

Returns:

  • (Array[Hash[::String, untyped]]) 


96
97
98
99
100
 
# File 'lib/cosmo/client.rb', line 96

def list_consumers(stream_name)
  response = nc.request("$JS.API.CONSUMER.LIST.#{stream_name}", "")
  data = Utils::Json.parse(response.data, default: {}, symbolize_names: false)
  Array(data["consumers"])
end

#list_streams ⇒ Array[Hash[::String, untyped]]

Returns:

  • (Array[Hash[::String, untyped]]) 


66
67
68
69
70
71
72
73
74
 
# File 'lib/cosmo/client.rb', line 66

def list_streams
  response = nc.request("$JS.API.STREAM.LIST", "")
  data = Utils::Json.parse(response.data, symbolize_names: false)
  return [] if data.nil? || data["streams"].nil?

  data["streams"]
rescue NATS::Error
  []
end

#pause_stream(name) ⇒ void

This method returns an undefined value.

Parameters:

  • name (::String) 


76
77
78
79
80
81
 
# File 'lib/cosmo/client.rb', line 76

def pause_stream(name)
  config = stream_info(name).config.to_h
  config[:metadata] ||= {}
  config[:metadata][:"_cosmo.paused"] = "true"
  update_stream(name, config)
end

#publish(subject, payload, **params) ⇒ Object

Parameters:

  • subject (::String)
  • payload (::String)
  • params (Object)

Returns:

  • (Object) 


21
22
23
 
# File 'lib/cosmo/client.rb', line 21

def publish(subject, payload, **params)
  js.publish(subject, payload, **params)
end

#purge(stream_name, subject) ⇒ ::Integer?

Parameters:

  • stream_name (::String)
  • subject (::String, nil)

Returns:

  • (::Integer, nil)

Raises:

  • (NATS::JetStream::Error) 


115
116
117
118
119
120
121
122
 
# File 'lib/cosmo/client.rb', line 115

def purge(stream_name, subject)
  payload = subject ? Utils::Json.dump({ filter: subject }) : ""
  response = @nc.request("$JS.API.STREAM.PURGE.#{stream_name}", payload)
  result = Utils::Json.parse(response.data, default: {}, symbolize_names: false)
  raise NATS::JetStream::Error, result.dig("error", "description") if result["error"]

  result["purged"] # number of messages purged
end

#setup_stream(name, config) ⇒ Object

Create/update a stream, falling back to create when there's no stream.

Parameters:

  • name (String) —

    Stream name

  • config (Hash) —

    Full desired stream configuration

Returns:

  • (Object) 


48
49
50
51
52
 
# File 'lib/cosmo/client.rb', line 48

def setup_stream(name, config)
  update_stream(name, config)
rescue NATS::JetStream::Error::StreamNotFound
  create_stream(name, config)
end

#stream_info(name) ⇒ Object

Parameters:

  • name (::String, Symbol)

Returns:

  • (Object) 


29
30
31
 
# File 'lib/cosmo/client.rb', line 29

def stream_info(name)
  js.stream_info(name)
end

#stream_paused?(name) ⇒ Boolean

Parameters:

  • name (::String)

Returns:

  • (Boolean) 


90
91
92
93
94
 
# File 'lib/cosmo/client.rb', line 90

def stream_paused?(name)
  stream_info(name).config.metadata&.[](:"_cosmo.paused") == "true"
rescue NATS::IO::Timeout
  false
end

#subscribe(subject, consumer_name, config) ⇒ Object

Parameters:

  • subject (::String, Array[::String])
  • consumer_name (::String)
  • config (Hash[Symbol, untyped])

Returns:

  • (Object) 


25
26
27
 
# File 'lib/cosmo/client.rb', line 25

def subscribe(subject, consumer_name, config)
  js.pull_subscribe(subject, consumer_name, config: config)
end

#unpause_stream(name) ⇒ void

This method returns an undefined value.

Parameters:

  • name (::String) 


83
84
85
86
87
88
 
# File 'lib/cosmo/client.rb', line 83

def unpause_stream(name)
  config = stream_info(name).config.to_h
  config[:metadata] ||= {}
  config[:metadata].delete(:"_cosmo.paused")
  update_stream(name, config)
end

#update_stream(name, config) ⇒ Object

Parameters:

  • name (::String, Symbol)
  • config (Hash[Symbol, untyped])

Returns:

  • (Object) 


41
42
43
 
# File 'lib/cosmo/client.rb', line 41

def update_stream(name, config)
  js.update_stream(name: name, **config)
end