Implement an FSEvents based watcher for macOS.

Provides efficient and low latency directory watching on macOS when using the CFRunLoop based event loop.

dub.sdl

@@ -27,6 +27,7 @@ configuration "cfrunloop" {
 	platforms "osx"
 	versions "EventcoreCFRunLoopDriver"
 	lflags "-framework" "CoreFoundation"
+	lflags "-framework" "CoreServices"
 }
 
 configuration "kqueue" {

source/eventcore/drivers/posix/driver.d

@@ -54,7 +54,7 @@ final class PosixEventDriver(Loop : PosixEventLoop) : EventDriver {
 		version (Posix) alias PipeDriver = PosixEventDriverPipes!Loop;
 		else alias PipeDriver = DummyEventDriverPipes!Loop;
 		version (linux) alias WatcherDriver = InotifyEventDriverWatchers!EventsDriver;
-		//else version (OSX) alias WatcherDriver = FSEventsEventDriverWatchers!EventsDriver;
+		else version (EventcoreCFRunLoopDriver) alias WatcherDriver = FSEventsEventDriverWatchers!EventsDriver;
 		else alias WatcherDriver = PollEventDriverWatchers!EventsDriver;
 		version (Posix) alias ProcessDriver = PosixEventDriverProcesses!Loop;
 		else alias ProcessDriver = DummyEventDriverProcesses!Loop;

source/eventcore/drivers/posix/watchers.d

@@ -196,51 +196,214 @@ final class InotifyEventDriverWatchers(Events : EventDriverEvents) : EventDriver
 	}
 }
 
-version (OSX)
+version (darwin)
 final class FSEventsEventDriverWatchers(Events : EventDriverEvents) : EventDriverWatchers {
 @safe: /*@nogc:*/ nothrow:
-	private Events m_events;
+	import eventcore.internal.corefoundation;
+	import eventcore.internal.coreservices;
+	import std.string : toStringz;
+
+	private {
+		static struct WatcherSlot {
+			FSEventStreamRef stream;
+			string path;
+			string fullPath;
+			FileChangesCallback callback;
+			WatcherID id;
+			int refCount = 1;
+			bool recursive;
+			FSEventStreamEventId lastEvent;
+			ubyte[16 * size_t.sizeof] userData;
+			DataInitializer userDataDestructor;
+		}
+		//HashMap!(void*, WatcherSlot) m_watches;
+		WatcherSlot[WatcherID] m_watches;
+		WatcherID[void*] m_streamMap;
+		Events m_events;
+		size_t m_handleCounter = 1;
+		uint m_validationCounter;
+	}
 
 	this(Events events) { m_events = events; }
 
 	final override WatcherID watchDirectory(string path, bool recursive, FileChangesCallback on_change)
-	{
+	@trusted {
-		/*FSEventStreamCreate
+		import std.path : absolutePath;
-		FSEventStreamScheduleWithRunLoop
+
-		FSEventStreamStart*/
+		FSEventStreamContext ctx;
-		assert(false, "TODO!");
+		ctx.info = () @trusted { return cast(void*)this; } ();
+
+		string abspath;
+		try abspath = absolutePath(path);
+		catch (Exception e) assert(false, e.msg);
+
+		if (m_handleCounter == 0) {
+			m_handleCounter++;
+			m_validationCounter++;
+		}
+		auto id = WatcherID(cast(size_t)m_handleCounter++, m_validationCounter);
+
+		WatcherSlot slot = {
+			path: path,
+			fullPath: abspath,
+			callback: on_change,
+			id: id,
+			recursive: recursive,
+			lastEvent: kFSEventStreamEventIdSinceNow
+		};
+
+		startStream(slot, kFSEventStreamEventIdSinceNow);
+
+		m_events.loop.m_waiterCount++;
+		m_watches[id] = slot;
+		return id;
 	}
 
 	final override bool isValid(WatcherID handle)
 	const {
-		return false;
+		return !!(handle in m_watches);
 	}
 
 	final override void addRef(WatcherID descriptor)
 	{
 		if (!isValid(descriptor)) return;
 
-		assert(false, "TODO!");
+		auto slot = descriptor in m_watches;
+		slot.refCount++;
 	}
 
 	final override bool releaseRef(WatcherID descriptor)
 	{
 		if (!isValid(descriptor)) return true;
 
-		/*FSEventStreamStop
+		auto slot = descriptor in m_watches;
-		FSEventStreamUnscheduleFromRunLoop
+		if (!--slot.refCount) {
-		FSEventStreamInvalidate
+			destroyStream(slot.stream);
-		FSEventStreamRelease*/
+			m_watches.remove(descriptor);
-		assert(false, "TODO!");
+			m_events.loop.m_waiterCount--;
+			return false;
+		}
+
+		return true;
 	}
 
 	final protected override void* rawUserData(WatcherID descriptor, size_t size, DataInitializer initialize, DataInitializer destroy)
 	@system {
 		if (!isValid(descriptor)) return null;
 
-		return m_loop.rawUserDataImpl(descriptor, size, initialize, destroy);
+		auto slot = descriptor in m_watches;
+
+		if (size > WatcherSlot.userData.length) assert(false);
+		if (!slot.userDataDestructor) {
+			initialize(slot.userData.ptr);
+			slot.userDataDestructor = destroy;
+		}
+		return slot.userData.ptr;
 	}
 
+	private static extern(C) void onFSEvent(ConstFSEventStreamRef streamRef,
+		void* clientCallBackInfo, size_t numEvents, void* eventPaths,
+		const(FSEventStreamEventFlags)* eventFlags,
+		const(FSEventStreamEventId)* eventIds)
+	{
+		import std.conv : to;
+		import std.path : asRelativePath, baseName, dirName;
+
+		if (!numEvents) return;
+
+		auto this_ = () @trusted { return cast(FSEventsEventDriverWatchers)clientCallBackInfo; } ();
+		auto h = () @trusted { return cast(void*)streamRef; } ();
+		auto ps = h in this_.m_streamMap;
+		if (!ps) return;
+		auto id = *ps;
+		auto slot = id in this_.m_watches;
+
+		auto patharr = () @trusted { return (cast(const(char)**)eventPaths)[0 .. numEvents]; } ();
+		auto flagsarr = () @trusted { return eventFlags[0 .. numEvents]; } ();
+		auto idarr = () @trusted { return eventIds[0 .. numEvents]; } ();
+
+		// A new stream needs to be created after every change, because events
+		// get coalesced per file (event flags get or'ed together) and it becomes
+		// impossible to determine the actual event
+		this_.startStream(*slot, idarr[$-1]);
+
+		foreach (i; 0 .. numEvents) {
+			auto pathstr = () @trusted { return to!string(patharr[i]); } ();
+			auto f = flagsarr[i];
+
+			string rp;
+			try rp = pathstr.asRelativePath(slot.fullPath).to!string;
+			catch (Exception e) assert(false, e.msg);
+
+			if (rp == "." || rp == "") continue;
+
+			FileChange ch;
+			ch.baseDirectory = slot.path;
+			ch.directory = dirName(rp);
+			ch.name = baseName(rp);
+
+			if (ch.directory == ".") ch.directory = "";
+
+			if (!slot.recursive && ch.directory != "") continue;
+
+			void emit(FileChangeKind k)
+			{
+				ch.kind = k;
+				slot.callback(id, ch);
+			}
+
+			import std.file : exists;
+			bool does_exist = exists(pathstr);
+
+			// The order of tests is important to properly lower the more
+			// complex flags system to the three event types provided by
+			// eventcore
+			if (f & kFSEventStreamEventFlagItemRenamed) {
+				if (f & kFSEventStreamEventFlagItemCreated)
+					emit(FileChangeKind.removed);
+				else emit(FileChangeKind.added);
+			} else if (f & kFSEventStreamEventFlagItemRemoved && !does_exist) {
+				emit(FileChangeKind.removed);
+			} else if (f & kFSEventStreamEventFlagItemModified && does_exist) {
+				emit(FileChangeKind.modified);
+			} else if (f & kFSEventStreamEventFlagItemCreated && does_exist) {
+				emit(FileChangeKind.added);
+			}
+		}
+	}
+
+	private void startStream(ref WatcherSlot slot, FSEventStreamEventId since_when)
+	@trusted {
+		if (slot.stream) {
+			destroyStream(slot.stream);
+			slot.stream = null;
+		}
+
+		FSEventStreamContext ctx;
+		ctx.info = () @trusted { return cast(void*)this; } ();
+
+		auto pstr = CFStringCreateWithBytes(null,
+			cast(const(ubyte)*)slot.path.ptr, slot.path.length,
+			kCFStringEncodingUTF8, false);
+		scope (exit) CFRelease(pstr);
+		auto paths = CFArrayCreate(null, cast(const(void)**)&pstr, 1, null);
+		scope (exit) CFRelease(paths);
+
+		slot.stream = FSEventStreamCreate(null, &onFSEvent, () @trusted { return &ctx; } (),
+			paths, since_when, 0.1, kFSEventStreamCreateFlagFileEvents|kFSEventStreamCreateFlagNoDefer);
+		FSEventStreamScheduleWithRunLoop(slot.stream, CFRunLoopGetMain(), kCFRunLoopDefaultMode);
+		FSEventStreamStart(slot.stream);
+
+		m_streamMap[cast(void*)slot.stream] = slot.id;
+	}
+
+	private void destroyStream(FSEventStreamRef stream)
+	@trusted {
+		FSEventStreamStop(stream);
+		FSEventStreamInvalidate(stream);
+		FSEventStreamRelease(stream);
+		m_streamMap.remove(cast(void*)stream);
+	}
 }
 
 

source/eventcore/internal/corefoundation.d

@@ -1,18 +1,79 @@
 module eventcore.internal.corefoundation;
 
-version (Darwin):
+version (darwin):
 
-extern(C):
+nothrow extern(C):
 
-static if (!is(typeof(CFRelease))) {
+static if (!is(CFTypeRef)) {
 	alias CFTypeRef = const(void)*;
 	alias CFTypeRef CFAllocatorRef;
 	extern const CFAllocatorRef kCFAllocatorDefault;
-	CFTypeRef CFRetain(CFTypeRef cf);
+
-	void CFRelease(CFTypeRef cf);
+	CFTypeRef CFRetain(CFTypeRef cf) @nogc;
+	void CFRelease(CFTypeRef cf) @nogc;
+
+	struct __CFString;
+	alias CFStringRef = const(__CFString)*;
+
+	alias Boolean = ubyte;
+	alias UniChar = ushort;
+	alias CFTypeID = size_t;
+	alias CFIndex = sizediff_t;
+
+	alias CFAllocatorRetainCallBack = const(void)* function(const void *info);
+	alias CFAllocatorReleaseCallBack = void function(const void *info);
+	alias CFAllocatorCopyDescriptionCallBack = CFStringRef function(const void *info);
 }
 
-static if (!is(typeof(CFRunLoop))) {
+static if (!is(CFArrayRef)) {
+	struct __CFArray;
+	alias CFArrayRef = const(__CFArray)*;
+
+	alias CFArrayRetainCallBack = const(void)* function(CFAllocatorRef allocator, const(void)* value);
+	alias CFArrayReleaseCallBack = void function(CFAllocatorRef allocator, const(void)* value);
+	alias CFArrayCopyDescriptionCallBack = CFStringRef function(const(void)* value);
+	alias CFArrayEqualCallBack = Boolean function(const(void)* value1, const(void)* value2);
+
+	struct CFArrayCallBacks {
+		CFIndex version_;
+		CFArrayRetainCallBack retain;
+		CFArrayReleaseCallBack release;
+		CFArrayCopyDescriptionCallBack copyDescription;
+		CFArrayEqualCallBack equal;
+	}
+
+	CFArrayRef CFArrayCreate(CFAllocatorRef allocator, const(void)** values, CFIndex numValues, const(CFArrayCallBacks)* callBacks);
+}
+
+static if (!is(CFStringEncoding)) {
+	alias CFStringEncoding = uint;
+
+	enum kCFStringEncodingInvalidId = 0xffffffffU;
+	enum {
+	    kCFStringEncodingMacRoman = 0,
+	    kCFStringEncodingWindowsLatin1 = 0x0500,
+	    kCFStringEncodingISOLatin1 = 0x0201,
+	    kCFStringEncodingNextStepLatin = 0x0B01,
+	    kCFStringEncodingASCII = 0x0600,
+	    kCFStringEncodingUnicode = 0x0100,
+	    kCFStringEncodingUTF8 = 0x08000100,
+	    kCFStringEncodingNonLossyASCII = 0x0BFF,
+
+	    kCFStringEncodingUTF16 = 0x0100,
+	    kCFStringEncodingUTF16BE = 0x10000100,
+	    kCFStringEncodingUTF16LE = 0x14000100,
+
+	    kCFStringEncodingUTF32 = 0x0c000100,
+	    kCFStringEncodingUTF32BE = 0x18000100,
+	    kCFStringEncodingUTF32LE = 0x1c000100
+	}
+
+	CFStringRef CFStringCreateWithCString(CFAllocatorRef alloc, const(char)* cStr, CFStringEncoding encoding);
+	CFStringRef CFStringCreateWithBytes(CFAllocatorRef alloc, const(ubyte)* bytes, CFIndex numBytes, CFStringEncoding encoding, Boolean isExternalRepresentation);
+	CFStringRef CFStringCreateWithCharacters(CFAllocatorRef alloc, const(UniChar)* chars, CFIndex numChars);
+}
+
+static if (!is(CFRunLoopRef)) {
 	alias CFRunLoopMode = CFStringRef;
 	struct __CFRunLoop;
 	alias CFRunLoopRef = __CFRunLoop*;
@@ -20,16 +81,51 @@ static if (!is(typeof(CFRunLoop))) {
 	alias CFRunLoopSourceRef = __CFRunLoopSource*;
 
 	alias CFTimeInterval = double;
-	alias Boolean = bool;
+
+	enum CFRunLoopRunResult : int {
+	    kCFRunLoopRunFinished = 1,
+	    kCFRunLoopRunStopped = 2,
+	    kCFRunLoopRunTimedOut = 3,
+	    kCFRunLoopRunHandledSource = 4
+	}
 
 	extern const CFStringRef kCFRunLoopDefaultMode;
 	extern const CFStringRef kCFRunLoopCommonModes;
 
-	void CFRunLoopAddSource(CFRunLoopRef rl, CFRunLoopSourceRef source, CFRunLoopMode mode);
+	CFRunLoopRef CFRunLoopGetMain() @nogc;
+	void CFRunLoopAddSource(CFRunLoopRef rl, CFRunLoopSourceRef source, CFRunLoopMode mode) @nogc;
 	CFRunLoopRunResult CFRunLoopRunInMode(CFRunLoopMode mode, CFTimeInterval seconds, Boolean returnAfterSourceHandled);
 }
 
-static if (!is(CFFileDescriptor)) {
+static if (!is(CFFileDescriptorRef)) {
-	alias FSEventStreamRef = x;
+	alias CFFileDescriptorNativeDescriptor = int;
-	FSEventStreamRef FSEventStreamCreate(CFAllocatorRef allocator, FSEventStreamCallback callback, FSEventStreamContext *context, CFArrayRef pathsToWatch, FSEventStreamEventId sinceWhen, CFTimeInterval latency, FSEventStreamCreateFlags flags);
+
+	struct __CFFileDescriptor;
+	alias CFFileDescriptorRef = __CFFileDescriptor*;
+
+	/* Callback Reason Types */
+	enum CFOptionFlags {
+	    kCFFileDescriptorReadCallBack = 1UL << 0,
+	    kCFFileDescriptorWriteCallBack = 1UL << 1
+	}
+
+	alias CFFileDescriptorCallBack = void function(CFFileDescriptorRef f, CFOptionFlags callBackTypes, void* info);
+
+	struct CFFileDescriptorContext {
+	    CFIndex version_;
+	    void* info;
+	    void* function(void *info) retain;
+	    void function(void *info) release;
+	    CFStringRef function(void *info) copyDescription;
+	}
+
+	CFTypeID CFFileDescriptorGetTypeID() @nogc;
+	CFFileDescriptorRef	CFFileDescriptorCreate(CFAllocatorRef allocator, CFFileDescriptorNativeDescriptor fd, Boolean closeOnInvalidate, CFFileDescriptorCallBack callout, const(CFFileDescriptorContext)* context) @nogc;
+	CFFileDescriptorNativeDescriptor CFFileDescriptorGetNativeDescriptor(CFFileDescriptorRef f) @nogc;
+	void CFFileDescriptorGetContext(CFFileDescriptorRef f, CFFileDescriptorContext* context) @nogc;
+	void CFFileDescriptorEnableCallBacks(CFFileDescriptorRef f, CFOptionFlags callBackTypes) @nogc;
+	void CFFileDescriptorDisableCallBacks(CFFileDescriptorRef f, CFOptionFlags callBackTypes) @nogc;
+	void CFFileDescriptorInvalidate(CFFileDescriptorRef f) @nogc;
+	Boolean CFFileDescriptorIsValid(CFFileDescriptorRef f) @nogc;
+	CFRunLoopSourceRef CFFileDescriptorCreateRunLoopSource(CFAllocatorRef allocator, CFFileDescriptorRef f, CFIndex order) @nogc;
 }

source/eventcore/internal/coreservices.d

@@ -0,0 +1,100 @@
+module eventcore.internal.coreservices;
+
+import eventcore.internal.corefoundation;
+
+version (darwin):
+
+nothrow extern(C):
+
+static if (!is(FSEventStreamRef)) {
+	alias FSEventStreamCreateFlags = uint;
+
+	enum {
+		kFSEventStreamCreateFlagNone  = 0x00000000,
+		kFSEventStreamCreateFlagUseCFTypes = 0x00000001,
+		kFSEventStreamCreateFlagNoDefer = 0x00000002,
+		kFSEventStreamCreateFlagWatchRoot = 0x00000004,
+		kFSEventStreamCreateFlagIgnoreSelf = 0x00000008,
+		kFSEventStreamCreateFlagFileEvents = 0x00000010,
+		kFSEventStreamCreateFlagMarkSelf = 0x00000020,
+		kFSEventStreamCreateFlagUseExtendedData = 0x00000040
+	}
+
+	//#define kFSEventStreamEventExtendedDataPathKey      CFSTR("path")
+	//#define kFSEventStreamEventExtendedFileIDKey        CFSTR("fileID")
+
+	alias FSEventStreamEventFlags = uint;
+
+	enum {
+		kFSEventStreamEventFlagNone   = 0x00000000,
+		kFSEventStreamEventFlagMustScanSubDirs = 0x00000001,
+		kFSEventStreamEventFlagUserDropped = 0x00000002,
+		kFSEventStreamEventFlagKernelDropped = 0x00000004,
+		kFSEventStreamEventFlagEventIdsWrapped = 0x00000008,
+		kFSEventStreamEventFlagHistoryDone = 0x00000010,
+		kFSEventStreamEventFlagRootChanged = 0x00000020,
+		kFSEventStreamEventFlagMount  = 0x00000040,
+		kFSEventStreamEventFlagUnmount = 0x00000080,
+		kFSEventStreamEventFlagItemCreated = 0x00000100,
+		kFSEventStreamEventFlagItemRemoved = 0x00000200,
+		kFSEventStreamEventFlagItemInodeMetaMod = 0x00000400,
+		kFSEventStreamEventFlagItemRenamed = 0x00000800,
+		kFSEventStreamEventFlagItemModified = 0x00001000,
+		kFSEventStreamEventFlagItemFinderInfoMod = 0x00002000,
+		kFSEventStreamEventFlagItemChangeOwner = 0x00004000,
+		kFSEventStreamEventFlagItemXattrMod = 0x00008000,
+		kFSEventStreamEventFlagItemIsFile = 0x00010000,
+		kFSEventStreamEventFlagItemIsDir = 0x00020000,
+		kFSEventStreamEventFlagItemIsSymlink = 0x00040000,
+		kFSEventStreamEventFlagOwnEvent = 0x00080000,
+		kFSEventStreamEventFlagItemIsHardlink = 0x00100000,
+		kFSEventStreamEventFlagItemIsLastHardlink = 0x00200000,
+		kFSEventStreamEventFlagItemCloned = 0x00400000
+	}
+
+	alias FSEventStreamEventId = ulong;
+
+	enum kFSEventStreamEventIdSinceNow = 0xFFFFFFFFFFFFFFFFUL;
+
+
+	struct __FSEventStream;
+	alias FSEventStreamRef = __FSEventStream*;
+	alias ConstFSEventStreamRef = const(__FSEventStream)*;
+
+	struct FSEventStreamContext {
+	  CFIndex version_;
+	  void* info;
+	  CFAllocatorRetainCallBack retain;
+	  CFAllocatorReleaseCallBack release;
+	  CFAllocatorCopyDescriptionCallBack copyDescription;
+	}
+
+	alias FSEventStreamCallback = void function(ConstFSEventStreamRef streamRef,
+		void* clientCallBackInfo, size_t numEvents, void* eventPaths,
+		const(FSEventStreamEventFlags)* eventFlags,
+		const(FSEventStreamEventId)* eventIds);
+
+	FSEventStreamRef FSEventStreamCreate(CFAllocatorRef allocator,
+		FSEventStreamCallback callback, FSEventStreamContext* context,
+		CFArrayRef pathsToWatch, FSEventStreamEventId sinceWhen,
+		CFTimeInterval latency, FSEventStreamCreateFlags flags);
+
+	void FSEventStreamRetain(FSEventStreamRef streamRef);
+	void FSEventStreamRelease(FSEventStreamRef streamRef);
+
+	void FSEventStreamScheduleWithRunLoop(FSEventStreamRef streamRef,
+		CFRunLoopRef runLoop, CFStringRef runLoopMode);
+
+	void FSEventStreamUnscheduleFromRunLoop(FSEventStreamRef streamRef,
+		CFRunLoopRef runLoop, CFStringRef runLoopMode);
+
+	void FSEventStreamInvalidate(FSEventStreamRef streamRef);
+
+	Boolean FSEventStreamStart(FSEventStreamRef streamRef);
+
+	FSEventStreamEventId FSEventStreamFlushAsync(FSEventStreamRef streamRef);
+
+	void FSEventStreamFlushSync(FSEventStreamRef streamRef);
+
+	void FSEventStreamStop(FSEventStreamRef streamRef);
+}