Mockup implementation with swift-nio

Package.resolved

@@ -9,6 +9,24 @@
         "version" : "0.0.7"
       }
     },
+    {
+      "identity" : "swift-atomics",
+      "kind" : "remoteSourceControl",
+      "location" : "https://github.com/apple/swift-atomics.git",
+      "state" : {
+        "revision" : "6c89474e62719ddcc1e9614989fff2f68208fe10",
+        "version" : "1.1.0"
+      }
+    },
+    {
+      "identity" : "swift-collections",
+      "kind" : "remoteSourceControl",
+      "location" : "https://github.com/apple/swift-collections.git",
+      "state" : {
+        "revision" : "937e904258d22af6e447a0b72c0bc67583ef64a2",
+        "version" : "1.0.4"
+      }
+    },
     {
       "identity" : "swift-docc-plugin",
       "kind" : "remoteSourceControl",
@@ -17,6 +35,24 @@
         "revision" : "3303b164430d9a7055ba484c8ead67a52f7b74f6",
         "version" : "1.0.0"
       }
+    },
+    {
+      "identity" : "swift-nio",
+      "kind" : "remoteSourceControl",
+      "location" : "https://github.com/apple/swift-nio",
+      "state" : {
+        "revision" : "3db5c4aeee8100d2db6f1eaf3864afdad5dc68fd",
+        "version" : "2.59.0"
+      }
+    },
+    {
+      "identity" : "swift-nio-ssl",
+      "kind" : "remoteSourceControl",
+      "location" : "https://github.com/apple/swift-nio-ssl",
+      "state" : {
+        "revision" : "320bd978cceb8e88c125dcbb774943a92f6286e9",
+        "version" : "2.25.0"
+      }
     }
   ],
   "version" : 2

Package.swift

@@ -7,7 +7,7 @@ let package = Package(
     name: "SwiftyRedis",
     platforms: [
         .iOS(.v13),
-        .macOS(.v10_15),
+        .macOS(.v13),
         .macCatalyst(.v13),
         .tvOS(.v13),
         .watchOS(.v6),
@@ -24,13 +24,22 @@ let package = Package(
         // .package(url: /* package url */, from: "1.0.0"),
         .package(url: "https://github.com/apple/swift-docc-plugin", from: "1.0.0"),
         .package(url: "https://github.com/groue/Semaphore", from: "0.0.7"),
+        .package(url: "https://github.com/apple/swift-nio", from: "2.59.0"),
+        .package(url: "https://github.com/apple/swift-nio-ssl", from: "2.25.0")
     ],
     targets: [
         // Targets are the basic building blocks of a package. A target can define a module or a test suite.
         // Targets can depend on other targets in this package, and on products in packages this package depends on.
         .target(
             name: "SwiftyRedis",
-            dependencies: ["Semaphore"]
+            dependencies: [
+                "Semaphore",
+                .product(name: "NIOCore", package: "swift-nio"),
+                .product(name: "NIOPosix", package: "swift-nio"),
+                .product(name: "NIO", package: "swift-nio"),
+                .product(name: "NIOConcurrencyHelpers", package: "swift-nio"),
+                .product(name: "NIOSSL", package: "swift-nio-ssl")
+            ]
         ),
         .testTarget(
             name: "SwiftyRedisTests",

Sources/SwiftyRedis/AsyncUtils.swift

@@ -0,0 +1,104 @@
+//
+//  AsyncNetworking.swift
+//
+//
+//  Created by Michael van Straten on 27.05.23.
+//
+
+import Foundation
+import Network
+@_spi(AsyncChannel) import NIOCore
+
+/**
+ A struct representing an asynchronous data stream.
+ */
+struct AsyncDataStream {
+    var con: NIOAsyncChannelInboundStream<ByteBuffer>.AsyncIterator
+    var buffer: ByteBuffer = ByteBuffer()
+
+    /**
+     Polls for new data and appends it to the buffer.
+     */
+    private mutating func poll() async throws {
+        if let new_data = try await con.next() {
+            buffer.writeImmutableBuffer(new_data)
+        }
+    }
+
+    /**
+     Checks if there is insufficient data in the buffer.
+     - Parameter requiredAmount: The required amount of data.
+     - Returns: `true` if there is insufficient data, `false` otherwise.
+     */
+    private func hasInsufficientData(requiredAmount: Int) -> Bool {
+        return requiredAmount > buffer.readableBytes
+    }
+}
+
+extension AsyncDataStream {
+    /**
+     Retrieves the next byte from the data stream.
+     - Returns: The next byte.
+     */
+    mutating func next() async throws -> UInt8 {
+        while hasInsufficientData(requiredAmount: 1) {
+            try await poll()
+        }
+
+        return buffer.readBytes(length: 1)![0]
+    }
+
+    /**
+     Retrieves the next `n` bytes from the data stream.
+     - Parameter n: The number of bytes to retrieve.
+     - Returns: The retrieved data.
+     */
+    mutating func next(n: Int) async throws -> Data {
+        if n == 0 {
+            return Data()
+        }
+
+        while hasInsufficientData(requiredAmount: n) {
+            try await poll()
+        }
+
+        return Data(self.buffer.readBytes(length: n)!)
+    }
+
+    /**
+     Retrieves data from the data stream until a specific subsequence is found.
+     - Parameter subsequence: The subsequence to search for.
+     - Returns: The retrieved data.
+     */
+    mutating func nextUntil<S: DataProtocol>(subsequence: S) async throws -> Data {
+        var range = buffer.readableBytesView.firstRange(of: subsequence)
+        while range == nil {
+            try await poll()
+            range = buffer.readableBytesView.firstRange(of: subsequence)
+        }
+
+        return try await next(n: range!.lowerBound - buffer.readableBytesView.startIndex)
+    }
+}
+
+/**
+ Maps each element of the sequence asynchronously.
+
+ - Parameter transform: A closure that takes an element of the sequence as a parameter and returns a transformed value.
+ - Returns: An array containing the transformed values.
+ - Throws: Errors thrown by the `transform` closure.
+ - Note: This extension method is used internally by the ``ResponseValueParser`` and ``RedisPipeline`` class.
+  */
+extension Sequence {
+    func mapAsync<T>(
+        _ transform: (Element) async throws -> T
+    ) async rethrows -> [T] {
+        var values = [T]()
+
+        for element in self {
+            try await values.append(transform(element))
+        }
+
+        return values
+    }
+}