RUBY-2281 Use local event subscribers in test suite (#1953)

Co-authored-by: Oleg Pudeyev <[email protected]>

Author: p-mongo

spec/integration/cursor_reaping_spec.rb

@@ -5,19 +5,23 @@
   # in MRI, I don't currently know how to force GC to run in JRuby
   only_mri
 
-  let(:client) { subscribed_client }
+  let(:subscriber) { EventSubscriber.new }
+
+  let(:client) do
+    authorized_client.tap do |client|
+      client.subscribe(Mongo::Monitoring::COMMAND, subscriber)
+    end
+  end
+
   let(:collection) { client['cursor_reaping_spec'] }
 
-  before(:all) do
+  before do
     data = [{a: 1}] * 10
-    ClientRegistry.instance.global_client('subscribed')['cursor_reaping_spec'].insert_many(data)
+    authorized_client['cursor_reaping_spec'].delete_many
+    authorized_client['cursor_reaping_spec'].insert_many(data)
   end
 
   context 'a no-timeout cursor' do
-    before do
-      EventSubscriber.clear_events!
-    end
-
     it 'reaps nothing when we do not query' do
       # this is a base line test to ensure that the reaps in the other test
       # aren't done on some global cursor
@@ -26,7 +30,7 @@
       # just the scope, no query is happening
       collection.find.batch_size(2).no_cursor_timeout
 
-      events = EventSubscriber.started_events.select do |event|
+      events = subscriber.started_events.select do |event|
         event.command['killCursors']
       end
 
@@ -58,14 +62,14 @@
       # force periodic executor to run because its frequency is not configurable
       client.cluster.instance_variable_get('@periodic_executor').execute
 
-      started_event = EventSubscriber.started_events.detect do |event|
+      started_event = subscriber.started_events.detect do |event|
         event.command['killCursors'] &&
         event.command['cursors'].map { |c| Utils.int64_value(c) }.include?(cursor_id)
       end
 
       expect(started_event).not_to be_nil
 
-      succeeded_event = EventSubscriber.succeeded_events.detect do |event|
+      succeeded_event = subscriber.succeeded_events.detect do |event|
         event.command_name == 'killCursors' && event.request_id == started_event.request_id
       end
 

spec/integration/get_more_spec.rb

@@ -4,8 +4,16 @@
   # https://jira.mongodb.org/browse/RUBY-1987
   min_server_fcv '3.2'
 
+  let(:subscriber) { EventSubscriber.new }
+
+  let(:client) do
+    authorized_client.tap do |client|
+      client.subscribe(Mongo::Monitoring::COMMAND, subscriber)
+    end
+  end
+
   let(:collection) do
-    subscribed_client['get_more_spec']
+    client['get_more_spec']
   end
 
   let(:scope) do
@@ -16,11 +24,10 @@
     collection.delete_many
     collection.insert_one(a: 1)
     #collection.insert_one(a: 2)
-    EventSubscriber.clear_events!
   end
 
   let(:get_more_command) do
-    event = EventSubscriber.single_command_started_event('getMore')
+    event = subscriber.single_command_started_event('getMore')
     event.command['getMore']
   end
 

spec/integration/retryable_errors_spec.rb

@@ -4,8 +4,16 @@
   # Requirement for fail point
   min_server_fcv '4.0'
 
+  let(:subscriber) { EventSubscriber.new }
+
+  let(:client_options) do
+    {}
+  end
+
   let(:client) do
-    subscribed_client
+    authorized_client.with(client_options).tap do |client|
+      client.subscribe(Mongo::Monitoring::COMMAND, subscriber)
+    end
   end
 
   let(:collection) do
@@ -83,7 +91,7 @@
       end
 
       let(:events) do
-        EventSubscriber.command_started_events('find')
+        subscriber.command_started_events('find')
       end
     end
 
@@ -128,7 +136,7 @@
       end
 
       let(:events) do
-        EventSubscriber.command_started_events('insert')
+        subscriber.command_started_events('insert')
       end
     end
 
@@ -141,6 +149,7 @@
       end
 
       it 'publishes two events' do
+        operation_exception
 
         expect(events.length).to eq(2)
       end
@@ -155,6 +164,7 @@
       end
 
       it 'publishes one event' do
+        operation_exception
 
         expect(events.length).to eq(1)
       end
@@ -217,17 +227,17 @@
       context 'modern read retries' do
         require_wired_tiger_on_36
 
-        let(:client) do
-          subscribed_client.with(retry_reads: true)
+        let(:client_options) do
+          {retry_reads: true}
         end
 
         it_behaves_like 'failing retry'
         it_behaves_like 'modern retry'
       end
 
       context 'legacy read retries' do
-        let(:client) do
-          subscribed_client.with(retry_reads: false, read_retry_interval: 0)
+        let(:client_options) do
+          {retry_reads: false, read_retry_interval: 0}
         end
 
         it_behaves_like 'failing retry'
@@ -236,8 +246,8 @@
     end
 
     context 'when read retries are disabled' do
-      let(:client) do
-        subscribed_client.with(retry_reads: false, max_read_retries: 0)
+      let(:client_options) do
+        {retry_reads: false, max_read_retries: 0}
       end
 
       include_context 'read operation'
@@ -252,17 +262,17 @@
       context 'modern write retries' do
         require_wired_tiger_on_36
 
-        let(:client) do
-          subscribed_client.with(retry_writes: true)
+        let(:client_options) do
+          {retry_writes: true}
         end
 
         it_behaves_like 'failing retry'
         it_behaves_like 'modern retry'
       end
 
       context 'legacy write' do
-        let(:client) do
-          subscribed_client.with(retry_writes: false)
+        let(:client_options) do
+          {retry_writes: false}
         end
 
         it_behaves_like 'failing retry'

spec/integration/retryable_writes/retryable_writes_36_and_older_spec.rb

@@ -32,7 +32,7 @@
   let(:check_collection) do
     # Verify data in the collection using another client instance to avoid
     # having the verification read trigger cluster scans on the writing client
-    subscribed_client[TEST_COLL]
+    root_authorized_client[TEST_COLL]
   end
 
   let(:primary_connection) do

spec/integration/step_down_spec.rb

@@ -29,11 +29,11 @@
     end
   end
 
-  let(:event_subscriber) { EventSubscriber.new }
+  let(:subscriber) { EventSubscriber.new }
 
   let(:test_client) do
     authorized_client_without_any_retries.with(server_selection_timeout: 20).tap do |client|
-      client.subscribe(Mongo::Monitoring::CONNECTION_POOL, event_subscriber)
+      client.subscribe(Mongo::Monitoring::CONNECTION_POOL, subscriber)
     end
   end
 
@@ -48,8 +48,8 @@
 
     let(:subscribed_client) do
       test_client.tap do |client|
-        client.subscribe(Mongo::Monitoring::COMMAND, EventSubscriber)
-        client.subscribe(Mongo::Monitoring::CONNECTION_POOL, EventSubscriber)
+        client.subscribe(Mongo::Monitoring::COMMAND, subscriber)
+        client.subscribe(Mongo::Monitoring::CONNECTION_POOL, subscriber)
       end
     end
 
@@ -65,21 +65,21 @@
     it 'continues through step down' do
 
       subscribed_client.cluster.next_primary.pool.clear
-      event_subscriber.clear_events!
+      subscriber.clear_events!
 
       # get the first item
       item = enum.next
       expect(item['test']).to eq(1)
 
-      connection_created_events = EventSubscriber.published_events.select do |event|
+      connection_created_events = subscriber.published_events.select do |event|
         event.is_a?(Mongo::Monitoring::Event::Cmap::ConnectionCreated)
       end
       expect(connection_created_events).not_to be_empty
 
       current_primary = subscribed_client.cluster.next_primary
       ClusterTools.instance.change_primary
 
-      EventSubscriber.clear_events!
+      subscriber.clear_events!
 
       # exhaust the batch
       9.times do
@@ -90,14 +90,14 @@
       item = enum.next
       expect(item['test']).to eq(1)
 
-      get_more_events = EventSubscriber.started_events.select do |event|
+      get_more_events = subscriber.started_events.select do |event|
         event.command['getMore']
       end
 
       expect(get_more_events.length).to eq(1)
 
       # getMore should have been sent on the same connection as find
-      connection_created_events = EventSubscriber.published_events.select do |event|
+      connection_created_events = subscriber.published_events.select do |event|
         event.is_a?(Mongo::Monitoring::Event::Cmap::ConnectionCreated)
       end
       expect(connection_created_events).to be_empty
@@ -155,13 +155,13 @@
       let(:fail_point_code) { 10107 }
 
       it 'keeps connection open' do
-        event_subscriber.clear_events!
+        subscriber.clear_events!
 
         expect do
           collection.insert_one(test: 1)
         end.to raise_error(Mongo::Error::OperationFailure, /10107/)
 
-        expect(event_subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(0)
+        expect(subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(0)
       end
     end
 
@@ -174,13 +174,13 @@
       let(:fail_point_code) { 10107 }
 
       it 'closes the connection' do
-        event_subscriber.clear_events!
+        subscriber.clear_events!
 
         expect do
           collection.insert_one(test: 1)
         end.to raise_error(Mongo::Error::OperationFailure, /10107/)
 
-        expect(event_subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(1)
+        expect(subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(1)
       end
     end
 
@@ -191,13 +191,13 @@
       let(:fail_point_code) { 11600 }
 
       it 'closes the connection' do
-        event_subscriber.clear_events!
+        subscriber.clear_events!
 
         expect do
           collection.insert_one(test: 1)
         end.to raise_error(Mongo::Error::OperationFailure, /11600/)
 
-        expect(event_subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(1)
+        expect(subscriber.select_published_events(Mongo::Monitoring::Event::Cmap::PoolCleared).count).to eq(1)
       end
     end
   end

spec/lite_spec_helper.rb

@@ -136,8 +136,6 @@
   end
 end
 
-EventSubscriber.initialize
-
 if SpecConfig.instance.active_support?
   require "active_support/time"
   require 'mongo/active_support'

spec/mongo/client_construction_spec.rb

@@ -3,6 +3,8 @@
 describe Mongo::Client do
   clean_slate
 
+  let(:subscriber) { EventSubscriber.new }
+
   describe '.new' do
     context 'with scan: false' do
       it 'does not perform i/o' do
@@ -1253,7 +1255,7 @@
 
     it 'copies monitoring subscribers' do
       monitoring.subscribers.clear
-      client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, EventSubscriber.new)
+      client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, subscriber)
       expect(monitoring.present_subscribers.length).to eq(1)
       expect(monitoring.subscribers[Mongo::Monitoring::SERVER_HEARTBEAT].length).to eq(1)
 
@@ -1264,12 +1266,12 @@
 
     it 'does not change subscribers on original client' do
       monitoring.subscribers.clear
-      client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, EventSubscriber.new)
+      client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, subscriber)
       expect(monitoring.present_subscribers.length).to eq(1)
       expect(monitoring.subscribers[Mongo::Monitoring::SERVER_HEARTBEAT].length).to eq(1)
 
-      new_client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, EventSubscriber.new)
-      new_client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, EventSubscriber.new)
+      new_client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, subscriber)
+      new_client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, subscriber)
       expect(new_monitoring.present_subscribers.length).to eq(1)
       expect(new_monitoring.subscribers[Mongo::Monitoring::SERVER_HEARTBEAT].length).to eq(3)
       # original client should not have gotten any of the new subscribers
@@ -1297,7 +1299,7 @@
 
     it 'resets monitoring subscribers' do
       monitoring.subscribers.clear
-      client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, EventSubscriber.new)
+      client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, subscriber)
       expect(monitoring.present_subscribers.length).to eq(1)
       expect(monitoring.subscribers[Mongo::Monitoring::SERVER_HEARTBEAT].length).to eq(1)
 
@@ -1310,12 +1312,12 @@
 
     it 'does not change subscribers on original client' do
       monitoring.subscribers.clear
-      client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, EventSubscriber.new)
+      client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, subscriber)
       expect(monitoring.present_subscribers.length).to eq(1)
       expect(monitoring.subscribers[Mongo::Monitoring::SERVER_HEARTBEAT].length).to eq(1)
 
-      new_client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, EventSubscriber.new)
-      new_client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, EventSubscriber.new)
+      new_client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, subscriber)
+      new_client.subscribe(Mongo::Monitoring::SERVER_HEARTBEAT, subscriber)
       # 7 default subscribers + heartbeat
       expect(new_monitoring.present_subscribers.length).to eq(8)
       # the heartbeat subscriber on the original client is not inherited
@@ -1758,7 +1760,6 @@
     # in #with, the consistent behavior is to never transfer sdam_proc to
     # the new client.
     context 'when sdam_proc is given on original client' do
-      let(:subscriber) { EventSubscriber.new }
 
       let(:sdam_proc) do
         Proc.new do |client|