diff --git a/Sources/Inotify/Inotify.swift b/Sources/Inotify/Inotify.swift index 3d5d856..475588a 100644 --- a/Sources/Inotify/Inotify.swift +++ b/Sources/Inotify/Inotify.swift @@ -11,12 +11,24 @@ public actor Inotify { 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.Continuation.BufferingPolicy = .unbounded) throws { self.fd = inotify_init1(CInt(IN_NONBLOCK | IN_CLOEXEC)) guard self.fd >= 0 else { 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 { @@ -99,10 +111,13 @@ public actor Inotify { let _ = try? await self.addWatchWithAutomaticSubtreeWatching(forDirectory: event.path.string, mask: mask) } - private static func createEventReader(forFileDescriptor fd: CInt) -> (any DispatchSourceRead, AsyncStream) { + private static func createEventReader( + forFileDescriptor fd: CInt, + bufferingPolicy: AsyncStream.Continuation.BufferingPolicy + ) -> (any DispatchSourceRead, AsyncStream) { let (stream, continuation) = AsyncStream.makeStream( of: RawInotifyEvent.self, - bufferingPolicy: .bufferingNewest(512) + bufferingPolicy: bufferingPolicy ) let reader = DispatchSource.makeReadSource( diff --git a/Tests/InotifyIntegrationTests/BufferingTests.swift b/Tests/InotifyIntegrationTests/BufferingTests.swift new file mode 100644 index 0000000..5155952 --- /dev/null +++ b/Tests/InotifyIntegrationTests/BufferingTests.swift @@ -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..