Keep every event by default
Events were buffered with `bufferingNewest(512)`, so a burst of changes silently lost all but the newest 512 events whenever the consumer lagged. The stream now buffers without limit, and `init(bufferingPolicy:)` lets callers choose a bounded policy.
This commit is contained in:
@@ -11,12 +11,24 @@ public actor Inotify {
|
|||||||
self.eventStream.compactMap(self.transform(_:))
|
self.eventStream.compactMap(self.transform(_:))
|
||||||
}
|
}
|
||||||
|
|
||||||
public init() throws {
|
/// Creates an inotify instance.
|
||||||
|
///
|
||||||
|
/// Events are read from the kernel as soon as they arrive and buffered
|
||||||
|
/// until they are consumed from ``events``.
|
||||||
|
///
|
||||||
|
/// - Parameter bufferingPolicy: How events are kept while no consumer is
|
||||||
|
/// reading ``events``. The default `.unbounded` keeps every event, so a
|
||||||
|
/// burst of changes is never lost; a bounded policy trades memory for
|
||||||
|
/// dropped events.
|
||||||
|
public init(bufferingPolicy: AsyncStream<RawInotifyEvent>.Continuation.BufferingPolicy = .unbounded) throws {
|
||||||
self.fd = inotify_init1(CInt(IN_NONBLOCK | IN_CLOEXEC))
|
self.fd = inotify_init1(CInt(IN_NONBLOCK | IN_CLOEXEC))
|
||||||
guard self.fd >= 0 else {
|
guard self.fd >= 0 else {
|
||||||
throw InotifyError.initFailed(errno: cinotify_get_errno())
|
throw InotifyError.initFailed(errno: cinotify_get_errno())
|
||||||
}
|
}
|
||||||
(self.eventReader, self.eventStream) = Self.createEventReader(forFileDescriptor: fd)
|
(self.eventReader, self.eventStream) = Self.createEventReader(
|
||||||
|
forFileDescriptor: fd,
|
||||||
|
bufferingPolicy: bufferingPolicy
|
||||||
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
public func isExcluded(_ name: String) -> Bool {
|
public func isExcluded(_ name: String) -> Bool {
|
||||||
@@ -99,10 +111,13 @@ public actor Inotify {
|
|||||||
let _ = try? await self.addWatchWithAutomaticSubtreeWatching(forDirectory: event.path.string, mask: mask)
|
let _ = try? await self.addWatchWithAutomaticSubtreeWatching(forDirectory: event.path.string, mask: mask)
|
||||||
}
|
}
|
||||||
|
|
||||||
private static func createEventReader(forFileDescriptor fd: CInt) -> (any DispatchSourceRead, AsyncStream<RawInotifyEvent>) {
|
private static func createEventReader(
|
||||||
|
forFileDescriptor fd: CInt,
|
||||||
|
bufferingPolicy: AsyncStream<RawInotifyEvent>.Continuation.BufferingPolicy
|
||||||
|
) -> (any DispatchSourceRead, AsyncStream<RawInotifyEvent>) {
|
||||||
let (stream, continuation) = AsyncStream<RawInotifyEvent>.makeStream(
|
let (stream, continuation) = AsyncStream<RawInotifyEvent>.makeStream(
|
||||||
of: RawInotifyEvent.self,
|
of: RawInotifyEvent.self,
|
||||||
bufferingPolicy: .bufferingNewest(512)
|
bufferingPolicy: bufferingPolicy
|
||||||
)
|
)
|
||||||
|
|
||||||
let reader = DispatchSource.makeReadSource(
|
let reader = DispatchSource.makeReadSource(
|
||||||
|
|||||||
@@ -0,0 +1,32 @@
|
|||||||
|
import Foundation
|
||||||
|
import Testing
|
||||||
|
@testable import Inotify
|
||||||
|
|
||||||
|
@Suite("Event Buffering")
|
||||||
|
struct BufferingTests {
|
||||||
|
@Test func deliversEveryEventOfABurstToALateConsumer() async throws {
|
||||||
|
try await withTempDir { dir in
|
||||||
|
let fileCount = 1000
|
||||||
|
let watcher = try Inotify()
|
||||||
|
try await watcher.addWatch(path: dir, mask: .create)
|
||||||
|
|
||||||
|
for index in 0..<fileCount {
|
||||||
|
try createFile(at: "\(dir)/file-\(index).txt")
|
||||||
|
}
|
||||||
|
try await Task.sleep(for: .milliseconds(500))
|
||||||
|
|
||||||
|
let eventTask = Task {
|
||||||
|
var events: [InotifyEvent] = []
|
||||||
|
for await event in await watcher.events {
|
||||||
|
events.append(event)
|
||||||
|
}
|
||||||
|
return events
|
||||||
|
}
|
||||||
|
try await Task.sleep(for: .seconds(1))
|
||||||
|
eventTask.cancel()
|
||||||
|
let events = await eventTask.value
|
||||||
|
|
||||||
|
#expect(events.count == fileCount, "Expected \(fileCount) CREATE events, got \(events.count)")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user