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.



139
140
141
 
# File 'lib/cosmo/client.rb', line 139

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) 


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

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) 


173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/cosmo/client.rb', line 173

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) 


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

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


62
63
64
65
66
67
68
69
 
# File 'lib/cosmo/client.rb', line 62

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_consumer(stream_name, consumer_name) ⇒ Boolean

Parameters:

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

Returns:

  • (Boolean) 


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

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

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

Parameters:

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

Returns:

  • (Hash[::String, untyped]) 


119
120
121
122
 
# File 'lib/cosmo/client.rb', line 119

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) 


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

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

#ephemeral_subscribe(subject, config) ⇒ Object

NOTE: nats-pure's #pull_subscribe has no path to a true ephemeral pull consumer. Its rescue branch unconditionally sets config[:durable_name] = durable (jetstream.rb).

Parameters:

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

Returns:

  • (Object)

Raises:



147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/cosmo/client.rb', line 147

def ephemeral_subscribe(subject, config) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
  config = config.dup
  stream = config.delete(:stream) or raise ArgumentError, "stream required for ephemeral consumers"
  raise ArgumentError, "inactive_threshold required for ephemeral consumers" unless config[:inactive_threshold]

  subject = subject.first if subject.is_a?(Array) && subject.size == 1
  consumer_config = NATS::JetStream::API::ConsumerConfig.new(config)
  if subject.is_a?(Array)
    consumer_config[:filter_subjects] ||= subject
  else
    consumer_config[:filter_subject] ||= subject
  end

  info = js.add_consumer(stream, consumer_config)

  sub = nc.subscribe(nc.new_inbox)
  sub.extend(NATS::JetStream.const_get(:PullSubscription))
  sub.jsi = NATS::JetStream.const_get(:JS)::Sub.new(
    js: js, stream: stream, consumer: info.name,
    nms: "#{js.prefix}.CONSUMER.MSG.NEXT.#{stream}.#{info.name}"
  )
  sub
end

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

Parameters:

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

Returns:

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


115
116
117
 
# File 'lib/cosmo/client.rb', line 115

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) 


133
134
135
136
137
 
# File 'lib/cosmo/client.rb', line 133

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


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

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


71
72
73
74
75
76
77
78
79
 
# File 'lib/cosmo/client.rb', line 71

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) 


81
82
83
84
85
86
 
# File 'lib/cosmo/client.rb', line 81

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) 


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

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) 


53
54
55
56
57
 
# File 'lib/cosmo/client.rb', line 53

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) 


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

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

#stream_paused?(name) ⇒ Boolean

Parameters:

  • name (::String)

Returns:

  • (Boolean) 


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

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

#subscribe(subject, consumer_name, config) ⇒ Object

Create a pull subscription. Durable with consumer_name, and ephemeral without.

Parameters:

  • config (Hash) —

    Consumer config. Ephemeral consumers additionally require :stream and :inactive_threshold (seconds the consumer survives without a fetch).

  • subject (::String, Array[::String])
  • consumer_name (::String, nil)

Returns:

  • (Object) 


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

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

  ephemeral_subscribe(subject, config)
end

#unpause_stream(name) ⇒ void

This method returns an undefined value.

Parameters:

  • name (::String) 


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

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) 


46
47
48
 
# File 'lib/cosmo/client.rb', line 46

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