Implement Service API

[vankiru]

Context

This is an implementation of Service API.

The structure of the Service API is the following:

• NATS::Service is a starting point that connects all service parts together.
• NATS::Service::Group represents a service group.
• NATS::Service::Endpoint contains the logic of an individual endpoint.
• NATS::Service::Monitoring implements service discovery and status.
• NATS::Service::Stats is responsible for gathering statistics about endpoint usage.
• NATS::Service::Status prepares data for PING/INFO/STATS requests.
• NATS::Service::Callbacks handles on_stats and on_stop callbacks.
• NATS::Service::Validator validates name, version, subject and queue values.

Examples

General

client = NATS.connect

service = client.add_service(
  name: "calc",
  version: "1.0.0",
  description: "description"
)

service.on_stop do
  puts "Service stopped at #{Time.now}"
end

service.add_endpoint("min") do |message|
  min = JSON.parse(message.data).min
  message.respond(min.to_json)
end

service.add_endpoint("max") do |message|
  max = JSON.parse(message.data).max
  message.respond(max.to_json)
end

min = client.request("min", [5, 100, -7, 34].to_json)
max = client.request("max", [5, 100, -7, 34].to_json)

puts "min = #{min.data}, max = #{max.data}"

service.stop

Output will be:

min = -7, max = 100
Service stopped at 2025-01-24 20:53:07 +0400

Groups

arth = service.add_group("arth")
agg = service.add_group("agg")

arth.add_endpoint("sum") do |message|
  sum = JSON.parse(message.data).sum
  message.respond(sum.to_json)
end

agg.add_endpoint("avg") do |message|
  data = JSON.parse(message.data)
  message.respond((data.sum / data.size.to_f).to_json)
end

sum = client.request("arth.sum", [3, 4, 5, 7].to_json)
avg = client.request("agg.avg", [3, 4, 5, 7].to_json)

puts "sum = #{sum.data}, avg = #{avg.data}"

Output will be:

sum = 19, avg = 4.75

Stats

service.on_stats do |endpoint|
  errors = endpoint.stats.num_errors
  requests = endpoint.stats.num_requests

  { error_rate: (100.0 * errors / requests).round(2) }
end

service.add_endpoint("divide") do |message|
  data = JSON.parse(message.data)
  message.respond((data["dividend"] / data["divisor"]).to_json)
end

client.request("divide", {dividend: 5, divisor: 2}.to_json)
client.request("divide", {dividend: 7, divisor: 0}.to_json)
client.request("divide", {dividend: 3, divisor: 1}.to_json)
client.request("divide", {dividend: 4, divisor: 2}.to_json)
client.request("divide", {dividend: 8, divisor: 0}.to_json)

puts <<~INFO
=== Info ===
#{service.info}
INFO

puts <<~STATS
=== Stats ===
#{service.stats}
STATS

Output will be:

=== Info ===
{
  :name=>"calc", 
  :id=>"eQ5Ot2qpt7KlceCgvCqI59", 
  :version=>"1.0.0", 
  :metadata=>nil, 
  :description=>"description", 
  :endpoints=>[
    {
      :name=>"divide", 
      :subject=>"divide", 
      :queue_group=>"q", 
      :metadata=>nil
    }
  ]
}
=== Stats ===
{
  :name=>"calc", 
  :id=>"eQ5Ot2qpt7KlceCgvCqI59", 
  :version=>"1.0.0", 
  :metadata=>nil, 
  :started=>"2025-01-24T17:02:45Z", 
  :endpoints=>[
    {
      :name=>"divide", 
      :subject=>"divide", 
      :queue_group=>"q", 
      :num_requests=>5, 
      :processing_time=>92000, 
      :average_processing_time=>18400, 
      :num_errors=>2, 
      :last_error=>"divided by 0", 
      :data=>{:error_rate=>40.0}
    }
  ]
}

Discovery

client = NATS.connect

service = client.add_service(
  name: "calc",
  version: "1.0.0",
  description: "description"
)

service.add_endpoint("nothing") do |message|
  message.respond("nothing")
end

puts <<~PING
=== PING ===
#{client.request("$SRV.PING.calc").data}
PING

puts <<~INFO
=== INFO ===
#{client.request("$SRV.INFO.calc").data}
INFO

puts <<~STATS
=== STATS ===
#{client.request("$SRV.STATS.calc").data}
STATS

service.stop

Output will be

=== PING ===
{"type":"io.nats.micro.v1.ping_response","name":"calc","id":"PZCG7ddvNpVhIW31ES4JdJ","version":"1.0.0","metadata":null}
=== INFO ===
{"type":"io.nats.micro.v1.info_response","name":"calc","id":"PZCG7ddvNpVhIW31ES4JdJ","version":"1.0.0","metadata":null,"description":"description","endpoints":[{"name":"nothing","subject":"nothing","queue_group":"q","metadata":null}]}
=== STATS ===
{"type":"io.nats.micro.v1.stats_response","name":"calc","id":"PZCG7ddvNpVhIW31ES4JdJ","version":"1.0.0","metadata":null,"started":"2025-01-24T17:18:32Z","endpoints":[{"name":"nothing","subject":"nothing","queue_group":"q","num_requests":0,"processing_time":0,"average_processing_time":0,"num_errors":0,"last_error":null,"data":null}]}

Errors

client = NATS.connect

service = client.add_service(
  name: "calc",
  version: "1.0.0",
  description: "description"
)

service.add_endpoint("standard-error") do |message|
  message.respond_with_error(StandardError.new("standard error"))
end

service.add_endpoint("hash-error") do |message|
  message.respond_with_error(code: 503, description: "hash error")
end

service.add_endpoint("string-error") do |message|
  message.respond_with_error("string error")
end

puts client.request("standard-error").header
puts client.request("hash-error").header
puts client.request("string-error").header

Output will be

{"Nats-Service-Error"=>"standard error", "Nats-Service-Error-Code"=>"500"}
{"Nats-Service-Error"=>"hash error", "Nats-Service-Error-Code"=>"503"}
{"Nats-Service-Error"=>"string error", "Nats-Service-Error-Code"=>"500"}