102 lines
3.2 KiB
Swift
102 lines
3.2 KiB
Swift
|
import Foundation
|
||
|
|
|
||
|
|
/// Tracks files already handed off — shared across watcher restarts.
|
||
|
|
public actor ProcessedSkipList {
|
||
|
|
private var processed = Set<String>()
|
||
|
|
|
||
|
|
public init() {}
|
||
|
|
|
||
|
|
public func contains(_ path: String) -> Bool {
|
||
|
|
processed.contains(path)
|
||
|
|
}
|
||
|
|
|
||
|
|
public func mark(_ path: String) {
|
||
|
|
processed.insert(path)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
/// Polling watch folder. Emits file paths once they are:
|
||
|
|
/// - matching extension
|
||
|
|
/// - size-stable for `stabilityChecks` consecutive polls (partial-write guard)
|
||
|
|
/// - not in the skip list
|
||
|
|
/// Polling (not inotify): cheap at DIT-cart scale, identical behavior on
|
||
|
|
/// network volumes where inotify is unreliable.
|
||
|
|
public actor WatchFolder {
|
||
|
|
private let path: String
|
||
|
|
private let extensions: Set<String>
|
||
|
|
private let pollInterval: Duration
|
||
|
|
private let stabilityChecks: Int
|
||
|
|
private let skipList: ProcessedSkipList
|
||
|
|
|
||
|
|
private var sizeHistory: [String: (size: UInt64, stableCount: Int)] = [:]
|
||
|
|
private var task: Task<Void, Never>?
|
||
|
|
private var continuation: AsyncStream<String>.Continuation?
|
||
|
|
|
||
|
|
public init(
|
||
|
|
path: String,
|
||
|
|
extensions: [String],
|
||
|
|
pollInterval: Duration = .seconds(2),
|
||
|
|
stabilityChecks: Int = 2,
|
||
|
|
skipList: ProcessedSkipList = ProcessedSkipList()
|
||
|
|
) {
|
||
|
|
self.path = path
|
||
|
|
self.extensions = Set(extensions.map { $0.lowercased() })
|
||
|
|
self.pollInterval = pollInterval
|
||
|
|
self.stabilityChecks = stabilityChecks
|
||
|
|
self.skipList = skipList
|
||
|
|
}
|
||
|
|
|
||
|
|
/// Start watching; returns stream of ready file paths.
|
||
|
|
public func start() -> AsyncStream<String> {
|
||
|
|
let stream = AsyncStream<String> { continuation in
|
||
|
|
self.continuation = continuation
|
||
|
|
}
|
||
|
|
task = Task {
|
||
|
|
while !Task.isCancelled {
|
||
|
|
await pollOnce()
|
||
|
|
try? await Task.sleep(for: pollInterval)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return stream
|
||
|
|
}
|
||
|
|
|
||
|
|
public func stop() {
|
||
|
|
task?.cancel()
|
||
|
|
task = nil
|
||
|
|
continuation?.finish()
|
||
|
|
continuation = nil
|
||
|
|
}
|
||
|
|
|
||
|
|
private func pollOnce() async {
|
||
|
|
let fm = FileManager.default
|
||
|
|
guard let entries = try? fm.contentsOfDirectory(atPath: path) else { return }
|
||
|
|
|
||
|
|
for entry in entries {
|
||
|
|
let ext = (entry as NSString).pathExtension.lowercased()
|
||
|
|
guard extensions.contains(ext) else { continue }
|
||
|
|
let full = path + "/" + entry
|
||
|
|
if await skipList.contains(full) { continue }
|
||
|
|
|
||
|
|
guard let attrs = try? fm.attributesOfItem(atPath: full),
|
||
|
|
let size = attrs[.size] as? UInt64 else { continue }
|
||
|
|
|
||
|
|
if let prev = sizeHistory[full] {
|
||
|
|
if prev.size == size {
|
||
|
|
let stable = prev.stableCount + 1
|
||
|
|
if stable >= stabilityChecks {
|
||
|
|
await skipList.mark(full)
|
||
|
|
sizeHistory.removeValue(forKey: full)
|
||
|
|
continuation?.yield(full)
|
||
|
|
} else {
|
||
|
|
sizeHistory[full] = (size, stable)
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
sizeHistory[full] = (size, 0) // still growing
|
||
|
|
}
|
||
|
|
} else {
|
||
|
|
sizeHistory[full] = (size, 0)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|