Scope connection pool to clusteR

Author: durran

lib/mongo/address.rb

@@ -63,6 +63,10 @@ def ==(other)
       host == other.host && port == other.port
     end
 
+    def eql?(other)
+      self == other
+    end
+
     # Calculate the hash value for the address.
     #
     # @example Calculate the hash value.

lib/mongo/cluster.rb

@@ -106,6 +106,7 @@ def initialize(seeds, monitoring, options = Options::Redacted.new)
       @options = options.freeze
       @topology = Topology.initial(seeds, options)
       @update_lock = Mutex.new
+      @pool_lock = Mutex.new
 
       subscribe_to(Event::STANDALONE_DISCOVERED, Event::StandaloneDiscovered.new(self))
       subscribe_to(Event::DESCRIPTION_CHANGED, Event::DescriptionChanged.new(self))
@@ -168,6 +169,22 @@ def max_read_retries
       options[:max_read_retries] || MAX_READ_RETRIES
     end
 
+    # Get the scoped connection pool for the server.
+    #
+    # @example Get the connection pool.
+    #   cluster.pool(server)
+    #
+    # @param [ Server ] server The server.
+    #
+    # @return [ Server::ConnectionPool ] The connection pool.
+    #
+    # @since 2.2.0
+    def pool(server)
+      @pool_lock.synchronize do
+        pools[server.address] ||= Server::ConnectionPool.get(server)
+      end
+    end
+
     # Get the interval, in seconds, in which a mongos read operation is
     # retried.
     #
@@ -339,6 +356,10 @@ def addition_allowed?(address)
       !@topology.single? || direct_connection?(address)
     end
 
+    def pools
+      @pools ||= {}
+    end
+
     def servers_list
       @update_lock.synchronize do
         @servers.reduce([]) do |servers, server|

lib/mongo/server.rb

@@ -182,7 +182,7 @@ def inspect
     #
     # @since 2.0.0
     def pool
-      @pool ||= ConnectionPool.get(self)
+      @pool ||= cluster.pool(self)
     end
 
     # Determine if the provided tags are a subset of the server's tags.

lib/mongo/server/connection_pool.rb

@@ -23,9 +23,6 @@ class Server
     class ConnectionPool
       include Loggable
 
-      # Used for synchronization of pools access.
-      MUTEX = Mutex.new
-
       # @return [ Hash ] options The pool options.
       attr_reader :options
 
@@ -131,22 +128,10 @@ class << self
         #
         # @since 2.0.0
         def get(server)
-          MUTEX.synchronize do
-            pools[server.address] ||= create_pool(server)
-          end
-        end
-
-        private
-
-        def create_pool(server)
           ConnectionPool.new(server.options) do
             Connection.new(server, server.options)
           end
         end
-
-        def pools
-          @pools ||= {}
-        end
       end
     end
   end

spec/mongo/server/connection_pool_spec.rb

@@ -18,17 +18,22 @@
     Mongo::Event::Listeners.new
   end
 
+  let(:cluster) do
+    double('cluster')
+  end
+
   describe '#checkin' do
 
     let(:server) do
-      Mongo::Server.new(address, double('cluster'), monitoring, listeners, options)
+      Mongo::Server.new(address, cluster, monitoring, listeners, options)
     end
 
     let!(:pool) do
       described_class.get(server)
     end
 
     after do
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end
 
@@ -55,7 +60,7 @@
   describe '#checkout' do
 
     let(:server) do
-      Mongo::Server.new(address, double('cluster'), monitoring, listeners, options)
+      Mongo::Server.new(address, cluster, monitoring, listeners, options)
     end
 
     let!(:pool) do
@@ -103,53 +108,52 @@
   describe '#disconnect!' do
 
     let(:server) do
-      Mongo::Server.new(address, double('cluster'), monitoring, listeners, options)
+      Mongo::Server.new(address, cluster, monitoring, listeners, options)
     end
 
     let!(:pool) do
       described_class.get(server)
     end
 
-    after do
-      server.disconnect!
-    end
-
     it 'disconnects the queue' do
-      expect(pool.send(:queue)).to receive(:disconnect!).twice.and_call_original
-      pool.disconnect!
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
+      expect(pool.send(:queue)).to receive(:disconnect!).once.and_call_original
+      server.disconnect!
     end
   end
 
   describe '.get' do
 
     let(:server) do
-      Mongo::Server.new(address, double('cluster'), monitoring, listeners, options)
+      Mongo::Server.new(address, cluster, monitoring, listeners, options)
     end
 
     let!(:pool) do
       described_class.get(server)
     end
 
     after do
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end
 
     it 'returns the pool for the server' do
-      expect(pool).to eql(described_class.get(server))
+      expect(pool).to_not be_nil
     end
   end
 
   describe '#inspect' do
 
     let(:server) do
-      Mongo::Server.new(address, double('cluster'), monitoring, listeners, options)
+      Mongo::Server.new(address, cluster, monitoring, listeners, options)
     end
 
     let!(:pool) do
       described_class.get(server)
     end
 
     after do
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end
 
@@ -165,7 +169,7 @@
   describe '#with_connection' do
 
     let(:server) do
-      Mongo::Server.new(address, double('cluster'), monitoring, listeners, options)
+      Mongo::Server.new(address, cluster, monitoring, listeners, options)
     end
 
     let!(:pool) do

spec/mongo/server/connection_spec.rb

@@ -14,35 +14,34 @@
     Mongo::Event::Listeners.new
   end
 
+  let(:cluster) do
+    double('cluster')
+  end
+
   let(:server) do
-    Mongo::Server.new(address, double('cluster'), monitoring, listeners, TEST_OPTIONS)
+    Mongo::Server.new(address, cluster, monitoring, listeners, TEST_OPTIONS)
+  end
+
+  let(:pool) do
+    double('pool')
   end
 
   after do
+    expect(cluster).to receive(:pool).with(server).and_return(pool)
+    expect(pool).to receive(:disconnect!).and_return(true)
     server.disconnect!
   end
 
   describe '#connectable?' do
 
-    # context 'when the connection is connectable' do
-
-    #   let(:connection) do
-    #     described_class.new(server)
-    #   end
-
-    #   it 'returns true' do
-    #     expect(connection).to be_connectable
-    #   end
-    # end
-
     context 'when the connection is not connectable' do
 
       let(:bad_address) do
         Mongo::Address.new('127.0.0.1:666')
       end
 
       let(:bad_server) do
-        Mongo::Server.new(bad_address, double('cluster'), monitoring, listeners, TEST_OPTIONS)
+        Mongo::Server.new(bad_address, cluster, monitoring, listeners, TEST_OPTIONS)
       end
 
       let(:connection) do

spec/mongo/server_spec.rb

@@ -18,13 +18,18 @@
     Mongo::Address.new('127.0.0.1:27017')
   end
 
+  let(:pool) do
+    Mongo::Server::ConnectionPool.get(server)
+  end
+
   describe '#==' do
 
     let(:server) do
       described_class.new(address, cluster, monitoring, listeners, TEST_OPTIONS)
     end
 
     after do
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end
 
@@ -81,6 +86,10 @@
         server.disconnect!
       end
 
+      before do
+        expect(cluster).to receive(:pool).with(server).and_return(pool)
+      end
+
       it 'returns true' do
         expect(server).to be_connectable
       end
@@ -97,6 +106,7 @@
       end
 
       before do
+        expect(cluster).to receive(:pool).with(server).and_return(pool)
         server.disconnect!
       end
 
@@ -117,6 +127,7 @@
     end
 
     after do
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end
 
@@ -133,6 +144,7 @@
 
     it 'stops the monitor instance' do
       expect(server.instance_variable_get(:@monitor)).to receive(:stop!).and_return(true)
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end
   end
@@ -150,6 +162,7 @@
     end
 
     after do
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end
 
@@ -166,32 +179,14 @@
     end
   end
 
-  describe '#pool' do
-
-    let(:server) do
-      described_class.new(address, cluster, monitoring, listeners, TEST_OPTIONS)
-    end
-
-    let(:pool) do
-      server.pool
-    end
-
-    after do
-      server.disconnect!
-    end
-
-    it 'returns the connection pool for the server' do
-      expect(pool).to be_a(Mongo::Server::ConnectionPool)
-    end
-  end
-
   describe '#scan!' do
 
     let(:server) do
       described_class.new(address, cluster, monitoring, listeners, TEST_OPTIONS)
     end
 
     after do
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end
 
@@ -211,6 +206,7 @@
     end
 
     after do
+      expect(cluster).to receive(:pool).with(server).and_return(pool)
       server.disconnect!
     end