This document shows how to do following:

1. Get a csv from a local folder,
2. Unmarshal it to a Java pojo object array
3. Split the array to each object
4. Handle each pojo object by a customized code

Following is the pojo definition with camel csv format annotations

package com.zzyan.domain;

import org.apache.camel.dataformat.bindy.annotation.CsvRecord;
import org.apache.camel.dataformat.bindy.annotation.DataField;

@CsvRecord(separator = ",", skipFirstLine = true)
public class Item {

    @DataField(pos=1)
    private String transactionType;

    @DataField(pos=2)
    private String sku;

    @DataField(pos=3)
    private String itemDescrition;

    @DataField(pos=4)
    private String price;

/*
    here ignores all the getter and setters
*/

    @Override
    public String toString() {
        return "Item{" +
                "transactionType='" + transactionType + '\'' +
                ", sku='" + sku + '\'' +
                ", itemDescrition='" + itemDescrition + '\'' +
                ", price='" + price + '\'' +
                '}';
    }
}

We also define a customized processor, the processor will take in 1 Item object, and do some handling. Here we do nothing but just print it out.

package com.zzyan.processor;

import com.zzyan.domain.Item;
import lombok.extern.slf4j.Slf4j;
import org.apache.camel.Exchange;
import org.springframework.stereotype.Component;

@Component
@Slf4j
public class ItemProcessor implements org.apache.camel.Processor {
    @Override
    public void process(Exchange exchange) throws Exception {
        Item item = (Item)exchange.getIn().getBody();
        System.out.println(item.getItemDescrition());
    }
}

Finally we define the Camel Route to link these together.

package com.zzyan.route;

import com.zzyan.domain.Item;
import com.zzyan.processor.ItemProcessor;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.dataformat.bindy.csv.BindyCsvDataFormat;
import org.apache.camel.spi.DataFormat;
import org.springframework.stereotype.Component;

//@Component
public class UnmarshalCsvRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {

        DataFormat bindy = new BindyCsvDataFormat(Item.class);

        from("timer:hello?period=10s")
                .log("Timer Invoked and the body is ${body}")
                .pollEnrich("file:c:/data/input?noop=true&readLock=none")
                .log("body: ${body}")
                .unmarshal(bindy)
                .log("The unmarshaled object is ${body}")
                .split(body())
                  .log("new Item ${body}")
                  .process(new ItemProcessor())
                .end();
    }
}

Put the csv file with the same header to c:/data/input folder, and run the spring boot project, the Route will pickup the file and do the transformation.

In real time integration, especially file based integration, we normally starts from fetching file from a sftp server.

It’s common that the files put on the sftp server is a zipped file with multiple files compressed together.

Following shows how to use a Camel route to fetch a zip file from the sftp server, and after that unzip the file and put to local folders.

Please refer to Spring boot + Camel hello world to see how to setup the project before creating the Route.

The code contains both getting file from a sftp, and fetch from a local folder(commented). It also includes a Choice to do the file filtering and switch to different sub-Routes if required.

package com.zzyan.route;

import org.apache.camel.Predicate;
import org.apache.camel.builder.PredicateBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.model.dataformat.ZipFileDataFormat;
import org.springframework.stereotype.Component;

import java.util.Iterator;

//@Component
public class UnzipRoute extends RouteBuilder {
    @Override
    public void configure() throws Exception {

        // Define a Zip File format, since our Zip file have multiple zip files
        ZipFileDataFormat zipFile = new ZipFileDataFormat();
        zipFile.setUsingIterator(true);

        from("timer:hello?period=10s") // Triggered by Timer every 10 seconds
                .log("Timer Invoked and the body is ${body}")
                .pollEnrich("sftp://USERNAME@HOST/FILEPATH?password=PASSWORD&noop=true")
				// following shows pull the file from local c:/data/zipInput
                //.pollEnrich("file:c:/data/zipInput?noop=true&readLock=none")
                .log("${file:name}")
                .unmarshal(zipFile)
                .split(bodyAs(Iterator.class)).streaming()
                  .log("${file:name}")
                  .choice()
                    .when(header("CamelFileName").startsWith("Headers"))
                      .log("This is the Header")
                      .to("file:c:/data/output")
                    .otherwise()
                      .to("file:c:/data/output")
                  .endChoice()
                .end();
    }
}