Refactor test client (#5)

Package.swift

@@ -12,7 +12,7 @@ let package = Package(
         .library(name: "LCLSpeedtest", targets: ["LCLSpeedtest"])
     ],
     dependencies: [
-        .package(url: "https://github.com/apple/swift-nio.git", from: "2.62.0"),
+        .package(url: "https://github.com/johnnzhou/swift-nio.git", branch: "main"),
         .package(url: "https://github.com/apple/swift-log.git", from: "1.5.3"),
         .package(url: "https://github.com/johnnzhou/websocket-kit.git", branch: "main")
     ],
@@ -23,10 +23,11 @@ let package = Package(
                 .product(name: "NIO", package: "swift-nio"),
                 .product(name: "NIOCore", package: "swift-nio"),
                 .product(name: "NIOPosix", package: "swift-nio"),
-                .product(name: "NIOHTTP1", package: "swift-nio"),
+                .product(name: "NIOConcurrencyHelpers", package: "swift-nio"),
                 .product(name: "Logging", package: "swift-log"),
                 .product(name: "WebSocketKit", package: "websocket-kit")
             ]
-        )
+        ),
+        .executableTarget(name: "Demo", dependencies: ["LCLSpeedtest"])
     ]
 )

Sources/Demo/Client.swift

@@ -0,0 +1,71 @@
+import Foundation
+import LCLSpeedtest
+
+@main
+struct Client {
+    static func main() async throws {
+        var client = SpeedTestClient()
+
+        client.onDownloadProgress = { progress in
+            print("progress: \(progress.convertTo(unit: .Mbps)) mbps")
+        }
+        try await client.start(with: .download)
+    }
+}
+
+enum MeasurementUnit: String, CaseIterable, Identifiable, Encodable {
+
+    case Mbps
+    case MBps
+
+    var id: Self {self}
+
+    var string: String {
+        switch self {
+        case .Mbps:
+            return "mbps"
+        case .MBps:
+            return "MB/s"
+        }
+    }
+}
+
+extension MeasurementProgress {
+
+    /// data in Mbps
+    var defaultValueInMegaBits: Double {
+        get {
+            self.convertTo(unit: .Mbps)
+        }
+    }
+
+    /// data in MB/s
+    var defaultValueInMegaBytes: Double {
+        get {
+            self.convertTo(unit: .MBps)
+        }
+    }
+
+    /**
+     Convert the measurement data to the given unit
+
+     - Parameters:
+        unit: the target unit to convert to
+     - Returns: the value in `Double` under the specified unit measurement
+     */
+    func convertTo(unit: MeasurementUnit) -> Double {
+        let elapsedTime = appInfo.elapsedTime
+        let numBytes = appInfo.numBytes
+        let time = Float64(elapsedTime) / 1000000
+        var speed = Float64(numBytes) / time
+        switch unit {
+        case .Mbps:
+            speed *= 8
+        case .MBps:
+            speed *= 1
+        }
+
+        speed /= 1000000
+        return speed
+    }
+}

Sources/Constants.swift → Sources/LCLSpeedtest/Constants.swift

@@ -19,13 +19,15 @@ let MAX_RETRY_COUNT: UInt8 = 5
 let DISCOVER_SERVER_URL = "https://locate.measurementlab.net/v2/nearest/ndt/ndt7?client_name=ndt7-client-ios"
 
 /// Maximum message size to send in one websocket frame.
-let MAX_MESSAGE_SIZE: Int = 1 << 24
+let MAX_MESSAGE_SIZE: Int = 1 << 23
 
 /// Minimum message size to send in one websocket frame.
 let MIN_MESSAGE_SIZE: Int = 1 << 13
 
 /// The number of second to update measurement report to the caller
-let MEASUREMENT_REPORT_INTERVAL: Int64 = 250000 // 250 ms in microsecond
+let MEASUREMENT_REPORT_INTERVAL: Int64 = 250 // 250 ms
 
 /// The number of second to measure the throughput.
-let MEASUREMENT_DURATION: Int64 = 10000000 // 10 second in microsecond
+let MEASUREMENT_DURATION: Int64 = 10 // 10 seconds
+
+let SCALING_FACTOR = 16

Sources/DownloadClient.swift → Sources/LCLSpeedtest/DownloadClient.swift

@@ -18,44 +18,38 @@ import NIOWebSocket
 
 internal final class DownloadClient: SpeedTestable {
     private let url: URL
-    private let eventloop: MultiThreadedEventLoopGroup
+    private let eventloopGroup: MultiThreadedEventLoopGroup
 
-    private var startTime: Int64
-    private var numBytes: Int64
-    private var previousTimeMark: Int64
+    private var startTime: NIODeadline
+    private var totalBytes: Int
+    private var previousTimeMark: NIODeadline
     private let jsonDecoder: JSONDecoder
+    private let emitter = DispatchQueue(label: "downloader", qos: .userInteractive)
 
     required init(url: URL) {
         self.url = url
-        self.eventloop = MultiThreadedEventLoopGroup.singleton
-        self.startTime = 0
-        self.previousTimeMark = 0
-        self.numBytes = 0
+        self.eventloopGroup = MultiThreadedEventLoopGroup(numberOfThreads: 4)
+        self.startTime = .now()
+        self.previousTimeMark = .now()
+        self.totalBytes = 0
         self.jsonDecoder = JSONDecoder()
     }
 
-    deinit {
-        do {
-            try self.eventloop.syncShutdownGracefully()
-        } catch {
-            fatalError("Failed to close channel gracefully: \(error)")
-        }
-    }
-
     var onMeasurement: ((SpeedTestMeasurement) -> Void)?
     var onProgress: ((MeasurementProgress) -> Void)?
     var onFinish: ((MeasurementProgress, Error?) -> Void)?
 
     func start() throws -> EventLoopFuture<Void> {
-        let promise = self.eventloop.next().makePromise(of: Void.self)
-        try WebSocket.connect(
+        let promise = self.eventloopGroup.next().makePromise(of: Void.self)
+        WebSocket.connect(
             to: self.url,
             headers: self.httpHeaders,
             configuration: self.configuration,
-            on: self.eventloop
+            on: self.eventloopGroup
         ) { ws in
             print("websocket connected")
-            self.startTime = Date.nowInMicroSecond
+
+            self.startTime = .now()
 
             ws.onText(self.onText)
             ws.onBinary(self.onBinary)
@@ -68,36 +62,40 @@ internal final class DownloadClient: SpeedTestable {
                 switch closeResult {
                 case .success:
                     if let onFinish = self.onFinish {
-                        onFinish(
-                            DownloadClient.generateMeasurementProgress(
-                                startTime: self.startTime,
-                                numBytes: self.numBytes,
-                                direction: .download
-                            ),
-                            nil
-                        )
+                        self.emitter.async {
+                            onFinish(
+                                DownloadClient.generateMeasurementProgress(
+                                    startTime: self.startTime,
+                                    numBytes: self.totalBytes,
+                                    direction: .download
+                                ),
+                                nil
+                            )
+                        }
                     }
                     ws.close(code: .normalClosure, promise: promise)
                 case .failure(let error):
                     if let onFinish = self.onFinish {
-                        onFinish(
-                            DownloadClient.generateMeasurementProgress(
-                                startTime: self.startTime,
-                                 numBytes: self.numBytes,
-                                  direction: .download
-                            ),
-                            error
-                        )
+                        self.emitter.async {
+                            onFinish(
+                                DownloadClient.generateMeasurementProgress(
+                                    startTime: self.startTime,
+                                     numBytes: self.totalBytes,
+                                      direction: .download
+                                ),
+                                error
+                            )
+                        }
                     }
                     ws.close(code: .goingAway, promise: promise)
                 }
             }
-        }.wait()
+        }.cascadeFailure(to: promise)
         return promise.futureResult
     }
 
     func stop() throws {
-        var itr = self.eventloop.makeIterator()
+        var itr = self.eventloopGroup.makeIterator()
         while let next = itr.next() {
             try next.close()
         }
@@ -107,28 +105,32 @@ internal final class DownloadClient: SpeedTestable {
         let buffer = ByteBuffer(string: text)
         do {
             let measurement: SpeedTestMeasurement = try jsonDecoder.decode(SpeedTestMeasurement.self, from: buffer)
-            self.numBytes += Int64(buffer.readableBytes)
+            self.totalBytes += buffer.readableBytes
             if let onMeasurement = self.onMeasurement {
-                onMeasurement(measurement)
+                self.emitter.async {
+                    onMeasurement(measurement)
+                }
             }
         } catch {
             print("onText Error: \(error)")
         }
     }
 
     func onBinary(ws: WebSocket, bytes: ByteBuffer) {
-        self.numBytes += Int64(bytes.readableBytes)
+        self.totalBytes += bytes.readableBytes
         if let onProgress = self.onProgress {
-            let current = Date.nowInMicroSecond
-            if current - previousTimeMark >= MEASUREMENT_REPORT_INTERVAL {
-                onProgress(
-                    DownloadClient.generateMeasurementProgress(
-                        startTime: self.startTime,
-                         numBytes: self.numBytes,
-                         direction: .download
+            let current = NIODeadline.now()
+            if (current - self.previousTimeMark) > TimeAmount.milliseconds(MEASUREMENT_REPORT_INTERVAL) {
+                self.emitter.async {
+                    onProgress(
+                        DownloadClient.generateMeasurementProgress(
+                            startTime: self.startTime,
+                            numBytes: self.totalBytes,
+                            direction: .download
+                        )
                     )
-                )
-                previousTimeMark = current
+                }
+                self.previousTimeMark = current
             }
         }
     }

Sources/Models/Measurement.swift → Sources/LCLSpeedtest/Models/Measurement.swift

@@ -127,4 +127,10 @@ extension MeasurementProgress {
             test: direction.rawValue
         )
     }
+
+    var mbps: Double {
+        let elapsedTime = appInfo.elapsedTime  // microsecond
+        let numBytes = appInfo.numBytes
+        return Double(numBytes * 8) / Double(elapsedTime)
+    }
 }

Sources/SpeedTestClient.swift → Sources/LCLSpeedtest/SpeedTestClient.swift

@@ -66,11 +66,11 @@ public struct SpeedTestClient {
     }
 
     /// Run the download test using the available test servers
-    private mutating func runDownloadTest(using testServers: [TestServer]) async throws {
-        guard let downloadPath = testServers.first?.urls.downloadPath,
-                let downloadURL = URL(string: downloadPath) else {
-            throw SpeedTestError.invalidTestURL("Cannot locate URL for download test")
-        }
+   private mutating func runDownloadTest(using testServers: [TestServer]) async throws {
+       guard let downloadPath = testServers.first?.urls.downloadPath,
+               let downloadURL = URL(string: downloadPath) else {
+           throw SpeedTestError.invalidTestURL("Cannot locate URL for download test")
+       }
 
         downloader = DownloadClient(url: downloadURL)
         downloader?.onProgress = self.onDownloadProgress

Sources/SpeedTestable.swift → Sources/LCLSpeedtest/SpeedTestable.swift

@@ -13,7 +13,7 @@
 import Foundation
 import WebSocketKit
 import NIOWebSocket
-import NIOCore
+ import NIOCore
 
 /// This protocol defines callbacks to monitor the speed test progress, including the measurement progress,
 /// measurement result, and potential errors when test finishes.
@@ -91,13 +91,13 @@ extension SpeedTestable {
     ///
     /// - Returns: a `MeasurementProgress` containing the sampling period, number of bytes transmitted and test direction.
     static func generateMeasurementProgress(
-        startTime: Int64,
-        numBytes: Int64,
+        startTime: NIODeadline,
+        numBytes: Int,
         direction: TestDirection
     ) -> MeasurementProgress {
         return MeasurementProgress.create(
-            elapedTime: Date.nowInMicroSecond - startTime,
-            numBytes: numBytes,
+            elapedTime: (NIODeadline.now() - startTime).nanoseconds / 1000,
+            numBytes: Int64(numBytes),
             direction: direction
         )
     }