RUBY-1875 Pass read concern options through normal driver mechanisms

.evergreen/config.yml

@@ -781,7 +781,7 @@ buildvariants:
   matrix_spec:
     lint: on
     ruby: "ruby-2.6"
-    mongodb-version: ["4.0"]
+    mongodb-version: ["4.0", "4.2"]
     topology: [replica-set, sharded-cluster]
   display_name: "${mongodb-version} ${topology} ${lint} ${ruby}"
   run_on:

lib/mongo/collection/view/builder/aggregation.rb

@@ -93,7 +93,10 @@ def write?
           def aggregation_command
             command = BSON::Document.new(:aggregate => collection.name, :pipeline => pipeline)
             command[:cursor] = cursor if cursor
-            command[:readConcern] = collection.read_concern if collection.read_concern
+            if collection.read_concern
+              command[:readConcern] = Options::Mapper.transform_values_to_strings(
+                collection.read_concern)
+            end
             command.merge!(Options::Mapper.transform_documents(options, MAPPINGS))
             command
           end

lib/mongo/collection/view/builder/find_command.rb

@@ -96,7 +96,10 @@ def specification
 
           def find_command
             document = BSON::Document.new('find' => collection.name, 'filter' => filter)
-            document[:readConcern] = collection.read_concern if collection.read_concern
+            if collection.read_concern
+              document[:readConcern] = Options::Mapper.transform_values_to_strings(
+                collection.read_concern)
+            end
             command = Options::Mapper.transform_documents(convert_flags(options), MAPPINGS, document)
             convert_limit_and_batch_size(command)
             command

lib/mongo/collection/view/builder/map_reduce.rb

@@ -139,7 +139,10 @@ def map_reduce_command
               :query => filter,
               :out => { inline: 1 }
             )
-            command[:readConcern] = collection.read_concern if collection.read_concern
+            if collection.read_concern
+              command[:readConcern] = Options::Mapper.transform_values_to_strings(
+                collection.read_concern)
+            end
             command.merge!(view_options)
             command.merge!(Options::Mapper.transform_documents(options, MAPPINGS))
             command

lib/mongo/collection/view/readable.rb

@@ -138,7 +138,10 @@ def count(opts = {})
           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
+          if read_concern
+            cmd[:readConcern] = Options::Mapper.transform_values_to_strings(
+              read_concern)
+          end
           cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
           Mongo::Lint.validate_underscore_read_preference(opts[:read])
           read_pref = opts[:read] || read_preference
@@ -205,7 +208,10 @@ def count_documents(opts = {})
         def estimated_document_count(opts = {})
           cmd = { count: collection.name }
           cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
-          cmd[:readConcern] = read_concern if read_concern
+          if read_concern
+            cmd[:readConcern] = Options::Mapper.transform_values_to_strings(
+              read_concern)
+          end
           Mongo::Lint.validate_underscore_read_preference(opts[:read])
           read_pref = opts[:read] || read_preference
           selector = ServerSelector.get(read_pref || server_selector)
@@ -242,7 +248,10 @@ def distinct(field_name, opts = {})
                   :key => field_name.to_s,
                   :query => filter }
           cmd[:maxTimeMS] = opts[:max_time_ms] if opts[:max_time_ms]
-          cmd[:readConcern] = read_concern if read_concern
+          if read_concern
+            cmd[:readConcern] = Options::Mapper.transform_values_to_strings(
+              read_concern)
+          end
           Mongo::Lint.validate_underscore_read_preference(opts[:read])
           read_pref = opts[:read] || read_preference
           selector = ServerSelector.get(read_pref || server_selector)

lib/mongo/lint.rb

@@ -67,11 +67,18 @@ def validate_read_concern_option(read_concern)
       end
       return if read_concern.empty?
       keys = read_concern.keys
-      if keys != [:level]
+      if read_concern.is_a?(BSON::Document)
+        # Permits indifferent access
+        allowed_keys = ['level']
+      else
+        # Does not permit indifferent access
+        allowed_keys = [:level]
+      end
+      if keys != allowed_keys
         raise Error::LintError, "Read concern has invalid keys: #{keys.inspect}"
       end
       level = read_concern[:level]
-      return if [:local, :majority, :snapshot].include?(level)
+      return if [:local, :available, :majority, :linearizable, :snapshot].include?(level)
       raise Error::LintError, "Read concern level is invalid: #{level.inspect}"
     end
     module_function :validate_read_concern_option

lib/mongo/operation/parallel_scan/command.rb

@@ -32,7 +32,10 @@ class Command
 
         def selector(server)
           sel = { :parallelCollectionScan => coll_name, :numCursors => cursor_count }
-          sel[:readConcern] = read_concern if read_concern
+          if read_concern
+            sel[:readConcern] = Options::Mapper.transform_values_to_strings(
+              read_concern)
+          end
           sel[:maxTimeMS] = max_time_ms if max_time_ms
           update_selector_for_read_pref(sel, server)
           sel

lib/mongo/operation/parallel_scan/op_msg.rb

@@ -31,7 +31,10 @@ class OpMsg < OpMsgBase
         def selector(server)
           sel = { :parallelCollectionScan => coll_name, :numCursors => cursor_count }
           sel[:maxTimeMS] = max_time_ms if max_time_ms
-          sel[:readConcern] = read_concern if read_concern
+          if read_concern
+            sel[:readConcern] = Options::Mapper.transform_values_to_strings(
+              read_concern)
+          end
           sel
         end
       end

lib/mongo/operation/shared/sessions_supported.rb

@@ -62,7 +62,9 @@ def apply_causal_consistency_if_possible(selector, server)
         if !server.standalone?
           cc_doc = session.send(:causal_consistency_doc)
           if cc_doc
-            selector[:readConcern] = (selector[:readConcern] || read_concern || {}).merge(cc_doc)
+            rc_doc = (selector[:readConcern] || read_concern || {}).merge(cc_doc)
+            selector[:readConcern] = Options::Mapper.transform_values_to_strings(
+              rc_doc)
           end
         end
       end

lib/mongo/session.rb

@@ -804,13 +804,13 @@ def add_txn_opts!(command, read)
           if rc.nil? || rc.empty?
             c.delete(:readConcern)
           else
-            c[:readConcern ] = rc
+            c[:readConcern ] = Options::Mapper.transform_values_to_strings(rc)
           end
         end
 
         # We need to send the read concern level as a string rather than a symbol.
-        if c[:readConcern] && c[:readConcern][:level]
-          c[:readConcern][:level] = c[:readConcern][:level].to_s
+        if c[:readConcern]
+          c[:readConcern] = Options::Mapper.transform_values_to_strings(c[:readConcern])
         end
 
         # The write concern should be added to any abortTransaction or commitTransaction command.

lib/mongo/uri.rb

@@ -554,7 +554,7 @@ def self.uri_option(uri_key, name, extra = {})
     # Client Options
     uri_option 'appname', :app_name
     uri_option 'compressors', :compressors, :type => :array
-    uri_option 'readconcernlevel', :level, group: :read_concern
+    uri_option 'readconcernlevel', :level, group: :read_concern, type: :symbol
     uri_option 'retryreads', :retry_reads, :type => :retry_reads
     uri_option 'retrywrites', :retry_writes, :type => :retry_writes
     uri_option 'zlibcompressionlevel', :zlib_compression_level, :type => :zlib_compression_level

spec/integration/read_concern_spec.rb

@@ -8,6 +8,10 @@
   end
 
   let(:specified_read_concern) do
+    { :level => :local }
+  end
+
+  let(:expected_read_concern) do
     { 'level' => 'local' }
   end
 
@@ -19,7 +23,7 @@
 
   shared_examples_for 'a read concern is specified' do
     it 'sends a read concern to the server' do
-      expect(sent_read_concern).to eq(specified_read_concern)
+      expect(sent_read_concern).to eq(expected_read_concern)
     end
   end
 

spec/mongo/collection_spec.rb

@@ -101,7 +101,8 @@
   describe '#with' do
 
     let(:client) do
-      new_local_client(SpecConfig.instance.addresses, SpecConfig.instance.test_options)
+      new_local_client(SpecConfig.instance.addresses,
+        SpecConfig.instance.test_options.merge(monitoring_io: false))
     end
 
     let(:database) do
@@ -133,7 +134,8 @@
       context 'when the client has a server selection timeout setting' do
 
         let(:client) do
-          new_local_client(SpecConfig.instance.addresses, SpecConfig.instance.test_options.merge(server_selection_timeout: 2))
+          new_local_client(SpecConfig.instance.addresses,
+            SpecConfig.instance.test_options.merge(server_selection_timeout: 2, monitoring_io: false))
         end
 
         it 'passes the the server_selection_timeout to the cluster' do
@@ -144,7 +146,11 @@
       context 'when the client has a read preference set' do
 
         let(:client) do
-          new_local_client(SpecConfig.instance.addresses, SpecConfig.instance.test_options.merge(read: { mode: :primary_preferred }))
+          new_local_client(SpecConfig.instance.addresses,
+            SpecConfig.instance.test_options.merge(
+              read: { mode: :primary_preferred },
+              monitoring_io: false,
+          ))
         end
 
         it 'sets the new read options on the new collection' do
@@ -156,7 +162,12 @@
       context 'when the client has a read preference and server selection timeout set' do
 
         let(:client) do
-          new_local_client(SpecConfig.instance.addresses, SpecConfig.instance.test_options.merge(read: { mode: :primary_preferred }, server_selection_timeout: 2))
+          new_local_client(SpecConfig.instance.addresses,
+            SpecConfig.instance.test_options.merge(
+              read: { mode: :primary_preferred },
+              server_selection_timeout: 2,
+              monitoring_io: false
+          ))
         end
 
         it 'sets the new read options on the new collection' do
@@ -186,7 +197,11 @@
       context 'when the client has a write concern set' do
 
         let(:client) do
-          new_local_client(SpecConfig.instance.addresses, SpecConfig.instance.test_options.merge(write: INVALID_WRITE_CONCERN))
+          new_local_client(SpecConfig.instance.addresses,
+            SpecConfig.instance.test_options.merge(
+              write: INVALID_WRITE_CONCERN,
+              monitoring_io: false,
+          ))
         end
 
         it 'sets the new write options on the new collection' do
@@ -219,7 +234,11 @@
       context 'when the client has a server selection timeout setting' do
 
         let(:client) do
-          new_local_client(SpecConfig.instance.addresses, SpecConfig.instance.test_options.merge(server_selection_timeout: 2))
+          new_local_client(SpecConfig.instance.addresses,
+            SpecConfig.instance.test_options.merge(
+              server_selection_timeout: 2,
+              monitoring_io: false,
+          ))
         end
 
         it 'passes the server_selection_timeout setting to the cluster' do
@@ -230,7 +249,11 @@
       context 'when the client has a read preference set' do
 
         let(:client) do
-          new_local_client(SpecConfig.instance.addresses, SpecConfig.instance.test_options.merge(read: { mode: :primary_preferred }))
+          new_local_client(SpecConfig.instance.addresses,
+            SpecConfig.instance.test_options.merge(
+              read: { mode: :primary_preferred },
+              monitoring_io: false,
+          ))
         end
 
         it 'sets the new read options on the new collection' do

spec/mongo/uri_option_parsing_spec.rb

@@ -319,8 +319,8 @@
 
     it_behaves_like 'parses successfully'
 
-    it 'is a string' do
-      expect(uri.uri_options[:read_concern]).to eq(BSON::Document.new(level: 'snapshot'))
+    it 'is a symbol' do
+      expect(uri.uri_options[:read_concern]).to eq(BSON::Document.new(level: :snapshot))
     end
   end
 

spec/support/connection_string.rb

@@ -123,6 +123,7 @@ def tests
     end
 
     class Test
+      include RSpec::Core::Pending
 
       attr_reader :description
       attr_reader :uri_string
@@ -153,6 +154,10 @@ def options
 
       def client
         @client ||= ClientRegistry.instance.new_local_client(@spec['uri'], monitoring_io: false)
+      rescue Mongo::Error::LintError => e
+        if e.message =~ /arbitraryButStillValid/
+          skip 'Test uses a read concern that fails linter'
+        end
       end
 
       def uri

spec/support/transactions/operation.rb

@@ -139,6 +139,12 @@ def with_transaction(session, context, collection)
 
         if arguments['options']
           options = Utils.snakeize_hash(arguments['options'])
+          if options[:read_concern]
+            options[:read_concern] = Options::Mapper.transform_keys_to_symbols(options[:read_concern])
+            if options[:read_concern][:level] == 'majority'
+              options[:read_concern][:level] = :majority
+            end
+          end
         else
           options = nil
         end

spec/support/transactions/test.rb

@@ -54,6 +54,12 @@ def initialize(data, test, spec)
         @data = data
         @description = test['description']
         @client_options = Utils.convert_client_options(test['clientOptions'] || {})
+        if @client_options[:read_concern]
+          @client_options[:read_concern] = Options::Mapper.transform_keys_to_symbols(@client_options[:read_concern])
+          if @client_options[:read_concern][:level].is_a?(String)
+            @client_options[:read_concern][:level] = @client_options[:read_concern][:level].to_sym
+          end
+        end
         @session_options = Utils.snakeize_hash(test['sessionOptions'] || {})
         @fail_point = test['failPoint']
         @operations = test['operations']