cli: IO dependencies

.swiftlint.yml

@@ -42,6 +42,15 @@ opt_in_rules:
 analyzer_rules:
   - unused_import
 custom_rules:
+  prefer_dependency_print:
+    name: "Dependency Output"
+    included: '/Sources/'
+    excluded: '/(Tests|Examples)/'
+    regex: '(?<![\w.])(?:Swift\s*\.\s*)?print\s*\('
+    match_kinds:
+      - identifier
+    message: "Use d.out.print instead of print or Swift.print."
+    severity: error
   spaces_not_tabs:
     name: "Spaces not Tabs"
     regex: ^\t

Package.swift

@@ -108,6 +108,8 @@ let package = Package(
         .target(
             name: "XUtils",
             dependencies: [
+                "CXKit",
+                .product(name: "Dependencies", package: "swift-dependencies"),
                 .product(name: "SystemPackage", package: "swift-system"),
                 .product(
                     name: "Subprocess",
@@ -161,21 +163,34 @@ let package = Package(
             dependencies: [
                 "XKit",
                 "XToolSupport",
+                "XCLIUtils",
                 "PackLib",
                 .product(name: "Crypto", package: "swift-crypto"),
                 .product(name: "X509", package: "swift-certificates"),
             ],
             swiftSettings: swiftSettings,
         ),
+        .target(
+            name: "XCLIUtils",
+            dependencies: [
+                "XUtils",
+                .product(name: "Dependencies", package: "swift-dependencies"),
+                .product(name: "ConcurrencyExtras", package: "swift-concurrency-extras"),
+                .product(name: "NIOCore", package: "swift-nio"),
+                .product(name: "NIOPosix", package: "swift-nio"),
+            ],
+            swiftSettings: swiftSettings,
+        ),
         .target(
             name: "XToolSupport",
             dependencies: [
                 "SwiftyMobileDevice",
                 "XKit",
+                "XCLIUtils",
                 "PackLib",
                 .product(name: "ArgumentParser", package: "swift-argument-parser"),
                 .product(name: "SystemPackage", package: "swift-system"),
-                .product(name: "NIOPosix", package: "swift-nio"),
+                .product(name: "Dependencies", package: "swift-dependencies"),
                 .product(name: "libunxip", package: "unxip"),
             ],
             cSettings: cSettings,

Sources/XCLIUtils/CommandInput.swift

@@ -0,0 +1,212 @@
+import ConcurrencyExtras
+import Dependencies
+import Foundation
+import NIOCore
+import NIOPosix
+import XUtils
+
+/// A single consumer's command input. A nil chunk means EOF.
+package struct CommandInput: DependencyKey, Sendable {
+    package var read: @Sendable () async throws -> Data?
+    package var close: @Sendable () async -> Void
+    private let lines = LineBuffer()
+
+    package init(
+        read: @escaping @Sendable () async throws -> Data?,
+        close: @escaping @Sendable () async -> Void = {}
+    ) {
+        self.read = read
+        self.close = close
+    }
+
+    package func readLine() async throws -> String? {
+        try await lines.readLine(read: read)
+    }
+
+    package static var liveValue: Self {
+        let input = StandardCommandInput()
+        return Self(read: { try await input.read() }, close: { await input.close() })
+    }
+
+    package static var testValue: Self { Self(read: unimplemented("CommandInput.read", placeholder: nil)) }
+
+    private actor LineBuffer {
+        var buffer = Data()
+        var ended = false
+
+        func readLine(read: @Sendable () async throws -> Data?) async throws -> String? {
+            while true {
+                try Task.checkCancellation()
+                if let newline = buffer.firstIndex(of: UInt8(ascii: "\n")) {
+                    let line = String(decoding: buffer[..<newline], as: UTF8.self)
+                    buffer.removeSubrange(...newline)
+                    return line.trimmingTrailingCarriageReturn()
+                }
+                if ended {
+                    guard !buffer.isEmpty else { return nil }
+                    defer { buffer.removeAll() }
+                    return String(decoding: buffer, as: UTF8.self).trimmingTrailingCarriageReturn()
+                }
+                if let chunk = try await read() {
+                    buffer.append(chunk)
+                } else {
+                    ended = true
+                }
+            }
+        }
+    }
+}
+
+private extension String {
+    func trimmingTrailingCarriageReturn() -> String {
+        hasSuffix("\r") ? String(dropLast()) : self
+    }
+}
+
+extension DependencyValues {
+    package var commandInput: CommandInput {
+        get { self[CommandInput.self] }
+        set { self[CommandInput.self] = newValue }
+    }
+}
+
+/// Buffers packets received before a command starts reading, without decoding bytes.
+package struct CommandInputQueue: Sendable {
+    private let stream: AsyncThrowingStream<Data, any Error>
+    private let continuation: AsyncThrowingStream<Data, any Error>.Continuation
+
+    package init() {
+        (stream, continuation) = AsyncThrowingStream.makeStream(of: Data.self)
+    }
+
+    package var input: CommandInput {
+        CommandInput(read: { try await read() }, close: { finish() })
+    }
+
+    package func read() async throws -> Data? {
+        try Task.checkCancellation()
+        var iterator = stream.makeAsyncIterator()
+        let chunk = try await iterator.next()
+        try Task.checkCancellation()
+        return chunk
+    }
+
+    package func receive(_ data: Data?) {
+        if let data { continuation.yield(data) } else { finish() }
+    }
+
+    package func finish(throwing error: (any Error)? = nil) {
+        continuation.finish(throwing: error)
+    }
+}
+
+/// Lazily opens stdin, retaining one reader across successive local prompts.
+actor StandardCommandInput {
+    static let chunkSize = 64 * 1024
+    private let descriptor: XUtils.FileDescriptor
+    private var source: Task<Source, any Error>?
+    private var fileRead: Task<Data?, any Error>?
+    private var isClosed = false
+
+    private enum Source: Sendable {
+        case file(FileHandle)
+        case pipe(any Channel, CommandInputQueue, flags: Int32)
+    }
+
+    init(descriptor: XUtils.FileDescriptor = .standardInput) {
+        self.descriptor = descriptor
+    }
+
+    func read() async throws -> Data? {
+        try Task.checkCancellation()
+        guard !isClosed else { return nil }
+        if source == nil {
+            source = Task { try await open() }
+            @Dependency(\.shutdownRegistry) var shutdownRegistry
+            await shutdownRegistry.register { await self.close() }
+        }
+        guard let opening = source else { return nil }
+        let source = try await opening.value
+        try Task.checkCancellation()
+        guard !isClosed else { return nil }
+        let data: Data?
+        switch source {
+        case .file(let handle):
+            let task = Task {
+                try await NIOThreadPool.singleton.runIfActive {
+                    let data = try handle.read(upToCount: Self.chunkSize)
+                    return data?.isEmpty == false ? data : nil
+                }
+            }
+            fileRead = task
+            defer { fileRead = nil }
+            data = try await task.cancellableValue
+        case .pipe(let channel, let queue, _):
+            channel.read()
+            data = try await queue.read()
+        }
+        try Task.checkCancellation()
+        return data
+    }
+
+    func close() async {
+        guard !isClosed else { return }
+        isClosed = true
+        guard let source = try? await source?.value else { return }
+        switch source {
+        case .file(let handle):
+            _ = try? await fileRead?.value
+            try? handle.close()
+        case .pipe(let channel, let queue, let flags):
+            queue.finish()
+            try? await channel.close().get()
+            // dup shares status flags with stdin; undo NIO's O_NONBLOCK change.
+            _ = fcntl(descriptor.rawValue, F_SETFL, flags)
+        }
+    }
+
+    private func open() async throws -> Source {
+        var info = stat()
+        guard fstat(descriptor.rawValue, &info) == 0 else { throw Errno(rawValue: errno) }
+        let duplicate = try descriptor.duplicate()
+        let kind = info.st_mode & mode_t(S_IFMT)
+        // Regular files and /dev/null cannot be registered with epoll on Linux.
+        if kind == mode_t(S_IFREG) || (kind == mode_t(S_IFCHR) && isatty(descriptor.rawValue) == 0) {
+            return .file(FileHandle(fileDescriptor: duplicate.rawValue, closeOnDealloc: true))
+        }
+        let flags = fcntl(descriptor.rawValue, F_GETFL)
+        let queue = CommandInputQueue()
+        do {
+            let channel = try await NIOPipeBootstrap(group: .singletonMultiThreadedEventLoopGroup)
+                .channelOption(ChannelOptions.autoRead, value: false)
+                .channelOption(ChannelOptions.maxMessagesPerRead, value: 1)
+                .channelOption(ChannelOptions.recvAllocator, value: FixedSizeRecvByteBufferAllocator(capacity: Self.chunkSize))
+                .takingOwnershipOfDescriptor(input: duplicate.rawValue) { channel in
+                    channel.pipeline.addHandler(Receiver(queue: queue)).map { channel }
+                }
+            return .pipe(channel, queue, flags: flags)
+        } catch {
+            try? duplicate.close()
+            _ = fcntl(descriptor.rawValue, F_SETFL, flags)
+            throw error
+        }
+    }
+
+    private final class Receiver: ChannelInboundHandler, Sendable {
+        typealias InboundIn = ByteBuffer
+        let queue: CommandInputQueue
+
+        init(queue: CommandInputQueue) { self.queue = queue }
+
+        func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+            queue.receive(Data(unwrapInboundIn(data).readableBytesView))
+        }
+
+        func channelInactive(context: ChannelHandlerContext) { queue.finish() }
+
+        func errorCaught(context: ChannelHandlerContext, error: any Error) {
+            queue.finish(throwing: error)
+            context.close(promise: nil)
+        }
+    }
+}

Sources/XToolSupport/Console.swift → Sources/XCLIUtils/Console.swift

@@ -1,48 +1,16 @@
 import Foundation
-import XKit
 import XUtils
-import NIOPosix
-import NIOCore
-
-enum Console {
-    private static func withStdio<T>(
-        _ body: (
-            _ stdin: NIOAsyncChannelInboundStream<ByteBuffer>,
-            _ stdout: NIOAsyncChannelOutboundWriter<ByteBuffer>
-        ) async throws -> T
-    ) async throws -> T {
-        try await NIOPipeBootstrap(group: .singletonMultiThreadedEventLoopGroup)
-            .takingOwnershipOfDescriptors(
-                input: FileDescriptor.standardInput.duplicate().rawValue,
-                output: FileDescriptor.standardOutput.duplicate().rawValue
-            )
-            .flatMapThrowing { try NIOAsyncChannel(wrappingChannelSynchronously: $0) }
-            .get()
-            .executeThenClose { try await body($0, $1) }
-    }
-
-    static func prompt(_ message: String) async throws -> String {
-        try await withStdio { stdin, stdout in
-            try await stdout.write(ByteBuffer(bytes: message.utf8))
-
-            fflush(stdoutSafe)
-
-            var data = Data()
-            for try await chunk in stdin {
-                let view = chunk.readableBytesView
-                if let endIndex = view.firstIndex(of: UInt8(ascii: "\n")) {
-                    data += view[..<endIndex]
-                    break
-                } else {
-                    data += view
-                }
-            }
-            let str = String(decoding: data, as: UTF8.self)
-            return str.hasSuffix("\r") ? String(str.dropLast()) : str
-        }
+import Dependencies
+
+package enum Console {
+    package static func prompt(_ message: String) async throws -> String {
+        @Dependency(\.commandInput) var input
+        d.out.print(message, terminator: "", flush: true)
+        guard let line = try await input.readLine() else { throw Console.Error("Input ended before a response was received.") }
+        return line
     }
 
-    static func promptRequired(_ message: String, existing: String?) async throws -> String {
+    package static func promptRequired(_ message: String, existing: String?) async throws -> String {
         let value: String
         if let existing {
             value = existing
@@ -55,12 +23,12 @@ enum Console {
         return value
     }
 
-    static func getPassword(_ message: String) async throws -> String {
+    package static func getPassword(_ message: String) async throws -> String {
         if !message.isEmpty {
-            print(message, terminator: "")
+            d.out.print(message, terminator: "")
         }
         let password = try await withoutEcho { try await prompt("") }
-        print()
+        d.out.print()
         return password
     }
 
@@ -91,7 +59,7 @@ enum Console {
         #endif
     }
 
-    static func chooseNumber(in range: Range<Int>) async throws -> Int {
+    package static func chooseNumber(in range: Range<Int>) async throws -> Int {
         let message = "Choice (\(range.lowerBound)-\(range.upperBound - 1)): "
         while true {
             if let choice = try await Int(prompt(message)), range.contains(choice) {
@@ -100,7 +68,7 @@ enum Console {
         }
     }
 
-    static func choose<T>(
+    package static func choose<T>(
         from elements: [T],
         onNoElement: () throws -> T,
         multiPrompt: @autoclosure () -> String,
@@ -112,9 +80,9 @@ enum Console {
         case 1:
             return elements[0]
         default:
-            print(multiPrompt())
+            d.out.print(multiPrompt())
             try elements.enumerated().forEach { index, element in
-                try print("\(index): \(formatter(element))")
+                try d.out.print("\(index): \(formatter(element))")
             }
             let choice = try await chooseNumber(in: elements.indices)
             return elements[choice]
@@ -124,7 +92,7 @@ enum Console {
     private static let yesSet: Set<String> = ["yes", "y"]
     private static let noSet: Set<String> = ["no", "n"]
 
-    static func confirm(_ message: String) async throws -> Bool {
+    package static func confirm(_ message: String) async throws -> Bool {
         while true {
             let resp = try await prompt("\(message) (yes/no): ").lowercased()
             if yesSet.contains(resp) {
@@ -135,9 +103,9 @@ enum Console {
         }
     }
 
-    struct Error: Swift.Error, CustomStringConvertible {
-        let description: String
-        init(_ description: String) {
+    package struct Error: Swift.Error, CustomStringConvertible {
+        package let description: String
+        package init(_ description: String) {
             self.description = description
         }
     }