cluster.rb in mongo-2.2.0

- old
+ new

@@ -104,18 +104,40 @@
       @monitoring = monitoring
       @event_listeners = Event::Listeners.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))
       subscribe_to(Event::PRIMARY_ELECTED, Event::PrimaryElected.new(self))
 
       seeds.each{ |seed| add(seed) }
+      ObjectSpace.define_finalizer(self, self.class.finalize(pools))
     end
 
+    # Finalize the cluster for garbage collection. Disconnects all the scoped
+    # connection pools.
+    #
+    # @example Finalize the cluster.
+    #   Cluster.finalize(pools)
+    #
+    # @param [ Hash<Address, Server::ConnectionPool> ] pools The connection
+    #   pools.
+    #
+    # @return [ Proc ] The Finalizer.
+    #
+    # @since 2.2.0
+    def self.finalize(pools)
+      proc do
+        pools.values.each do |pool|
+          pool.disconnect!
+        end
+      end
+    end
+
     # Get the nicer formatted string for use in inspection.
     #
     # @example Inspect the cluster.
     #   cluster.inspect
     #
@@ -166,10 +188,26 @@
     # @since 2.1.1
     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.
     #
     # @example Get the read retry interval.
     #   cluster.read_retry_interval
@@ -335,9 +373,13 @@
       address.seed == @topology.seed
     end
 
     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|