Add CSOT to synchronous GridFS operations.

driver-core/src/main/com/mongodb/internal/TimeoutContext.java

@@ -209,7 +209,7 @@ public int hashCode() {
     }
 
     @Nullable
-    private static Timeout calculateTimeout(@Nullable final Long timeoutMS) {
+    public static Timeout calculateTimeout(@Nullable final Long timeoutMS) {
         if (timeoutMS != null) {
             return timeoutMS == 0 ? Timeout.infinite() : Timeout.expiresIn(timeoutMS, MILLISECONDS);
         }

driver-core/src/test/functional/com/mongodb/client/CommandMonitoringTestHelper.java

@@ -120,11 +120,11 @@ static boolean isWriteCommand(final String commandName) {
         return asList("insert", "update", "delete").contains(commandName);
     }
 
-    public static void assertEventsEquality(final List<CommandEvent> expectedEvents, final List<CommandEvent> events) {
+    public static void assertEventsEquality(final List<CommandEvent> expectedEvents, final List<? extends CommandEvent> events) {
         assertEventsEquality(expectedEvents, events, null);
     }
 
-    public static void assertEventsEquality(final List<CommandEvent> expectedEvents, final List<CommandEvent> events,
+    public static void assertEventsEquality(final List<CommandEvent> expectedEvents, final List<? extends CommandEvent> events,
                                             @Nullable final Map<String, BsonDocument> lsidMap) {
         assertEquals(expectedEvents.size(), events.size());
 

driver-core/src/test/functional/com/mongodb/internal/connection/TestCommandListener.java

@@ -144,24 +144,27 @@ public CommandFailedEvent getCommandFailedEvent(final String commandName) {
                 .orElseThrow(() -> new IllegalArgumentException(commandName + " not found in command failed event list"));
     }
 
-    public List<CommandEvent> getCommandStartedEvents() {
+    public List<CommandStartedEvent> getCommandStartedEvents() {
         return getEvents(CommandStartedEvent.class, Integer.MAX_VALUE);
     }
 
-    public List<CommandEvent> getCommandSucceededEvents() {
+    public List<CommandSucceededEvent> getCommandSucceededEvents() {
         return getEvents(CommandSucceededEvent.class, Integer.MAX_VALUE);
     }
 
-    private List<CommandEvent> getEvents(final Class<? extends CommandEvent> type, final int maxEvents) {
+    private <T extends CommandEvent> List<T> getEvents(final Class<T> type, final int maxEvents) {
         lock.lock();
         try {
-            return getEvents().stream().filter(e -> e.getClass() == type).limit(maxEvents).collect(Collectors.toList());
+            return getEvents().stream()
+                    .filter(e -> e.getClass() == type)
+                    .map(type::cast)
+                    .limit(maxEvents).collect(Collectors.toList());
         } finally {
             lock.unlock();
         }
     }
 
-    public List<CommandEvent> waitForStartedEvents(final int numEvents) {
+    public List<CommandStartedEvent> waitForStartedEvents(final int numEvents) {
         lock.lock();
         try {
             while (!hasCompletedEvents(numEvents)) {

driver-core/src/test/resources/unified-test-format/client-side-operation-timeout/gridfs-download.json

@@ -299,7 +299,7 @@
                   "find"
                 ],
                 "blockConnection": true,
-                "blockTimeMS": 50
+                "blockTimeMS": 40
               }
             }
           }

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/gridfs/GridFSFindPublisher.java

@@ -16,12 +16,9 @@
 
 package com.mongodb.reactivestreams.client.gridfs;
 
-import com.mongodb.client.cursor.TimeoutMode;
 import com.mongodb.client.gridfs.model.GridFSFile;
 import com.mongodb.client.model.Collation;
 import com.mongodb.lang.Nullable;
-import com.mongodb.reactivestreams.client.MongoCollection;
-import com.mongodb.reactivestreams.client.MongoDatabase;
 import org.bson.conversions.Bson;
 import org.reactivestreams.Publisher;
 
@@ -127,17 +124,4 @@ public interface GridFSFindPublisher extends Publisher<GridFSFile> {
      * @mongodb.driver.manual reference/method/cursor.batchSize/#cursor.batchSize Batch Size
      */
     GridFSFindPublisher batchSize(int batchSize);
-
-    /**
-     * Sets the timeoutMode for the cursor.
-     *
-     * <p>
-     *     Requires the {@code timeout} to be set, either in the {@link com.mongodb.MongoClientSettings},
-     *     via {@link MongoDatabase} or via {@link MongoCollection}
-     * </p>
-     * @param timeoutMode the timeout mode
-     * @return this
-     * @since CSOT
-     */
-    GridFSFindPublisher timeoutMode(TimeoutMode timeoutMode);
 }

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/gridfs/GridFSBucketImpl.java

@@ -72,7 +72,7 @@ public GridFSBucketImpl(final MongoDatabase database, final String bucketName) {
              getChunksCollection(database, bucketName));
     }
 
-    GridFSBucketImpl(final String bucketName, final int chunkSizeBytes, final MongoCollection<GridFSFile> filesCollection,
+    private GridFSBucketImpl(final String bucketName, final int chunkSizeBytes, final MongoCollection<GridFSFile> filesCollection,
                      final MongoCollection<Document> chunksCollection) {
         this.bucketName = notNull("bucketName", bucketName);
         this.chunkSizeBytes = chunkSizeBytes;

driver-reactive-streams/src/main/com/mongodb/reactivestreams/client/internal/gridfs/GridFSFindPublisherImpl.java

@@ -16,7 +16,6 @@
 
 package com.mongodb.reactivestreams.client.internal.gridfs;
 
-import com.mongodb.client.cursor.TimeoutMode;
 import com.mongodb.client.gridfs.model.GridFSFile;
 import com.mongodb.client.model.Collation;
 import com.mongodb.lang.Nullable;
@@ -94,12 +93,6 @@ public GridFSFindPublisher batchSize(final int batchSize) {
         return this;
     }
 
-    @Override
-    public GridFSFindPublisher timeoutMode(final TimeoutMode timeoutMode) {
-        wrapped.timeoutMode(timeoutMode);
-        return this;
-    }
-
     @Override
     public void subscribe(final Subscriber<? super GridFSFile> s) {
         wrapped.subscribe(s);

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/ClientSideOperationTimeoutProseTest.java

@@ -19,17 +19,44 @@
 import com.mongodb.MongoClientSettings;
 import com.mongodb.client.AbstractClientSideOperationsTimeoutProseTest;
 import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.gridfs.GridFSBucket;
+import com.mongodb.reactivestreams.client.gridfs.GridFSBuckets;
+import com.mongodb.reactivestreams.client.syncadapter.SyncGridFSBucket;
 import com.mongodb.reactivestreams.client.syncadapter.SyncMongoClient;
+import org.junit.jupiter.api.Disabled;
 
 
 /**
  * See https://github.com/mongodb/specifications/blob/master/source/client-side-operations-timeout/tests/README.rst#prose-tests
  */
 public final class ClientSideOperationTimeoutProseTest extends AbstractClientSideOperationsTimeoutProseTest {
+    private com.mongodb.reactivestreams.client.MongoClient wrapped;
 
     @Override
     protected MongoClient createMongoClient(final MongoClientSettings mongoClientSettings) {
-        return new SyncMongoClient(MongoClients.create(mongoClientSettings));
+        wrapped = MongoClients.create(mongoClientSettings);
+        return new SyncMongoClient(wrapped);
+    }
+
+    @Override
+    protected GridFSBucket createGridFsBucket(final MongoDatabase mongoDatabase, final String bucketName) {
+        return new SyncGridFSBucket(GridFSBuckets.create(wrapped.getDatabase(mongoDatabase.getName()), bucketName));
+    }
+
+    @Override
+    @Disabled("TODO (CSOT) - JAVA-4057")
+    public void testGridFSUploadViaOpenUploadStreamTimeout() {
+    }
+
+    @Disabled("TODO (CSOT) - JAVA-4057")
+    @Override
+    public void testAbortingGridFsUploadStreamTimeout() {
+    }
+
+    @Disabled("TODO (CSOT) - JAVA-4057")
+    @Override
+    public void testGridFsDownloadStreamTimeout() {
     }
 
     @Override

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/ReadConcernTest.java

@@ -17,7 +17,6 @@
 package com.mongodb.reactivestreams.client;
 
 import com.mongodb.ReadConcern;
-import com.mongodb.event.CommandEvent;
 import com.mongodb.event.CommandStartedEvent;
 import com.mongodb.internal.connection.TestCommandListener;
 import org.bson.BsonDocument;
@@ -65,7 +64,7 @@ public void shouldIncludeReadConcernInCommand() throws InterruptedException {
                 .find())
                 .block(TIMEOUT_DURATION);
 
-        List<CommandEvent> events = commandListener.getCommandStartedEvents();
+        List<CommandStartedEvent> events = commandListener.getCommandStartedEvents();
 
         BsonDocument commandDocument = new BsonDocument("find", new BsonString("test"))
                 .append("readConcern", ReadConcern.LOCAL.asDocument())

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/internal/BatchCursorFluxTest.java

@@ -373,7 +373,7 @@ public void testBatchCursorReportsCursorErrors() {
 
         BsonDocument getMoreCommand = commandListener.getCommandStartedEvents().stream()
                 .filter(e -> e.getCommandName().equals("getMore"))
-                .map(e -> ((CommandStartedEvent) e).getCommand())
+                .map(CommandStartedEvent::getCommand)
                 .findFirst()
                 .get();
 

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/syncadapter/SyncGridFSBucket.java

@@ -42,6 +42,7 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 import static com.mongodb.ClusterFixture.TIMEOUT_DURATION;
 import static com.mongodb.reactivestreams.client.syncadapter.ContextHelper.CONTEXT;
@@ -79,6 +80,11 @@ public ReadConcern getReadConcern() {
         return wrapped.getReadConcern();
     }
 
+    @Override
+    public Long getTimeout(final TimeUnit timeUnit) {
+        throw new UnsupportedOperationException("Not implemented yet!");
+    }
+
     @Override
     public GridFSBucket withChunkSizeBytes(final int chunkSizeBytes) {
         return new SyncGridFSBucket(wrapped.withChunkSizeBytes(chunkSizeBytes));
@@ -99,6 +105,11 @@ public GridFSBucket withReadConcern(final ReadConcern readConcern) {
         return new SyncGridFSBucket(wrapped.withReadConcern(readConcern));
     }
 
+    @Override
+    public GridFSBucket withTimeout(final long timeout, final TimeUnit timeUnit) {
+        throw new UnsupportedOperationException("Not implemented yet!");
+    }
+
     @Override
     public GridFSUploadStream openUploadStream(final String filename) {
         return openUploadStream(filename, new GridFSUploadOptions());

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/unified/ClientSideOperationTimeoutTest.java

@@ -57,6 +57,7 @@ public ClientSideOperationTimeoutTest(final String fileDescription, final String
         assumeFalse(testDescription.endsWith("createChangeStream on client"));
         assumeFalse(testDescription.endsWith("createChangeStream on database"));
         assumeFalse(testDescription.endsWith("createChangeStream on collection"));
+        assumeFalse("TODO (CSOT) - JAVA-4057", fileDescription.contains("GridFS"));
         checkSkipCSOTTest(fileDescription, testDescription);
 
         if (testDescription.equals("timeoutMS is refreshed for close")) {

driver-scala/src/main/scala/org/mongodb/scala/gridfs/GridFSFindObservable.scala

@@ -121,27 +121,6 @@ case class GridFSFindObservable(private val wrapped: GridFSFindPublisher) extend
     this
   }
 
-  /**
-   * Sets the timeoutMode for the cursor.
-   *
-   * Requires the `timeout` to be set, either in the `MongoClientSettings`,
-   * via `MongoDatabase` or via `MongoCollection`
-   *
-   * If the `timeout` is set then:
-   *
-   * - For non-tailable cursors, the default value of timeoutMode is `TimeoutMode.CURSOR_LIFETIME`
-   * - For tailable cursors, the default value of timeoutMode is `TimeoutMode.ITERATION` and its an error
-   * to configure it as: `TimeoutMode.CURSOR_LIFETIME`
-   *
-   * @param timeoutMode the timeout mode
-   * @return this
-   * @since CSOT
-   */
-  def timeoutMode(timeoutMode: TimeoutMode): GridFSFindObservable = {
-    wrapped.timeoutMode(timeoutMode)
-    this
-  }
-
   /**
    * Helper to return a single observable limited to the first result.
    *

driver-sync/src/main/com/mongodb/client/gridfs/CollectionTimeoutHelper.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2008-present 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.
+ */
+
+package com.mongodb.client.gridfs;
+
+import com.mongodb.MongoOperationTimeoutException;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.internal.time.Timeout;
+import com.mongodb.lang.Nullable;
+
+import static java.util.concurrent.TimeUnit.MILLISECONDS;
+
+final class CollectionTimeoutHelper {
+    private CollectionTimeoutHelper(){
+        //NOP
+    }
+
+    public static <T> MongoCollection<T> collectionWithTimeout(final MongoCollection<T> collection,
+                                                               final String message,
+                                                               @Nullable final Timeout timeout) {
+        if (timeout != null && !timeout.isInfinite()) {
+            long remainingMs = timeout.remaining(MILLISECONDS);
+            if (timeout.hasExpired()) {
+                // TODO (CSOT) - JAVA-5248 Update to MongoOperationTimeoutException
+                throw new MongoOperationTimeoutException(message);
+            }
+            return collection.withTimeout(remainingMs, MILLISECONDS);
+        }
+        return collection;
+    }
+}