fix: return and copy only max files

Author: fdelbrayelle

docker-compose-ci.yml

@@ -11,7 +11,7 @@ services:
       - "6622:22"
     command: "foo:pass*+=:1001:1001:upload"
     volumes:
-      - ${PWD}/src/test/resources/ssh/id_rsa.pub:/home/foo/.ssh/authorized_keys
+      - ${PWD}/src/test/resources/ssh/id_ed25519.pub:/home/foo/.ssh/authorized_keys
 
   # https://github.com/fauria/docker-vsftpd/issues/41
   ftp:

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

@@ -98,6 +98,18 @@ public class Trigger extends AbstractTrigger implements PollingTriggerInterface,
     )
     private Property<Integer> maxFiles = Property.ofValue(25);
 
+    private static class PendingFile {
+        private final Path path;
+        private final Entry candidate;
+        private final ChangeType changeType;
+
+        private PendingFile(Path path, Entry candidate, ChangeType changeType) {
+            this.path = path;
+            this.candidate = candidate;
+            this.changeType = changeType;
+        }
+    }
+
     @Override
     public Duration getInterval() {
         return this.interval;
@@ -117,7 +129,7 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
         Optional<Duration> rStateTtl = runContext.render(stateTtl).as(Duration.class);
 
         Map<String, StatefulTriggerService.Entry> state = readState(runContext, rStateKey, rStateTtl);
-        List<TriggeredFile> toFire = new ArrayList<>();
+        List<PendingFile> pendingFiles = new ArrayList<>();
 
         logger.debug("Evaluating trigger for path: {}", fromPath);
 
@@ -127,49 +139,57 @@ public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerCo
                 .filter(path -> rRegExp == null || path.toString().matches(rRegExp))
                 .toList();
 
-            for(Path path: paths) {
-                 try {
-                     BasicFileAttributes attrs = Files.readAttributes(path, BasicFileAttributes.class);
-                     var modifiedAt = attrs.lastModifiedTime().toInstant();
-                     var key = path.toUri().toString();
-                     var version = String.format("%d_%d", modifiedAt.toEpochMilli(), attrs.size());
-
-                     var candidate = StatefulTriggerService.Entry.candidate(key, version, modifiedAt);
-                     var change = computeAndUpdateState(state, candidate, rOn);
-
-                     if (change.fire()) {
-                         var changeType = change.isNew() ? ChangeType.CREATE : ChangeType.UPDATE;
-                         
-                         io.kestra.plugin.fs.nfs.List.File fileModel = mapToFile(path);
-                         toFire.add(TriggeredFile.builder()
-                             .file(fileModel)
-                             .changeType(changeType)
-                             .build());
-                         logger.info("File {} detected with change type: {}", path, changeType);
-                     }
-                 } catch (IOException e) {
-                     logger.warn("Error processing path {}: {}", path, e.getMessage(), e);
-                 }
+            for (Path path : paths) {
+                try {
+                    BasicFileAttributes attrs = Files.readAttributes(path, BasicFileAttributes.class);
+                    var modifiedAt = attrs.lastModifiedTime().toInstant();
+                    var key = path.toUri().toString();
+                    var version = String.format("%d_%d", modifiedAt.toEpochMilli(), attrs.size());
+                    var candidate = StatefulTriggerService.Entry.candidate(key, version, modifiedAt);
+                    var prev = state.get(key);
+
+                    if (!shouldFire(prev, version, rOn)) {
+                        computeAndUpdateState(state, candidate, rOn);
+                        continue;
+                    }
+
+                    var changeType = prev == null ? ChangeType.CREATE : ChangeType.UPDATE;
+                    pendingFiles.add(new PendingFile(path, candidate, changeType));
+                } catch (IOException e) {
+                    logger.warn("Error processing path {}: {}", path, e.getMessage(), e);
+                }
             }
         } catch (IOException e) {
             logger.error("Error walking/listing path {}: {}", fromPath, e.getMessage(), e);
             return Optional.empty();
         }
 
-        writeState(runContext, rStateKey, state, rStateTtl);
-
-        if (toFire.isEmpty()) {
-            logger.debug("No new or updated files found.");
-            return Optional.empty();
+        int rMaxFiles = runContext.render(this.maxFiles).as(Integer.class).orElse(25);
+        if (pendingFiles.size() > rMaxFiles) {
+            logger.warn("Too many files to process ({}), limiting to {}", pendingFiles.size(), rMaxFiles);
+            pendingFiles = pendingFiles.subList(0, Math.min(rMaxFiles, pendingFiles.size()));
         }
 
-        int rMaxFiles = runContext.render(this.maxFiles).as(Integer.class).orElse(25);
-        if (toFire.size() > rMaxFiles) {
-            logger.warn("Too many files to process ({}), limiting to {}", toFire.size(), rMaxFiles);
-            toFire = toFire.subList(0, Math.min(rMaxFiles, toFire.size()));
+        List<TriggeredFile> toFire = new ArrayList<>();
+
+        for (PendingFile pending : pendingFiles) {
+            try {
+                io.kestra.plugin.fs.nfs.List.File fileModel = mapToFile(pending.path);
+                computeAndUpdateState(state, pending.candidate, rOn);
+                toFire.add(TriggeredFile.builder()
+                    .file(fileModel)
+                    .changeType(pending.changeType)
+                    .build());
+                logger.info("File {} detected with change type: {}", pending.path, pending.changeType);
+            } catch (IOException e) {
+                logger.warn("Error processing path {}: {}", pending.path, e.getMessage(), e);
+            }
         }
 
+        writeState(runContext, rStateKey, state, rStateTtl);
+
         if (toFire.isEmpty()) {
+            logger.debug("No new or updated files found.");
             return Optional.empty();
         }