File Manager (#46)

File Manager — an Akka-based service providing a basic filesystem-like API over network. Currently it just wraps the local filesystem operations. In the future support for other kinds of storage might get added. Ref #32

FileManager/src/main/scala/org/enso/FileManager.scala

@@ -0,0 +1,80 @@
+package org.enso
+
+import java.nio.file.Path
+import java.util.UUID
+
+import akka.actor.Scheduler
+import akka.actor.typed.ActorRef
+import akka.actor.typed.Behavior
+import akka.actor.typed.scaladsl.AbstractBehavior
+import akka.actor.typed.scaladsl.ActorContext
+import akka.actor.typed.scaladsl.AskPattern.Askable
+import akka.actor.typed.scaladsl.Behaviors
+import akka.util.Timeout
+import io.methvin.watcher.DirectoryWatcher
+import org.enso.filemanager.API
+import org.enso.filemanager.API._
+
+import scala.collection.mutable
+import scala.concurrent.Future
+import scala.reflect.ClassTag
+import scala.util.Failure
+import scala.util.Success
+import scala.util.Try
+
+/** The main actor class.
+  *
+  * Implements an RPC-like protocol. Please see member types of
+  * [[org.enso.filemanager.API]] for a list of supported operations and their
+  * respective request-response packages.
+  */
+case class FileManager(projectRoot: Path, context: ActorContext[InputMessage])
+    extends AbstractBehavior[API.InputMessage] {
+
+  /** Active filesystem subtree watchers */
+  val watchers: mutable.Map[UUID, DirectoryWatcher] = mutable.Map()
+
+  def onMessageTyped[response <: Response.Success: ClassTag](
+    message: Request[response]
+  ): Unit = {
+    val response = try {
+      message.contents.validate(projectRoot)
+      val result = message.contents.handle(this)
+      Success(result)
+    } catch { case ex: Throwable => Failure(ex) }
+    context.log.debug(s"Responding with $response")
+    message.replyTo ! response
+  }
+
+  override def onMessage(message: InputMessage): this.type = {
+    context.log.debug(s"Received $message")
+    message.handle(this)
+    this
+  }
+}
+
+object FileManager {
+  val API: org.enso.filemanager.API.type = org.enso.filemanager.API
+
+  /** Factory function for [[FileManager]] [[akka.actor.typed.Behavior]]. */
+  def apply(projectRoot: Path): Behavior[InputMessage] =
+    Behaviors.setup(context => FileManager(projectRoot, context))
+
+  /** Convenience wrapper for
+    * [[akka.actor.typed.scaladsl.AskPattern.Askable.ask]].
+    *
+    * It takes only the request payload (i.e. operation specific part of the
+    * request) and takes care of the rest, automatically deducing the expected
+    * response type.
+    */
+  def ask[response <: Response.Success: ClassTag](
+    actor: ActorRef[API.InputMessage],
+    payload: Request.Payload[response]
+  )(implicit timeout: Timeout,
+    scheduler: Scheduler
+  ): Future[Try[response]] = {
+    actor.ask { replyTo: ActorRef[Try[response]] =>
+      Request(replyTo, payload)
+    }
+  }
+}

FileManager/src/main/scala/org/enso/filemanager/API.scala

@@ -0,0 +1,355 @@
+package org.enso.filemanager
+
+import akka.actor.typed.ActorRef
+import io.methvin.watcher.DirectoryChangeEvent
+import io.methvin.watcher.DirectoryWatcher
+import java.nio.file.Files
+import java.nio.file.NoSuchFileException
+import java.nio.file.NotDirectoryException
+import java.nio.file.Path
+import java.nio.file.attribute.BasicFileAttributes
+import java.util.UUID
+
+import org.apache.commons.io.FileUtils
+import org.enso.FileManager
+import org.enso.filemanager.Detail.EventNotifier
+
+import scala.reflect.ClassTag
+import scala.util.Try
+
+/** Container for types defined for File Manager API.
+  *
+  * Each File Manager operation is implemented as nested type with further
+  * `Request` and `Response` subtypes.
+  */
+object API {
+  import Request.Payload
+  import Response.Success
+
+  /** Base class for messages received by the [[FileManager]]. */
+  type InputMessage = Request[_]
+
+  /** Base class for messages that [[FileManager]] responds with. */
+  type OutputMessage = Try[Response.Success]
+
+  /**
+    * Exception type that is raised on attempt to access to file outside the
+    * project subtree.
+    */
+  final case class PathOutsideProjectException(
+    projectRoot: Path,
+    accessedPath: Path)
+      extends Exception(
+        s"""Cannot access path $accessedPath because it does not belong to 
+           |the project under root directory $projectRoot""".stripMargin
+          .replaceAll("\n", " ")
+      )
+
+  ////////////////////////
+  //// RPC Definition ////
+  ////////////////////////
+
+  /** Request template, parametrised by the response type.
+    */
+  sealed case class Request[ResponseType <: Success: ClassTag](
+    replyTo: ActorRef[Try[ResponseType]],
+    contents: Payload[ResponseType]) {
+
+    def handle(fileManager: FileManager): Unit =
+      fileManager.onMessageTyped(this)
+
+    /** Throws a [[PathOutsideProjectException]] if request involves paths
+      * outside the project subtree. */
+    def validate(projectRoot: Path): Unit =
+      contents.validate(projectRoot)
+  }
+
+  object Request {
+
+    /** Base class for all the operation-specific contents of [[Request]]. */
+    abstract class Payload[+ResponseType <: Success: ClassTag] {
+      def touchedPaths: Seq[Path]
+      def handle(fileManager: FileManager): ResponseType
+
+      def validate(projectRoot: Path): Unit =
+        touchedPaths.foreach(Detail.validatePath(_, projectRoot))
+    }
+  }
+
+  object Response {
+    sealed abstract class Success
+  }
+
+  //////////////////////////////
+  //// Requests / Responses ////
+  //////////////////////////////
+
+  object CopyDirectory {
+    case class Response() extends Success
+    case class Request(from: Path, to: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(from, to)
+      override def handle(fileManager: FileManager): Response = {
+        FileUtils.copyDirectory(from.toFile, to.toFile)
+        Response()
+      }
+    }
+  }
+
+  object CopyFile {
+    case class Response() extends Success
+    case class Request(from: Path, to: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] =
+        Seq(from, to)
+      override def handle(fileManager: FileManager): Response = {
+        Files.copy(from, to)
+        Response()
+      }
+    }
+  }
+
+  object DeleteDirectory {
+    case class Response() extends Success
+    case class Request(path: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(path)
+      override def handle(fileManager: FileManager): Response = {
+        // Despite what commons-io documentation says, the exception is not
+        // thrown when directory is missing, so we do it by hand.
+        if (Files.notExists(path))
+          throw new NoSuchFileException(path.toString)
+
+        FileUtils.deleteDirectory(path.toFile)
+        Response()
+      }
+    }
+  }
+
+  object DeleteFile {
+    case class Response() extends Success
+    case class Request(path: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(path)
+      override def handle(fileManager: FileManager): Response = {
+        Files.delete(path)
+        Response()
+      }
+    }
+  }
+
+  object Exists {
+    case class Response(exists: Boolean) extends Success
+    case class Request(path: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(path)
+      override def handle(fileManager: FileManager) =
+        Response(Files.exists(path))
+    }
+  }
+
+  object List {
+    case class Response(entries: Seq[Path]) extends Success
+    case class Request(path: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(path)
+      override def handle(fileManager: FileManager): Response = {
+        val str = Files.list(path)
+        try {
+          Response(str.toArray.to[Vector].map(_.asInstanceOf[Path]))
+        } finally str.close()
+      }
+    }
+  }
+
+  object MoveDirectory {
+    case class Response() extends Success
+    case class Request(from: Path, to: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(from, to)
+      override def handle(fileManager: FileManager): Response = {
+        FileUtils.moveDirectory(from.toFile, to.toFile)
+        Response()
+      }
+    }
+  }
+
+  object MoveFile {
+    case class Response() extends Success
+    case class Request(from: Path, to: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(from, to)
+      override def handle(fileManager: FileManager): Response = {
+        Files.move(from, to)
+        Response()
+      }
+    }
+  }
+
+  object Read {
+    case class Response(contents: Array[Byte]) extends Success
+    case class Request(path: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(path)
+      override def handle(fileManager: FileManager): Response = {
+        val contents = Files.readAllBytes(path)
+        Response(contents)
+      }
+    }
+  }
+
+  object Status {
+    case class Response(attributes: BasicFileAttributes) extends Success
+    case class Request(path: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(path)
+      override def handle(fileManager: FileManager): Response = {
+        val attributes =
+          Files.readAttributes(path, classOf[BasicFileAttributes])
+        Response(attributes)
+      }
+    }
+  }
+
+  object Touch {
+    case class Response() extends Success
+    case class Request(path: Path) extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(path)
+      override def handle(fileManager: FileManager): Response = {
+        FileUtils.touch(path.toFile)
+        Response()
+      }
+    }
+  }
+
+  /** Operations for managing filesystem watches, please see details.
+    *
+    * The watch will send [[FileSystemEvent]] to the observing agent when any
+    * entry in the observed filesystem subtree is created, modified or deleted.
+    * As this mechanism is built on top of [[java.nio.file.WatchService]] its
+    * limitations and caveats apply. In particular:
+    *  - events may come in different order;
+    *  - events may not come at all if they undo each other (e.g. create and
+    *  delete file in short time period, one modification may overshadow
+    *  another);
+    *  - duplicate notifications may be emitted for a single event;
+    *  - deletion of child entries may not be observed if parent entry is;
+    *  deleted;
+    *  - all of the behaviors listed above are highly system dependent.
+    *
+    * Additionally:
+    *  - watching is always recursive and must target a directory
+    *  - the watched path must not be a symlink (though its parent path
+    *  components are allowed to be symlinks)
+    *  - if the observed path contains symlink, it will remain unresolved in the
+    *  notification events (i.e. the event path prefix shall be the same as the
+    *  observed subtree root).
+    * */
+  object Watch {
+
+    object Create {
+      case class Response(id: UUID) extends Success
+      case class Request(
+        observedDirPath: Path,
+        observer: ActorRef[FileSystemEvent])
+          extends Payload[Response] {
+        override def touchedPaths: Seq[Path] = Seq(observedDirPath)
+        override def handle(fileManager: FileManager): Response = {
+          // Watching a symlink target works only on Windows, presumably thanks
+          // to recursive watch being natively supported. We block this to keep
+          // thinks uniform between platforms.
+          if (Files.isSymbolicLink(observedDirPath))
+            throw new NotDirectoryException(observedDirPath.toString)
+
+          // Watching ordinary file throws an exception on Windows.
+          // To unify behavior, we do this on all platforms.
+          if (!Files.isDirectory(observedDirPath))
+            throw new NotDirectoryException(observedDirPath.toString)
+
+          val handler =
+            EventNotifier(observedDirPath, observer, fileManager)
+          val id = UUID.randomUUID()
+          val watcher = DirectoryWatcher.builder
+            .path(observedDirPath)
+            .listener(handler.notify(_))
+            .build()
+          watcher.watchAsync()
+          fileManager.watchers += (id -> watcher)
+          Response(id)
+        }
+      }
+    }
+
+    object Remove {
+      case class Response() extends Success
+      case class Request(id: UUID) extends Payload[Response] {
+        override def touchedPaths: Seq[Path] = Seq()
+        override def handle(fileManager: FileManager): Response = {
+          val watcher = fileManager.watchers(id)
+          watcher.close()
+          fileManager.watchers -= id
+          Response()
+        }
+      }
+    }
+  }
+
+  object Write {
+    case class Response() extends Success
+    case class Request(path: Path, contents: Array[Byte])
+        extends Payload[Response] {
+      override def touchedPaths: Seq[Path] = Seq(path)
+      override def handle(fileManager: FileManager): Response = {
+        Files.write(path, contents)
+        Response()
+      }
+    }
+  }
+
+  case class FileSystemEvent(
+    eventType: DirectoryChangeEvent.EventType,
+    path: Path)
+}
+
+/** Implementation details, not expected to be relied on as path of API. */
+object Detail {
+  import API._
+
+  def validatePath(validatedPath: Path, projectRoot: Path): Unit = {
+    val normalized = validatedPath.toAbsolutePath.normalize()
+    if (!normalized.startsWith(projectRoot))
+      throw PathOutsideProjectException(projectRoot, validatedPath)
+  }
+
+  /** Helper class used for sending filesystem event notifications. */
+  case class EventNotifier(
+    observedPath: Path,
+    observer: ActorRef[FileSystemEvent],
+    fileManager: FileManager) {
+
+    val realObservedPath: Path           = observedPath.toRealPath()
+    val observingUnresolvedPath: Boolean = observedPath != realObservedPath
+
+    /** If the path prefix got resolved, restores the observed one.
+      *
+      * macOS generates events containing resolved path, i.e. with symlinks
+      * resolved. We don't really want this, as we want to be completely
+      * indifferent to symlink presence and still be able to easily compare
+      * paths. Therefore if we are under symlink and generated event uses
+      * real path, we replace it with path prefix that was observation
+      * target.
+      */
+    def fixedPath(path: Path): Path = {
+      val needsFixing = observingUnresolvedPath && path.startsWith(
+          realObservedPath
+        )
+      needsFixing match {
+        case true  => observedPath.resolve(realObservedPath.relativize(path))
+        case false => path
+      }
+    }
+
+    /** Notifies the observer about a given filesystem event. */
+    def notify(event: DirectoryChangeEvent): Unit = {
+      val message = FileSystemEvent(
+        event.eventType,
+        fixedPath(event.path)
+      )
+      if (message.path != observedPath) {
+        val logText = s"Notifying $observer with $message"
+        fileManager.context.log.debug(logText)
+        observer ! message
+      }
+    }
+  }
+}

FileManager/src/test/scala/org/enso/filemanager/BehaviorTests.scala

@@ -0,0 +1,279 @@
+package org.enso.filemanager
+
+import akka.actor.testkit.typed.scaladsl.BehaviorTestKit
+import akka.actor.testkit.typed.scaladsl.TestInbox
+
+import java.nio.file.FileAlreadyExistsException
+import java.nio.file.Files
+import java.nio.file.NoSuchFileException
+import java.nio.file.Path
+
+import org.apache.commons.io.FileExistsException
+import org.enso.FileManager
+import org.enso.FileManager.API._
+import org.scalatest.FunSuite
+import org.scalatest.Matchers
+import org.scalatest.Outcome
+
+import scala.reflect.ClassTag
+import scala.util.Failure
+import scala.util.Success
+
+class BehaviorTests extends FunSuite with Matchers with Helpers {
+  var testKit: BehaviorTestKit[InputMessage] = _
+  var inbox: TestInbox[OutputMessage]        = _
+
+  override def withFixture(test: NoArgTest): Outcome = {
+    withTemporaryDirectory(_ => {
+      testKit = BehaviorTestKit(FileManager(tempDir))
+      inbox   = TestInbox[OutputMessage]()
+      test()
+    })
+  }
+
+  def expectSuccess[T <: Response.Success: ClassTag](): T = {
+    inbox.receiveMessage() match {
+      case Failure(err) =>
+        fail(s"Unexpected error message: $err")
+      case Success(msg) =>
+        msg shouldBe a[T]
+        msg.asInstanceOf[T]
+    }
+  }
+
+  def expectError[T <: Throwable: ClassTag](): T = {
+    inbox.receiveMessage() match {
+      case Failure(exception) =>
+        exception shouldBe a[T]
+        exception.asInstanceOf[T]
+      case Success(msg) =>
+        fail(s"Unexpected non-error message: $msg")
+    }
+  }
+
+  def runRequest(contents: Request.Payload[Response.Success]): Unit =
+    testKit.run(Request(inbox.ref, contents))
+
+  def ask[res <: Response.Success: ClassTag](
+    contents: Request.Payload[res]
+  ): res = {
+    runRequest(contents)
+    expectSuccess[res]()
+  }
+
+  // ask for something that is not allowed and is expected to cause exception
+  def abet[exception <: Throwable: ClassTag](
+    contents: Request.Payload[Response.Success]
+  ): exception = {
+    runRequest(contents)
+    expectError[exception]()
+  }
+
+  test("Copy directory: empty directory") {
+    val subdir      = createSubDir()
+    val destination = tempDir.resolve("target")
+    ask(CopyDirectory.Request(subdir, destination))
+
+    expectExist(subdir)
+    expectExist(destination)
+  }
+
+  test("Copy directory: non-empty directory") {
+    val subtree     = createSubtree()
+    val destination = tempDir.resolve("target")
+    ask(CopyDirectory.Request(subtree.root, destination))
+    val subtreeExpected = subtree.rebase(destination)
+    expectSubtree(subtree)
+    expectSubtree(subtreeExpected)
+  }
+
+  test("Copy directory: target already exists") {
+    val subtree     = createSubtree()
+    val destination = tempDir.resolve("target")
+    Files.createDirectory(destination)
+    // no exception should happen, but merge
+    ask(CopyDirectory.Request(subtree.root, destination))
+    val subtreeExpected = subtree.rebase(destination)
+    expectSubtree(subtree)
+    expectSubtree(subtreeExpected)
+  }
+
+  test("Copy file: plain") {
+    val srcFile = createSubFile()
+    val dstFile = tempDir.resolve("file2")
+    ask(CopyFile.Request(srcFile, dstFile))
+    expectExist(srcFile)
+    expectExist(dstFile)
+    assert(Files.readAllBytes(dstFile).sameElements(contents))
+  }
+
+  test("Copy file: target already exists") {
+    val srcFile = createSubFile()
+    val dstFile = createSubFile()
+    abet[FileAlreadyExistsException](CopyFile.Request(srcFile, dstFile))
+    expectExist(srcFile)
+  }
+
+  test("Delete directory: empty directory") {
+    val dir = createSubDir()
+    ask(DeleteDirectory.Request(dir))
+    expectNotExist(dir)
+  }
+
+  test("Delete directory: non-empty directory") {
+    val subtree = createSubtree()
+    ask(DeleteDirectory.Request(subtree.root))
+    expectNotExist(subtree.root)
+  }
+
+  test("Delete directory: missing directory") {
+    val missingPath = tempDir.resolve("foo")
+    abet[NoSuchFileException](DeleteDirectory.Request(missingPath))
+  }
+
+  test("Delete file: simple") {
+    val file = createSubFile()
+    expectExist(file)
+    ask(DeleteFile.Request(file))
+    expectNotExist(file)
+  }
+
+  test("Delete file: missing file") {
+    val missingPath = tempDir.resolve("foo")
+    expectNotExist(missingPath)
+    abet[NoSuchFileException](DeleteFile.Request(missingPath))
+    expectNotExist(missingPath)
+  }
+
+  test("Exists: outside project by relative path") {
+    val path = tempDir.resolve("../foo")
+    // Make sure that our path seemingly may look like something under the project.
+    assert(path.startsWith(tempDir))
+    abet[PathOutsideProjectException](Exists.Request(path))
+  }
+
+  test("Exists: outside project by absolute path") {
+    abet[PathOutsideProjectException](Exists.Request(homeDirectory()))
+  }
+
+  test("Exists: existing file") {
+    val filePath = createSubFile()
+    val response = ask(Exists.Request(filePath))
+    response.exists should be(true)
+  }
+
+  test("Exists: existing directory") {
+    val dirPath  = createSubDir()
+    val response = ask(Exists.Request(dirPath))
+    response.exists should be(true)
+  }
+
+  test("Exists: missing file") {
+    val filePath = tempDir.resolve("bar")
+    val response = ask(Exists.Request(filePath))
+    response.exists should be(false)
+  }
+
+  test("List: empty directory") {
+    val requestContents = List.Request(tempDir)
+    val response        = ask(requestContents)
+    response.entries should have length 0
+  }
+
+  test("List: missing directory") {
+    val path = tempDir.resolve("bar")
+    abet[NoSuchFileException](List.Request(path))
+  }
+
+  test("List: non-empty directory") {
+    val filePath   = createSubFile()
+    val subdirPath = createSubDir()
+    val response   = ask(List.Request(tempDir))
+
+    def expectPath(path: Path): Path = {
+      response.entries.find(_.toString == path.toString) match {
+        case Some(entry) => entry
+        case _           => fail(s"cannot find entry for path $path")
+      }
+    }
+
+    response.entries should have length 2
+    expectPath(filePath)
+    expectPath(subdirPath)
+  }
+
+  test("List: outside project") {
+    abet[PathOutsideProjectException](List.Request(homeDirectory()))
+  }
+
+  test("Move directory: empty directory") {
+    val subdir      = createSubDir()
+    val destination = tempDir.resolve("target")
+    ask(MoveDirectory.Request(subdir, destination))
+    assert(!Files.exists(subdir))
+    assert(Files.exists(destination))
+  }
+
+  test("Move directory: non-empty directory") {
+    val subtree     = createSubtree()
+    val destination = tempDir.resolve("target")
+    ask(MoveDirectory.Request(subtree.root, destination))
+    val subtreeExpected = subtree.rebase(destination)
+    assert(!Files.exists(subtree.root))
+    expectSubtree(subtreeExpected)
+  }
+
+  test("Move directory: target already exists") {
+    val subtree     = createSubtree()
+    val destination = tempDir.resolve("target")
+    Files.createDirectory(destination)
+    abet[FileExistsException](MoveDirectory.Request(subtree.root, destination))
+    // Source was not destroyed by failed move.
+    expectSubtree(subtree)
+  }
+
+  test("Stat: missing file") {
+    val filePath = tempDir.resolve("bar")
+    abet[NoSuchFileException](Status.Request(filePath))
+  }
+
+  test("Read: file") {
+    val filePath = tempDir.resolve("bar")
+    Files.write(filePath, contents)
+    val response = ask(Read.Request(filePath))
+    response.contents should be(contents)
+  }
+
+  test("Touch: new file") {
+    val filePath = tempDir.resolve("bar")
+    ask(Touch.Request(filePath))
+    expectExist(filePath)
+    Files.size(filePath) should be(0)
+  }
+
+  test("Touch: update file") {
+    val filePath         = createSubFile()
+    val initialTimestamp = Files.getLastModifiedTime(filePath).toInstant
+    Thread.sleep(1000)
+    ask(Touch.Request(filePath))
+    val finalTimestamp = Files.getLastModifiedTime(filePath).toInstant
+    assert(initialTimestamp.isBefore(finalTimestamp))
+    expectExist(filePath)
+  }
+
+  test("Write: file") {
+    val filePath = tempDir.resolve("bar")
+    ask(Write.Request(filePath, contents))
+    val actualFileContents = Files.readAllBytes(filePath)
+    actualFileContents should be(contents)
+  }
+
+  test("Status: normal file") {
+    val filePath = createSubFile()
+    val contents = "aaa"
+    Files.write(filePath, contents.getBytes())
+    val response = ask(Status.Request(filePath))
+    response.attributes.isDirectory should be(false)
+    response.attributes.size should be(contents.length)
+  }
+}

FileManager/src/test/scala/org/enso/filemanager/Helpers.scala

@@ -0,0 +1,85 @@
+package org.enso.filemanager
+
+import java.nio.file.Files
+import java.nio.file.Path
+import java.nio.file.Paths
+
+import org.apache.commons.io.FileUtils
+
+import org.scalatest.Matchers
+
+trait Helpers extends Matchers {
+  var tempDir: Path = _
+
+  val contents: Array[Byte] =
+    "葦垣の中の和草にこやかに我れと笑まして人に知らゆな\nzażółć gęślą jaźń".getBytes
+
+  def createSubFile(): Path = {
+    val path = Files.createTempFile(tempDir, "foo", "")
+    Files.write(path, contents)
+  }
+
+  def createSubDir(): Path = {
+    Files.createTempDirectory(tempDir, "foo")
+  }
+
+  def homeDirectory(): Path = Paths.get(System.getProperty("user.home"))
+
+  def setupTemp(): Unit = {
+    tempDir = Files.createTempDirectory("file-manager-test")
+  }
+
+  def cleanTemp(): Unit = {
+    FileUtils.deleteDirectory(tempDir.toFile)
+    tempDir = null
+  }
+
+  def withTemporaryDirectory[ret](f: Path => ret): ret = {
+    setupTemp()
+    try f(tempDir)
+    finally cleanTemp()
+  }
+
+  case class Subtree(
+    root: Path,
+    childrenFiles: Seq[Path],
+    childrenDirs: Seq[Path]) {
+
+    val elements: Seq[Path] =
+      (Seq(root) ++ childrenDirs ++ childrenFiles).map(root.resolve)
+
+    def rebase(otherRoot: Path): Subtree =
+      Subtree(otherRoot, childrenFiles, childrenDirs)
+  }
+
+  def createSubtree(): Subtree = {
+    val root       = createSubDir()
+    val rootFile1  = Paths.get("file1")
+    val rootSubDir = Paths.get("dir")
+    val rootFile2  = Paths.get("dir/file2")
+
+    Files.write(root.resolve(rootFile1), contents)
+    Files.createDirectory(root.resolve(rootSubDir))
+    Files.write(root.resolve(rootFile2), contents)
+    Subtree(root, Seq(rootFile1, rootFile2), Seq(rootSubDir))
+  }
+
+  def expectSubtree(subtree: Subtree): Unit = {
+    assert(Files.exists(subtree.root))
+    subtree.elements.foreach(
+      elem => expectExist(subtree.root.resolve(elem))
+    )
+
+    val listStream = Files.list(subtree.root)
+    try listStream.count() should be(2)
+    finally listStream.close()
+  }
+
+  def expectExist(path: Path): Unit = {
+    assert(Files.exists(path), s"$path is expected to exist")
+  }
+
+  def expectNotExist(path: Path): Unit = {
+    assert(!Files.exists(path), s"$path is expected to not exist")
+  }
+}

FileManager/src/test/scala/org/enso/filemanager/WatchTests.scala

@@ -0,0 +1,223 @@
+package org.enso.filemanager
+
+import akka.actor.Scheduler
+import akka.actor.testkit.typed.scaladsl.ActorTestKit
+import akka.actor.testkit.typed.scaladsl.TestProbe
+import akka.actor.typed.ActorRef
+import akka.util.Timeout
+import io.methvin.watcher.DirectoryChangeEvent
+import java.nio.file.Files
+import java.nio.file.NotDirectoryException
+import java.nio.file.Path
+import java.util.UUID
+
+import org.apache.commons.io.FileUtils
+import org.enso.FileManager
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.FunSuite
+import org.scalatest.Matchers
+import org.scalatest.Outcome
+
+import scala.concurrent.Await
+import scala.concurrent.Future
+import scala.concurrent.duration._
+import scala.reflect.ClassTag
+import scala.util.Try
+
+// needs to be separate because watcher message are asynchronous
+class WatchTests
+    extends FunSuite
+    with BeforeAndAfterAll
+    with Matchers
+    with Helpers {
+  import FileManager.API._
+
+  var testKit: ActorTestKit         = ActorTestKit()
+  implicit val timeout: Timeout     = 3.seconds
+  implicit val scheduler: Scheduler = testKit.scheduler
+
+  var fileManager: ActorRef[InputMessage]   = _
+  var testProbe: TestProbe[FileSystemEvent] = _
+  var watcherID: UUID                       = _
+
+  override def withFixture(test: NoArgTest): Outcome = {
+    withTemporaryDirectory(_ => {
+      fileManager = testKit.spawn(FileManager(tempDir))
+      testProbe   = testKit.createTestProbe[FileSystemEvent]("file-observer")
+      watcherID   = observe(tempDir)
+
+      try super.withFixture(test)
+      finally if (watcherID != null)
+        // Otherwise directory would stay blocked on Windows.
+        unobserve(watcherID)
+    })
+  }
+
+  override def afterAll() {
+    testKit.shutdownTestKit()
+  }
+
+  def matchesEvent(
+    path: Path,
+    eventType: DirectoryChangeEvent.EventType
+  )(message: FileSystemEvent
+  ): Boolean = {
+    message.path == path && message.eventType == eventType
+  }
+
+  def expectEventFor(
+    eventType: DirectoryChangeEvent.EventType,
+    events: Seq[FileSystemEvent]
+  )(path: Path
+  ): Unit = {
+    assert(
+      events.exists(matchesEvent(path, eventType)),
+      s"not received message about $path"
+    )
+  }
+
+  def expectNextEvent(
+    path: Path,
+    eventType: DirectoryChangeEvent.EventType,
+    probe: TestProbe[FileSystemEvent] = testProbe
+  ): Unit = {
+    val message = probe.receiveMessage()
+    assert(
+      matchesEvent(path, eventType)(message),
+      s"expected of type $eventType for $path, got $message"
+    )
+  }
+
+  def ask[response <: Response.Success: ClassTag](
+    requestPayload: Request.Payload[response]
+  ): Future[Try[response]] = {
+    FileManager.ask(fileManager, requestPayload)
+  }
+
+  def observe(
+    path: Path,
+    replyTo: ActorRef[FileSystemEvent] = testProbe.ref
+  ): UUID = {
+    val futureResponse = ask(Watch.Create.Request(path, replyTo))
+    Await.result(futureResponse, timeout.duration).get.id
+  }
+
+  def unobserve(id: UUID): Watch.Remove.Response = {
+    val futureResponse = ask(Watch.Remove.Request(id))
+    Await.result(futureResponse, timeout.duration).get
+  }
+
+  test("Watcher: observe subtree creation and deletion") {
+    val subtree = createSubtree()
+    val events  = testProbe.receiveMessages(subtree.elements.size)
+    subtree.elements.foreach(
+      expectEventFor(DirectoryChangeEvent.EventType.CREATE, events)
+    )
+
+    FileUtils.deleteDirectory(subtree.root.toFile)
+
+    val deletionEvents = testProbe.receiveMessages(subtree.elements.size)
+    subtree.elements.foreach(
+      expectEventFor(
+        DirectoryChangeEvent.EventType.DELETE,
+        deletionEvents
+      )
+    )
+
+    testProbe.expectNoMessage(50.millis)
+  }
+
+  test("Watcher: observe file modification") {
+    val dir10 = tempDir.resolve("dir10")
+    Files.createDirectory(dir10)
+    expectNextEvent(dir10, DirectoryChangeEvent.EventType.CREATE)
+
+    val dir20 = dir10.resolve("dir20")
+    Files.createDirectories(dir20)
+    expectNextEvent(dir20, DirectoryChangeEvent.EventType.CREATE)
+
+    val someFile = dir20.resolve("file.dat")
+    Files.createFile(someFile)
+    expectNextEvent(someFile, DirectoryChangeEvent.EventType.CREATE)
+
+    // Need to wait a moment, as change soon after creation might be missed
+    // otherwise by some subpar watch implementations.
+    Thread.sleep(2000)
+    Files.write(someFile, "blahblah".getBytes)
+    expectNextEvent(someFile, DirectoryChangeEvent.EventType.MODIFY)
+
+    Files.delete(someFile)
+    expectNextEvent(someFile, DirectoryChangeEvent.EventType.DELETE)
+
+    FileUtils.deleteDirectory(dir20.toFile)
+    expectNextEvent(dir20, DirectoryChangeEvent.EventType.DELETE)
+    testProbe.expectNoMessage(50.millis)
+  }
+
+  test("Watcher: disabling watch") {
+    val subtree = createSubtree()
+    testProbe.receiveMessages(subtree.elements.size)
+    testProbe.expectNoMessage(50.millis)
+    val stopResponse = unobserve(watcherID)
+    watcherID = null
+    stopResponse should be(Watch.Remove.Response())
+
+    // Watch has been disabled, no further messages should come
+    FileUtils.deleteDirectory(subtree.root.toFile)
+    testProbe.expectNoMessage(50.millis)
+  }
+
+  test("Watcher: cannot watch ordinary file") {
+    val file = createSubFile()
+    assertThrows[NotDirectoryException]({ observe(file, testProbe.ref) })
+  }
+
+  test("Watcher: cannot watch symlink") {
+    val dir     = createSubDir()
+    val dirLink = tempDir.resolve("mylink")
+    Files.createSymbolicLink(dirLink, dir)
+    assertThrows[NotDirectoryException]({ observe(dirLink, testProbe.ref) })
+  }
+
+  test("Watcher: can watch under symlink") {
+    // The observed directory is not and does not contain symlink,
+    // however the path we observe it through contains symlink
+    val top       = createSubDir()
+    val linkToTop = top.resolve("link")
+    Files.createSymbolicLink(linkToTop, top)
+
+    val realSub = top.resolve("sub")
+    val linkSub = linkToTop.resolve("sub")
+
+    Files.createDirectory(realSub)
+
+    val symlinkEventProbe =
+      testKit.createTestProbe[FileSystemEvent]("observe-symlink-dir")
+
+    val id = observe(linkSub, symlinkEventProbe.ref)
+    try {
+      // Create file using "real" path.
+      val filename         = "testfile"
+      val realFilePath     = realSub.resolve(filename)
+      val observedFilePath = linkSub.resolve(filename)
+
+      val expectedOfType = (eventType: DirectoryChangeEvent.EventType) =>
+        expectNextEvent(
+          observedFilePath,
+          eventType,
+          symlinkEventProbe
+        )
+
+      Files.createFile(realFilePath)
+      expectedOfType(DirectoryChangeEvent.EventType.CREATE)
+
+      Files.write(realFilePath, contents)
+      expectedOfType(DirectoryChangeEvent.EventType.MODIFY)
+
+      Files.delete(realFilePath)
+      expectedOfType(DirectoryChangeEvent.EventType.DELETE)
+
+      symlinkEventProbe.expectNoMessage(50.millis)
+    } finally unobserve(id)
+  }
+}

build.sbt

@@ -23,7 +23,8 @@ lazy val enso = (project in file("."))
     syntax,
     pkg,
     interpreter,
-    projectManager
+    projectManager,
+    fileManager
   )
 
 // Sub-Projects
@@ -141,6 +142,9 @@ val akkaStream = "com.typesafe.akka" %% "akka-stream"          % "2.5.23"
 val akkaHttp         = "com.typesafe.akka" %% "akka-http"                % "10.1.8"
 val akkaSpray        = "com.typesafe.akka" %% "akka-http-spray-json"     % "10.1.8"
 val akkaTyped        = "com.typesafe.akka" %% "akka-actor-typed"         % "2.5.23"
+val akkaTestkit      = "com.typesafe.akka" %% "akka-testkit"             % "2.5.23"
+val akkaSLF4J        = "com.typesafe.akka" %% "akka-slf4j"               % "2.5.23"
+val akkaTestkitTyped = "com.typesafe.akka" %% "akka-actor-testkit-typed" % "2.5.23" % Test
 
 val akka = Seq(akkaActor, akkaStream, akkaHttp, akkaSpray, akkaTyped)
 
@@ -148,6 +152,21 @@ val circe = Seq("circe-core", "circe-generic", "circe-yaml").map(
   "io.circe" %% _ % "0.10.0"
 )
 
+lazy val fileManager = (project in file("FileManager"))
+  .settings(
+    (Compile / mainClass) := Some("org.enso.filemanager.FileManager")
+  )
+  .settings(
+    libraryDependencies ++= akka,
+    libraryDependencies += akkaSLF4J,
+    libraryDependencies += "ch.qos.logback" % "logback-classic" % "1.2.3",
+    libraryDependencies += "org.scalatest"  %% "scalatest"      % "3.2.0-SNAP10" % Test,
+    libraryDependencies += "org.scalacheck" %% "scalacheck"     % "1.14.0" % Test,
+    libraryDependencies += akkaTestkitTyped,
+    libraryDependencies += "commons-io" % "commons-io"        % "2.6",
+    libraryDependencies += "io.methvin" % "directory-watcher" % "0.9.6"
+  )
+
 lazy val projectManager = (project in file("project-manager"))
   .settings(
     (Compile / mainClass) := Some("org.enso.projectmanager.Server")

project/plugins.sbt

@@ -1 +1,4 @@
 addSbtPlugin("de.sciss" % "sbt-jflex" % "0.4.0")
+addSbtPlugin(
+  "com.thoughtworks.sbt-api-mappings" % "sbt-api-mappings" % "2.0.0"
+)