Adapt to use the new `WatchService`

This commit adapts `Watched` so that it supports the new `WatchService`
infrastructure introduced in sbt/io. The goal of this infrastructure is
to provide and API for and several implementations of services that
monitor changes to the file system.

The service to use to monitor the file system can be configured with the
key `watchService`.

main-command/src/main/scala/sbt/Watched.scala

@@ -6,30 +6,35 @@ package sbt
 import BasicCommandStrings.ClearOnFailure
 import State.FailureWall
 import annotation.tailrec
-import java.io.File
+import java.nio.file.FileSystems
 
-import sbt.io.PathFinder
+import scala.concurrent.duration._
-import sbt.internal.io.{ SourceModificationWatch, WatchState }
+
+import sbt.io.WatchService
+import sbt.internal.io.{ Source, SourceModificationWatch, WatchState }
 import sbt.internal.util.AttributeKey
 import sbt.internal.util.Types.const
 
 trait Watched {
 
   /** The files watched when an action is run with a preceeding ~ */
-  def watchPaths(s: State): Seq[File] = Nil
+  def watchSources(s: State): Seq[Source] = Nil
   def terminateWatch(key: Int): Boolean = Watched.isEnter(key)
 
   /**
    * The time in milliseconds between checking for changes.  The actual time between the last change made to a file and the
    * execution time is between `pollInterval` and `pollInterval*2`.
    */
-  def pollInterval: Int = Watched.PollDelayMillis
+  def pollInterval: FiniteDuration = Watched.PollDelay
 
   /** The message to show when triggered execution waits for sources to change.*/
   private[sbt] def watchingMessage(s: WatchState): String = Watched.defaultWatchingMessage(s)
 
   /** The message to show before an action is run. */
   private[sbt] def triggeredMessage(s: WatchState): String = Watched.defaultTriggeredMessage(s)
+
+  /** The `WatchService` to use to monitor the file system. */
+  private[sbt] def watchService(): WatchService = FileSystems.getDefault.newWatchService()
 }
 
 object Watched {
@@ -43,7 +48,7 @@ object Watched {
 
   def multi(base: Watched, paths: Seq[Watched]): Watched =
     new AWatched {
-      override def watchPaths(s: State) = (base.watchPaths(s) /: paths)(_ ++ _.watchPaths(s))
+      override def watchSources(s: State) = (base.watchSources(s) /: paths)(_ ++ _.watchSources(s))
       override def terminateWatch(key: Int): Boolean = base.terminateWatch(key)
       override val pollInterval = (base +: paths).map(_.pollInterval).min
       override def watchingMessage(s: WatchState) = base.watchingMessage(s)
@@ -51,15 +56,16 @@ object Watched {
     }
   def empty: Watched = new AWatched
 
-  val PollDelayMillis = 500
+  val PollDelay: FiniteDuration = 500.milliseconds
   def isEnter(key: Int): Boolean = key == 10 || key == 13
   def printIfDefined(msg: String) = if (!msg.isEmpty) System.out.println(msg)
 
   def executeContinuously(watched: Watched, s: State, next: String, repeat: String): State = {
     @tailrec def shouldTerminate: Boolean =
       (System.in.available > 0) && (watched.terminateWatch(System.in.read()) || shouldTerminate)
-    val sourcesFinder = PathFinder { watched watchPaths s }
+    val sources = watched.watchSources(s)
-    val watchState = s get ContinuousState getOrElse WatchState.empty
+    val service = watched.watchService()
+    val watchState = s get ContinuousState getOrElse WatchState.empty(service, sources)
 
     if (watchState.count > 0)
       printIfDefined(watched watchingMessage watchState)
@@ -67,8 +73,7 @@ object Watched {
     val (triggered, newWatchState) =
       try {
         val (triggered, newWatchState) =
-          SourceModificationWatch.watch(sourcesFinder, watched.pollInterval, watchState)(
+          SourceModificationWatch.watch(watched.pollInterval, watchState)(shouldTerminate)
-            shouldTerminate)
         (triggered, newWatchState)
       } catch {
         case e: Exception =>
@@ -84,7 +89,8 @@ object Watched {
       (ClearOnFailure :: next :: FailureWall :: repeat :: s).put(ContinuousState, newWatchState)
     } else {
       while (System.in.available() > 0) System.in.read()
-      s.put(ContinuousState, WatchState.empty)
+      service.close()
+      s.remove(ContinuousState)
     }
   }
   val ContinuousState =

main/src/main/scala/sbt/Defaults.scala

@@ -5,6 +5,7 @@ package sbt
 
 import Def.{ Initialize, ScopedKey, Setting, SettingsDefinition }
 import java.io.{ File, PrintWriter }
+import java.nio.file.FileSystems
 import java.net.{ URI, URL }
 import java.util.Optional
 import java.util.concurrent.{ TimeUnit, Callable }
@@ -23,7 +24,7 @@ import sbt.internal._
 import sbt.internal.CommandStrings.ExportStream
 import sbt.internal.inc.ZincUtil
 import sbt.internal.inc.JavaInterfaceUtil._
-import sbt.internal.io.WatchState
+import sbt.internal.io.{ Source, WatchState }
 import sbt.internal.librarymanagement._
 import sbt.internal.librarymanagement.mavenint.{
   PomExtraDependencyAttributes,
@@ -47,7 +48,8 @@ import sbt.io.{
   PathFinder,
   SimpleFileFilter,
   DirectoryFilter,
-  Hash
+  Hash,
+  WatchService
 }, Path._
 import sbt.librarymanagement.Artifact.{ DocClassifier, SourceClassifier }
 import sbt.librarymanagement.Configurations.{
@@ -250,7 +252,10 @@ object Defaults extends BuildCommon {
       Previous.references :== new Previous.References,
       concurrentRestrictions := defaultRestrictions.value,
       parallelExecution :== true,
-      pollInterval :== 500,
+      pollInterval :== new FiniteDuration(500, TimeUnit.MILLISECONDS),
+      watchService :== { () =>
+        FileSystems.getDefault.newWatchService
+      },
       logBuffered :== false,
       commands :== Nil,
       showSuccess :== true,
@@ -317,7 +322,12 @@ object Defaults extends BuildCommon {
     unmanagedSources := collectFiles(unmanagedSourceDirectories,
                                      includeFilter in unmanagedSources,
                                      excludeFilter in unmanagedSources).value,
-    watchSources in ConfigGlobal ++= unmanagedSources.value,
+    watchSources in ConfigGlobal ++= {
+      val bases = unmanagedSourceDirectories.value
+      val include = (includeFilter in unmanagedSources).value
+      val exclude = (excludeFilter in unmanagedSources).value
+      bases.map(b => new Source(b, include, exclude))
+    },
     managedSourceDirectories := Seq(sourceManaged.value),
     managedSources := generate(sourceGenerators).value,
     sourceGenerators :== Nil,
@@ -337,7 +347,12 @@ object Defaults extends BuildCommon {
     unmanagedResources := collectFiles(unmanagedResourceDirectories,
                                        includeFilter in unmanagedResources,
                                        excludeFilter in unmanagedResources).value,
-    watchSources in ConfigGlobal ++= unmanagedResources.value,
+    watchSources in ConfigGlobal ++= {
+      val bases = unmanagedResourceDirectories.value
+      val include = (includeFilter in unmanagedResources).value
+      val exclude = (excludeFilter in unmanagedResources).value
+      bases.map(b => new Source(b, include, exclude))
+    },
     resourceGenerators :== Nil,
     resourceGenerators += Def.task {
       PluginDiscovery.writeDescriptors(discoveredSbtPlugins.value, resourceManaged.value)
@@ -537,7 +552,7 @@ object Defaults extends BuildCommon {
   def generate(generators: SettingKey[Seq[Task[Seq[File]]]]): Initialize[Task[Seq[File]]] =
     generators { _.join.map(_.flatten) }
 
-  def watchTransitiveSourcesTask: Initialize[Task[Seq[File]]] = {
+  def watchTransitiveSourcesTask: Initialize[Task[Seq[Source]]] = {
     import ScopeFilter.Make.{ inDependencies => inDeps, _ }
     val selectDeps = ScopeFilter(inAggregates(ThisProject) || inDeps(ThisProject))
     val allWatched = (watchSources ?? Nil).all(selectDeps)
@@ -554,6 +569,7 @@ object Defaults extends BuildCommon {
 
   def watchSetting: Initialize[Watched] =
     Def.setting {
+      val getService = watchService.value
       val interval = pollInterval.value
       val base = thisProjectRef.value
       val msg = watchingMessage.value
@@ -564,11 +580,13 @@ object Defaults extends BuildCommon {
         override def pollInterval = interval
         override def watchingMessage(s: WatchState) = msg(s)
         override def triggeredMessage(s: WatchState) = trigMsg(s)
-        override def watchPaths(s: State) = EvaluateTask(Project structure s, key, s, base) match {
+        override def watchService() = getService()
-          case Some((_, Value(ps))) => ps
+        override def watchSources(s: State) =
-          case Some((_, Inc(i)))    => throw i
+          EvaluateTask(Project structure s, key, s, base) match {
-          case None                 => sys.error("key not found: " + Def.displayFull(key))
+            case Some((_, Value(ps))) => ps
-        }
+            case Some((_, Inc(i)))    => throw i
+            case None                 => sys.error("key not found: " + Def.displayFull(key))
+          }
       }
     }
 

main/src/main/scala/sbt/Keys.scala

@@ -36,8 +36,8 @@ import sbt.internal.{
   SessionSettings,
   LogManager
 }
-import sbt.io.FileFilter
+import sbt.io.{ FileFilter, WatchService }
-import sbt.internal.io.WatchState
+import sbt.internal.io.{ Source, WatchState }
 import sbt.internal.util.{ AttributeKey, SourcePosition }
 
 import sbt.librarymanagement.Configurations.CompilerPlugin
@@ -130,9 +130,10 @@ object Keys {
   val analysis = AttributeKey[CompileAnalysis]("analysis", "Analysis of compilation, including dependencies and generated outputs.", DSetting)
   val watch = SettingKey(BasicKeys.watch)
   val suppressSbtShellNotification = SettingKey[Boolean]("suppressSbtShellNotification", """True to suppress the "Executing in batch mode.." message.""", CSetting)
-  val pollInterval = SettingKey[Int]("poll-interval", "Interval between checks for modified sources by the continuous execution command.", BMinusSetting)
+  val pollInterval = SettingKey[FiniteDuration]("poll-interval", "Interval between checks for modified sources by the continuous execution command.", BMinusSetting)
-  val watchSources = TaskKey[Seq[File]]("watch-sources", "Defines the sources in this project for continuous execution to watch for changes.", BMinusSetting)
+  val watchService = SettingKey[() => WatchService]("watch-service", "Service to use to monitor file system changes.", BMinusSetting)
-  val watchTransitiveSources = TaskKey[Seq[File]]("watch-transitive-sources", "Defines the sources in all projects for continuous execution to watch.", CSetting)
+  val watchSources = TaskKey[Seq[Source]]("watch-sources", "Defines the sources in this project for continuous execution to watch for changes.", BMinusSetting)
+  val watchTransitiveSources = TaskKey[Seq[Source]]("watch-transitive-sources", "Defines the sources in all projects for continuous execution to watch.", CSetting)
   val watchingMessage = SettingKey[WatchState => String]("watching-message", "The message to show when triggered execution waits for sources to change.", DSetting)
   val triggeredMessage = SettingKey[WatchState => String]("triggered-message", "The message to show before triggered execution executes an action after sources change.", DSetting)