SWIFT-719 Add a forEach method to async cursors and change streams (#408)

Author: kmahar

Sources/MongoSwift/ChangeStream.swift

@@ -108,13 +108,11 @@ public class ChangeStream<T: Codable>: CursorProtocol {
         }
     }
 
-    deinit {
-        assert(!self.isAlive, "change stream wasn't closed")
-    }
-
     /// Indicates whether this change stream has the potential to return more data.
-    public var isAlive: Bool {
-        return self.wrappedCursor.isAlive
+    public func isAlive() -> EventLoopFuture<Bool> {
+        return self.client.operationExecutor.execute {
+            self.wrappedCursor.isAlive
+        }
     }
 
     /// The `ResumeToken` associated with the most recent event seen by the change stream.
@@ -128,10 +126,13 @@ public class ChangeStream<T: Codable>: CursorProtocol {
      * (specified on the `ChangeStreamOptions` passed  to the method that created this change stream) before trying
      * again.
      *
-     * A thread from the pool will be occupied until the returned future is completed, so performance degradation
-     * is possible if the number of polling change streams is too close to the total number of threads in the thread
-     * pool. To configure the total number of threads in the pool, set the `ClientOptions.threadPoolSize` option
-     * during client creation.
+     * A thread from the driver's internal thread pool will be occupied until the returned future is completed, so
+     * performance degradation is possible if the number of polling change streams is too close to the total number of
+     * threads in the thread pool. To configure the total number of threads in the pool, set the
+     * `ClientOptions.threadPoolSize` option during client creation.
+     *
+     * Note: You *must not* call any change stream methods besides `kill` and `isAlive` while the future returned from
+     * this method is unresolved. Doing so will result in undefined behavior.
      *
      * - Returns:
      *   An `EventLoopFuture<T?>` evaluating to the next `T` in this change stream, `nil` if the change stream is
@@ -158,6 +159,9 @@ public class ChangeStream<T: Codable>: CursorProtocol {
      *
      * This method may be called repeatedly while `isAlive` is true to retrieve new data.
      *
+     * Note: You *must not* call any change stream methods besides `kill` and `isAlive` while the future returned from
+     * this method is unresolved. Doing so will result in undefined behavior.
+     *
      * - Returns:
      *    An `EventLoopFuture<T?>` containing the next `T` in this change stream, an error if one occurred, or `nil` if
      *    there was no data.
@@ -180,6 +184,9 @@ public class ChangeStream<T: Codable>: CursorProtocol {
      * Since `toArray` will only fetch the currently available results, it may return more data if it is called again
      * while the change stream is still alive.
      *
+     * Note: You *must not* call any change stream methods besides `kill` and `isAlive` while the future returned from
+     * this method is unresolved. Doing so will result in undefined behavior.
+     *
      * - Returns:
      *    An `EventLoopFuture<[T]>` evaluating to the results currently available in this change stream, or an error.
      *
@@ -195,6 +202,35 @@ public class ChangeStream<T: Codable>: CursorProtocol {
         }
     }
 
+    /**
+     * Calls the provided closure with each event in the change stream as it arrives.
+     *
+     * A thread from the driver's internal thread pool will be occupied until the returned future is completed, so
+     * performance degradation is possible if the number of polling change streams is too close to the total number of
+     * threads in the thread pool. To configure the total number of threads in the pool, set the
+     * `ClientOptions.threadPoolSize` option during client creation.
+     *
+     * Note: You *must not* call any change stream methods besides `kill` and `isAlive` while the future returned from
+     * this method is unresolved. Doing so will result in undefined behavior.
+     *
+     * - Returns:
+     *     An `EventLoopFuture<Void>` which will complete once the change stream is closed or once an error is
+     *     encountered.
+     *
+     *     If the future evaluates to an error, that error is likely one of the following:
+     *     - `CommandError` if an error occurs while fetching more results from the server.
+     *     - `LogicError` if this function is called after the change stream has died.
+     *     - `LogicError` if this function is called and the session associated with this change stream is inactive.
+     *     - `DecodingError` if an error occurs decoding the server's responses.
+     */
+    public func forEach(_ body: @escaping (T) throws -> Void) -> EventLoopFuture<Void> {
+        return self.client.operationExecutor.execute {
+            while let next = try self.processEvent(self.wrappedCursor.next()) {
+                try body(next)
+            }
+        }
+    }
+
     /**
      * Kill this change stream.
      *

Sources/MongoSwift/CursorCommon.swift

@@ -11,14 +11,14 @@ internal protocol CursorProtocol {
     /**
      * 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
+     * This method 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 }
+    func isAlive() -> EventLoopFuture<Bool>
 
     /**
      * Get the next `T` from the cursor.
@@ -242,7 +242,13 @@ internal class Cursor<CursorKind: MongocCursorWrapper> {
         }
     }
 
-    /// Whether the cursor is alive.
+    /// Asserts that the cursor was closed.
+    deinit {
+        assert(!self._isAlive, "cursor or change stream wasn't closed before it went out of scope")
+    }
+
+    /// Whether the cursor is alive. This property can block while waiting for the lock and should only be accessed
+    /// from within the executor.
     internal var isAlive: Bool {
         return self.lock.withLock {
             self._isAlive
@@ -301,7 +307,7 @@ internal class Cursor<CursorKind: MongocCursorWrapper> {
         }
     }
 
-    /// Retreive all the currently available documents in the result set.
+    /// Retrieve all the currently available documents in the result set.
     /// This will not exhaust the cursor.
     /// This method is blocking and should only be run via the executor.
     internal func toArray() throws -> [Document] {

Sources/MongoSwift/MongoCursor.swift

@@ -41,6 +41,10 @@ public class MongoCursor<T: Codable>: CursorProtocol {
     /// Decoder from the client, database, or collection that created this cursor.
     internal let decoder: BSONDecoder
 
+    /// The ID used by the server to track the cursor over time. If all of the cursor's results were returnable in a
+    /// single batch, or if the cursor contained no results, this value will be nil.
+    public let id: Int64?
+
     /**
      * Initializes a new `MongoCursor` instance. Not meant to be instantiated directly by a user. When `forceIO` is
      * true, this initializer will force a connection to the server if one is not already established.
@@ -66,11 +70,14 @@ public class MongoCursor<T: Codable>: CursorProtocol {
             session: session,
             type: cursorType ?? .nonTailable
         )
-    }
 
-    /// Asserts that the cursor was closed.
-    deinit {
-        assert(!self.isAlive, "cursor wasn't closed before it went out of scope")
+        self.id = self.wrappedCursor.withUnsafeMongocPointer { ptr in
+            guard let ptr = ptr else {
+                return nil
+            }
+            let id = mongoc_cursor_get_id(ptr)
+            return id == 0 ? nil : id
+        }
     }
 
     /// Decodes a result to the generic type or `nil` if no result were returned.
@@ -89,25 +96,16 @@ public class MongoCursor<T: Codable>: CursorProtocol {
     /**
      * 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
+     * This method 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`.
      */
-    public var isAlive: Bool {
-        return self.wrappedCursor.isAlive
-    }
-
-    /// Returns the ID used by the server to track the cursor. `nil` once all results have been fetched from the server.
-    public var id: Int64? {
-        return self.wrappedCursor.withUnsafeMongocPointer { ptr in
-            guard let ptr = ptr else {
-                return nil
-            }
-            let id = mongoc_cursor_get_id(ptr)
-            return id == 0 ? nil : id
+    public func isAlive() -> EventLoopFuture<Bool> {
+        return self.client.operationExecutor.execute {
+            self.wrappedCursor.isAlive
         }
     }
 
@@ -120,6 +118,9 @@ public class MongoCursor<T: Codable>: CursorProtocol {
      * 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`).
      *
+     * Note: You *must not* call any cursor methods besides `kill` and `isAlive` while the future returned from this
+     * method is unresolved. Doing so will result in undefined behavior.
+     *
      * - Returns:
      *    An `EventLoopFuture<T?>` containing the next `T` in this cursor, an error if one occurred, or `nil` if
      *    there was no data.
@@ -143,6 +144,14 @@ public class MongoCursor<T: Codable>: CursorProtocol {
      * If this cursor is tailable, this method will continue polling until a non-empty batch is returned from the server
      * or the cursor is closed.
      *
+     * A thread from the driver's internal thread pool will be occupied until the returned future is completed, so
+     * performance degradation is possible if the number of polling cursors is too close to the total number of threads
+     * in the thread pool. To configure the total number of threads in the pool, set the `ClientOptions.threadPoolSize`
+     * option during client creation.
+     *
+     * Note: You *must not* call any cursor methods besides `kill` and `isAlive` while the future returned from this
+     * method is unresolved. Doing so will result in undefined behavior.
+     *
      * - Returns:
      *   An `EventLoopFuture<T?>` evaluating to the next `T` in this cursor, or `nil` if the cursor is exhausted. If
      *   the underlying cursor is tailable, the future will not resolve until data is returned (potentially after
@@ -168,6 +177,9 @@ public class MongoCursor<T: Codable>: CursorProtocol {
      * If this cursor is tailable, `toArray` will only fetch the currently available results, and it
      * may return more data if it is called again while the cursor is still alive.
      *
+     * Note: You *must not* call any cursor methods besides `kill` and `isAlive` while the future returned from this
+     * method is unresolved. Doing so will result in undefined behavior.
+     *
      * - Returns:
      *    An `EventLoopFuture<[T]>` evaluating to the results currently available in this cursor, or an error.
      *
@@ -183,6 +195,39 @@ public class MongoCursor<T: Codable>: CursorProtocol {
         }
     }
 
+    /**
+     * Calls the provided closure with each element in the cursor.
+     *
+     * If the cursor is not tailable, this method will exhaust it, calling the closure with every document.
+     *
+     * If the cursor is tailable, the method will call the closure with each new document as it arrives.
+     *
+     * A thread from the driver's internal thread pool will be occupied until the returned future is completed, so
+     * performance degradation is possible if the number of polling cursors is too close to the total number of threads
+     * in the thread pool. To configure the total number of threads in the pool, set the `ClientOptions.threadPoolSize`
+     * option during client creation.
+     *
+     * Note: You *must not* call any cursor methods besides `kill` and `isAlive` while the future returned from this
+     * method is unresolved. Doing so will result in undefined behavior.
+     *
+     * - Returns:
+     *     An `EventLoopFuture<Void>` which will succeed when the end of the cursor is reached, or in the case of a
+     *     tailable cursor, when the cursor is killed via `kill`.
+     *
+     *     If the future evaluates to an error, that error is likely one of the following:
+     *     - `CommandError` if an error occurs while fetching more results from the server.
+     *     - `LogicError` if this function is called after the cursor has died.
+     *     - `LogicError` if this function is called and the session associated with this cursor is inactive.
+     *     - `DecodingError` if an error occurs decoding the server's responses.
+     */
+    public func forEach(_ body: @escaping (T) throws -> Void) -> EventLoopFuture<Void> {
+        return self.client.operationExecutor.execute {
+            while let next = try self.decode(result: self.wrappedCursor.next()) {
+                try body(next)
+            }
+        }
+    }
+
     /**
      * Kill this cursor.
      *

Sources/MongoSwiftSync/ChangeStream.swift

@@ -29,8 +29,12 @@ public class ChangeStream<T: Codable>: CursorProtocol {
      * This change stream will be dead if `next` returns `nil` or an error. It will also be dead if `tryNext` returns
      * an error, but will still be alive if `tryNext` returns `nil`.
      */
-    public var isAlive: Bool {
-        return self.asyncChangeStream.isAlive
+    public func isAlive() -> Bool {
+        do {
+            return try self.asyncChangeStream.isAlive().wait()
+        } catch {
+            return false
+        }
     }
 
     /**

Sources/MongoSwiftSync/CursorCommon.swift

@@ -6,14 +6,14 @@ internal protocol CursorProtocol: LazySequenceProtocol, IteratorProtocol {
     /**
      * 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
+     * This method 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 }
+    func isAlive() throws -> Bool
 
     /**
      * Get the next `T` from the cursor.

Sources/MongoSwiftSync/MongoCursor.swift

@@ -16,15 +16,19 @@ public class MongoCursor<T: Codable>: CursorProtocol {
     /**
      * 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
+     * This method 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`.
      */
-    public var isAlive: Bool {
-        return self.asyncCursor.isAlive
+    public func isAlive() -> Bool {
+        do {
+            return try self.asyncCursor.isAlive().wait()
+        } catch {
+            return false
+        }
     }
 
     /// Returns the ID used by the server to track the cursor. `nil` once all results have been fetched from the server.

Tests/LinuxMain.swift

@@ -14,6 +14,8 @@ extension AsyncMongoCursorTests {
         ("testTailableAsyncCursor", testTailableAsyncCursor),
         ("testAsyncNext", testAsyncNext),
         ("testCursorToArray", testCursorToArray),
+        ("testForEach", testForEach),
+        ("testCursorId", testCursorId),
     ]
 }
 
@@ -51,6 +53,7 @@ extension ChangeStreamTests {
         ("testChangeStreamError", testChangeStreamError),
         ("testChangeStreamEmpty", testChangeStreamEmpty),
         ("testChangeStreamToArray", testChangeStreamToArray),
+        ("testChangeStreamForEach", testChangeStreamForEach),
     ]
 }