Improvements to oslib.watch

build.mill

@@ -11,11 +11,11 @@ import de.tobiasroeser.mill.vcs.version.VcsVersion
 
 val communityBuildDottyVersion = sys.props.get("dottyVersion").toList
 
-val scala213Version = "2.13.14"
+val scala213Version = "2.13.16"
 
 val scalaVersions = Seq(
-  "3.3.1",
-  "2.12.17",
+  "3.3.5",
+  "2.12.20",
   scala213Version
 ) ++ communityBuildDottyVersion
 

os/watch/src/FSEventsWatcher.scala

@@ -2,14 +2,16 @@ package os.watch
 
 import com.sun.jna.{NativeLong, Pointer}
 
+import scala.util.control.NonFatal
+
 class FSEventsWatcher(
     srcs: Seq[os.Path],
     onEvent: Set[os.Path] => Unit,
-    logger: (String, Any) => Unit = (_, _) => (),
+    filter: os.Path => Boolean,
+    logger: (String, Any) => Unit,
     latency: Double
 ) extends Watcher {
   private[this] var closed = false
-  private[this] val existingFolders = collection.mutable.Set.empty[os.Path]
   private[this] val callback = new FSEventStreamCallback {
     def invoke(
         streamRef: FSEventStreamRef,
@@ -22,20 +24,18 @@ class FSEventsWatcher(
       val length = numEvents.intValue
       val pathStrings = eventPaths.getStringArray(0, length)
       logger("FSEVENT", pathStrings)
-      val paths = pathStrings.map(os.Path(_))
+      val paths = pathStrings.iterator.map(os.Path(_)).filter(filter).toArray
       val nestedPaths = collection.mutable.Buffer.empty[os.Path]
       // When folders are moved, OS-X does not emit file events for all sub-paths
       // within the new folder, so we are forced to walk that folder and emit the
       // paths ourselves
       for (p <- paths) {
-        if (!os.isDir(p, followLinks = false)) existingFolders.remove(p)
-        else {
-          existingFolders.add(p)
-          try os.walk.stream(p).foreach(nestedPaths.append(_))
-          catch { case e: Throwable => /*do nothing*/ }
+        if (os.isDir(p, followLinks = false)) {
+          try os.walk.stream(p).foreach(p => if (filter(p)) nestedPaths.append(p))
+          catch { case NonFatal(_) => /*do nothing*/ }
         }
       }
-      onEvent((paths ++ nestedPaths).toSet)
+      onEvent((paths.iterator ++ nestedPaths.iterator).toSet)
     }
   }
 

os/watch/src/WatchServiceWatcher.scala

@@ -14,9 +14,9 @@ import scala.util.Properties.isWin
 class WatchServiceWatcher(
     roots: Seq[os.Path],
     onEvent: Set[os.Path] => Unit,
-    logger: (String, Any) => Unit = (_, _) => ()
+    filter: os.Path => Boolean,
+    logger: (String, Any) => Unit
 ) extends Watcher {
-
   val nioWatchService = FileSystems.getDefault.newWatchService()
   val currentlyWatchedPaths = mutable.Map.empty[os.Path, WatchKey]
   val newlyWatchedPaths = mutable.Buffer.empty[os.Path]
@@ -46,7 +46,7 @@ class WatchServiceWatcher(
           modifiers: _*
         )
       )
-      newlyWatchedPaths.append(p)
+      if (filter(p)) newlyWatchedPaths.append(p)
     }
     bufferedEvents.add(p)
   }
@@ -61,12 +61,33 @@ class WatchServiceWatcher(
 
     logger("WATCH KINDS", events.map(_.kind()))
 
+    def logWarning(msg: String): Unit = {
+      System.err.println(s"[oslib.watch] (path=$p) $msg")
+    }
+
+    def logWarningContextNull(e: WatchEvent[_]): Unit = {
+      logWarning(
+        s"Context is null for event kind='${e.kind().name()}' of class ${e.kind().`type`().getName}, " +
+          s"this should never happen."
+      )
+    }
+
     for (e <- events) {
-      bufferedEvents.add(p / e.context().toString)
+      if (e.kind() == OVERFLOW) {
+        logWarning("Overflow detected, some filesystem changes may not be registered.")
+      } else {
+        contextSafe(e) match {
+          case Some(ctx) => bufferedEvents.add(p / ctx.toString)
+          case None => logWarningContextNull(e)
+        }
+      }
     }
 
     for (e <- events if e.kind() == ENTRY_CREATE) {
-      watchSinglePath(p / e.context().toString)
+      contextSafe(e) match {
+        case Some(ctx) => watchSinglePath(p / ctx.toString)
+        case None => logWarningContextNull(e)
+      }
     }
 
     watchKey.reset()
@@ -83,7 +104,8 @@ class WatchServiceWatcher(
         val listing =
           try os.list(top)
           catch {
-            case e: java.nio.file.NotDirectoryException => Nil
+            case _: java.nio.file.NotDirectoryException | _: java.nio.file.NoSuchFileException =>
+              Nil
           }
         for (p <- listing) watchSinglePath(p)
         bufferedEvents.add(top)
@@ -124,10 +146,10 @@ class WatchServiceWatcher(
 
       } catch {
         case e: InterruptedException =>
-          println("Interrupted, exiting: " + e)
+          logger("Interrupted, exiting.", e)
           isRunning.set(false)
         case e: ClosedWatchServiceException =>
-          println("Watcher closed, exiting: " + e)
+          logger("Watcher closed, exiting.", e)
           isRunning.set(false)
       }
   }
@@ -137,7 +159,7 @@ class WatchServiceWatcher(
       isRunning.set(false)
       nioWatchService.close()
     } catch {
-      case e: IOException => println("Error closing watcher: " + e)
+      case e: IOException => logger("Error closing watcher.", e)
     }
   }
 
@@ -146,4 +168,6 @@ class WatchServiceWatcher(
     onEvent(bufferedEvents.toSet)
     bufferedEvents.clear()
   }
+
+  def contextSafe[A](e: WatchEvent[A]): Option[A] = Option(e.context())
 }

os/watch/src/package.scala

@@ -24,15 +24,20 @@ package object watch {
    * changes happening within the watched roots folder, apart from the path
    * at which the change happened. It is up to the `onEvent` handler to query
    * the filesystem and figure out what happened, and what it wants to do.
+   *
+   * @param filter when new paths under `roots` are created, this function is
+   *               invoked with each path. If it returns `false`, the path is
+   *               not watched.
    */
   def watch(
       roots: Seq[os.Path],
       onEvent: Set[os.Path] => Unit,
-      logger: (String, Any) => Unit = (_, _) => ()
+      logger: (String, Any) => Unit = (_, _) => (),
+      filter: os.Path => Boolean = _ => true
   ): AutoCloseable = {
     val watcher = System.getProperty("os.name") match {
-      case "Mac OS X" => new os.watch.FSEventsWatcher(roots, onEvent, logger, 0.05)
-      case _ => new os.watch.WatchServiceWatcher(roots, onEvent, logger)
+      case "Mac OS X" => new os.watch.FSEventsWatcher(roots, onEvent, filter, logger, 0.05)
+      case _ => new os.watch.WatchServiceWatcher(roots, onEvent, filter, logger)
     }
 
     val thread = new Thread {