lib/mongo/collection/view/readable.rb in mongo-2.8.0 vs lib/mongo/collection/view/readable.rb in mongo-2.9.0.rc0

@@ -136,27 +136,26 @@
         def count(opts = {})
           cmd = { :count => collection.name, :query => filter }
           cmd[:skip] = opts[:skip] if opts[:skip]
           cmd[:hint] = opts[:hint] if opts[:hint]
           cmd[:limit] = opts[:limit] if opts[:limit]
+          cmd[:readConcern] = read_concern if read_concern
           cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
-          cmd[:readConcern] = collection.read_concern if collection.read_concern
           Mongo::Lint.validate_underscore_read_preference(opts[:read])
           read_pref = opts[:read] || read_preference
           selector = ServerSelector.get(read_pref || server_selector)
           with_session(opts) do |session|
-            read_with_retry(session) do
-              server = selector.select_server(cluster)
+            read_with_retry(session, selector) do |server|
               apply_collation!(cmd, server, opts)
-              Operation::Count.new({
+              Operation::Count.new(
                                      :selector => cmd,
                                      :db_name => database.name,
                                      :options => {:limit => -1},
                                      :read => read_pref,
                                      :session => session
-                                    }).execute(server)
-              end.n.to_i
+              ).execute(server)
+            end.n.to_i
           end
         end
 
         # Get a count of matching documents in the collection.
         #
@@ -179,11 +178,11 @@
         # @since 2.6.0
         def count_documents(opts = {})
           pipeline = [:'$match' => filter]
           pipeline << { :'$skip' => opts[:skip] } if opts[:skip]
           pipeline << { :'$limit' => opts[:limit] } if opts[:limit]
-          pipeline << { :'$group' => { _id: nil, n: { :'$sum' => 1 } } }
+          pipeline << { :'$group' => { _id: 1, n: { :'$sum' => 1 } } }
 
           opts.select! { |k, _| [:hint, :max_time_ms, :read, :collation, :session].include?(k) }
           first = aggregate(pipeline, opts).first
           return 0 unless first
           first['n'].to_i
@@ -204,17 +203,16 @@
         #
         # @since 2.6.0
         def estimated_document_count(opts = {})
           cmd = { count: collection.name }
           cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
-          cmd[:readConcern] = collection.read_concern if collection.read_concern
+          cmd[:readConcern] = read_concern if read_concern
           Mongo::Lint.validate_underscore_read_preference(opts[:read])
           read_pref = opts[:read] || read_preference
           selector = ServerSelector.get(read_pref || server_selector)
           with_session(opts) do |session|
-            read_with_retry(session) do
-              server = selector.select_server(cluster)
+            read_with_retry(session, selector) do |server|
               Operation::Count.new(
                 selector: cmd,
                 db_name: database.name,
                 read: read_pref,
                 session: session
@@ -242,17 +240,16 @@
         def distinct(field_name, opts = {})
           cmd = { :distinct => collection.name,
                   :key => field_name.to_s,
                   :query => filter }
           cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
-          cmd[:readConcern] = collection.read_concern if collection.read_concern
+          cmd[:readConcern] = read_concern if read_concern
           Mongo::Lint.validate_underscore_read_preference(opts[:read])
           read_pref = opts[:read] || read_preference
           selector = ServerSelector.get(read_pref || server_selector)
           with_session(opts) do |session|
-            read_with_retry(session) do
-              server = selector.select_server(cluster)
+            read_with_retry(session, selector) do |server|
               apply_collation!(cmd, server, opts)
               Operation::Distinct.new({
                                         :selector => cmd,
                                         :db_name => database.name,
                                         :options => {:limit => -1},
@@ -540,10 +537,18 @@
 
         def collation(doc = nil)
           configure(:collation, doc)
         end
 
+        def read_concern
+          if options[:session] && options[:session].in_transaction?
+            options[:session].send(:txn_read_concern) || collection.client.read_concern
+          else
+            collection.read_concern
+          end
+        end
+
         def read_preference
           rp = if options[:session] && options[:session].in_transaction?
             options[:session].txn_read_preference || collection.client.read_preference
           else
             @read_preference ||= (options[:read] || collection.read_preference)
@@ -569,10 +574,10 @@
           server = server_selector.select_server(cluster)
           cmd = Operation::ParallelScan.new({
                   :coll_name => collection.name,
                   :db_name => database.name,
                   :cursor_count => cursor_count,
-                  :read_concern => collection.read_concern,
+                  :read_concern => read_concern,
                   :session => session,
                 }.merge!(options))
           cmd.execute(server).cursor_ids.map do |cursor_id|
             result = if server.features.find_command_enabled?
               Operation::GetMore.new({