diff --git a/lib/mongo.rb b/lib/mongo.rb
index 2c707dd2e2..ed93036ce0 100644
--- a/lib/mongo.rb
+++ b/lib/mongo.rb
@@ -18,6 +18,7 @@
 require 'mongo/loggable'
 require 'mongo/monitoring'
 require 'mongo/logger'
+require 'mongo/retryable'
 require 'mongo/operation'
 require 'mongo/error'
 require 'mongo/event'
diff --git a/lib/mongo/address.rb b/lib/mongo/address.rb
index 6669151e6b..403a886441 100644
--- a/lib/mongo/address.rb
+++ b/lib/mongo/address.rb
@@ -28,10 +28,16 @@ class Address
     # Mapping from socket family to resolver class.
     #
     # @since 2.0.0
-    FAMILY_MAP = { ::Socket::PF_UNIX => Unix,
-                   ::Socket::AF_INET6 => IPv6,
-                   ::Socket::AF_INET => IPv4
-                 }
+    FAMILY_MAP = {
+      ::Socket::PF_UNIX => Unix,
+      ::Socket::AF_INET6 => IPv6,
+      ::Socket::AF_INET => IPv4
+    }.freeze
+
+    # The localhost constant.
+    #
+    # @since 2.1.0
+    LOCALHOST = 'localhost'.freeze
 
     # @return [ String ] seed The seed address.
     attr_reader :seed
@@ -146,7 +152,7 @@ def to_s
     def initialize_resolver!(timeout, ssl_options)
       return Unix.new(seed.downcase) if seed.downcase =~ Unix::MATCH
 
-      family = (host == 'localhost') ? ::Socket::AF_INET : ::Socket::AF_UNSPEC
+      family = (host == LOCALHOST) ? ::Socket::AF_INET : ::Socket::AF_UNSPEC
       error = nil
       ::Socket.getaddrinfo(host, nil, family, ::Socket::SOCK_STREAM).each do |info|
         begin
diff --git a/lib/mongo/address/ipv4.rb b/lib/mongo/address/ipv4.rb
index 9d21650cf9..4a73016e06 100644
--- a/lib/mongo/address/ipv4.rb
+++ b/lib/mongo/address/ipv4.rb
@@ -35,6 +35,11 @@ class IPv4
       # @since 2.0.0
       MATCH = Regexp.new('/\./').freeze
 
+      # Split value constant.
+      #
+      # @since 2.1.0
+      SPLIT = ':'.freeze
+
       # Parse an IPv4 address into its host and port.
       #
       # @example Parse the address.
@@ -46,7 +51,7 @@ class IPv4
       #
       # @since 2.0.0
       def self.parse(address)
-        parts = address.split(':')
+        parts = address.split(SPLIT)
         host = parts[0]
         port = (parts[1] || 27017).to_i
         [ host, port ]
diff --git a/lib/mongo/auth/cr/conversation.rb b/lib/mongo/auth/cr/conversation.rb
index dde5c2d5af..c80e256cd7 100644
--- a/lib/mongo/auth/cr/conversation.rb
+++ b/lib/mongo/auth/cr/conversation.rb
@@ -111,7 +111,7 @@ def initialize(user)
         private
 
         def validate!(reply)
-          raise Unauthorized.new(user) if reply.documents[0]['ok'] != 1
+          raise Unauthorized.new(user) if reply.documents[0][Operation::Result::OK] != 1
           @nonce = reply.documents[0][Auth::NONCE]
           @reply = reply
         end
diff --git a/lib/mongo/auth/ldap/conversation.rb b/lib/mongo/auth/ldap/conversation.rb
index eb57739448..c60b3b7324 100644
--- a/lib/mongo/auth/ldap/conversation.rb
+++ b/lib/mongo/auth/ldap/conversation.rb
@@ -87,7 +87,7 @@ def payload
         end
 
         def validate!(reply)
-          raise Unauthorized.new(user) if reply.documents[0]['ok'] != 1
+          raise Unauthorized.new(user) if reply.documents[0][Operation::Result::OK] != 1
           @reply = reply
         end
       end
diff --git a/lib/mongo/auth/scram/conversation.rb b/lib/mongo/auth/scram/conversation.rb
index 9a16388d17..e051b09e90 100644
--- a/lib/mongo/auth/scram/conversation.rb
+++ b/lib/mongo/auth/scram/conversation.rb
@@ -448,7 +448,7 @@ def validate_first_message!(reply)
         end
 
         def validate!(reply)
-          raise Unauthorized.new(user) unless reply.documents[0]['ok'] == 1
+          raise Unauthorized.new(user) unless reply.documents[0][Operation::Result::OK] == 1
           @reply = reply
         end
       end
diff --git a/lib/mongo/auth/user/view.rb b/lib/mongo/auth/user/view.rb
index f172582eaa..15cdfda9a2 100644
--- a/lib/mongo/auth/user/view.rb
+++ b/lib/mongo/auth/user/view.rb
@@ -90,8 +90,8 @@ def remove(name)
         def update(user_or_name, options = {})
           user = generate(user_or_name, options)
           Operation::Write::UpdateUser.new(
-              user: user,
-              db_name: database.name
+            user: user,
+            db_name: database.name
           ).execute(next_primary.context)
         end
 
diff --git a/lib/mongo/auth/x509/conversation.rb b/lib/mongo/auth/x509/conversation.rb
index 8e1e6107c9..11016a5a61 100644
--- a/lib/mongo/auth/x509/conversation.rb
+++ b/lib/mongo/auth/x509/conversation.rb
@@ -83,7 +83,7 @@ def initialize(user)
         private
 
         def validate!(reply)
-          raise Unauthorized.new(user) if reply.documents[0]['ok'] != 1
+          raise Unauthorized.new(user) if reply.documents[0][Operation::Result::OK] != 1
           @reply = reply
         end
       end
diff --git a/lib/mongo/bulk_write.rb b/lib/mongo/bulk_write.rb
index d7ff68d9f9..da13ee6353 100644
--- a/lib/mongo/bulk_write.rb
+++ b/lib/mongo/bulk_write.rb
@@ -23,6 +23,7 @@
 module Mongo
   class BulkWrite
     extend Forwardable
+    include Retryable
 
     # @return [ Mongo::Collection ] collection The collection.
     attr_reader :collection
@@ -45,17 +46,19 @@ class BulkWrite
     #
     # @since 2.1.0
     def execute
-      server = next_primary
       operation_id = Monitoring.next_operation_id
       result_combiner = ResultCombiner.new
-      operations.each do |operation|
-        execute_operation(
-          operation.keys.first,
-          operation.values.first,
-          server,
-          operation_id,
-          result_combiner
-        )
+      write_with_retry do
+        server = next_primary
+        operations.each do |operation|
+          execute_operation(
+            operation.keys.first,
+            operation.values.first,
+            server,
+            operation_id,
+            result_combiner
+          )
+        end
       end
       result_combiner.result
     end
diff --git a/lib/mongo/client.rb b/lib/mongo/client.rb
index 74fead3fe2..7e9d787b28 100644
--- a/lib/mongo/client.rb
+++ b/lib/mongo/client.rb
@@ -274,7 +274,7 @@ def reconnect
     #
     # @since 2.0.5
     def database_names
-      list_databases.collect{ |info| info['name'] }
+      list_databases.collect{ |info| info[Database::NAME] }
     end
 
     # Get info for each database.
@@ -286,7 +286,7 @@ def database_names
     #
     # @since 2.0.5
     def list_databases
-      use(Database::ADMIN).command(listDatabases: 1).first['databases']
+      use(Database::ADMIN).command(listDatabases: 1).first[Database::DATABASES]
     end
 
     private
diff --git a/lib/mongo/collection.rb b/lib/mongo/collection.rb
index f2f619bd46..3bca3b087a 100644
--- a/lib/mongo/collection.rb
+++ b/lib/mongo/collection.rb
@@ -23,6 +23,12 @@ module Mongo
   # @since 2.0.0
   class Collection
     extend Forwardable
+    include Retryable
+
+    # The capped option.
+    #
+    # @since 2.1.0
+    CAPPED = 'capped'.freeze
 
     # @return [ Mongo::Database ] The database the collection resides in.
     attr_reader :database
@@ -136,7 +142,7 @@ def with(new_options)
     #
     # @since 2.0.0
     def capped?
-      database.command(:collstats => name).documents[0]['capped']
+      database.command(:collstats => name).documents[0][CAPPED]
     end
 
     # Force the collection to be created in the database.
@@ -309,13 +315,15 @@ def inspect
     #
     # @since 2.0.0
     def insert_one(document, options = {})
-      Operation::Write::Insert.new(
-        :documents => [ document ],
-        :db_name => database.name,
-        :coll_name => name,
-        :write_concern => write_concern,
-        :options => options
-      ).execute(next_primary.context)
+      write_with_retry do
+        Operation::Write::Insert.new(
+          :documents => [ document ],
+          :db_name => database.name,
+          :coll_name => name,
+          :write_concern => write_concern,
+          :options => options
+        ).execute(next_primary.context)
+      end
     end
 
     # Insert the provided documents into the collection.
diff --git a/lib/mongo/collection/view.rb b/lib/mongo/collection/view.rb
index cc15686ecf..6009d0bb28 100644
--- a/lib/mongo/collection/view.rb
+++ b/lib/mongo/collection/view.rb
@@ -44,6 +44,7 @@ class View
       include Immutable
       include Iterable
       include Readable
+      include Retryable
       include Explainable
       include Writable
 
diff --git a/lib/mongo/collection/view/aggregation.rb b/lib/mongo/collection/view/aggregation.rb
index 0aa85afcbe..b6cf5bd606 100644
--- a/lib/mongo/collection/view/aggregation.rb
+++ b/lib/mongo/collection/view/aggregation.rb
@@ -26,6 +26,7 @@ class Aggregation
         include Iterable
         include Explainable
         include Loggable
+        include Retryable
 
         # @return [ View ] view The collection view.
         attr_reader :view
@@ -42,10 +43,15 @@ class Aggregation
         #
         # @since 2.1.0
         OPTIONS_MAP = {
-                       :allow_disk_use => :allowDiskUse,
-                       :max_time_ms => :maxTimeMS,
-                       :explain => :explain
-                      }
+          :allow_disk_use => :allowDiskUse,
+          :max_time_ms => :maxTimeMS,
+          :explain => :explain
+        }.freeze
+
+        # The reroute message.
+        #
+        # @since 2.1.0
+        REROUTE = 'Rerouting the Aggregation operation to the primary server.'.freeze
 
         # Set to true if disk usage is allowed during the aggregation.
         #
@@ -143,7 +149,7 @@ def secondary_ok?
 
         def send_initial_query(server)
           unless valid_server?(server)
-            log_warn('Rerouting the Aggregation operation to the primary server.')
+            log_warn(REROUTE)
             server = cluster.next_primary
           end
           initial_query_op.execute(server.context)
diff --git a/lib/mongo/collection/view/iterable.rb b/lib/mongo/collection/view/iterable.rb
index 9bcc5117e5..d5e3141448 100644
--- a/lib/mongo/collection/view/iterable.rb
+++ b/lib/mongo/collection/view/iterable.rb
@@ -35,8 +35,12 @@ module Iterable
         #
         # @yieldparam [ Hash ] Each matching document.
         def each
-          server = read.select_server(cluster)
-          @cursor = Cursor.new(view, send_initial_query(server), server)
+          @cursor = nil
+          read_with_retry do
+            server = read.select_server(cluster)
+            result = send_initial_query(server)
+            @cursor = Cursor.new(view, result, server)
+          end
           @cursor.each do |doc|
             yield doc
           end if block_given?
diff --git a/lib/mongo/collection/view/map_reduce.rb b/lib/mongo/collection/view/map_reduce.rb
index bee1d20caf..cf93fa82fd 100644
--- a/lib/mongo/collection/view/map_reduce.rb
+++ b/lib/mongo/collection/view/map_reduce.rb
@@ -24,8 +24,18 @@ class MapReduce
         extend Forwardable
         include Enumerable
         include Immutable
-        include Iterable
         include Loggable
+        include Retryable
+
+        # The inline option.
+        #
+        # @since 2.1.0
+        INLINE = 'inline'.freeze
+
+        # Reroute message.
+        #
+        # @since 2.1.0
+        REROUTE = 'Rerouting the MapReduce operation to the primary server.'.freeze
 
         # @return [ View ] view The collection view.
         attr_reader :view
@@ -42,6 +52,31 @@ class MapReduce
         # Delegate necessary operations to the collection.
         def_delegators :collection, :database
 
+        # Iterate through documents returned by the map/reduce.
+        #
+        # @example Iterate through the result of the map/reduce.
+        #   map_reduce.each do |document|
+        #     p document
+        #   end
+        #
+        # @return [ Enumerator ] The enumerator.
+        #
+        # @since 2.0.0
+        #
+        # @yieldparam [ Hash ] Each matching document.
+        def each
+          @cursor = nil
+          write_with_retry do
+            server = read.select_server(cluster)
+            result = send_initial_query(server)
+            @cursor = Cursor.new(view, result, server)
+          end
+          @cursor.each do |doc|
+            yield doc
+          end if block_given?
+          @cursor.to_enum
+        end
+
         # Set or get the finalize function for the operation.
         #
         # @example Set the finalize function.
@@ -148,7 +183,7 @@ def verbose(value = nil)
         private
 
         def inline?
-          out.nil? || out == { inline: 1 } || out == { 'inline' => 1 }
+          out.nil? || out == { inline: 1 } || out == { INLINE => 1 }
         end
 
         def map_reduce_spec
@@ -182,13 +217,12 @@ def valid_server?(server)
         end
 
         def secondary_ok?
-          out.respond_to?(:keys) &&
-              out.keys.first.to_s.downcase == 'inline'
+          out.respond_to?(:keys) && out.keys.first.to_s.downcase == INLINE
         end
 
         def send_initial_query(server)
           unless valid_server?(server)
-            log_warn('Rerouting the MapReduce operation to the primary server.')
+            log_warn(REROUTE)
             server = cluster.next_primary
           end
           result = initial_query_op.execute(server.context)
diff --git a/lib/mongo/collection/view/readable.rb b/lib/mongo/collection/view/readable.rb
index c1be99ffd6..50e6e49075 100644
--- a/lib/mongo/collection/view/readable.rb
+++ b/lib/mongo/collection/view/readable.rb
@@ -25,17 +25,17 @@ module Readable
         #
         # @since 2.0.0
         SPECIAL_FIELDS = {
-            :sort => :$orderby,
-            :hint => :$hint,
-            :comment => :$comment,
-            :snapshot => :$snapshot,
-            :max_scan => :$maxScan,
-            :max_value => :$max,
-            :min_value => :$min,
-            :max_time_ms => :$maxTimeMS,
-            :return_key => :$returnKey,
-            :show_disk_loc => :$showDiskLoc,
-            :explain => :$explain
+          :sort => :$orderby,
+          :hint => :$hint,
+          :comment => :$comment,
+          :snapshot => :$snapshot,
+          :max_scan => :$maxScan,
+          :max_value => :$max,
+          :min_value => :$min,
+          :max_time_ms => :$maxTimeMS,
+          :return_key => :$returnKey,
+          :show_disk_loc => :$showDiskLoc,
+          :explain => :$explain
         }.freeze
 
         # Options to cursor flags mapping.
diff --git a/lib/mongo/collection/view/writable.rb b/lib/mongo/collection/view/writable.rb
index 1bdfa8682b..abeab89d7c 100644
--- a/lib/mongo/collection/view/writable.rb
+++ b/lib/mongo/collection/view/writable.rb
@@ -35,7 +35,9 @@ def find_one_and_delete
           cmd[:fields] = projection if projection
           cmd[:sort] = sort if sort
           cmd[:maxTimeMS] = max_time_ms if max_time_ms
-          database.command(cmd).first['value']
+          write_with_retry do
+            database.command(cmd).first['value']
+          end
         end
 
         # Finds a single document and replaces it.
@@ -81,8 +83,10 @@ def find_one_and_update(document, opts = {})
           cmd[:new] = !!(opts[:return_document] && opts[:return_document] == :after)
           cmd[:upsert] = opts[:upsert] if opts[:upsert]
           cmd[:maxTimeMS] = max_time_ms if max_time_ms
-          value = database.command(cmd).first['value']
-          value unless value.nil? || value.empty?
+          write_with_retry do
+            value = database.command(cmd).first['value']
+            value unless value.nil? || value.empty?
+          end
         end
 
         # Remove documents from the collection.
@@ -166,24 +170,28 @@ def update_one(spec, opts = {})
         private
 
         def remove(value)
-          Operation::Write::Delete.new(
-            :delete => { q: selector, limit: value },
-            :db_name => collection.database.name,
-            :coll_name => collection.name,
-            :write_concern => collection.write_concern
-          ).execute(next_primary.context)
+          write_with_retry do
+            Operation::Write::Delete.new(
+              :delete => { q: selector, limit: value },
+              :db_name => collection.database.name,
+              :coll_name => collection.name,
+              :write_concern => collection.write_concern
+            ).execute(next_primary.context)
+          end
         end
 
         def update(spec, multi, opts)
-          Operation::Write::Update.new(
-            :update => { q: selector,
-                         u: spec,
-                         multi: multi,
-                         upsert: !!opts[:upsert] },
-            :db_name => collection.database.name,
-            :coll_name => collection.name,
-            :write_concern => collection.write_concern
-          ).execute(next_primary.context)
+          write_with_retry do
+            Operation::Write::Update.new(
+              :update => { q: selector,
+                           u: spec,
+                           multi: multi,
+                           upsert: !!opts[:upsert] },
+              :db_name => collection.database.name,
+              :coll_name => collection.name,
+              :write_concern => collection.write_concern
+            ).execute(next_primary.context)
+          end
         end
       end
     end
diff --git a/lib/mongo/database.rb b/lib/mongo/database.rb
index f4e2ae46fe..ea6744c67c 100644
--- a/lib/mongo/database.rb
+++ b/lib/mongo/database.rb
@@ -38,6 +38,16 @@ class Database
     # @since 2.0.0
     DEFAULT_OPTIONS = { :database => ADMIN }.freeze
 
+    # Database name field constant.
+    #
+    # @since 2.1.0
+    NAME = 'name'.freeze
+
+    # Databases constant.
+    #
+    # @since 2.1.0
+    DATABASES = 'databases'.freeze
+
     # The name of the collection that holds all the collection names.
     #
     # @since 2.0.0
diff --git a/lib/mongo/database/view.rb b/lib/mongo/database/view.rb
index 411def0b76..3d6e7bc732 100644
--- a/lib/mongo/database/view.rb
+++ b/lib/mongo/database/view.rb
@@ -53,7 +53,7 @@ def collection_names(options = {})
         server = next_primary
         collections_info(server).collect do |info|
           server.context.features.list_collections_enabled? ?
-            info['name'] : info['name'].sub("#{@database.name}.", '')
+            info[Database::NAME] : info[Database::NAME].sub("#{@database.name}.", '')
         end
       end
 
diff --git a/lib/mongo/dbref.rb b/lib/mongo/dbref.rb
index f7f65027eb..227560df73 100644
--- a/lib/mongo/dbref.rb
+++ b/lib/mongo/dbref.rb
@@ -53,8 +53,8 @@ class DBRef
     #
     # @since 2.1.0
     def as_json(*args)
-      document = { '$ref' => collection, '$id' => id }
-      document.merge!('$db' => database) if database
+      document = { COLLECTION => collection, ID => id }
+      document.merge!(DATABASE => database) if database
       document
     end
 
@@ -101,8 +101,8 @@ module ClassMethods
       # @since 2.0.0
       def from_bson(bson)
         decoded = super
-        if ref = decoded['$ref']
-          decoded = DBRef.new(ref, decoded['$id'], decoded['$db'])
+        if ref = decoded[COLLECTION]
+          decoded = DBRef.new(ref, decoded[ID], decoded[DATABASE])
         end
         decoded
       end
diff --git a/lib/mongo/index.rb b/lib/mongo/index.rb
index d368842458..dbbc7061a1 100644
--- a/lib/mongo/index.rb
+++ b/lib/mongo/index.rb
@@ -21,6 +21,11 @@ module Mongo
   # @since 2.0.0
   module Index
 
+    # Wildcard constant for all.
+    #
+    # @since 2.1.0
+    ALL = '*'.freeze
+
     # Specify ascending order for an index.
     #
     # @since 2.0.0
diff --git a/lib/mongo/index/view.rb b/lib/mongo/index/view.rb
index 722e42a1d5..c9cb583273 100644
--- a/lib/mongo/index/view.rb
+++ b/lib/mongo/index/view.rb
@@ -76,7 +76,7 @@ class View
       #
       # @since 2.0.0
       def drop_one(name)
-        raise Error::MultiIndexDrop.new if name == '*'
+        raise Error::MultiIndexDrop.new if name == Index::ALL
         drop_by_name(name)
       end
 
@@ -89,7 +89,7 @@ def drop_one(name)
       #
       # @since 2.0.0
       def drop_all
-        drop_by_name('*')
+        drop_by_name(Index::ALL)
       end
 
       # Creates an index on the collection.
diff --git a/lib/mongo/operation/result.rb b/lib/mongo/operation/result.rb
index cd76396763..28c9a58a28 100644
--- a/lib/mongo/operation/result.rb
+++ b/lib/mongo/operation/result.rb
@@ -190,12 +190,24 @@ def returned_count
       def successful?
         return true if !acknowledged?
         if first_document.has_key?(OK)
-          first_document[OK] == 1 && parser.message.empty?
+          ok? && parser.message.empty?
         else
           !query_failure? && parser.message.empty?
         end
       end
 
+      # Check the first document's ok field.
+      #
+      # @example Check the ok field.
+      #   result.ok?
+      #
+      # @return [ true, false ] If the command returned ok.
+      #
+      # @since 2.1.0
+      def ok?
+        first_document[OK] == 1
+      end
+
       # Validate the result by checking for any errors.
       #
       # @note This only checks for errors with writes since authentication is
diff --git a/lib/mongo/retryable.rb b/lib/mongo/retryable.rb
new file mode 100644
index 0000000000..5e67b2d58d
--- /dev/null
+++ b/lib/mongo/retryable.rb
@@ -0,0 +1,83 @@
+# Copyright (C) 2015 MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+module Mongo
+
+  # Defines basic behaviour around retrying operations.
+  #
+  # @since 2.1.0
+  module Retryable
+
+    # The not master error message.
+    #
+    # @since 2.1.0
+    NOT_MASTER = 'not master'.freeze
+
+    # Execute a read operation with a retry.
+    #
+    # @example Execute the read.
+    #   read_with_retry do
+    #     ...
+    #   end
+    #
+    # @note This only retries read operations on socket errors.
+    #
+    # @param [ Proc ] block The block to execute.
+    #
+    # @return [ Result ] The result of the operation.
+    #
+    # @since 2.1.0
+    def read_with_retry(&block)
+      begin
+        block.call
+      rescue Error::SocketError, Error::SocketTimeoutError
+        retry_operation(&block)
+      end
+    end
+
+    # Execute a write operation with a retry.
+    #
+    # @example Execute the write.
+    #   write_with_retry do
+    #     ...
+    #   end
+    #
+    # @note This only retries operations on not master failures, since it is
+    #   the only case we can be sure a partial write did not already occur.
+    #
+    # @param [ Proc ] block The block to execute.
+    #
+    # @return [ Result ] The result of the operation.
+    #
+    # @since 2.1.0
+    def write_with_retry(&block)
+      begin
+        block.call
+      rescue Error::OperationFailure => e
+        if e.message.include?(NOT_MASTER)
+          retry_operation(&block)
+        else
+          raise e
+        end
+      end
+    end
+
+    private
+
+    def retry_operation(&block)
+      cluster.scan!
+      block.call
+    end
+  end
+end
diff --git a/lib/mongo/server.rb b/lib/mongo/server.rb
index b0b1c794b7..410ff83a04 100644
--- a/lib/mongo/server.rb
+++ b/lib/mongo/server.rb
@@ -92,6 +92,21 @@ def context
       Context.new(self)
     end
 
+    # Determine if a connection to the server is able to be established and
+    # messages can be sent to it.
+    #
+    # @example Is the server connectable?
+    #   server.connectable?
+    #
+    # @return [ true, false ] If the server is connectable.
+    #
+    # @since 2.1.0
+    def connectable?
+      context.with_connection do |connection|
+        connection.connectable?
+      end
+    end
+
     # Disconnect the server from the connection.
     #
     # @example Disconnect the server.
@@ -101,9 +116,7 @@ def context
     #
     # @since 2.0.0
     def disconnect!
-      context.with_connection do |connection|
-        connection.disconnect!
-      end
+      pool.disconnect!
       monitor.stop! and true
     end
 
diff --git a/lib/mongo/server/connectable.rb b/lib/mongo/server/connectable.rb
index 03caa0ffbf..0385dc1491 100644
--- a/lib/mongo/server/connectable.rb
+++ b/lib/mongo/server/connectable.rb
@@ -20,6 +20,11 @@ class Server
     # @since 2.0.0
     module Connectable
 
+      # The ssl option prefix.
+      #
+      # @since 2.1.0
+      SSL = 'ssl'.freeze
+
       # The default time in seconds to timeout a connection attempt.
       #
       # @since 2.0.0
@@ -34,6 +39,19 @@ module Connectable
       # @return [ Integer ] pid The process id when the connection was created.
       attr_reader :pid
 
+      # Determine if the server is connectable. This will check not only if the
+      # connection exists, but if messages can sent to it successfully.
+      #
+      # @example Is the server connectable?
+      #   connection.connectable?
+      #
+      # @return [ true, false ] If the connection is connectable.
+      #
+      # @since 2.1.0
+      def connectable?
+        begin; ping; rescue; false; end
+      end
+
       # Determine if the connection is currently connected.
       #
       # @example Is the connection connected?
@@ -71,7 +89,7 @@ def ensure_connected
         connect! if socket.nil? || !socket.alive?
         begin
           yield socket
-        rescue Error::SocketError, Error::SocketTimeoutError => e
+        rescue Exception => e
           disconnect!
           raise e
         end
diff --git a/lib/mongo/server/connection.rb b/lib/mongo/server/connection.rb
index 76a46cc33e..bb3bc6e060 100644
--- a/lib/mongo/server/connection.rb
+++ b/lib/mongo/server/connection.rb
@@ -23,6 +23,21 @@ class Connection
       include Monitoring::Publishable
       extend Forwardable
 
+      # The ping command.
+      #
+      # @since 2.1.0
+      PING = { :ping => 1 }.freeze
+
+      # Ping message.
+      #
+      # @since 2.1.0
+      PING_MESSAGE = Protocol::Query.new(Database::ADMIN, Database::COMMAND, PING, :limit => -1)
+
+      # The ping message as raw bytes.
+      #
+      # @since 2.1.0
+      PING_BYTES = PING_MESSAGE.serialize.freeze
+
       # @return [ Mongo::Auth::CR, Mongo::Auth::X509, Mongo::Auth:LDAP, Mongo::Auth::SCRAM ]
       #   authenticator The authentication strategy.
       attr_reader :authenticator
@@ -133,12 +148,31 @@ def initialize(server, options = {})
         @monitoring = server.monitoring
         @options = options.freeze
         @server = server
-        @ssl_options = options.reject { |k, v| !k.to_s.start_with?('ssl') }
+        @ssl_options = options.reject { |k, v| !k.to_s.start_with?(SSL) }
         @socket = nil
         @pid = Process.pid
         setup_authentication!
       end
 
+      # Ping the connection to see if the server is responding to commands.
+      # This is non-blocking on the server side.
+      #
+      # @example Ping the connection.
+      #   connection.ping
+      #
+      # @note This uses a pre-serialized ping message for optimization.
+      #
+      # @return [ true, false ] If the server is accepting connections.
+      #
+      # @since 2.1.0
+      def ping
+        ensure_connected do |socket|
+          socket.write(PING_BYTES)
+          reply = Protocol::Reply.deserialize(socket)
+          reply.documents[0][Operation::Result::OK] == 1
+        end
+      end
+
       private
 
       def deliver(messages)
diff --git a/lib/mongo/server/connection_pool.rb b/lib/mongo/server/connection_pool.rb
index eefb2c11c6..19389455df 100644
--- a/lib/mongo/server/connection_pool.rb
+++ b/lib/mongo/server/connection_pool.rb
@@ -54,6 +54,18 @@ def checkout
         queue.dequeue
       end
 
+      # Disconnect the connection pool.
+      #
+      # @example Disconnect the connection pool.
+      #   pool.disconnect!
+      #
+      # @return [ true ] true.
+      #
+      # @since 2.1.0
+      def disconnect!
+        queue.disconnect!
+      end
+
       # Create the new connection pool.
       #
       # @example Create the new connection pool.
diff --git a/lib/mongo/server/connection_pool/queue.rb b/lib/mongo/server/connection_pool/queue.rb
index 9fce1aab56..c144d9663d 100644
--- a/lib/mongo/server/connection_pool/queue.rb
+++ b/lib/mongo/server/connection_pool/queue.rb
@@ -63,6 +63,21 @@ def dequeue
           end
         end
 
+        # Disconnect all connections in the queue.
+        #
+        # @example Disconnect all connections.
+        #   queue.disconnect!
+        #
+        # @return [ true ] Always true.
+        #
+        # @since 2.1.0
+        def disconnect!
+          mutex.synchronize do
+            queue.each{ |connection| connection.disconnect! }
+            true
+          end
+        end
+
         # Enqueue a connection in the queue.
         #
         # @example Enqueue a connection.
diff --git a/lib/mongo/server/monitor/connection.rb b/lib/mongo/server/monitor/connection.rb
index df306d22fa..9b3212d733 100644
--- a/lib/mongo/server/monitor/connection.rb
+++ b/lib/mongo/server/monitor/connection.rb
@@ -96,7 +96,7 @@ def dispatch(messages)
         def initialize(address, options = {})
           @address = address
           @options = options.freeze
-          @ssl_options = options.reject { |k, v| !k.to_s.start_with?('ssl') }
+          @ssl_options = options.reject { |k, v| !k.to_s.start_with?(SSL) }
           @socket = nil
           @pid = Process.pid
         end
diff --git a/lib/mongo/server_selector/selectable.rb b/lib/mongo/server_selector/selectable.rb
index b5f82c8da5..9518da1152 100644
--- a/lib/mongo/server_selector/selectable.rb
+++ b/lib/mongo/server_selector/selectable.rb
@@ -80,14 +80,11 @@ def initialize(options = {})
       def select_server(cluster)
         deadline = Time.now + server_selection_timeout
         while (deadline - Time.now) > 0
-          if cluster.single?
-            servers = cluster.servers
-          elsif cluster.sharded?
-            servers = near_servers(cluster.servers)
-          else
-            servers = select(cluster.servers)
+          servers = candidates(cluster)
+          if servers && !servers.compact.empty?
+            server = servers.first
+            return server if server.connectable?
           end
-          return servers.first if servers && !servers.compact.empty?
           cluster.scan!
         end
         raise Error::NoServerAvailable.new(self)
@@ -120,6 +117,16 @@ def local_threshold
 
       private
 
+      def candidates(cluster)
+        if cluster.single?
+          cluster.servers
+        elsif cluster.sharded?
+          near_servers(cluster.servers)
+        else
+          select(cluster.servers)
+        end
+      end
+
       # Select the primary from a list of provided candidates.
       #
       # @param [ Array ] candidates List of candidate servers to select the
diff --git a/spec/mongo/retryable_spec.rb b/spec/mongo/retryable_spec.rb
new file mode 100644
index 0000000000..3a622510bc
--- /dev/null
+++ b/spec/mongo/retryable_spec.rb
@@ -0,0 +1,147 @@
+require 'spec_helper'
+
+describe Mongo::Retryable do
+
+  let(:klass) do
+    Class.new do
+      include Mongo::Retryable
+
+      attr_reader :cluster
+      attr_reader :operation
+
+      def initialize(operation, cluster)
+        @operation = operation
+        @cluster = cluster
+      end
+
+      def read
+        read_with_retry do
+          operation.execute
+        end
+      end
+
+      def write
+        write_with_retry do
+          operation.execute
+        end
+      end
+    end
+  end
+
+  describe '#read_with_retry' do
+
+    let(:operation) do
+      double('operation')
+    end
+
+    let(:cluster) do
+      double('cluster')
+    end
+
+    let(:retryable) do
+      klass.new(operation, cluster)
+    end
+
+    context 'when no exception occurs' do
+
+      before do
+        expect(operation).to receive(:execute).and_return(true)
+      end
+
+      it 'executes the operation once' do
+        expect(retryable.read).to be true
+      end
+    end
+
+    context 'when a socket error occurs' do
+
+      before do
+        expect(operation).to receive(:execute).and_raise(Mongo::Error::SocketError).ordered
+        expect(cluster).to receive(:scan!).and_return(true).ordered
+        expect(operation).to receive(:execute).and_return(true).ordered
+      end
+
+      it 'executes the operation twice' do
+        expect(retryable.read).to be true
+      end
+    end
+
+    context 'when a socket timeout error occurs' do
+
+      before do
+        expect(operation).to receive(:execute).and_raise(Mongo::Error::SocketTimeoutError).ordered
+        expect(cluster).to receive(:scan!).and_return(true).ordered
+        expect(operation).to receive(:execute).and_return(true).ordered
+      end
+
+      it 'executes the operation twice' do
+        expect(retryable.read).to be true
+      end
+    end
+
+    context 'when an operation failure occurs' do
+
+      before do
+        expect(operation).to receive(:execute).and_raise(Mongo::Error::OperationFailure).ordered
+      end
+
+      it 'raises an exception' do
+        expect {
+          retryable.read
+        }.to raise_error(Mongo::Error::OperationFailure)
+      end
+    end
+  end
+
+  describe '#write_with_retry' do
+
+    let(:operation) do
+      double('operation')
+    end
+
+    let(:cluster) do
+      double('cluster')
+    end
+
+    let(:retryable) do
+      klass.new(operation, cluster)
+    end
+
+    context 'when no exception occurs' do
+
+      before do
+        expect(operation).to receive(:execute).and_return(true)
+      end
+
+      it 'executes the operation once' do
+        expect(retryable.write).to be true
+      end
+    end
+
+    context 'when a not master error occurs' do
+
+      before do
+        expect(operation).to receive(:execute).and_raise(Mongo::Error::OperationFailure.new('not master')).ordered
+        expect(cluster).to receive(:scan!).and_return(true).ordered
+        expect(operation).to receive(:execute).and_return(true).ordered
+      end
+
+      it 'executes the operation twice' do
+        expect(retryable.write).to be true
+      end
+    end
+
+    context 'when a normal operation failure occurs' do
+
+      before do
+        expect(operation).to receive(:execute).and_raise(Mongo::Error::OperationFailure).ordered
+      end
+
+      it 'raises an exception' do
+        expect {
+          retryable.write
+        }.to raise_error(Mongo::Error::OperationFailure)
+      end
+    end
+  end
+end
diff --git a/spec/mongo/server/connection_pool/queue_spec.rb b/spec/mongo/server/connection_pool/queue_spec.rb
index 00877e79fc..b96a07b918 100644
--- a/spec/mongo/server/connection_pool/queue_spec.rb
+++ b/spec/mongo/server/connection_pool/queue_spec.rb
@@ -47,6 +47,22 @@
     end
   end
 
+  describe '#disconnect!' do
+
+    let(:connection) do
+      double('connection')
+    end
+
+    let(:queue) do
+      described_class.new(:max_pool_size => 1) { connection }
+    end
+
+    it 'disconnects all connections in the queue' do
+      expect(connection).to receive(:disconnect!)
+      queue.disconnect!
+    end
+  end
+
   describe '#enqueue' do
 
     let(:connection) do
diff --git a/spec/mongo/server/connection_pool_spec.rb b/spec/mongo/server/connection_pool_spec.rb
index ae2188414b..a1e22c99f4 100644
--- a/spec/mongo/server/connection_pool_spec.rb
+++ b/spec/mongo/server/connection_pool_spec.rb
@@ -28,6 +28,10 @@
       described_class.get(server)
     end
 
+    after do
+      server.disconnect!
+    end
+
     context 'when a connection is checked out on the thread' do
 
       let!(:connection) do
@@ -96,6 +100,26 @@
     end
   end
 
+  describe '#disconnect!' do
+
+    let(:server) do
+      Mongo::Server.new(address, double('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!).and_call_original
+      pool.disconnect!
+    end
+  end
+
   describe '.get' do
 
     let(:server) do
@@ -106,6 +130,10 @@
       described_class.get(server)
     end
 
+    after do
+      server.disconnect!
+    end
+
     it 'returns the pool for the server' do
       expect(pool).to eql(described_class.get(server))
     end
@@ -121,6 +149,10 @@
       described_class.get(server)
     end
 
+    after do
+      server.disconnect!
+    end
+
     it 'includes the object id' do
       expect(pool.inspect).to include(pool.object_id.to_s)
     end
diff --git a/spec/mongo/server/connection_spec.rb b/spec/mongo/server/connection_spec.rb
index 3181cde77c..4b4b3441c6 100644
--- a/spec/mongo/server/connection_spec.rb
+++ b/spec/mongo/server/connection_spec.rb
@@ -18,6 +18,43 @@
     Mongo::Server.new(address, double('cluster'), monitoring, listeners, TEST_OPTIONS)
   end
 
+  after do
+    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:27016')
+      end
+
+      let(:bad_server) do
+        Mongo::Server.new(bad_address, double('cluster'), monitoring, listeners, TEST_OPTIONS)
+      end
+
+      let(:connection) do
+        described_class.new(bad_server)
+      end
+
+      it 'returns false' do
+        expect(connection).to_not be_connectable
+      end
+    end
+  end
+
   describe '#connect!' do
 
     context 'when no socket exists' do
diff --git a/spec/mongo/server_selection_spec.rb b/spec/mongo/server_selection_spec.rb
index da3930559e..24ef6fb2d2 100644
--- a/spec/mongo/server_selection_spec.rb
+++ b/spec/mongo/server_selection_spec.rb
@@ -39,6 +39,7 @@
             allow(s).to receive(:tags).and_return(server['tags'])
             allow(s).to receive(:secondary?).and_return(server['type'] == 'RSSecondary')
             allow(s).to receive(:primary?).and_return(server['type'] == 'RSPrimary')
+            allow(s).to receive(:connectable?).and_return(true)
           end
         end
       end
@@ -49,6 +50,7 @@
           Mongo::Server.new(address, double('cluster'), monitoring, listeners, TEST_OPTIONS).tap do |s|
             allow(s).to receive(:average_round_trip_time).and_return(server['avg_rtt_ms'])
             allow(s).to receive(:tags).and_return(server['tags'])
+            allow(s).to receive(:connectable?).and_return(true)
           end
         end
       end
diff --git a/spec/mongo/server_selector/primary_spec.rb b/spec/mongo/server_selector/primary_spec.rb
index bb0c03d7b4..6a37ad6025 100644
--- a/spec/mongo/server_selector/primary_spec.rb
+++ b/spec/mongo/server_selector/primary_spec.rb
@@ -20,10 +20,15 @@
     end
 
     context 'tag sets provided' do
-      let(:tag_sets) { [tag_set] }
+
+      let(:tag_sets) do
+        [ tag_set ]
+      end
 
       it 'raises an error' do
-        expect{selector.tag_sets}.to raise_error
+        expect {
+          selector.tag_sets
+        }.to raise_error(Mongo::Error::InvalidServerPreference)
       end
     end
   end
diff --git a/spec/mongo/server_spec.rb b/spec/mongo/server_spec.rb
index 7f443c1ff8..ef426d8465 100644
--- a/spec/mongo/server_spec.rb
+++ b/spec/mongo/server_spec.rb
@@ -24,6 +24,10 @@
       described_class.new(address, cluster, monitoring, listeners, TEST_OPTIONS)
     end
 
+    after do
+      server.disconnect!
+    end
+
     context 'when the other is not a server' do
 
       let(:other) do
@@ -65,6 +69,43 @@
     end
   end
 
+  describe '#connectable?' do
+
+    context 'when the server is connectable' do
+
+      let(:server) do
+        described_class.new(address, cluster, monitoring, listeners, TEST_OPTIONS)
+      end
+
+      after do
+        server.disconnect!
+      end
+
+      it 'returns true' do
+        expect(server).to be_connectable
+      end
+    end
+
+    context 'when the server is not connectable' do
+
+      let(:bad_address) do
+        Mongo::Address.new('127.0.0.1:27016')
+      end
+
+      let(:server) do
+        described_class.new(bad_address, cluster, monitoring, listeners, TEST_OPTIONS)
+      end
+
+      before do
+        server.disconnect!
+      end
+
+      it 'returns false' do
+        expect(server).to_not be_connectable
+      end
+    end
+  end
+
   describe '#context' do
 
     let(:server) do
@@ -75,6 +116,10 @@
       server.context
     end
 
+    after do
+      server.disconnect!
+    end
+
     it 'returns a new server context' do
       expect(context.server).to eq(server)
     end
@@ -95,7 +140,17 @@
   describe '#initialize' do
 
     let(:server) do
-      described_class.new(address, cluster, monitoring, listeners, TEST_OPTIONS.merge(:heartbeat_frequency => 5))
+      described_class.new(
+        address,
+        cluster,
+        monitoring,
+        listeners,
+        TEST_OPTIONS.merge(:heartbeat_frequency => 5)
+      )
+    end
+
+    after do
+      server.disconnect!
     end
 
     it 'sets the address host' do
@@ -121,6 +176,10 @@
       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
@@ -132,6 +191,10 @@
       described_class.new(address, cluster, monitoring, listeners, TEST_OPTIONS)
     end
 
+    after do
+      server.disconnect!
+    end
+
     it 'forces a scan on the monitor' do
       expect(server.scan!).to eq(server.description)
     end
@@ -147,6 +210,10 @@
       expect(server.monitor).to receive(:restart!).and_call_original
     end
 
+    after do
+      server.disconnect!
+    end
+
     it 'restarts the monitor and returns true' do
       expect(server.reconnect!).to be(true)
     end
