Synchronize Client#add_service

nats-io · Jan 27, 2025 · dcbf7fd · dcbf7fd
1 parent 5a7b2c5
commit dcbf7fd
Show file tree

Hide file tree

Showing 3 changed files with 16 additions and 12 deletions.
diff --git a/lib/nats/io/client.rb b/lib/nats/io/client.rb
@@ -101,7 +101,7 @@ class Client
     include MonitorMixin
     include Status
 
-    attr_reader :status, :server_info, :server_pool, :options, :stats, :uri, :subscription_executor, :reloader, :subs
+    attr_reader :status, :server_info, :server_pool, :options, :stats, :uri, :subscription_executor, :reloader, :services
 
     DEFAULT_PORT = {nats: 4222, ws: 80, wss: 443}.freeze
     DEFAULT_URI = "nats://localhost:#{DEFAULT_PORT[:nats]}".freeze
@@ -242,6 +242,9 @@ def initialize(uri = nil, opts = {})
       # Draining
       @drain_t = nil
 
+      # Service API
+      @services = []
+
       # Prepare for calling connect or automatic delayed connection
       parse_and_validate_options if uri || opts.any?
 
@@ -738,15 +741,13 @@ def flush(timeout = 10)
       end
     end
 
-    def services
-      @services ||= []
-    end
-
     def add_service(options)
-      service = NATS::Service.new(self, options)
+      synchronize do
+        service = NATS::Service.new(self, options)
+        services << service
 
-      services << service
-      service
+        service
+      end
     end
 
     alias_method :servers, :server_pool

diff --git a/spec/service/endpoint_spec.rb b/spec/service/endpoint_spec.rb
@@ -11,6 +11,8 @@
   end
 
   let(:client) { NATS.connect }
+  let(:subs) { client.instance_variable_get("@subs") }
+
   let(:service) { client.add_service(name: "foo", queue: "queue") }
 
   subject do
@@ -192,7 +194,7 @@
     it "drains handler subscription" do
       subject.stop
 
-      expect(client.subs.values).to include(
+      expect(subs.values).to include(
         having_attributes(subject: "foo.bar", drained: true)
       )
     end

diff --git a/spec/service/monitoring_spec.rb b/spec/service/monitoring_spec.rb
@@ -18,6 +18,7 @@
   let(:stats) { {**basic, started: "2025-01-24T05:40:37Z"} }
 
   let(:client) { NATS.connect }
+  let(:subs) { client.instance_variable_get("@subs") }
 
   before(:all) do
     @server = NatsServerControl.new
@@ -37,7 +38,7 @@
       it "starts monitoring with the specified prefix" do
         subject
 
-        expect(client.subs.values).to include(
+        expect(subs.values).to include(
           having_attributes(subject: "$FOO.PING"),
           having_attributes(subject: "$FOO.PING.foo"),
           having_attributes(subject: "$FOO.PING.foo.bar"),
@@ -57,7 +58,7 @@
       it "starts monitoring with the default prefix" do
         subject
 
-        expect(client.subs.values).to include(
+        expect(subs.values).to include(
           having_attributes(subject: "$SRV.PING"),
           having_attributes(subject: "$SRV.PING.foo"),
           having_attributes(subject: "$SRV.PING.foo.bar"),
@@ -119,7 +120,7 @@
       it "drains monitoring subscriptions" do
         subject.stop
 
-        expect(client.subs.values).to include(
+        expect(subs.values).to include(
           having_attributes(subject: "$SRV.PING", drained: true),
           having_attributes(subject: "$SRV.PING.foo", drained: true),
           having_attributes(subject: "$SRV.PING.foo.bar", drained: true),