RUBY-3332 Fix tailable cursors (#2793)

* RUBY-3332 Fix tailable cursors

* Extract method

* Fix code review remarks

lib/mongo/collection/view.rb

@@ -127,6 +127,7 @@ def hash
       #   return in each response from MongoDB.
       # @option options [ Hash ] :collation The collation to use.
       # @option options [ String ] :comment Associate a comment with the query.
+      # @option options [ :tailable, :tailable_await ] :cursor_type The type of cursor to use.
       # @option options [ Hash ] :explain Execute an explain with the provided
       #   explain options (known options are :verbose and :verbosity) rather
       #   than a find.

lib/mongo/collection/view/iterable.rb

@@ -186,6 +186,8 @@ def initial_query_op(session)
             collection.client.log_warn("The :oplog_replay option is deprecated and ignored by MongoDB 4.4 and later")
           end
 
+          maybe_set_tailable_options(spec)
+
           if explained?
             spec[:explain] = options[:explain]
             Operation::Explain.new(spec)
@@ -201,6 +203,19 @@ def send_initial_query(server, session = nil)
         def use_query_cache?
           QueryCache.enabled? && !collection.system_collection?
         end
+
+        # Add tailable cusror options to the command specifiction if needed.
+        #
+        # @param [ Hash ] spec The command specification.
+        def maybe_set_tailable_options(spec)
+          case cursor_type
+          when :tailable
+            spec[:tailable] = true
+          when :tailable_await
+            spec[:tailable] = true
+            spec[:await_data] = true
+          end
+        end
       end
     end
   end

spec/integration/find_options_spec.rb

@@ -13,10 +13,20 @@
     [ SpecConfig.instance.addresses.first ]
   end
 
+  let(:client_options) do
+    {}
+  end
+
+  let(:collection_options) do
+    {}
+  end
+
   let(:client) do
     ClientRegistry.instance.new_local_client(
       seeds,
-      SpecConfig.instance.test_options.merge(client_options)
+      SpecConfig.instance.test_options
+        .merge(database: SpecConfig.instance.test_db)
+        .merge(client_options)
     ).tap do |client|
       client.subscribe(Mongo::Monitoring::COMMAND, subscriber)
     end
@@ -30,8 +40,11 @@
     subscriber.started_events.find { |cmd| cmd.command_name == 'find' }
   end
 
+  let(:should_create_collection) { true }
+
   before do
-    ClientRegistry.instance.global_client('authorized')['find_options'].drop
+    client['find_options'].drop
+    collection.create if should_create_collection
     collection.insert_many([ { a: 1 }, { a: 2 }, { a: 3 } ])
   end
 
@@ -71,6 +84,8 @@
         { 'locale' => 'de_AT' }
       end
 
+      let(:should_create_collection) { false }
+
       it 'uses the collation defined on the collection' do
         collection.find({}, collation: collation).to_a
         expect(find_command.command['collation']).to eq(collation)
@@ -187,4 +202,26 @@
       end
     end
   end
+
+  describe 'cursor type' do
+    let(:collection_options) do
+      { capped: true, size: 1000 }
+    end
+
+    context 'when cursor type is :tailable' do
+      it 'sets the cursor type to tailable' do
+        collection.find({}, cursor_type: :tailable).first
+        expect(find_command.command['tailable']).to be true
+        expect(find_command.command['awaitData']).to be_falsey
+      end
+    end
+
+    context 'when cursor type is :tailable_await' do
+      it 'sets the cursor type to tailable' do
+        collection.find({}, cursor_type: :tailable_await).first
+        expect(find_command.command['tailable']).to be true
+        expect(find_command.command['awaitData']).to be true
+      end
+    end
+  end
 end