Class: Rubyists::Leopard::NatsServiceDiscovery

Inherits:
Object
  • Object
show all
Defined in:
lib/leopard/nats_service_discovery.rb,
lib/leopard/nats_service_discovery/cli.rb,
lib/leopard/nats_service_discovery/operation.rb,
lib/leopard/nats_service_discovery/operation/info.rb,
lib/leopard/nats_service_discovery/operation/ping.rb,
lib/leopard/nats_service_discovery/operation/verb.rb,
lib/leopard/nats_service_discovery/operation/stats.rb,
lib/leopard/nats_service_discovery/operation/collect.rb,
lib/leopard/nats_service_discovery/operation/connect.rb,
lib/leopard/nats_service_discovery/operation/services.rb,
lib/leopard/nats_service_discovery/operation/subject_map.rb,
lib/leopard/nats_service_discovery/operation/discovery_options.rb

Overview

Collects NATS Service API monitoring responses from a cluster.

The NATS Service API uses request/reply subjects such as $SRV.INFO. Cluster-wide requests can receive multiple replies, so this helper creates an inbox subscription, publishes the request, and collects replies until a short timeout elapses.

Defined Under Namespace

Modules: CLI, Operation

Constant Summary collapse

DEFAULT_PREFIX =

Default NATS Service API monitoring prefix.

'$SRV'
DEFAULT_TIMEOUT =

Default idle timeout used to decide that all service replies arrived.

0.25

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client:, prefix: DEFAULT_PREFIX) ⇒ NatsServiceDiscovery

Returns a new instance of NatsServiceDiscovery.

Parameters:

  • client (NATS::Client) —

    Connected NATS client.

  • prefix (String) (defaults to: DEFAULT_PREFIX) —

    Service API monitoring prefix.



24
25
26
27
 
# File 'lib/leopard/nats_service_discovery.rb', line 24

def initialize(client:, prefix: DEFAULT_PREFIX)
  @client = client
  @prefix = prefix
end

Instance Attribute Details

#client ⇒ Object (readonly)

Returns the value of attribute client.



20
21
22
 
# File 'lib/leopard/nats_service_discovery.rb', line 20

def client
  @client
end

#prefix ⇒ Object (readonly)

Returns the value of attribute prefix.



20
21
22
 
# File 'lib/leopard/nats_service_discovery.rb', line 20

def prefix
  @prefix
end

Instance Method Details

#collect_json(subject, timeout:) ⇒ Array<Hash> (private)

Collects JSON replies for a Service API subject until timeout elapses.

Parameters:

  • subject (String) —

    Service API subject to publish.

  • timeout (Numeric) —

    Idle timeout while waiting for replies.

Returns:

  • (Array<Hash>) —

    Parsed response payloads.



105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/leopard/nats_service_discovery.rb', line 105

def collect_json(subject, timeout:)
  replies = []
  sub = subscribe_to_inbox
  client.publish(subject, '', sub.subject)
  begin
    collect_replies(sub, replies, timeout)
  rescue NATS::Timeout
    replies
  end
ensure
  sub&.unsubscribe
end

#collect_replies(sub, replies, timeout) ⇒ void (private)

This method returns an undefined value.

Appends parsed replies to the provided accumulator until timeout.

Parameters:

  • sub (NATS::Subscription) —

    Reply subscription.

  • replies (Array<Hash>) —

    Response accumulator.

  • timeout (Numeric) —

    Idle timeout while waiting for replies.



134
135
136
137
138
139
140
141
 
# File 'lib/leopard/nats_service_discovery.rb', line 134

def collect_replies(sub, replies, timeout)
  loop do
    msg = sub.next_msg(timeout:)
    next if no_responders?(msg)

    replies << JSON.parse(msg.data)
  end
end

#endpoint_subject_map(name: nil, id: nil, timeout: DEFAULT_TIMEOUT) ⇒ Hash{String => Array<Hash>}

Builds a subject-to-listener map from service info responses.

Parameters:

  • name (String, nil) (defaults to: nil) —

    Optional service name filter.

  • id (String, nil) (defaults to: nil) —

    Optional service id filter; requires name.

  • timeout (Numeric) (defaults to: DEFAULT_TIMEOUT) —

    Idle timeout while collecting replies.

Returns:

  • (Hash{String => Array<Hash>}) —

    Endpoint listeners keyed by subject.



69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/leopard/nats_service_discovery.rb', line 69

def endpoint_subject_map(name: nil, id: nil, timeout: DEFAULT_TIMEOUT)
  info(name:, id:, timeout:).each_with_object({}) do |service, subjects|
    Array(service['endpoints']).each do |endpoint|
      subject = endpoint['subject']
      next if subject.to_s.empty?

      subjects[subject] ||= []
      subjects[subject] << listener_entry(service, endpoint)
    end
  end
end

#info(name: nil, id: nil, timeout: DEFAULT_TIMEOUT) ⇒ Array<Hash>

Collects $SRV.INFO responses.

Parameters:

  • name (String, nil) (defaults to: nil) —

    Optional service name filter.

  • id (String, nil) (defaults to: nil) —

    Optional service id filter; requires name.

  • timeout (Numeric) (defaults to: DEFAULT_TIMEOUT) —

    Idle timeout while collecting replies.

Returns:

  • (Array<Hash>) —

    Parsed info responses.



47
48
49
 
# File 'lib/leopard/nats_service_discovery.rb', line 47

def info(name: nil, id: nil, timeout: DEFAULT_TIMEOUT)
  collect_json(service_subject('INFO', name:, id:), timeout:)
end

#listener_entry(service, endpoint) ⇒ Hash (private)

Builds a subject-map listener entry from a service and endpoint payload.

Parameters:

  • service (Hash) —

    Parsed $SRV.INFO service response.

  • endpoint (Hash) —

    Endpoint payload from the service response.

Returns:

  • (Hash) —

    Listener entry suitable for subject maps.



158
159
160
161
162
163
164
165
166
167
 
# File 'lib/leopard/nats_service_discovery.rb', line 158

def listener_entry(service, endpoint)
  {
    'service' => service['name'],
    'service_id' => service['id'],
    'version' => service['version'],
    'endpoint' => endpoint['name'],
    'queue_group' => endpoint['queue_group'],
    'metadata' => endpoint['metadata'],
  }
end

#no_responders?(msg) ⇒ Boolean (private)

Reports whether a reply is the server no-responders status message.

Parameters:

  • msg (NATS::Msg) —

    Reply message.

Returns:

  • (Boolean) 


148
149
150
 
# File 'lib/leopard/nats_service_discovery.rb', line 148

def no_responders?(msg)
  msg.header && msg.header['Status'] == '503'
end

#ping(name: nil, id: nil, timeout: DEFAULT_TIMEOUT) ⇒ Array<Hash>

Collects $SRV.PING responses.

Parameters:

  • name (String, nil) (defaults to: nil) —

    Optional service name filter.

  • id (String, nil) (defaults to: nil) —

    Optional service id filter; requires name.

  • timeout (Numeric) (defaults to: DEFAULT_TIMEOUT) —

    Idle timeout while collecting replies.

Returns:

  • (Array<Hash>) —

    Parsed ping responses.



36
37
38
 
# File 'lib/leopard/nats_service_discovery.rb', line 36

def ping(name: nil, id: nil, timeout: DEFAULT_TIMEOUT)
  collect_json(service_subject('PING', name:, id:), timeout:)
end

#present?(value) ⇒ Boolean (private)

Reports whether a value is present for subject construction.

Parameters:

  • value (Object) —

    Value to check.

Returns:

  • (Boolean) 


174
175
176
 
# File 'lib/leopard/nats_service_discovery.rb', line 174

def present?(value)
  !value.nil? && !value.to_s.empty?
end

#service_subject(verb, name: nil, id: nil) ⇒ String

Builds a Service API monitoring subject.

Parameters:

  • verb (String, Symbol) —

    Monitoring verb such as INFO.

  • name (String, nil) (defaults to: nil) —

    Optional service name filter.

  • id (String, nil) (defaults to: nil) —

    Optional service id filter; requires name.

Returns:

  • (String) —

    Monitoring subject.

Raises:

  • (ArgumentError) 


88
89
90
91
92
93
94
95
 
# File 'lib/leopard/nats_service_discovery.rb', line 88

def service_subject(verb, name: nil, id: nil)
  raise ArgumentError, 'service id requires a service name' if present?(id) && !present?(name)

  parts = [prefix, verb.to_s.upcase]
  parts << name.to_s if present?(name)
  parts << id.to_s if present?(id)
  parts.join('.')
end

#stats(name: nil, id: nil, timeout: DEFAULT_TIMEOUT) ⇒ Array<Hash>

Collects $SRV.STATS responses.

Parameters:

  • name (String, nil) (defaults to: nil) —

    Optional service name filter.

  • id (String, nil) (defaults to: nil) —

    Optional service id filter; requires name.

  • timeout (Numeric) (defaults to: DEFAULT_TIMEOUT) —

    Idle timeout while collecting replies.

Returns:

  • (Array<Hash>) —

    Parsed stats responses.



58
59
60
 
# File 'lib/leopard/nats_service_discovery.rb', line 58

def stats(name: nil, id: nil, timeout: DEFAULT_TIMEOUT)
  collect_json(service_subject('STATS', name:, id:), timeout:)
end

#subscribe_to_inbox ⇒ NATS::Subscription (private)

Subscribes to an ephemeral reply inbox and flushes the subscription.

Returns:

  • (NATS::Subscription) —

    Subscription bound to the reply inbox.



121
122
123
124
125
 
# File 'lib/leopard/nats_service_discovery.rb', line 121

def subscribe_to_inbox
  sub = client.subscribe(client.new_inbox)
  client.flush
  sub
end