isolation

Signed-off-by: Adam Fowler <[email protected]>

Author: adam-fowler

Sources/Valkey/Subscriptions/ValkeyClient+subscribe.swift

@@ -25,11 +25,13 @@ extension ValkeyClient {
     ///
     /// - Parameters:
     ///   - channels: list of channels to subscribe to
+    ///   - isolation: Actor isolation
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
     public func subscribe<Value>(
         to channels: String...,
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscription) async throws -> sending Value
     ) async throws -> Value {
         try await self.subscribe(to: channels, process: process)
@@ -45,10 +47,12 @@ extension ValkeyClient {
     ///
     /// - Parameters:
     ///   - channels: list of channels to subscribe to
+    ///   - isolation: Actor isolation
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     public func subscribe<Value>(
         to channels: [String],
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscription) async throws -> sending Value
     ) async throws -> Value {
         try await self.subscribe(
@@ -67,11 +71,13 @@ extension ValkeyClient {
     ///
     /// - Parameters:
     ///   - patterns: list of channel patterns to subscribe to
+    ///   - isolation: Actor isolation
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
     public func psubscribe<Value>(
         to patterns: String...,
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscription) async throws -> sending Value
     ) async throws -> Value {
         try await self.psubscribe(to: patterns, process: process)
@@ -86,11 +92,13 @@ extension ValkeyClient {
     ///
     /// - Parameters:
     ///   - patterns: list of channel patterns to subscribe to
+    ///   - isolation: Actor isolation
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
     public func psubscribe<Value>(
         to patterns: [String],
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscription) async throws -> sending Value
     ) async throws -> Value {
         try await self.subscribe(
@@ -109,11 +117,13 @@ extension ValkeyClient {
     ///
     /// - Parameters:
     ///   - shardchannels: list of shard channels to subscribe to
+    ///   - isolation: Actor isolation
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
     public func ssubscribe<Value>(
         to shardchannels: String...,
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscription) async throws -> sending Value
     ) async throws -> Value {
         try await self.ssubscribe(to: shardchannels, process: process)
@@ -128,11 +138,13 @@ extension ValkeyClient {
     ///
     /// - Parameters:
     ///   - shardchannels: list of shard channels to subscribe to
+    ///   - isolation: Actor isolation
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
     public func ssubscribe<Value>(
         to shardchannels: [String],
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscription) async throws -> sending Value
     ) async throws -> Value {
         try await self.subscribe(
@@ -152,10 +164,12 @@ extension ValkeyClient {
     /// channel
     ///
     /// - Parameters:
+    ///   - isolation: Actor isolation
     ///   - process: Closure that is called with async sequence of key invalidations
     /// - Returns: Return value of closure
     @inlinable
     public func subscribeKeyInvalidations<Value>(
+        isolation: isolated (any Actor)? = #isolation,
         process: (AsyncMapSequence<ValkeySubscription, ValkeyKey>) async throws -> sending Value
     ) async throws -> Value {
         try await self.subscribe(to: [ValkeySubscriptions.invalidateChannel]) { subscription in

Sources/Valkey/Subscriptions/ValkeyConnection+subscribe.swift

@@ -167,13 +167,15 @@ extension ValkeyConnection {
     /// channel
     ///
     /// - Parameters:
+    ///   - isolation: Actor isolation
     ///   - process: Closure that is called with async sequence of key invalidations
     /// - Returns: Return value of closure
     @inlinable
     public func subscribeKeyInvalidations<Value>(
+        isolation: isolated (any Actor)? = #isolation,
         process: (AsyncMapSequence<ValkeySubscription, ValkeyKey>) async throws -> sending Value
     ) async throws -> sending Value {
-        try await self.subscribe(to: [ValkeySubscriptions.invalidateChannel]) { subscription in
+        try await self.subscribe(to: [ValkeySubscriptions.invalidateChannel], isolation: isolation) { subscription in
             let keys = subscription.map { ValkeyKey($0.message) }
             return try await process(keys)
         }