feat: add recursive listing option

closes #70

src/main/java/io/kestra/plugin/fs/vfs/Downloads.java

@@ -11,9 +11,9 @@
 import org.apache.commons.vfs2.impl.StandardFileSystemManager;
 import org.slf4j.Logger;
 
+import javax.validation.constraints.NotNull;
 import java.net.URI;
 import java.util.stream.Collectors;
-import javax.validation.constraints.NotNull;
 
 import static io.kestra.core.utils.Rethrow.throwFunction;
 
@@ -49,6 +49,12 @@ public abstract class Downloads extends AbstractVfsTask implements RunnableTask<
     @PluginProperty(dynamic = true)
     private String regExp;
 
+    @Schema(
+        title = "List file recursively"
+    )
+    @Builder.Default
+    private boolean recursive = false;
+
     public Output run(RunContext runContext) throws Exception {
         Logger logger = runContext.logger();
 
@@ -66,7 +72,8 @@ public Output run(RunContext runContext) throws Exception {
                 fsm,
                 fileSystemOptions,
                 this.uri(runContext, this.from),
-                this.regExp
+                this.regExp,
+                recursive
             );
 
             java.util.List<io.kestra.plugin.fs.vfs.models.File> files = run

src/main/java/io/kestra/plugin/fs/vfs/List.java

@@ -30,6 +30,12 @@ public abstract class List extends AbstractVfsTask implements RunnableTask<List.
     @PluginProperty(dynamic = true)
     private String regExp;
 
+    @Schema(
+        title = "List file recursively"
+    )
+    @Builder.Default
+    private boolean recursive = false;
+
     public Output run(RunContext runContext) throws Exception {
         try (StandardFileSystemManager fsm = new StandardFileSystemManager()) {
             fsm.init();
@@ -39,7 +45,8 @@ public Output run(RunContext runContext) throws Exception {
                 fsm,
                 this.fsOptions(runContext),
                 this.uri(runContext, this.from),
-                this.regExp
+                this.regExp,
+                this.recursive
             );
         }
     }

src/main/java/io/kestra/plugin/fs/vfs/Trigger.java

@@ -1,8 +1,6 @@
 package io.kestra.plugin.fs.vfs;
 
 import io.kestra.core.exceptions.IllegalVariableEvaluationException;
-import io.kestra.core.models.annotations.Example;
-import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.annotations.PluginProperty;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
@@ -13,8 +11,6 @@
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.models.triggers.TriggerOutput;
 import io.kestra.core.runners.RunContext;
-import io.kestra.core.utils.IdUtils;
-import io.kestra.plugin.fs.sftp.SftpInterface;
 import io.kestra.plugin.fs.vfs.models.File;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.*;
@@ -25,12 +21,12 @@
 import org.apache.commons.vfs2.impl.StandardFileSystemManager;
 import org.slf4j.Logger;
 
+import javax.validation.constraints.NotNull;
 import java.io.IOException;
 import java.net.URI;
 import java.time.Duration;
 import java.util.Optional;
 import java.util.stream.Collectors;
-import javax.validation.constraints.NotNull;
 
 import static io.kestra.core.utils.Rethrow.throwFunction;
 
@@ -76,8 +72,16 @@ public abstract class Trigger extends AbstractTrigger implements PollingTriggerI
     @PluginProperty(dynamic = true)
     private String regExp;
 
+    @Schema(
+        title = "List file recursively"
+    )
+    @Builder.Default
+    private boolean recursive = false;
+
     abstract public String getPort();
+
     abstract protected FileSystemOptions fsOptions(RunContext runContext) throws IllegalVariableEvaluationException, IOException;
+
     abstract protected String scheme();
 
     @Override
@@ -107,7 +111,8 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
                     fsm,
                     fileSystemOptions,
                     from,
-                    this.regExp
+                    this.regExp,
+                    this.recursive
                 );
             } catch (FileNotFolderException fileNotFolderException) {
                 logger.debug("From path doesn't exist '{}'", String.join(", ", fileNotFolderException.getInfo()));

src/main/java/io/kestra/plugin/fs/vfs/VfsService.java

@@ -6,9 +6,7 @@
 import org.apache.commons.io.FilenameUtils;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.StringUtils;
-import org.apache.commons.vfs2.FileObject;
-import org.apache.commons.vfs2.FileSystemOptions;
-import org.apache.commons.vfs2.Selectors;
+import org.apache.commons.vfs2.*;
 import org.apache.commons.vfs2.impl.StandardFileSystemManager;
 import org.apache.commons.vfs2.provider.AbstractFileObject;
 
@@ -17,6 +15,7 @@
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.NoSuchElementException;
+import java.util.Objects;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -88,10 +87,30 @@ public static List.Output list(
         StandardFileSystemManager fsm,
         FileSystemOptions fileSystemOptions,
         URI from,
-        String regExp
+        String regExp,
+        boolean recursive
     ) throws Exception {
         try (FileObject local = fsm.resolveFile(from.toString(), fileSystemOptions)) {
-            FileObject[] children = local.getChildren();
+            FileObject[] children = local.findFiles(new FileSelector() {
+                @Override
+                public boolean traverseDescendents(FileSelectInfo file) {
+                    // if not recursive only traverse "from"
+                    return recursive || Objects.equals(file.getFile().getName().getPath(), local.getName().getPath());
+                }
+
+                @Override
+                public boolean includeFile(FileSelectInfo file) throws Exception {
+                    // Do not include directories in the result and apply user's filter
+                    return file.getFile().isFile()
+                        && (regExp == null || file.getFile().getName().getPath().matches(regExp));
+                }
+            });
+
+            if (children == null) {
+                return List.Output.builder()
+                    .files(java.util.List.of())
+                    .build();
+            }
 
             java.util.List<File> list = Stream.of(children)
                 .map(throwFunction(r -> File.of((AbstractFileObject<?>) r)))

src/test/java/io/kestra/plugin/fs/ftp/ListTest.java

@@ -26,6 +26,7 @@ void all() throws Exception {
         for (int i = 0; i < 6; i++) {
             lastFile = IdUtils.create();
             ftpUtils.upload("upload" + dir + "/" + lastFile + ".yaml");
+            ftpUtils.upload("upload" + dir + "/subfolder/" + lastFile + ".yaml");
         }
         ftpUtils.upload("upload" + dir + "/file with space.yaml");
 
@@ -52,5 +53,33 @@ void all() throws Exception {
         run = task.run(TestsUtils.mockRunContext(runContextFactory, task, ImmutableMap.of()));
 
         assertThat(run.getFiles().size(), is(1));
+
+        task = List.builder()
+            .id(ListTest.class.getSimpleName())
+            .type(ListTest.class.getName())
+            .from("/upload" + dir)
+            .host("localhost")
+            .port("6621")
+            .username("guest")
+            .password("guest")
+            .recursive(true).build();
+
+        run = task.run(TestsUtils.mockRunContext(runContextFactory, task, ImmutableMap.of()));
+
+        assertThat(run.getFiles().size(), is(13));
+
+        task = List.builder()
+            .id(ListTest.class.getSimpleName())
+            .type(ListTest.class.getName())
+            .from("/" + dir)
+            .host("localhost")
+            .port("6621")
+            .username("guest")
+            .password("guest")
+            .recursive(true).build();
+
+        run = task.run(TestsUtils.mockRunContext(runContextFactory, task, ImmutableMap.of()));
+
+        assertThat(run.getFiles().size(), is(0));
     }
 }