[UPDATE] add beginGracefulShutdown() to stream response usecase.

Author: gradyzhuo

Sources/EventStoreDB/EventStoreDB.swift

@@ -377,23 +377,6 @@ extension EventStoreDBClient {
         try await underlyingClient.restartPersistentSubscriptionSubsystem()
     }
 
-    /// Subscribes to a persistent subscription for the specified stream selector and group name.
-    ///
-    /// - Parameters:
-    ///   - streamSelector: Identifies the stream or streams to subscribe to.
-    ///   - groupName: The name of the persistent subscription group.
-    ///   - configure: Optional closure to customize read options.
-    ///
-    /// - Returns: An active persistent subscription for the specified stream and group.
-    ///
-    /// - Throws: An error if the subscription cannot be established.
-    @available(*, deprecated)
-    public func subscribePersistentSubscription(to streamSelector: StreamSelector<StreamIdentifier>, groupName: String, configure: @Sendable (_ options: ReadOptions) -> ReadOptions = { $0 }) async throws -> PersistentSubscriptions<PersistentSubscription.AnyTarget>.Subscription {
-        let node = try await underlyingClient.selector.select()
-        let options = configure(.init())
-        let usecase = PersistentSubscriptions<PersistentSubscription.AnyTarget>.ReadAnyTarget(streamSelector: streamSelector, group: groupName, options: options)
-        return try await usecase.perform(node: node, callOptions: underlyingClient.defaultCallOptions)
-    }
 }
 
 public struct ReadOptions: EventStoreOptions {

Sources/EventStoreDB/PersistentSubscriptions/ReadAnyTarget.swift

@@ -8,5 +8,5 @@
 import GRPCCore
 
 package protocol UnaryStream: Usecase, UnaryRequestBuildable, StreamResponseHandlable {
-    func send(connection: GRPCClient<Transport>, request: ClientRequest<UnderlyingRequest>, callOptions: CallOptions) async throws -> Responses
+    func send(connection: GRPCClient<Transport>, request: ClientRequest<UnderlyingRequest>, callOptions: CallOptions, finished: @Sendable @escaping ()->Void) async throws -> Responses
 }

Sources/GRPCEncapsulates/Usecase/UnaryStream.swift

@@ -19,7 +19,9 @@ extension UnaryStream where Transport == HTTP2ClientTransport.Posix, Responses =
         return try await withRethrowingError(usage: "\(Self.self)\(#function)") {
             let metadata = Metadata(from: node.settings)
             let request = try request(metadata: metadata)
-            return try await send(connection: client, request: request, callOptions: callOptions)
+            return try await send(connection: client, request: request, callOptions: callOptions){
+                client.beginGracefulShutdown()
+            }
         }
     }
 
@@ -54,7 +56,9 @@ extension UnaryStream where Transport == HTTP2ClientTransport.Posix {
 
         return try await withRethrowingError(usage: "\(Self.self)\(#function)") {
             let request = try request(metadata: metadata)
-            return try await send(connection: client, request: request, callOptions: callOptions)
+            return try await send(connection: client, request: request, callOptions: callOptions){
+                client.beginGracefulShutdown()
+            }
         }
     }
 }

Sources/KurrentDB/Core/Additions/Usecase/UnaryStream.swift

@@ -30,7 +30,7 @@ extension Monitoring {
             }
         }
 
-        package func send(connection: GRPCClient<Transport>, request: ClientRequest<UnderlyingRequest>, callOptions: CallOptions) async throws -> Responses {
+        package func send(connection: GRPCClient<Transport>, request: ClientRequest<UnderlyingRequest>, callOptions: CallOptions, finished: @Sendable @escaping ()->Void) async throws -> Responses {
             let (stream, continuation) = AsyncThrowingStream.makeStream(of: Response.self)
             Task {
                 let client = ServiceClient(wrapping: connection)
@@ -39,6 +39,7 @@ extension Monitoring {
                         try continuation.yield(.init(from: message))
                     }
                     continuation.finish()
+                    finished()
                 }
             }
             return stream

Sources/KurrentDB/Monitoring/Usecase/Monitoring.Stats.swift

@@ -24,7 +24,7 @@ extension PersistentSubscriptions {
 
         /// The writer responsible for sending requests to the subscription service.
         let writer: Writer
-
+        
         /// The unique identifier of the subscription, if available.
         ///
         /// This is set during initialization based on the first response from the server.
@@ -39,7 +39,7 @@ extension PersistentSubscriptions {
         ///   - writer: The `Writer` instance used to send requests. Defaults to a new `Writer`.
         ///   - reader: An asynchronous stream of responses from the subscription service.
         /// - Throws: An error if the initialization process fails, such as when the response stream cannot be processed.
-        package init(requests writer: Writer = .init(), responses reader: AsyncThrowingStream<PersistentSubscriptions.ReadResponse, any Error>) async throws {
+        package init(requests writer: Writer = .init(), responses reader: AsyncThrowingStream<PersistentSubscriptions.ReadResponse, any Error>, onTermination: @Sendable @escaping ()->Void) async throws {
             self.writer = writer
 
             var iterator = reader.makeAsyncIterator()
@@ -50,6 +50,9 @@ extension PersistentSubscriptions {
             }
 
             let (stream, continuation) = AsyncThrowingStream.makeStream(of: PersistentSubscription.EventResult.self)
+            continuation.onTermination = { _ in
+                onTermination()
+            }
             Task {
                 while let response = try await iterator.next() {
                     if case let .readEvent(event, retryCount) = response {

Sources/KurrentDB/PersistentSubscriptions/PersistentSubscriptions.Subscription.swift

@@ -63,7 +63,9 @@ extension PersistentSubscriptions.AllStream {
                     }
                 }
             }
-            return try await .init(requests: writer, responses: responses.stream)
+            return try await .init(requests: writer, responses: responses.stream){
+                connection.beginGracefulShutdown()
+            }
         }
     }
 }

Sources/KurrentDB/PersistentSubscriptions/Usecase/AllStream/PersistentSubscriptions.AllStream.Read.swift

@@ -63,7 +63,9 @@ extension PersistentSubscriptions.SpecifiedStream {
                     }
                 }
             }
-            return try await .init(requests: writer, responses: responses.stream)
+            return try await .init(requests: writer, responses: responses.stream){
+                connection.beginGracefulShutdown()
+            }
         }
     }
 }

Sources/KurrentDB/PersistentSubscriptions/Usecase/Specified/PersistentSubscriptions.SpecifiedStream.Read.swift

@@ -42,7 +42,7 @@ extension Projections {
             }
         }
 
-        package func send(connection: GRPCClient<Transport>, request: ClientRequest<UnderlyingRequest>, callOptions: CallOptions) async throws -> Responses {
+        package func send(connection: GRPCClient<Transport>, request: ClientRequest<UnderlyingRequest>, callOptions: CallOptions, finished: @Sendable @escaping ()->Void) async throws -> Responses {
             try await withThrowingTaskGroup(of: Void.self) { _ in
                 let (stream, continuation) = AsyncThrowingStream.makeStream(of: Response.self)
                 let client = ServiceClient(wrapping: connection)
@@ -51,8 +51,8 @@ extension Projections {
                         try continuation.yield(handle(message: message))
                     }
                     continuation.finish()
+                    finished()
                 }
-                continuation.finish()
                 return stream
             }
         }

Sources/KurrentDB/Projections/Usecase/Projections.Statistics.swift

@@ -28,7 +28,7 @@ extension Streams where Target == AllStreams {
             }
         }
 
-        package func send(connection: GRPCClient<Transport>, request: ClientRequest<UnderlyingRequest>, callOptions: CallOptions) async throws -> Responses {
+        package func send(connection: GRPCClient<Transport>, request: ClientRequest<UnderlyingRequest>, callOptions: CallOptions, finished: @Sendable @escaping ()->Void) async throws -> Responses {
             try await withThrowingTaskGroup(of: Void.self) { _ in
                 let client = ServiceClient(wrapping: connection)
                 let (stream, continuation) = AsyncThrowingStream.makeStream(of: Response.self)
@@ -37,6 +37,7 @@ extension Streams where Target == AllStreams {
                         try continuation.yield(handle(message: message))
                     }
                     continuation.finish()
+                    finished()
                 }
                 return stream
             }