Add rtt method

Sources/NatsSwift/ConcurrentQueue.swift

@@ -0,0 +1,19 @@
+import NIOConcurrencyHelpers
+
+internal class ConcurrentQueue<T> {
+    private var elements: [T] = []
+    private let lock = NIOLock()
+
+    func enqueue(_ element: T) {
+        lock.lock()
+        defer { lock.unlock() }
+        elements.append(element)
+    }
+
+    func dequeue() -> T? {
+        lock.lock()
+        defer { lock.unlock() }
+        guard !elements.isEmpty else { return nil }
+        return elements.removeFirst()
+    }
+}

Sources/NatsSwift/NatsClient/NatsClient.swift

@@ -103,4 +103,13 @@ extension Client {
         return try await connectionHandler.subscribe(subject)
 
     }
+
+    public func rtt() async throws -> Duration {
+        guard let connectionHandler = self.connectionHandler else {
+            throw NatsClientError("internal error: empty connection handler")
+        }
+        let ping = RttCommand.makeFrom(channel: connectionHandler.channel)
+        connectionHandler.sendPing(ping)
+        return try await ping.getRoundTripTime	()
+    }
 }

Sources/NatsSwift/NatsConnection.swift

@@ -52,6 +52,8 @@ class ConnectionHandler: ChannelInboundHandler {
     private var serverInfoContinuation: CheckedContinuation<ServerInfo, Error>?
     private var connectionEstablishedContinuation: CheckedContinuation<Void, Error>?
 
+    private let pingQueue = ConcurrentQueue<RttCommand>()
+
     init(
         inputBuffer: ByteBuffer, urls: [URL], reconnectWait: TimeInterval, maxReconnects: Int?,
         retainServersOrder: Bool,
@@ -143,6 +145,7 @@ class ConnectionHandler: ChannelInboundHandler {
             case .pong:
                 logger.debug("pong")
                 self.outstandingPings.store(0, ordering: AtomicStoreOrdering.relaxed)
+                self.pingQueue.dequeue()?.setRoundTripTime()
             case .error(let err):
                 logger.debug("error \(err)")
 
@@ -414,7 +417,7 @@ class ConnectionHandler: ChannelInboundHandler {
         try await self.channel?.close().get()
     }
 
-    private func sendPing() {
+    internal func sendPing(_ rttCommand: RttCommand? = nil) {
         let pingsOut = self.outstandingPings.wrappingIncrementThenLoad(
             ordering: AtomicUpdateOrdering.relaxed)
         if pingsOut > 2 {
@@ -423,6 +426,7 @@ class ConnectionHandler: ChannelInboundHandler {
         }
         let ping = ClientOp.ping
         do {
+            self.pingQueue.enqueue(rttCommand ?? RttCommand.makeFrom(channel: self.channel))
             try self.write(operation: ping)
             logger.debug("sent ping: \(pingsOut)")
         } catch {

Sources/NatsSwift/RttCommand.swift

@@ -0,0 +1,24 @@
+import NIOCore
+
+internal class RttCommand {
+    let startTime = ContinuousClock().now
+    let promise: EventLoopPromise<Duration>?
+
+    static func makeFrom(channel: Channel?) -> RttCommand {
+        RttCommand(promise: channel?.eventLoop.makePromise(of: Duration.self))
+    }
+
+    private init(promise: EventLoopPromise<Duration>?) {
+        self.promise = promise
+    }
+
+    func setRoundTripTime() {
+        let now: ContinuousClock.Instant = ContinuousClock().now
+        let rtt: Duration = now - startTime
+        promise?.succeed(rtt)
+    }
+
+    func getRoundTripTime() async throws -> Duration {
+        try await promise?.futureResult.get() ?? Duration.zero
+    }
+}

Tests/NatsSwiftTests/Integration/ConnectionTests.swift

@@ -12,6 +12,7 @@ import XCTest
 class CoreNatsTests: XCTestCase {
 
     static var allTests = [
+        ("testRtt", testRtt),
         ("testPublish", testPublish),
         ("testPublishWithReply", testPublishWithReply),
         ("testSubscribe", testSubscribe),
@@ -30,6 +31,20 @@ class CoreNatsTests: XCTestCase {
         natsServer.stop()
     }
 
+    func testRtt() async throws {
+        natsServer.start()
+        logger.logLevel = .debug
+        let client = ClientOptions()
+            .url(URL(string: natsServer.clientURL)!)
+            .build()
+        try await client.connect()
+
+        let rtt: Duration = try await client.rtt()
+        XCTAssertGreaterThan(rtt, Duration.zero, "should have RTT")
+
+        try await client.close()
+    }
+
     func testPublish() async throws {
         natsServer.start()
         logger.logLevel = .debug