Add isolated parameter to subscribe functions

Author: adam-fowler

Sources/Valkey/Connection/ValkeyChannelHandler.swift

@@ -127,7 +127,7 @@ final class ValkeyChannelHandler: ChannelInboundHandler {
         }
     }
 
-    /// Add subscription, and call SUBSCRIBE command
+    /// Add subscription, and call SUBSCRIBE command if required
     func subscribe(
         command: some RESPCommand,
         continuation: ValkeySubscriptionSequence.Continuation,
@@ -150,8 +150,8 @@ final class ValkeyChannelHandler: ChannelInboundHandler {
                 }
                 .map { _ in subscriptionID }
 
-        case .doNothing(let subscription):
-            return self.eventLoop.makeSucceededFuture(subscription.id)
+        case .doNothing(let subscriptionID):
+            return self.eventLoop.makeSucceededFuture(subscriptionID)
         }
     }
 

Sources/Valkey/Subscriptions/ValkeyConnection+subscribe.swift

@@ -27,8 +27,9 @@ extension ValkeyConnection {
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
-    public func subscribe<Value>(
+    public func subscribe<Value: Sendable>(
         to channels: String...,
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscriptionSequence) async throws -> Value
     ) async throws -> Value {
         try await self.subscribe(to: channels, process: process)
@@ -46,7 +47,11 @@ extension ValkeyConnection {
     ///   - channels: list of channels to subscribe to
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
-    public func subscribe<Value>(to channels: [String], process: (ValkeySubscriptionSequence) async throws -> Value) async throws -> Value {
+    public func subscribe<Value: Sendable>(
+        to channels: [String],
+        isolation: isolated (any Actor)? = #isolation,
+        process: (ValkeySubscriptionSequence) async throws -> Value
+    ) async throws -> Value {
         let command = SUBSCRIBE(channel: channels)
         let (id, stream) = try await subscribe(command: command, filters: channels.map { .channel($0) })
         let value: Value
@@ -72,8 +77,9 @@ extension ValkeyConnection {
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
-    public func psubscribe<Value>(
+    public func psubscribe<Value: Sendable>(
         to patterns: String...,
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscriptionSequence) async throws -> Value
     ) async throws -> Value {
         try await self.psubscribe(to: patterns, process: process)
@@ -91,7 +97,11 @@ extension ValkeyConnection {
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
-    public func psubscribe<Value>(to patterns: [String], process: (ValkeySubscriptionSequence) async throws -> Value) async throws -> Value {
+    public func psubscribe<Value: Sendable>(
+        to patterns: [String],
+        isolation: isolated (any Actor)? = #isolation,
+        process: (ValkeySubscriptionSequence) async throws -> Value
+    ) async throws -> Value {
         let command = PSUBSCRIBE(pattern: patterns)
         let (id, stream) = try await subscribe(command: command, filters: patterns.map { .pattern($0) })
         let value: Value
@@ -117,8 +127,9 @@ extension ValkeyConnection {
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
-    public func ssubscribe<Value>(
+    public func ssubscribe<Value: Sendable>(
         to shardchannel: String...,
+        isolation: isolated (any Actor)? = #isolation,
         process: (ValkeySubscriptionSequence) async throws -> Value
     ) async throws -> Value {
         try await self.ssubscribe(to: shardchannel, process: process)
@@ -136,7 +147,11 @@ extension ValkeyConnection {
     ///   - process: Closure that is called with subscription async sequence
     /// - Returns: Return value of closure
     @inlinable
-    public func ssubscribe<Value>(to shardchannel: [String], process: (ValkeySubscriptionSequence) async throws -> Value) async throws -> Value {
+    public func ssubscribe<Value: Sendable>(
+        to shardchannel: [String],
+        isolation: isolated (any Actor)? = #isolation,
+        process: (ValkeySubscriptionSequence) async throws -> Value
+    ) async throws -> Value {
         let command = SSUBSCRIBE(shardchannel: shardchannel)
         let (id, stream) = try await subscribe(command: command, filters: shardchannel.map { .shardChannel($0) })
         let value: Value

Sources/Valkey/Subscriptions/ValkeySubscriptionCommands.swift

@@ -17,7 +17,8 @@ import DequeModule
 /// A subscription command does not return any values instead it pushes messages for each
 /// channel/pattern that has been subscribed/unsubscribed to. This struct catches each
 /// push notification and at the point we have received all the pushes required it returns
-/// the associated value
+/// the associated command. This can then be used to indicate the subscribe/unsubscribe has
+/// been successful
 struct ValkeySubscriptionCommandStack {
     struct SubscribeCommand {
         var filters: [ValkeySubscriptionFilter]

Sources/Valkey/Subscriptions/ValkeySubscriptions.swift

@@ -94,7 +94,7 @@ struct ValkeySubscriptions {
     }
 
     enum SubscribeAction {
-        case doNothing(ValkeySubscription)
+        case doNothing(Int)
         case subscribe(ValkeySubscription, [ValkeySubscriptionFilter])
     }
 
@@ -109,12 +109,12 @@ struct ValkeySubscriptions {
         let id = Self.getSubscriptionID()
         let subscription = ValkeySubscription(id: id, continuation: continuation, filters: filters, logger: self.logger)
         subscriptionIDMap[id] = subscription
-        var action = SubscribeAction.doNothing(subscription)
+        var action = SubscribeAction.doNothing(id)
         for filter in filters {
             switch subscriptionMap[filter, default: .init()].add(subscription: subscription) {
             case .subscribe:
                 switch action {
-                case .doNothing(let subscription):
+                case .doNothing:
                     action = .subscribe(subscription, [filter])
                 case .subscribe(let subscription, var filters):
                     filters.append(filter)