96 lines
4.2 KiB
Swift
96 lines
4.2 KiB
Swift
import Foundation
|
|
import TermindApplication
|
|
|
|
/// Turns the chunks a running command emits into progress reports.
|
|
///
|
|
/// Locked rather than actor-isolated, because it is written from `libssh2`'s own
|
|
/// serial queue between two reads — see `SSHExecOutputHandler`, which may not
|
|
/// await. The callback fires outside the lock, so a slow reader cannot stall the
|
|
/// command it is reading.
|
|
///
|
|
/// Reports are throttled. A build printing a thousand lines a second would
|
|
/// otherwise raise a thousand events, each one a SwiftUI invalidation, to show
|
|
/// text nobody can read at that speed. The final result is not throttled and
|
|
/// arrives moments later either way, so the only thing a dropped report costs is
|
|
/// an intermediate frame.
|
|
nonisolated public final class AgentCommandOutputStream: @unchecked Sendable {
|
|
private let lock = NSLock()
|
|
private var standardOut = Data()
|
|
private var standardError = Data()
|
|
private var lastReportedAt: ContinuousClock.Instant?
|
|
private let interval: Duration
|
|
private let limit: Int
|
|
private var report: @Sendable (String, String) -> Void
|
|
|
|
public init(
|
|
interval: Duration = .milliseconds(150),
|
|
limit: Int = 256 * 1_024,
|
|
report: @escaping @Sendable (String, String) -> Void
|
|
) {
|
|
self.interval = interval
|
|
self.limit = limit
|
|
self.report = report
|
|
}
|
|
|
|
/// Points the reports at whoever is now responsible for this command.
|
|
///
|
|
/// A command that outlives the `ssh_exec` which started it is handed on: the
|
|
/// call that started it has returned and stopped listening, and the
|
|
/// `ssh_wait` now blocked on it is the card the output belongs under. Output
|
|
/// follows the call that is currently answering for it, so it is never being
|
|
/// drawn into two places or none.
|
|
public func redirect(to report: @escaping @Sendable (String, String) -> Void) {
|
|
lock.withLock {
|
|
self.report = report
|
|
// The new listener has seen nothing, so it does not wait out a
|
|
// window opened by reports it never got.
|
|
lastReportedAt = nil
|
|
}
|
|
}
|
|
|
|
/// The handler to hand to a session. Retains this stream for as long as the
|
|
/// command runs, which is exactly its useful life.
|
|
public var handler: SSHExecOutputHandler {
|
|
{ [self] chunk, isStandardError in append(chunk, isStandardError: isStandardError) }
|
|
}
|
|
|
|
/// Everything captured so far, whether or not a report was due.
|
|
public var captured: (standardOut: String, standardError: String) {
|
|
lock.withLock { (decoded(standardOut), decoded(standardError)) }
|
|
}
|
|
|
|
private func append(_ chunk: Data, isStandardError: Bool) {
|
|
// The sink is read under the lock and called outside it: `redirect` can
|
|
// replace it from another thread, and a slow reader must not be holding
|
|
// the lock the command's own queue needs to append its next chunk.
|
|
typealias Due = (String, String, @Sendable (String, String) -> Void)
|
|
let due: Due? = lock.withLock {
|
|
if isStandardError {
|
|
Self.appendCapped(chunk, to: &standardError, limit: limit)
|
|
} else {
|
|
Self.appendCapped(chunk, to: &standardOut, limit: limit)
|
|
}
|
|
let now = ContinuousClock.now
|
|
if let lastReportedAt, now - lastReportedAt < interval { return nil }
|
|
lastReportedAt = now
|
|
return (decoded(standardOut), decoded(standardError), report)
|
|
}
|
|
guard let due else { return }
|
|
due.2(due.0, due.1)
|
|
}
|
|
|
|
/// A chunk boundary can fall inside a multi-byte character, which decodes as
|
|
/// a replacement character and corrects itself on the next report. Only the
|
|
/// live view ever sees it: the finished result is decoded once, whole, by
|
|
/// the session.
|
|
private func decoded(_ data: Data) -> String { String(decoding: data, as: UTF8.self) }
|
|
|
|
/// The same head-keeping rule the session's own capture uses, so the live
|
|
/// view and the finished result are never showing different ends of the
|
|
/// same output.
|
|
private static func appendCapped(_ chunk: Data, to buffer: inout Data, limit: Int) {
|
|
guard buffer.count < limit else { return }
|
|
buffer.append(chunk.prefix(limit - buffer.count))
|
|
}
|
|
}
|