File Inbound Channel unable to read file

Question

I use Spring Integration to process CSV files manually added to a folder.

The beginning of my integration is :

<bean id="recursiveScanner" class="org.springframework.integration.file.RecursiveLeafOnlyDirectoryScanner" />
<int-file:inbound-channel-adapter channel="filesIn" directory="file:${integration.directory}" scanner="recursiveScanner">
    <int:poller id="poller" fixed-delay="${integration.delay}" />
</int-file:inbound-channel-adapter>

Followed by custom service activators.

When I move a small file (few megs) in the scanned directory it works perfectly, but when it's a big file (93 Mio) I get an exception :

org.springframework.batch.item.ItemStreamException: Failed to initialize the reader
..........
Caused by: java.io.FileNotFoundException: D:\nreco-import-batch-2014\.\home\entries.csv (Le processus ne peut pas accéder au fichier car ce fichier est utilisé par un autre processus)

It says the file can't be read because already openned by another process.

My guess is that the Spring channel is triggered at the begining of the file transfer, when the file is still being copied by the operating system, thus is not yet readable.

So my question is: is there a way to configure the file inbound channel to be triggered only when the file copy is complete ?

For now I test in local, with very good transfer rates between by two HDD, but the app will be deployed on a server where the scanned directory will be accessed by FTP, I am sure it will fail for much smaller files then...

Thanks

---

edit: forgot to explain one part

one of my service activators (the final one) is a Spring Batch job which parses the newly added file, process the data and insert them in a MongoDB database

it's the Batch step actually reading the file which throws the Exception

Answer 1

So I ended up using the method given in http://blog.on-x.com/2010/11/spring-integration-implementation-d%E2%80%99un-scenario-35 but with only one custom filter which inherits AcceptOnceFileListFilter, for some reason I can't get the CompositeFileListFilter working...

public class CustomFileListFilter extends AcceptOnceFileListFilter<File> {

    @Override
    public boolean accept(File file) {

        try {
            FileInputStream stream = new FileInputStream(file);
            stream.close();
        }
        catch (IOException e) {
            log.error("File not readable");
            return false;
        }

        return super.accept(file);
    }

}

I'm still interested in other solutions

Answer 2

I remember working on an app similar to this... we had 2 directories...

\in\ was used for the file transfers... every x minutes, it would check if the file was there. Once the file was present, it would rename it and move it in the \process\ directory.

Were were using scp under aix and it did the job processing very large files.

Answer 3

i also came with this scenario. any how i used the same logic above to make my code work

    <?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:integration="http://www.springframework.org/schema/integration"
    xmlns:file="http://www.springframework.org/schema/integration/file"
    xsi:schemaLocation="http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans.xsd
            http://www.springframework.org/schema/integration
            http://www.springframework.org/schema/integration/spring-integration.xsd
            http://www.springframework.org/schema/integration/file
            http://www.springframework.org/schema/integration/file/spring-integration-file.xsd">

    <bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>

    <file:inbound-channel-adapter id="filesIn"
                                 directory="D:\SpringPoller\Input" prevent-duplicates="false">
        <integration:poller id="poller" fixed-delay="5000"/>
    </file:inbound-channel-adapter>

    <integration:service-activator input-channel="filesIn"
                                   output-channel="filesOut"
                                   ref="handler"/>

    <file:outbound-channel-adapter id="filesOut"
                                   directory="D:\SpringPoller\Output"
                                   delete-source-files="true"/>

    <bean id="handler" class="dummy.Handler"/>

</beans>



    package dummy;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;


public class Handler {



    public File handleFile(File input) {
        System.out.println("Copying file: " + input.getAbsolutePath() + "with length :" + input.length());

        // check the file is fully available (full file was copied and no part files processed ...
        FileInputStream stream = null;
        try {
            stream = new FileInputStream(input);

            System.out.println("***********  Stream available now");

        }
        catch (IOException e) {
            System.out.println("##########  Stream not available now");
        }
        finally { 
            try {

                stream.close();
            }catch(Exception e) {

            }
        }


        return input;
    }



}