Merge 413c858 into cdf3a26

lib/moped/authenticatable.rb

@@ -60,14 +60,19 @@ def credentials
     # @since 2.0.0
     def login(database, username, password)
       getnonce = Protocol::Command.new(database, getnonce: 1)
-      result = Operation::Read.new(getnonce).execute(self)
-      authenticate = Protocol::Commands::Authenticate.new(database, username, password, result["nonce"])
-      connection do |conn|
-        conn.write([ authenticate ])
-        document = conn.read.documents.first
-        raise Errors::AuthenticationFailure.new(authenticate, document) unless document["ok"] == 1
-        credentials[database] = [username, password]
+      self.write([getnonce])
+      reply = self.receive_replies([getnonce]).first
+      if getnonce.failure?(reply)
+        return
       end
+      result = getnonce.results(reply)
+
+      authenticate = Protocol::Commands::Authenticate.new(database, username, password, result["nonce"])
+      self.write([ authenticate ])
+      document = self.read.documents.first
+
+      raise Errors::AuthenticationFailure.new(authenticate, document) unless document["ok"] == 1
+      credentials[database] = [username, password]
     end
 
     # Logout the user from the provided database.
@@ -82,7 +87,11 @@ def login(database, username, password)
     # @since 2.0.0
     def logout(database)
       command = Protocol::Command.new(database, logout: 1)
-      Operation::Read.new(command).execute(self)
+      self.write([command])
+      reply = self.receive_replies([command]).first
+      if command.failure?(reply)
+        return
+      end
       credentials.delete(database)
     end
   end

lib/moped/cluster.rb

@@ -224,7 +224,8 @@ def with_primary(&block)
       if node = nodes.find(&:primary?)
         begin
           node.ensure_primary do
-            return yield(node.apply_credentials(credentials))
+            node.credentials = credentials
+            return yield(node)
           end
         rescue Errors::ConnectionFailure, Errors::ReplicaSetReconfigured
         end
@@ -251,7 +252,8 @@ def with_secondary(&block)
       available_nodes = nodes.select(&:secondary?).shuffle!
       while node = available_nodes.shift
         begin
-          return yield(node.apply_credentials(credentials))
+          node.credentials = credentials
+          return yield(node)
         rescue Errors::ConnectionFailure => e
           next
         end

lib/moped/connection.rb

@@ -11,6 +11,7 @@ module Moped
   #
   # @since 2.0.0
   class Connection
+    include Authenticatable
 
     # The default connection timeout, in seconds.
     #

lib/moped/connection/pool.rb

@@ -132,6 +132,26 @@ def size
         end
       end
 
+      # Disconnect all connections including the ones
+      # waiting on the pool and the ones in use.
+      #
+      # @example Disconnect all connections from the pool
+      #   pool.disconnect
+      #
+      # @since 2.0.0
+      def disconnect
+        mutex.synchronize do
+          pinned.values.each(&:disconnect)
+          conns = []
+          while unpinned.size > 0
+            conn = unpinned.shift
+            conn.disconnect
+            conns.push(conn)
+          end
+          conns.each { |c| unpinned.push(c) }
+        end
+      end
+
       # Get the timeout when attempting to check out items from the pool.
       #
       # @example Get the checkout timeout.

lib/moped/failover/retry.rb

@@ -26,7 +26,9 @@ module Retry
       def execute(exception, node)
         node.disconnect
         begin
-          yield if block_given?
+          node.connection do |conn|
+            yield(conn) if block_given?
+          end
         rescue Exception => e
           node.down!
           raise(e)

lib/moped/node.rb

@@ -13,7 +13,6 @@ module Moped
   #
   # @since 1.0.0
   class Node
-    include Authenticatable
     include Executable
     include Instrumentable
 
@@ -29,6 +28,10 @@ class Node
     #   @return [ Time ] The last time the node did a refresh.
     attr_reader :address, :down_at, :latency, :options, :refreshed_at
 
+    # @!attribute credentials
+    #   @return [ Hash ] The credentials of the node.
+    attr_accessor :credentials
+
     # Is this node equal to another?
     #
     # @example Is the node equal to another.
@@ -87,24 +90,6 @@ def command(database, cmd, options = {})
       read(Protocol::Command.new(database, cmd, options))
     end
 
-    # Connect the node on the underlying connection.
-    #
-    # @example Connect the node.
-    #   node.connect
-    #
-    # @raise [ Errors::ConnectionFailure ] If connection failed.
-    #
-    # @return [ true ] If the connection suceeded.
-    #
-    # @since 2.0.0
-    def connect
-      start = Time.now
-      connection{ |conn| conn.connect }
-      @latency = Time.now - start
-      @down_at = nil
-      true
-    end
-
     # Is the node currently connected?
     #
     # @example Is the node connected?
@@ -131,20 +116,6 @@ def connection
       end
     end
 
-    # Force the node to disconnect from the server.
-    #
-    # @example Disconnect the node.
-    #   node.disconnect
-    #
-    # @return [ true ] If the disconnection succeeded.
-    #
-    # @since 1.2.0
-    def disconnect
-      credentials.clear
-      connection{ |conn| conn.disconnect }
-      true
-    end
-
     # Is the node down?
     #
     # @example Is the node down?
@@ -157,6 +128,20 @@ def down?
       @down_at
     end
 
+    # Force the node to disconnect from the server.
+    #
+    # @example Disconnect the node.
+    #   node.disconnect
+    #
+    # @return [ true ] If the disconnection succeeded.
+    #
+    # @since 1.2.0
+    def disconnect
+      @credentials.clear
+      pool.disconnect
+      true
+    end
+
     # Mark the node as down.
     #
     # @example Mark the node as down.
@@ -168,7 +153,7 @@ def down?
     def down!
       @down_at = Time.new
       @latency = nil
-      disconnect if connected?
+      disconnect
     end
 
     # Yields the block if a connection can be established, retrying when a
@@ -185,15 +170,23 @@ def down!
     #
     # @since 1.0.0
     def ensure_connected(&block)
-      return yield if executing?(:connection)
-      execute(:connection) do
-        begin
-          connect unless connected?
-          yield(self)
-        rescue Exception => e
-          Failover.get(e).execute(e, self, &block)
+      unless (conn = stack(:connection)).empty?
+        return yield(conn.first)
+      end
+
+      begin
+        connection do |conn|
+          stack(:connection) << conn
+          connect(conn) unless conn.connected?
+          conn.apply_credentials(@credentials)
+          yield(conn)
         end
+      rescue Exception => e
+        Failover.get(e).execute(e, self, &block)
+      ensure
+        end_execution(:connection)
       end
+
     end
 
     # Set a flag on the node for the duration of provided block so that an
@@ -259,6 +252,7 @@ def initialize(address, options = {})
       @latency = nil
       @primary = nil
       @secondary = nil
+      @credentials = {}
       @instrumenter = options[:instrumenter] || Instrumentable::Log
       @address = Address.new(address, timeout)
       @address.resolve(self)
@@ -523,6 +517,24 @@ def inspect
 
     private
 
+    # Connect the node on the underlying connection.
+    #
+    # @example Connect the node.
+    #   node.connect
+    #
+    # @raise [ Errors::ConnectionFailure ] If connection failed.
+    #
+    # @return [ true ] If the connection suceeded.
+    #
+    # @since 2.0.0
+    def connect(conn)
+      start = Time.now
+      conn.connect
+      @latency = Time.now - start
+      @down_at = nil
+      true
+    end
+
     # Configure the node based on the return from the ismaster command.
     #
     # @api private
@@ -554,7 +566,7 @@ def configure(settings)
     def discover(*nodes)
       nodes.flatten.compact.each do |peer|
         node = Node.new(peer, options)
-        node.credentials.merge!(credentials)
+        node.credentials.merge!(@credentials)
         peers.push(node)
       end
     end
@@ -574,16 +586,14 @@ def discover(*nodes)
     def flush(ops = queue)
       operations, callbacks = ops.transpose
       logging(operations) do
-        ensure_connected do
-          replies = nil
-          connection do |conn|
-            conn.write(operations)
-            replies = conn.receive_replies(operations)
-          end
-          replies.zip(callbacks).map do |reply, callback|
-            callback ? callback[reply] : reply
-          end.last
+        replies = nil
+        ensure_connected do |conn|
+          conn.write(operations)
+          replies = conn.receive_replies(operations)
         end
+        replies.zip(callbacks).map do |reply, callback|
+          callback ? callback[reply] : reply
+        end.last
       end
     ensure
       ops.clear

lib/moped/operation/read.rb

@@ -47,12 +47,7 @@ def initialize(operation)
       def execute(node)
         node.process(operation) do |reply|
           if operation.failure?(reply)
-            if reply.unauthorized? && node.credentials.has_key?(database)
-              node.login(database, *node.credentials[database])
-              return execute(node)
-            else
-              raise operation.failure_exception(reply)
-            end
+            raise operation.failure_exception(reply)
           end
           operation.results(reply)
         end