Copying multiple files to watchedFolder causes app to grab zero byte files

[seakrebel]

Copying multiple files to watchedFolder causes app to grab zero byte files

[kkdlau]

Hi @Frooodle, i think this is a good issue that can assign someone like me who wish to contribute to open source 😃

My initial idea of solving this issue is to update collectFilesForProcessing to ensure we only collects files that are fully copied, either by checking if the size is growing, or using some os level features (e.g. lsof)

let me if there is comment for the solution 😆

[Frooodle]

A good callout and bug
I will work on this over weekend

[Frooodle]

@kkdlau hows this going?

[seakrebel]

Here is an example. Haven't tested it. Also not an java expert. But could be leading into right direction.

PipelineDirectoryProcessor.java

// [...]
import java.util.concurrent.TimeUnit;
// [...]
public class PipelineDirectoryProcessor {
    // [...]

    private static final long STABILITY_CHECK_DELAY = 1000; // 1 second
    private static final long STABILITY_CHECK_COUNT = 5; // Check 5 times

    private File[] collectFilesForProcessing(Path dir, Path jsonFile, PipelineOperation operation) throws IOException {
        try (Stream<Path> paths = Files.list(dir)) {
            if ("automated".equals(operation.getParameters().get("fileInput"))) {
                return paths.filter(path -> !Files.isDirectory(path) && !path.equals(jsonFile) && isFileStable(path))
                            .map(Path::toFile)
                            .toArray(File[]::new);
            } else {
                String fileInput = (String) operation.getParameters().get("fileInput");
                return new File[] { new File(fileInput) };
            }
        }
    }

    private boolean isFileStable(Path path) throws IOException {
        long initialSize = Files.size(path);
        for (int i = 0; i < STABILITY_CHECK_COUNT; i++) {
            try {
                TimeUnit.MILLISECONDS.sleep(STABILITY_CHECK_DELAY);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new IOException("Thread interrupted during stability check", e);
            }
            long newSize = Files.size(path);
            if (initialSize != newSize) {
                return false;
            }
        }
        return initialSize > 0; // Also ensuring the file is not zero bytes
    }
    // [...]
}
// [...]

[kkdlau]

@kkdlau hows this going?

Hi, was busy with my full-time work 😞
But I already have the draft of the PR
Just need to go through couple of regression testing to ensure it doesn't break the existing features

Will create a PR tnt (APAC time)👍🏻

[kkdlau]

Here is an example. Haven't tested it. Also not an java expert. But could be leading into right direction.

PipelineDirectoryProcessor.java

// [...]

import java.util.concurrent.TimeUnit;

// [...]

public class PipelineDirectoryProcessor {

    // [...]



    private static final long STABILITY_CHECK_DELAY = 1000; // 1 second

    private static final long STABILITY_CHECK_COUNT = 5; // Check 5 times



    private File[] collectFilesForProcessing(Path dir, Path jsonFile, PipelineOperation operation) throws IOException {

        try (Stream<Path> paths = Files.list(dir)) {

            if ("automated".equals(operation.getParameters().get("fileInput"))) {

                return paths.filter(path -> !Files.isDirectory(path) && !path.equals(jsonFile) && isFileStable(path))

                            .map(Path::toFile)

                            .toArray(File[]::new);

            } else {

                String fileInput = (String) operation.getParameters().get("fileInput");

                return new File[] { new File(fileInput) };

            }

        }

    }



    private boolean isFileStable(Path path) throws IOException {

        long initialSize = Files.size(path);

        for (int i = 0; i < STABILITY_CHECK_COUNT; i++) {

            try {

                TimeUnit.MILLISECONDS.sleep(STABILITY_CHECK_DELAY);

            } catch (InterruptedException e) {

                Thread.currentThread().interrupt();

                throw new IOException("Thread interrupted during stability check", e);

            }

            long newSize = Files.size(path);

            if (initialSize != newSize) {

                return false;

            }

        }

        return initialSize > 0; // Also ensuring the file is not zero bytes

    }

    // [...]

}

// [...]

Thanks for the idea 👍🏻
My draft is quite similar except for isFileStable implementation
will share more details when I open the PR