How to get Streams of File Attributes from the FileSystem?

If there's some way of directly getting those attributes, I don't know.

On converting the FileVisitor API to reactive streams:

The mechanism of reactive streams backpressure is a pull-push model, where a demand is first signaled by downstream (the pull part) and then the upstream is allowed to send no more items than the demand signaled (the push part).

The problem with the FileVisitor API is that there's no way to directly hookup with such control flow mechanism. Once you set it off it just goes calling your callback and not caring too much about anything else.

There's no clean way of bridging this, but one way you could do that is using the Source.queue (https://doc.akka.io/docs/akka/current/stream/operators/Source/queue.html) to isolate that API from the rest of your stream like that:

val queue = 
Source.queue[BasiFileAttributes](bufferSize, OverflowStrategy.backpressure)
      //the rest of your Akka Streams pipeline
      .run(system);

This will materialize to queue that you can now pass to your FileVisitor. Anything you offer to that queue will go down the stream. If when you offer there's no demand and the queue is full the Future returned by offer will not complete until such time that there's space in the queue. So in the API you could simply do:

//inside the FileVisitor API callback
Await.result(queue.offer(attrs), Duration.Inf)

And that would block the callback thread when the stream is backpressured. Ugly but isolated.

You can access file attributes with their path by using Files.find which accepts a BiPredicate<Path, BasicFileAttributes> and store the value as it tests each path.

The side effect action inside the BiPredicate will enable operations on both objects without needing to touch the file system per item in the path. With your predicate condition yourPred, side effect predicate below will collect the attributes for you to retrieve inside the stream processing:

public static void main(String[] args) throws IOException {
    Path dir = Path.of(args[0]);

    // Use `ConcurrentHashMap` if using `stream.parallel()`
    HashMap <Path,BasicFileAttributes> attrs = new HashMap<>();

    BiPredicate<Path, BasicFileAttributes> yourPred = (p,a) -> true;

    BiPredicate<Path, BasicFileAttributes> predicate = (p,a) -> {
        return yourPred.test(p, a)
                // && p.getNameCount() == dir.getNameCount()+1 // Simulates Files.list
                && attrs.put(p, a) == null;
    };
    try(var stream = Files.find(dir, Integer.MAX_VALUE, predicate)) {
        stream.forEach(p-> System.out.println(p.toString()+" => "+attrs.get(p)));
        // Or: if your put all your handling code in the predicate use stream.count();
    }
}

To similate the effect of File.list use a one level find scanner:

 BiPredicate<Path, BasicFileAttributes> yourPred = (p,a) -> p.getNameCount() == dir.getNameCount()+1;

For a large folder scan you should clean up the attrs map as you go by inserting attrs.remove(p); after consuming the path.

Edit

The answer above can be refactored to a 3 line call returning stream of Map.Entry<Path, BasicFileAttributes>, or it's easy to add a class/record to hold the Path/BasicFileAttribute pair and return Stream<PathInfo> instead:

/**
 * Call Files.find() returning a stream with both Path+BasicFileAttributes
 * as type Map.Entry<Path, BasicFileAttributes>
 * <p>Could declare a specific record to replace Map.Entry as:
 *    record PathInfo(Path path, BasicFileAttributes attr) { };
 */
public static Stream<Map.Entry<Path, BasicFileAttributes>>
find(Path dir, int maxDepth, BiPredicate<Path, BasicFileAttributes> matcher, FileVisitOption... options) throws IOException {

    HashMap <Path,BasicFileAttributes> attrs = new HashMap<>();
    BiPredicate<Path, BasicFileAttributes> predicate = (p,a) -> (matcher == null || matcher.test(p, a)) && attrs.put(p, a) == null;

    return Files.find(dir, maxDepth, predicate, options).map(p -> Map.entry(p, attrs.remove(p)));
}

Starting from DuncG's answer I got the following to work in Scala3 as a pretty generic Akka Stream class. This is actually very neat as it creates a sideffect of the Files.find function which it immediately encapsulates back into a clean functional reactive stream.

class DirectoryList(
    dir: Path, 
    matcher: (Path, BasicFileAttributes) => Boolean = (p,a) => true, 
    maxDepth: Int = 1
) extends GraphStage[SourceShape[(Path,BasicFileAttributes)]]:
    import scala.jdk.FunctionConverters.*
    import scala.jdk.OptionConverters.*
    
    val out: Outlet[(Path,BasicFileAttributes)] = Outlet("PathAttributeSource")
    override val shape = SourceShape(out)


    override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
        new GraphStageLogic(shape) {
            private var next: (Path,BasicFileAttributes) = _

            def append(path: Path, att: BasicFileAttributes): Boolean = 
                val matched = matcher(path,att)
                if matched then next = (path,att)
                matched
            
            private val pathStream = Files.find(dir, maxDepth, append.asJava)
            private val sit = pathStream.iterator()
            
            setHandler(out, new OutHandler {
                override def onPull(): Unit = { 
                    if sit.hasNext then
                        sit.next()
                        push(out,next)
                    else
                        pathStream.close()  
                        complete(out)
                }

                override def onDownstreamFinish(cause: Throwable): Unit =
                    pathStream.close()  
                    super.onDownstreamFinish(cause)
            })
        }
end DirectoryList

This can then be used as follows:

val sourceGraph = DirectoryList(Path.of("."),depth=10)
val result = Source.fromGraph(sourceGraph).map{ (p: Path,att: BasicFileAttributes) => 
        println(s"received <$p> : dir=${att.isDirectory}")}.run()

The full source code is here on github and an initial test here. Perhaps one could improve it by tuning the answers so that a certain number of path attribute pairs are passed along in bulk.