Close file tree repository registrations

Upon successful registration with a FileTreeRepository, an Observable is
returned by the FileTreeRepository that can be used to observer the
specific globs that were registered. The FileTreeRepository also has a
global Observable that can be used to monitor _all_ events. In order to
implement this feature, internally the FileTreeRepository needs to hold
a reference to the registered Observable so that it forwards relevant
file change events. If we do not close the Observable, it leaks memory
inside of FileTreeRepository. There were a number of places within sbt
where we registered globs and did nothing with the returned Observable.
It was thus straightforward to fix the leak by just closing the returned
Observables.

This came up because I was looking at a heap dump of
https://github.com/jtjeferreira/sbt-multi-module-sample after running
1000 no-op compiles and noticed that the FileTreeRepository.observables
were taking up 75MB out of a total heap of about 300MB.

As a side note, it would be nice if sbt had a warning for unused return
values when a statement is not the last in a block. It's possible that
these leaks wouldn't have happened if we were forced to handle the
returned Observables.

main/src/main/scala/sbt/internal/Continuous.scala

@@ -178,7 +178,7 @@ private[sbt] object Continuous extends DeprecatedContinuous {
     val repository = getRepository(state)
     dynamicInputs ++= inputs
     logger.debug(s"[watch] [${scopedKey.show}] Found inputs: ${inputs.map(_.glob).mkString(",")}")
-    inputs.foreach(i => repository.register(i.glob))
+    inputs.foreach(i => repository.register(i.glob).foreach(_.close()))
     val watchSettings = new WatchSettings(scopedKey)
     new Config(
       scopedKey.show,
@@ -462,7 +462,7 @@ private[sbt] object Continuous extends DeprecatedContinuous {
       if (trackMetaBuild) {
         state.get(CheckBuildSources.CheckBuildSourcesKey).flatMap(_.fileTreeRepository) match {
           case Some(r) => buildGlobs.foreach(r.register(_).foreach(observers.addObservable))
-          case _       => buildGlobs.foreach(repo.register)
+          case _       => buildGlobs.foreach(repo.register(_).foreach(_.close()))
         }
       }
 
@@ -505,6 +505,7 @@ private[sbt] object Continuous extends DeprecatedContinuous {
       override def close(): Unit = {
         configHandle.close()
         handles.forEach(_.close())
+        observers.close()
       }
     }
     val watchLogger: WatchLogger = msg => logger.debug(msg.toString)

main/src/main/scala/sbt/internal/nio/CheckBuildSources.scala

@@ -75,13 +75,13 @@ private[sbt] class CheckBuildSources extends AutoCloseable {
           val repo = FileTreeRepository.default
           repo.addObserver(_ => needUpdate.set(true))
           repository.set(repo)
-          newSources.foreach(g => repo.register(g))
+          newSources.foreach(g => repo.register(g).foreach(_.close()))
         case r =>
       }
     }
     val previousSources = sources.getAndSet(newSources)
     if (previousSources != newSources) {
-      fileTreeRepository.foreach(r => newSources.foreach(g => r.register(g)))
+      fileTreeRepository.foreach(r => newSources.foreach(g => r.register(g).foreach(_.close())))
       previousStamps.set(getStamps(force = true))
     }
   }

main/src/main/scala/sbt/nio/Settings.scala

@@ -149,7 +149,7 @@ private[sbt] object Settings {
       // This makes watch work by ensuring that the input glob is registered with the
       // repository used by the watch process.
       state.value.get(globalFileTreeRepository).foreach { repo =>
-        inputs.foreach(repo.register)
+        inputs.foreach(repo.register(_).foreach(_.close()))
       }
       dynamicInputs.foreach(_ ++= inputs.map(g => DynamicInput(g, stamper, forceTrigger)))
       view.list(inputs)