Add cluster node selection for readonly commands

Signed-off-by: Adam Fowler <adamfowler71@gmail.com>

Author: adam-fowler

Sources/Valkey/Cluster/ValkeyClusterClient.swift

@@ -258,9 +258,16 @@ public final class ValkeyClusterClient: Sendable {
         _ commands: [any ValkeyCommand]
     ) async -> [Result<RESPToken, any Error>] {
         guard commands.count > 0 else { return [] }
+        let readOnlyCommand = commands.reduce(true) { $0 && $1.isReadOnly }
+        let nodeSelection =
+            if readOnlyCommand {
+                self.clientConfiguration.readOnlyReplicaSelection.clusterNodeSelection
+            } else {
+                ValkeyClusterNodeSelection.primary
+            }
         // get a list of nodes and the commands that should be run on them
         do {
-            let nodes = try await self.splitCommandsAcrossNodes(commands: commands)
+            let nodes = try await self.splitCommandsAcrossNodes(commands: commands, nodeSelection: nodeSelection)
             // if this list has one element, then just run the pipeline on that single node
             if nodes.count == 1 {
                 do {
@@ -570,7 +577,10 @@ public final class ValkeyClusterClient: Sendable {
     /// These array of indices are then used to create collections of commands to
     /// run on each node
     @usableFromInline
-    func splitCommandsAcrossNodes(commands: [any ValkeyCommand]) async throws -> [ValkeyServerAddress: NodeAndCommands].Values {
+    func splitCommandsAcrossNodes(
+        commands: [any ValkeyCommand],
+        nodeSelection: ValkeyClusterNodeSelection
+    ) async throws -> [ValkeyServerAddress: NodeAndCommands].Values {
         var nodeMap: [ValkeyServerAddress: NodeAndCommands] = [:]
         var index = commands.startIndex
         var prevAddress: ValkeyServerAddress? = nil
@@ -583,7 +593,7 @@ public final class ValkeyClusterClient: Sendable {
                 // Get hash slot for key and add all the commands you have iterated through so far to the
                 // node associated with that key and break out of loop
                 let hashSlot = try self.hashSlot(for: keysAffected)
-                let node = try await self.nodeClient(for: hashSlot.map { [$0] } ?? [], nodeSelection: .primary)
+                let node = try await self.nodeClient(for: hashSlot.map { [$0] } ?? [], nodeSelection: nodeSelection)
                 let address = node.serverAddress
                 let nodeAndCommands = NodeAndCommands(node: node, commandIndices: .init(commands.startIndex..<index))
                 nodeMap[address] = nodeAndCommands
@@ -599,7 +609,7 @@ public final class ValkeyClusterClient: Sendable {
                 if keysAffected.count > 0 {
                     // If command affects a key get hash slot for key and add command to the node associated with that key
                     let hashSlot = try self.hashSlot(for: keysAffected)
-                    let node = try await self.nodeClient(for: hashSlot.map { [$0] } ?? [], nodeSelection: .primary)
+                    let node = try await self.nodeClient(for: hashSlot.map { [$0] } ?? [], nodeSelection: nodeSelection)
                     prevAddress = node.serverAddress
                     nodeMap[prevAddress, default: .init(node: node, commandIndices: [])].commandIndices.append(index)
                 } else {
@@ -610,7 +620,7 @@ public final class ValkeyClusterClient: Sendable {
             }
         } else {
             // if none of the commands affect any keys then choose a random node
-            let node = try await self.nodeClient(for: [], nodeSelection: .primary)
+            let node = try await self.nodeClient(for: [], nodeSelection: nodeSelection)
             let address = node.serverAddress
             let nodeAndCommands = NodeAndCommands(node: node, commandIndices: .init(commands.startIndex..<index))
             nodeMap[address] = nodeAndCommands

Sources/Valkey/Cluster/ValkeyClusterReplicaSelection.swift renamed to Sources/Valkey/Cluster/ValkeyClusterNodeSelection.swift

@@ -610,13 +610,16 @@ struct ClusterIntegrationTests {
                 [(host: firstNodeHostname, port: firstNodePort)],
                 logger: logger
             ) { client in
-                let nodesAndIndices = try await client.splitCommandsAcrossNodes(commands: values.commands)
-                #expect(nodesAndIndices.count == values.selection.count)
-                let sortedNodeAndIndics = nodesAndIndices.sorted { $0.commandIndices[0] < $1.commandIndices[0] }
-                var iterator = sortedNodeAndIndics.makeIterator()
-                var expectedIterator = values.selection.makeIterator()
-                while let result = iterator.next() {
-                    #expect(result.commandIndices == expectedIterator.next())
+                let nodeSelections: [ValkeyClusterNodeSelection] = [.primary, .cycleReplicas(234)]
+                for selection in nodeSelections {
+                    let nodesAndIndices = try await client.splitCommandsAcrossNodes(commands: values.commands, nodeSelection: selection)
+                    #expect(nodesAndIndices.count == values.selection.count)
+                    let sortedNodeAndIndics = nodesAndIndices.sorted { $0.commandIndices[0] < $1.commandIndices[0] }
+                    var iterator = sortedNodeAndIndics.makeIterator()
+                    var expectedIterator = values.selection.makeIterator()
+                    while let result = iterator.next() {
+                        #expect(result.commandIndices == expectedIterator.next())
+                    }
                 }
             }
         }
@@ -665,6 +668,40 @@ struct ClusterIntegrationTests {
             }
         }
 
+        @Test
+        @available(valkeySwift 1.0, *)
+        func testClientPipelineMultipleNodesReadonly() async throws {
+            var logger = Logger(label: "ValkeyCluster")
+            logger.logLevel = .trace
+            let firstNodeHostname = clusterFirstNodeHostname!
+            let firstNodePort = clusterFirstNodePort ?? 6379
+            try await ClusterIntegrationTests.withValkeyCluster([(host: firstNodeHostname, port: firstNodePort)], logger: logger) {
+                client in
+                let keys = (0..<100).map { ValkeyKey("Test\($0)") }
+                var setCommands: [any ValkeyCommand] = .init()
+                for key in keys {
+                    setCommands.append(SET(key, value: key.description))
+                }
+                _ = await client.execute(setCommands)
+                var getCommands: [any ValkeyCommand] = .init()
+                for i in 0..<100 {
+                    let key = ValkeyKey("Test\(i)")
+                    getCommands.append(GET(key))
+                }
+                let results = await client.execute(getCommands)
+                let response = try results[0].get().decode(as: String.self)
+                #expect(response == "Test0")
+                let response2 = try results[2].get().decode(as: String.self)
+                #expect(response2 == "Test2")
+
+                var delCommands: [any ValkeyCommand] = .init()
+                for key in keys {
+                    delCommands.append(DEL(keys: [key]))
+                }
+                _ = await client.execute(delCommands)
+            }
+        }
+
         @Test
         @available(valkeySwift 1.0, *)
         func testClientPipelineMultipleNodesParameterPack() async throws {