Using DirectoryChangesSource recursively

Akka Akka Streams & Alpakka
1

We are trying to use the the DirectoryChangesSource to monitor directory changes but having some issue with it where we found no solutions yet in the documentation.

The directory structure:

  • dirA
    ** dir1
  • dirB
    ** dir2

The use case we are trying to solve is:

  1. Track all changes in all directories including the nested ones
  2. A full directory with files is dropped in the nested dirs where we also need to start tracking changes + capture the files in the nested directories as changes

The first can be fixed by using a Files.walk, filter on directories and apply the DirectoryChangesSource on each. Not sure yet what this means for resources / system load.

  Files
    .walk(Paths.get(applicationSettings.sourceSettings.sourcePath))
    .filter(path => Files.isDirectory(path))
    .toScala[Stream]
    .map(path => {
      DirectoryChangesSource(path, pollInterval = 1 second, maxBufferSize = 1000)
    })
    .reduce(_ concat _)

Does anybody have an idea on how to tackle the 2nd issues?