Converting CSV file to Java Objects (POJO) and send it to ActiveMQ queue

Question

My aim is to read the csv file, convert it to Java Objects (POJO) and send the Java Objects one by one to ActiveMQ queue. Below is the code:

public void configure() throws Exception {
    from("file:src/main/resources?fileName=data.csv")               
    .unmarshal(bindy)
    .split(body())
    .to("file:src/main/resources/?fileName=equityfeeds.txt")
    .split().tokenize(",").streaming().to("jms:queue:javaobjects.upstream.queue");          
}

Issues: 1.When I execute the code no file(equityfeeds.txt) gets created and no objects goes to the queue. What's wrong? I don't need to do any processing right now. I just need to unmarshal the csv to POJOs and send the Java Objects one by one to the ActiveMQ queue.

EquityFeeds (POJO)

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

    @DataField(pos = 1) 
    private String externalTransactionId;

    @DataField(pos = 2)
    private String clientId;

    @DataField(pos = 3)
    private String securityId;

    @DataField(pos = 4)
    private String transactionType;

    @DataField(pos = 5, pattern = "dd/MM/YY")
    private Date transactionDate;

    @DataField(pos = 6)
    private float marketValue; 

    @DataField(pos = 7)
    private String priorityFlag;

Please kindly help. Please tell me where I am going wrong.

@pvpkiran:Below is my Camel Code for producer:

public void configure() throws Exception {
            from("file:src/main/resources?fileName=data.csv")               
                .unmarshal(bindy)
                .split(body())
                .streaming().to("jms:queue:javaobjects.upstream.queue");
}

Below is my Consumer Code (Using JMS API):

@JmsListener(destination = "javaobjects.upstream.queue")
public void javaObjectsListener(final Message objectMessage) throws JMSException {
        Object messageData = null;
        if(objectMessage instanceof ObjectMessage) {
            ObjectMessage objMessage = (ObjectMessage) objectMessage;
            messageData = objMessage.getObject();
        }
        System.out.println("Object: "+messageData.toString());
    }

I am not using Camel for consuming the JMSMessage. In the consumer I am using JMS API for consuming the message. Also I am not testing the code. The messages have come in ActiveMQ and I am using JMS API (as above) to consume the message. In the terminal in am getting NullPointerException. Also 2 message have gone into ActiveMQ.DLQ giving the below Error Message:

java.lang.Throwable: Delivery[7] exceeds redelivery policy limit:RedeliveryPolicy {destination = null, collisionAvoidanceFactor = 0.15, maximumRedeliveries = 6, maximumRedeliveryDelay = -1, initialRedeliveryDelay = 1000, useCollisionAvoidance = false, useExponentialBackOff = false, backOffMultiplier = 5.0, redeliveryDelay = 1000, preDispatchCheck = true}, cause:null

what is bindy in this line .unmarshal(bindy) also can u give a sample csv — pvpkiran
final BindyCsvDataFormat bindy=new BindyCsvDataFormat(camelproject.EquityFeeds.class); sample csv **externalTransactionId,clientId,securityId,transactionType,transactionDate,marketValue,priorityFlag SAPEXTXN1,GS,ICICI,BUY,23/11/13,101.9,Y SAPEXTXN2,AS,REL,SELL,20/11/13,121.9,N ** — sidd

pvpkiran pvpkiran · Accepted Answer · 2020-03-26T10:17:28

Try this. This should work

from("file:src/main/resources?fileName=equityfeeds.csv")
                    .unmarshal(new BindyCsvDataFormat(EquityFeeds.class))
                    .split(body())
                    .streaming().to("jms:queue:javaobjects.upstream.queue");
// This route is for Testing
from("jms:queue:javaobjects.upstream.queue").to("bean:camelBeanComponent?method=processRoute");

And write a consumer component bean

@Component
public class CamelBeanComponent {
    public void processRoute(Exchange exchange) {
        System.out.println(exchange.getIn().getBody());
    }
}

This printed(You need to add toString() if you need output like this)

EquityFeeds(externalTransactionId=SAPEXTXN1, clientId=GS, securityId=ICICI, transactionType=BUY, transactionDate=Sun Dec 30 00:00:00 CET 2012, marketValue=101.9, priorityFlag=Y)
EquityFeeds(externalTransactionId=SAPEXTXN2, clientId=AS, securityId=REL, transactionType=SELL, transactionDate=Sun Dec 30 00:00:00 CET 2012, marketValue=121.9, priorityFlag=N)

If you use .split().tokenize(",") then, each field in each line(not complete line) is converted to EquityFeeds object (with other fields as null) is sent as a message to the queue

Converting CSV file to Java Objects (POJO) and send it to ActiveMQ queue

1 Answers