Mule : Reading a large file line by line

This page last changed on Apr 29, 2006 by mlusetti.

A solution for reading from a large file and generating one event per line.
Suggestion by Jay Lessik:

This example uses the following Mule components:

Additionally, it uses the following custom Java classes (Source code is at the end of the example):

FileSplitter
BufferedFileSplitter
FileMessageSplitter

Step 1: Configure The File Provider

Set the autoDelete property to false and specify the NoActionTransformeras the inbound transformer, it is likely you would also like to set the moveToDirectory property to any directory (other than the input directory) otherwise you've to take care of removing or moving the file yourself. This will cause the File Provider to pass a File object to the component (instead of a String containing the entire contents of the file).

<connector name="myFileConnector" className="org.mule.providers.file.FileConnector">
    <properties>
        <property name="autoDelete" value="false"/>
        <property name="moveToDirectory" value="file:///myOutputDir"/>
        <map name="serviceOverrides">
            <property name="inbound.transformer" value="org.mule.transformers.NoActionTransformer"/>
        </map>
    </properties>
</connector>

NOTE: The File Provider will not delete the file after processing so if you want it to be deleted you will have to handle that yourself.

Step 2: Configure The Endpoints

Configure a file endpoint specifying the File Provider configured in Step 1 and a VM endpoint for the resulting records.

<global-endpoints>
    <endpoint name="myFileEndpoint" address="file:///myInputDir?connector=myFileConnector"/>
</global-endpoints>

<endpoint-identifiers>
    <endpoint-identifier name="myOutputEndpoint" value="vm://myOutputEndpoint.queue"/>
</endpoint-identifiers>

NOTE: The outbound endpoint does not have to be a VM endpoint, it can be any type of endpoint.

Step 3: Configure A File Splitter Mule Component

In the mule configuration file, configure a file splitter component to split the file. The file splitter component will recieve the File message from the File endpoint configured in Step 2 and route it through the FileMessageSplitter outbound router (using the Mule BridgeComponent). The FileMessageSplitter outbound router will split the file into records and dispatch them to the outbound endpoint.

<mule-descriptor name="myBigFileSplitter" implementation="org.mule.components.simple.BridgeComponent">
    <inbound-router>
        <global-endpoint name="myFileEndpoint"/>
    </inbound-router>
    <outbound-router>
        <router className="foo.bar.FileMessageSplitter">
            <endpoint address="myOutputEndpoint" />
        </router>
    </outbound-router>
    <interceptor name="default"/>
</mule-descriptor>

Java Source Code

FileSplitter.java

public interface FileSplitter {
    public void setFile(File file) throws FileNotFoundException;

    public File getFile();

    public String getNextRecord() throws IOException;

    public void close();
}

BufferedFileSplitter.java

public class BufferedFileSplitter implements FileSplitter {
    private File file;
    private BufferedReader bufferedReader;

    public BufferedFileSplitter() {
    }

    /**
     * Constructor.
     * @param file
     * @throws FileNotFoundException If the file is null or cannot be found.
     */
    public BufferedFileSplitter(File file) throws FileNotFoundException {
        if (file == null) {
            throw new FileNotFoundException("The file cannot be null.");
        }
        this.file = file;
        this.bufferedReader = new BufferedReader(new FileReader(file));
    }

    /**
     * Returns the next record from the file.
     *
     * @return A String containing the contents of the next record, not including any line-termination characters,
     *         or null if the end of file has been reached.
     * @throws IOException If an I/O error occurs.
     */
    public String getNextRecord() throws IOException {
        if (file == null) {
            throw new IOException("The file is null. The file must be set before it can be read.");
        }
        if (bufferedReader == null) {
            throw new IOException("The buffered reader is null.");
        }
        return bufferedReader.readLine();
    }

    /**
     * Closes the BufferedFileSplitter.
     */
    public void close() {
        try {
            bufferedReader.close();
        } catch (IOException e) {
            log.error("Unable to close buffered reader.", e);
        }
    }

    /**
     * Gets the file being split.
     *
     * @return The File.
     */
    public File getFile() {
        return file;
    }

    /**
     * Sets the file to split.
     * @param file
     * @throws FileNotFoundException If the file is null or cannot be found.
     */
    public void setFile(File file) throws FileNotFoundException {
        if (file == null) {
            throw new FileNotFoundException("The File cannot be null.");
        }
        this.file = file;
        this.bufferedReader = new BufferedReader(new FileReader(file));
    }
}

FileMessageSplitter.java

public class FileMessageSplitter extends AbstractMessageSplitter {
    private Map properties;
    private FileSplitter fileSplitter;

    protected void initialise(UMOMessage message) {
        super.initialise(message);

        File payloadFile;

        // Check the message type
        Object payload = message.getPayload();
        if (payload instanceof File) {
            payloadFile = (File) payload;
        } else {
            throw new IllegalArgumentException("The payload for this router must be of type java.io.File");
        }

        fileSplitter = new BufferedFileSplitter(payloadFile);

        // Cache the properties here because for some message types getting the properties can be expensive
        properties = message.getProperties();
    }

    protected UMOMessage getMessagePart(UMOMessage message, UMOEndpoint endpoint) {
        String record = null;

        try {
            record = fileSplitter.getNextRecord();
        } catch (IOException e) {
            log.fatal("Fatal Error: " + e.getMessage(), e);
        }

        if (record != null) {
            return new MuleMessage(record, new HashMap(properties));
        } else {
            return null;
        }
    }
}