SWIFT-672 Make MongoSwift.MongoCursor async, implement MongoSwiftSync.MongoCursor (#392)

Author: patrickfreed

Sources/MongoSwift/CursorCommon.swift

@@ -0,0 +1,60 @@
+import NIO
+
+/// A protocol describing the common behavior between cursor-like objects in the driver.
+internal protocol Cursor {
+    /// The decoded type iterated over by the cursor.
+    associatedtype T: Codable
+
+    /**
+     * Indicates whether this cursor has the potential to return more data.
+     *
+     * This property is mainly useful if this cursor is tailable, since in that case `tryNext` may return more results
+     * even after returning `nil`.
+     *
+     * If this cursor is non-tailable, it will always be dead as soon as either `tryNext` returns `nil` or an error.
+     *
+     * This cursor will be dead as soon as `next` returns `nil` or an error, regardless of the `CursorType`.
+     */
+    var isAlive: Bool { get }
+
+    /**
+     * Get the next `T` from the cursor.
+     *
+     * If this cursor is tailable, this method will continue retrying until a non-empty batch is returned or the cursor
+     * is closed
+     */
+    func next() -> EventLoopFuture<T?>
+
+    /**
+     * Attempt to get the next `T` from the cursor, returning `nil` if there are no results.
+     *
+     * If this cursor is tailable and `isAlive` is true, this may be called multiple times to attempt to retrieve more
+     * elements.
+     *
+     * If this cursor is a tailable await cursor, it will wait for results server side for a maximum of `maxAwaitTimeMS`
+     * before evaluating to `nil`. This option can be configured via options passed to the method that created this
+     * cursor (e.g. the `maxAwaitTimeMS` option on the `FindOptions` passed to `find`).
+     */
+    func tryNext() -> EventLoopFuture<T?>
+
+    /**
+     * Kills this cursor.
+     *
+     * This method MUST be called before this cursor goes out of scope to prevent leaking resources.
+     * This method may be called even if there are unresolved futures created from other `Cursor` methods.
+     *
+     * This method should not fail.
+     */
+    func kill() -> EventLoopFuture<Void>
+}
+
+extension EventLoopFuture {
+    /// Run the provided callback after this future succeeds, preserving the succeeded value.
+    internal func afterSuccess(f: @escaping (Value) -> EventLoopFuture<Void>) -> EventLoopFuture<Value> {
+        return self.flatMap { value in
+            f(value).and(value: value)
+        }.map { _, value in
+            value
+        }
+    }
+}

Sources/MongoSwift/MongoCollection+Indexes.swift

@@ -362,9 +362,9 @@ extension MongoCollection {
      *
      * - Throws: `LogicError` if the provided session is inactive.
      */
-    public func listIndexes(session: ClientSession? = nil) throws -> MongoCursor<IndexModel> {
+    public func listIndexes(session: ClientSession? = nil) -> EventLoopFuture<MongoCursor<IndexModel>> {
         let operation = ListIndexesOperation(collection: self)
-        return try self._client.executeOperation(operation, session: session)
+        return self._client.operationExecutor.execute(operation, client: self._client, session: session)
     }
 
     /**
@@ -373,19 +373,20 @@ extension MongoCollection {
      * - Parameters:
      *   - session: Optional `ClientSession` to use when executing this command
      *
-     * - Returns: A `MongoCursor` over the index names.
+     * - Returns: An `EventLoopFuture<[String]>` containing the index names.
      *
      * - Throws: `LogicError` if the provided session is inactive.
      */
-    public func listIndexNames(session: ClientSession? = nil) throws -> [String] {
-        let operation = ListIndexesOperation(collection: self)
-        let models = try self._client.executeOperation(operation, session: session)
-        let names: [String] = try models.all().map { model in
-            guard let name = model.options?.name else {
-                throw InternalError(message: "Server response missing a 'name' field")
+    public func listIndexNames(session _: ClientSession? = nil) throws -> EventLoopFuture<[String]> {
+        return self.listIndexes().flatMap { cursor in
+            cursor.all()
+        }.flatMapThrowing { models in
+            try models.map { model in
+                guard let name = model.options?.name else {
+                    throw InternalError(message: "Server response missing a 'name' field")
+                }
+                return name
             }
-            return name
         }
-        return names
     }
 }

Sources/MongoSwift/MongoCollection+Read.swift

@@ -22,9 +22,9 @@ extension MongoCollection {
         _ filter: Document = [:],
         options: FindOptions? = nil,
         session: ClientSession? = nil
-    ) throws -> MongoCursor<CollectionType> {
+    ) -> EventLoopFuture<MongoCursor<CollectionType>> {
         let operation = FindOperation(collection: self, filter: filter, options: options)
-        return try self._client.executeOperation(operation, session: session)
+        return self._client.operationExecutor.execute(operation, client: self._client, session: session)
     }
 
     /**
@@ -46,10 +46,11 @@ extension MongoCollection {
         _ filter: Document = [:],
         options: FindOneOptions? = nil,
         session: ClientSession? = nil
-    ) throws -> T? {
+    ) throws -> EventLoopFuture<T?> {
         let options = options.map { FindOptions(from: $0) }
-        let cursor = try self.find(filter, options: options, session: session)
-        return try cursor.next()?.get()
+        return self.find(filter, options: options, session: session).flatMap { cursor in
+            cursor.next().afterSuccess { _ in cursor.kill() }
+        }
     }
 
     /**
@@ -71,9 +72,9 @@ extension MongoCollection {
         _ pipeline: [Document],
         options: AggregateOptions? = nil,
         session: ClientSession? = nil
-    ) throws -> MongoCursor<Document> {
+    ) -> EventLoopFuture<MongoCursor<Document>> {
         let operation = AggregateOperation(collection: self, pipeline: pipeline, options: options)
-        return try self._client.executeOperation(operation, session: session)
+        return self._client.operationExecutor.execute(operation, client: self._client, session: session)
     }
 
     /**

Sources/MongoSwift/MongoCollection.swift

@@ -19,7 +19,7 @@ public struct MongoCollection<T: Codable> {
     internal let _client: MongoClient
 
     /// The namespace for this collection.
-    internal let namespace: MongoNamespace
+    public let namespace: MongoNamespace
 
     /// Encoder used by this collection for BSON conversions. (e.g. converting `CollectionType`s, indexes, and options
     /// to documents).