Spring Batch Flat File Reader reads multiple files

Problem :
Implement a standalone flight search engine that lists matching flights for a traveller who is looking to fly between 2 locations on a given date. The input locations should be in 3-letter location code format as present in the airlines flight data.
Specifications: You are given 3 CSV files, each containing the available flights data for British Airways, Lufthansa Airlines and Air France respectively.
Each file contains the following fields:

1. FLIGHT_NUM - Unique flight number, starting with 2-letter airline code.
   
2. DEP_LOC - Departure location code of the flight.
   
3. ARR_LOC - Arrival location code of the flight.
   
4. VALID_TILL - Date (DD-MM-YYYY) till which each flight is available. It means that this flight would fly once every day till this date.
   
5. FLIGHT_TIME - Local time (HHmm) at which the flight departs from the departure location.
   
6. FLIGHT_DURN - Flight duration (HH.mm) from departure location to the arrival location.
   
7. FARE - This is the fare of the flight per person in $ (US Dollars).

Following are the constraints should apply:

• The engine should accept 4 input parameters: Departure Location, Arrival Location, Flight Date, Output Preference. “Output Preference” is a String suggesting whether the flight results should be sorted only by fare, or by both fare and flight duration.
  
• The engine should then search for the flights in all 3 CSV files, and list the aggregate results on standard output, sorted by Fare and Flight Duration, depending on “Output Preference”.
  
• The engine should be written considering that there could be more CSV files in future, and each CSV file might contain more data than present.
  
• If the Departure location or/and Arrival location is not present in any of the CSV files, then engine should return a user friendly error.
  
• If there are no flights available for the user entered input parameters, then engine should return a different user friendly error.

Input data: The following PIPE delimited sample data is given to you. You are allowed to extend on this data and create heavier input files. Flight_Details_1.csv

FLIGHT_NUM|DEP_LOC|ARR_LOC|VALID_TILL|FLIGHT_TIME|FLIGHT_DURN|FARE AF299|FRA|LHR|20-11-2010|0600|4.10|480 AF118|DUB|MUC|21-12-2010|1410|5.40|580 AF371|AMS|MAD|30-11-2010|1210|3.45|320 AF453|BOS|CDG|20-11-2010|1350|7.30|1000 AF544|BOM|LHR|10-12-2010|1150|8.10|950 AF271|AMS|MAD|27-10-2010|1100|3.30|500 AF249|JFK|LHR|01-12-2010|1550|8.40|1030

Flight_Details_2.csv

FLIGHT_NUM|DEP_LOC|ARR_LOC|VALID_TILL|FLIGHT_TIME|FLIGHT_DURN|FARE AF299|FRA|LHR|20-11-2010|0600|4.10|480 AF118|DUB|MUC|21-12-2010|1410|5.40|580 AF371|AMS|MAD|30-11-2010|1210|3.45|320 AF453|BOS|CDG|20-11-2010|1350|7.30|1000 AF544|BOM|LHR|10-12-2010|1150|8.10|950 AF271|AMS|MAD|27-10-2010|1100|3.30|500 AF249|JFK|LHR|01-12-2010|1550|8.40|1030

Flight_Details_3.csv

FLIGHT_NUM|DEP_LOC|ARR_LOC|VALID_TILL|FLIGHT_TIME|FLIGHT_DURN|FARE BA123|DEL|AMS|12-10-2010|0050|8.00|950 BA412|BOS|CDG|31-12-2010|0210|7.50|800 BA413|BOS|AMS|30-11-2010|1530|7.00|750 BA111|LHR|PEK|30-10-2010|2340|12.50|1200 BA765|LHR|BOM|31-12-2010|1420|8.50|825 BA322|CDG|NRT|15-11-2010|0010|13.00|1150 BA438|DEL|AMS|30-11-2010|1325|10.50|920

Problem Solution:

1. Create a Data TransferObject/ValueObject which can hold the properties of the flight.
   
2. Create another Data Transfer Objects which suits for search criteria.
   
3. Create a mapper class which splits the each line at specified delimiter and map the tokens to ValueObject.
   
4. Create a Writer class and implement the business logic.
   
5. Make necessary configurations to run the program.
   
6. Create a client program and to provide search criteria.

FlightDetailsVO.java: A Data Transfer/Value Object which has the flight properties.

package com.flight.dto;

import java.io.Serializable;
import java.util.Date;

public class FlightDetailsVO implements Serializable {

 /**
  * 
  */
 private static final long serialVersionUID = 1L;
 
 private String flightNum;
 private String depLoc;
 private String arrLoc;
 private Date validTill;
 private String flightTime;
 private double flightDuration;
 private double fare;
 
 public String getFlightNum() {
  return flightNum;
 }
 public void setFlightNum(String flightNum) {
  this.flightNum = flightNum;
 }
 public String getDepLoc() {
  return depLoc;
 }
 public void setDepLoc(String depLoc) {
  this.depLoc = depLoc;
 }
 public String getArrLoc() {
  return arrLoc;
 }
 public void setArrLoc(String arrLoc) {
  this.arrLoc = arrLoc;
 }
 public Date getValidTill() {
  return validTill;
 }
 public void setValidTill(Date validTill) {
  this.validTill = validTill;
 }
 public String getFlightTime() {
  return flightTime;
 }
 public void setFlightTime(String flightTime) {
  this.flightTime = flightTime;
 }
 public double getFlightDuration() {
  return flightDuration;
 }
 public void setFlightDuration(double flightDuration) {
  this.flightDuration = flightDuration;
 }
 public double getFare() {
  return fare;
 }
 public void setFare(double fare) {
  this.fare = fare;
 }
 
 @Override
 public String toString() {
  return flightNum + "|" 
    + depLoc + "|" 
    + arrLoc + "|" 
    + validTill + "|" 
    + flightTime + "|" 
    + flightDuration + "|" 
    + fare;
 }
 
}

SearchVO.java: It is used to pass the user given search criteria to search engine.

package com.flight.dto;

import java.io.Serializable;
import java.util.Date;

public class SearchVO implements Serializable {
 
 /**
  * 
  */
 private static final long serialVersionUID = 1L;
 
 private String depLoc;
 private String arrLoc;
 private Date flightDate;
 private String sortedBy;
 
 public String getDepLoc() {
  return depLoc;
 }
 public void setDepLoc(String depLoc) {
  this.depLoc = depLoc;
 }
 public String getArrLoc() {
  return arrLoc;
 }
 public void setArrLoc(String arrLoc) {
  this.arrLoc = arrLoc;
 }
 public Date getFlightDate() {
  return flightDate;
 }
 public void setFlightDate(Date flightDate) {
  this.flightDate = flightDate;
 }
 public String getSortedBy() {
  return sortedBy;
 }
 public void setSortedBy(String sortedBy) {
  this.sortedBy = sortedBy;
 }

}

FlightSearchFieldSetMapper.java: It’s a field set mapper and it will fetch all the lines from multiple files and will set to a list and will provide the same to writer.

package com.flight.mapper;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.validation.BindException;

import com.flight.dto.FlightDetailsVO;

public class FlightSearchFieldSetMapper implements FieldSetMapper<FlightDetailsVO> {

 private static final String PARSE_ERROR = "PARSE_ERROR";
 @Override
 public FlightDetailsVO mapFieldSet(FieldSet fieldSet) throws BindException {
  
  FlightDetailsVO flightDetailsVO = new FlightDetailsVO();
  
  flightDetailsVO.setFlightNum(fieldSet.readString(0));
  flightDetailsVO.setDepLoc(fieldSet.readRawString(1));
  flightDetailsVO.setArrLoc(fieldSet.readRawString(2));
  
  String validTill = fieldSet.readRawString(3);

  try {
   flightDetailsVO.setValidTill(convertStringToDate(validTill));
  } catch(ParseException pe) {
   flightDetailsVO.setFlightNum(PARSE_ERROR);
  }
  
  flightDetailsVO.setFlightTime(fieldSet.readRawString(4));
           flightDetailsVO.setFlightDuration(Double.parseDouble(fieldSet.readRawString(5)));
  flightDetailsVO.setFare(Double.parseDouble(fieldSet.readRawString(6)));
  
  return flightDetailsVO;
 }
 
 public Date convertStringToDate(String strDate) throws ParseException {
  
  SimpleDateFormat formatter = new SimpleDateFormat("dd-MM-yyyy");
  return formatter.parse(strDate);
 }
 
 public Date converStringToTime(String StrTime) throws ParseException {
  SimpleDateFormat formatter = new SimpleDateFormat("HH:mm");
  return formatter.parse(StrTime);
 }
 
}

FlightSearchItemWriter.java: It’s Item Writer, in our program it will act as search engine.

package com.flight.writer;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;

import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.batch.item.ItemWriter;

import com.flight.dto.FlightDetailsVO;
import com.flight.dto.SearchVO;

public class FlightSearchItemWriter implements ItemWriter<FlightDetailsVO>, ItemStream {
 
 private SearchVO searchVO;

 @Override
 public void write(List<? extends FlightDetailsVO> flightDetailsVOList) throws Exception {
  ArrayList<FlightDetailsVO> flightDetailsList = new ArrayList>FlightDetailsVO<(flightDetailsVOList);
  
  ArrayList<FlightDetailsVO> availFlightsList = new ArrayList>FlightDetailsVO<();
  
  if(getSearchVO() != null) {
   for(FlightDetailsVO currVO : flightDetailsList) {
    if(currVO.getValidTill().compareTo(searchVO.getFlightDate()) >= 0) {
     availFlightsList.add(currVO);
    }
   }
   
   if(availFlightsList != null && availFlightsList.size() > 0) {
    System.out.println("FLIGHT_NUM|DEP_LOC|ARR_LOC|VALID_TILL|FLIGHT_TIME|FLIGHT_DURN|FARE");
    System.out.println("------------------------------------------------------------------");
    
    Collections.sort(availFlightsList, new Comparator<FlightDetailsVO>() {  
        @Override  
        public int compare(FlightDetailsVO o1, FlightDetailsVO o2) {  
            int fareComp = Double.compare(o1.getFare(), o2.getFare());  
            if(searchVO.getSortedBy().equalsIgnoreCase("both")) {
             if (fareComp != 0) {  
                 return fareComp;  
             }  
             int durComp = Double.compare(o1.getFlightDuration(), o2.getFlightDuration());
             return durComp;
            } 
            return fareComp; 
        }  
    });    
    for(FlightDetailsVO availFlightVO : availFlightsList) {
     System.out.println(availFlightVO.toString());
    }
   }
  } else {
   System.out.println("Search criteria is not provided.....");
  }
 }

 @Override
 public void close() throws ItemStreamException {
  
 }

 @Override
 public void open(ExecutionContext arg0) throws ItemStreamException {
  
 }

 @Override
 public void update(ExecutionContext arg0) throws ItemStreamException {
  
 }

 public SearchVO getSearchVO() {
  return searchVO;
 }

 public void setSearchVO(SearchVO searchVO) {
  this.searchVO = searchVO;
 }


}

flight-seaarch-context.xml: It’s a configuration file, which contain all spring beans definitions.

FlightSearchRun.java: Its client program for Flight Search engine.

package com.flight.main;

import java.text.SimpleDateFormat;

import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.flight.dto.SearchVO;
import com.flight.writer.FlightSearchItemWriter;

public class FlightSearchRun {
 
 public static void main(String[] args) {
  String[] springConfig  = {"resources/xml/flight-seaarch-context.xml"};
  
  ApplicationContext context = new ClassPathXmlApplicationContext(springConfig);
  
  JobLauncher jobLauncher = (JobLauncher) context.getBean("jobLauncher");
  Job job = (Job) context.getBean("flightSearchJob");
  try {
   SearchVO searchVO = new SearchVO();
   searchVO.setDepLoc("DUB");
   searchVO.setArrLoc("MUC");
   searchVO.setSortedBy("both");
   
   SimpleDateFormat formatter = new SimpleDateFormat("dd-MM-yyyy");
   searchVO.setFlightDate(formatter.parse("21-12-2013"));
   
   FlightSearchItemWriter writer = (FlightSearchItemWriter) context.getBean("flightSearchItemWriter");
   writer.setSearchVO(searchVO);
   
   JobExecution execution = jobLauncher.run(job, new JobParameters());
   System.out.println("Exit Status : " + execution.getStatus());

  } catch (Exception e) {
   e.printStackTrace();
  }
 }

}