GEO custom command responses (#128)

* GEO custom command responses

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

* Fix reading integers from bulkString

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

* Fix tests after rebase

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

---------

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

Author: adam-fowler

Sources/Valkey/Commands/Custom/GeoCustomCommands.swift

@@ -0,0 +1,65 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the valkey-swift project
+//
+// Copyright (c) 2025 the valkey-swift authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See valkey-swift/CONTRIBUTORS.txt for the list of valkey-swift authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+public struct GeoCoordinates: RESPTokenDecodable, Sendable {
+    public let longitude: Double
+    public let latitude: Double
+
+    public init(fromRESP token: RESPToken) throws {
+        (self.longitude, self.latitude) = try token.decodeArrayElements()
+    }
+}
+
+public typealias GEODISTResponse = Double?
+extension GEODIST {
+    public typealias Response = GEODISTResponse
+}
+
+extension GEOPOS {
+    public typealias Response = [GeoCoordinates?]
+}
+
+extension GEOSEARCH {
+    /// Search entry for GEOSEARCH command.
+    ///
+    /// Given the response for GEOSEARCH is dependent on which `with` attributes flags
+    /// are set in the command it is not possible to know the structure of the response
+    /// beforehand. The order the attributes are in the array, if relevant with flag is
+    /// set, is distance, hash and coordinates. These can be decoded respectively as a
+    /// `Double`, `String` and ``GeoCoordinate``
+    public struct SearchEntry: RESPTokenDecodable, Sendable {
+        public let member: String
+        public let attributes: [RESPToken]
+
+        public init(fromRESP token: RESPToken) throws {
+            switch token.value {
+            case .array(let array):
+                var arrayIterator = array.makeIterator()
+                guard let member = arrayIterator.next() else {
+                    throw RESPParsingError(code: .unexpectedType, buffer: token.base)
+                }
+                self.member = try String(fromRESP: member)
+                self.attributes = array.dropFirst().map { $0 }
+
+            case .bulkString(let buffer):
+                self.member = String(buffer: buffer)
+                self.attributes = []
+
+            default:
+                throw RESPParsingError(code: .unexpectedType, buffer: token.base)
+            }
+        }
+    }
+    public typealias Response = [SearchEntry]
+}

Sources/Valkey/Commands/GeoCommands.swift

@@ -106,8 +106,6 @@ public struct GEODIST<Member1: RESPStringRenderable, Member2: RESPStringRenderab
             }
         }
     }
-    public typealias Response = ByteBuffer?
-
     public var key: ValkeyKey
     public var member1: Member1
     public var member2: Member2
@@ -154,8 +152,6 @@ public struct GEOHASH: ValkeyCommand {
 /// Returns the longitude and latitude of members from a geospatial index.
 @_documentation(visibility: internal)
 public struct GEOPOS: ValkeyCommand {
-    public typealias Response = RESPToken.Array
-
     public var key: ValkeyKey
     public var members: [String]
 
@@ -844,8 +840,6 @@ public struct GEOSEARCH: ValkeyCommand {
             RESPPureToken("ANY", any).encode(into: &commandEncoder)
         }
     }
-    public typealias Response = RESPToken.Array
-
     public var key: ValkeyKey
     public var from: From
     public var by: By
@@ -1141,7 +1135,7 @@ extension ValkeyConnectionProtocol {
         member1: Member1,
         member2: Member2,
         unit: GEODIST<Member1, Member2>.Unit? = nil
-    ) async throws -> ByteBuffer? {
+    ) async throws -> GEODISTResponse {
         try await send(command: GEODIST(key, member1: member1, member2: member2, unit: unit))
     }
 
@@ -1163,7 +1157,7 @@ extension ValkeyConnectionProtocol {
     /// - Complexity: O(1) for each member requested.
     /// - Response: [Array]: An array where each element is a two elements array representing longitude and latitude (x,y) of each member name passed as argument to the command
     @inlinable
-    public func geopos(_ key: ValkeyKey, members: [String] = []) async throws -> RESPToken.Array {
+    public func geopos(_ key: ValkeyKey, members: [String] = []) async throws -> GEOPOS.Response {
         try await send(command: GEOPOS(key, members: members))
     }
 
@@ -1341,7 +1335,7 @@ extension ValkeyConnectionProtocol {
         withcoord: Bool = false,
         withdist: Bool = false,
         withhash: Bool = false
-    ) async throws -> RESPToken.Array {
+    ) async throws -> GEOSEARCH.Response {
         try await send(
             command: GEOSEARCH(
                 key,

Sources/Valkey/RESP/RESPTokenDecodable.swift

@@ -175,8 +175,13 @@ extension Int: RESPTokenDecodable {
             }
             self = value
 
-        case .bulkString,
-            .simpleString,
+        case .bulkString(let buffer):
+            guard let value = Int(String(buffer: buffer)) else {
+                throw RESPParsingError(code: .canNotParseInteger, buffer: token.base)
+            }
+            self = value
+
+        case .simpleString,
             .bulkError,
             .simpleError,
             .verbatimString,
@@ -200,6 +205,19 @@ extension Double: RESPTokenDecodable {
         switch token.value {
         case .double(let value):
             self = value
+
+        case .number(let value):
+            guard let double = Double(exactly: value) else {
+                throw RESPParsingError(code: .unexpectedType, buffer: token.base)
+            }
+            self = double
+
+        case .bulkString(let buffer):
+            guard let value = Double(String(buffer: buffer)) else {
+                throw RESPParsingError(code: .canNotParseDouble, buffer: token.base)
+            }
+            self = value
+
         default:
             throw RESPParsingError(code: .unexpectedType, buffer: token.base)
         }

Sources/_ValkeyCommandsBuilder/ValkeyCommandsRender.swift

@@ -18,6 +18,9 @@ func renderValkeyCommands(_ commands: [String: ValkeyCommand], fullCommandList:
         "BZPOPMAX",
         "BZPOPMIN",
         "CLUSTER SHARDS",
+        "GEODIST",
+        "GEOPOS",
+        "GEOSEARCH",
         "LMPOP",
         "SPOP",
         "SSCAN",

Tests/IntegrationTests/ValkeyTests.swift

@@ -798,4 +798,32 @@ struct GeneratedCommands {
             }
         }
     }
+
+    @available(valkeySwift 1.0, *)
+    @Test
+    func testGEOPOS() async throws {
+        var logger = Logger(label: "Valkey")
+        logger.logLevel = .trace
+        try await withValkeyConnection(.hostname(valkeyHostname, port: 6379), logger: logger) { connection in
+            try await withKey(connection: connection) { key in
+                let count = try await connection.geoadd(
+                    key,
+                    datas: [.init(longitude: 1.0, latitude: 53.0, member: "Edinburgh"), .init(longitude: 1.4, latitude: 53.5, member: "Glasgow")]
+                )
+                #expect(count == 2)
+                let search = try await connection.geosearch(
+                    key,
+                    from: .fromlonlat(.init(longitude: 0.0, latitude: 53.0)),
+                    by: .circle(.init(radius: 10000, unit: .mi)),
+                    withcoord: true,
+                    withdist: true,
+                    withhash: true
+                )
+                print(search.map { $0.member })
+                try print(search.map { try $0.attributes[0].decode(as: Double.self) })
+                try print(search.map { try $0.attributes[1].decode(as: String.self) })
+                try print(search.map { try $0.attributes[2].decode(as: GeoCoordinates.self) })
+            }
+        }
+    }
 }

Tests/ValkeyTests/CommandTests.swift

@@ -700,4 +700,67 @@ struct CommandTests {
             }
         }
     }
+
+    struct GeoCommands {
+        @Test
+        @available(valkeySwift 1.0, *)
+        func geopos() async throws {
+            let channel = NIOAsyncTestingChannel()
+            let logger = Logger(label: "test")
+            let connection = try await ValkeyConnection.setupChannelAndConnect(channel, configuration: .init(), logger: logger)
+            try await channel.processHello()
+
+            try await withThrowingTaskGroup(of: Void.self) { group in
+                group.addTask {
+                    let values = try await connection.geopos("key1", members: ["Edinburgh", "Glasgow", "Dundee"])
+                    #expect(values.count == 3)
+                    let edinburgh = try #require(values[0])
+                    #expect(values[1] == nil)
+                    let dundee = try #require(values[2])
+                    #expect(edinburgh.longitude == 1)
+                    #expect(edinburgh.latitude == 2)
+                    #expect(dundee.longitude == 3)
+                    #expect(dundee.latitude == 4)
+                }
+                group.addTask {
+                    let outbound = try await channel.waitForOutboundWrite(as: ByteBuffer.self)
+                    #expect(outbound == RESPToken(.command(["GEOPOS", "key1", "Edinburgh", "Glasgow", "Dundee"])).base)
+                    try await channel.writeInbound(
+                        RESPToken(
+                            .array([
+                                .array([.bulkString("1.0"), .bulkString("2.0")]),
+                                .null,
+                                .array([.bulkString("3.0"), .bulkString("4.0")]),
+                            ])
+                        ).base
+                    )
+                }
+                try await group.waitForAll()
+            }
+        }
+
+        @Test
+        @available(valkeySwift 1.0, *)
+        func geodist() async throws {
+            let channel = NIOAsyncTestingChannel()
+            let logger = Logger(label: "test")
+            let connection = try await ValkeyConnection.setupChannelAndConnect(channel, configuration: .init(), logger: logger)
+            try await channel.processHello()
+
+            try await withThrowingTaskGroup(of: Void.self) { group in
+                group.addTask {
+                    let distance = try await connection.geodist("key1", member1: "Edinburgh", member2: "Glasgow")
+                    #expect(distance == 42)
+                }
+                group.addTask {
+                    let outbound = try await channel.waitForOutboundWrite(as: ByteBuffer.self)
+                    #expect(outbound == RESPToken(.command(["GEODIST", "key1", "Edinburgh", "Glasgow"])).base)
+                    try await channel.writeInbound(
+                        RESPToken(.bulkString("42.0")).base
+                    )
+                }
+                try await group.waitForAll()
+            }
+        }
+    }
 }