Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RESOLVED - issue BATCH-581: Add filter capability to item oriented paradigm

Browse Source 
http://jira.springframework.org/browse/BATCH-581

Added a new sample job: 'CustomerFilterJob' that represents a use case for the filtering of items in a processor.
This commit is contained in:
lucasward
2008-09-29 05:29:29 +00:00
parent 97773129eb
commit 2bc0c17140
17 changed files with 885 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/CompositeCustomerUpdateLineTokenizer.java Normal file
View File

@@ -0,0 +1,61 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade;
import org.springframework.batch.item.file.mapping.FieldSet;
import org.springframework.batch.item.file.transform.LineTokenizer;
/**
* Composite {@link LineTokenizer} that delegates the tokenization of a line to one of two potential
* tokenizers. The file format in this case uses one character, either F, A, U, or D to indicate
* whether or not the line is an a footer record, or a customer add, update, or delete, and
* will delegate accordingly.
*
* @author Lucas Ward
* @since 2.0
*/
public class CompositeCustomerUpdateLineTokenizer implements LineTokenizer {
private LineTokenizer customerTokenizer;
private LineTokenizer footerTokenizer;
/* (non-Javadoc)
* @see org.springframework.batch.item.file.transform.LineTokenizer#tokenize(java.lang.String)
*/
public FieldSet tokenize(String line) {
if(line.charAt(0) == 'F'){
//line starts with F, so the footer tokenizer should tokenize it.
return footerTokenizer.tokenize(line);
}
else if(line.charAt(0) == 'A' || line.charAt(0) == 'U' || line.charAt(0) == 'D'){
//line starts with A,U, or D, so it must be a customer operation.
return customerTokenizer.tokenize(line);
}
else{
//If the line doesn't start with any of the characters above, it must obviously be invalid.
throw new IllegalArgumentException("Invalid line encountered for tokenizing: " + line);
}
}
/**
* Set the {@link LineTokenizer} that will be used to tokenize any lines that begin with
* A, U, or D, and are thus a customer operation.
*
* @param customerTokenizer
*/
public void setCustomerTokenizer(LineTokenizer customerTokenizer) {
this.customerTokenizer = customerTokenizer;
}
/**
* Set the {@link LineTokenizer} that will be used to tokenize any lines that being with
* F and is thus a footer record.
*
* @param footerTokenizer
*/
public void setFooterTokenizer(LineTokenizer footerTokenizer) {
this.footerTokenizer = footerTokenizer;
}
}

19
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/CustomerDao.java Normal file
View File

@@ -0,0 +1,19 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade;
import java.math.BigDecimal;
/**
* @author Lucas Ward
*
*/
public interface CustomerDao {
CustomerCredit getCustomerByName(String name);
void insertCustomer(String name, BigDecimal credit);
void updateCustomer(String name, BigDecimal credit);
}

45
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/CustomerOperation.java Normal file
View File

@@ -0,0 +1,45 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade;
import java.util.HashMap;
import java.util.Map;
/**
* Enum representing on of 3 possible actions on a customer update:
* Add, update, or delete
*
* @author Lucas Ward
*
*/
public enum CustomerOperation {
ADD('A'), UPDATE('U'), DELETE('D');
private final char code;
private static final Map<Character,CustomerOperation> codeMap;
private CustomerOperation(char code) {
this.code = code;
}
static{
codeMap = new HashMap<Character,CustomerOperation>();
for(CustomerOperation operation:values()){
codeMap.put(operation.getCode(), operation);
}
}
public static CustomerOperation fromCode(char code){
if(codeMap.containsKey(code)){
return codeMap.get(code);
}
else{
throw new IllegalArgumentException("Invalid code: [" + code + "]");
}
}
public char getCode() {
return code;
}
}

46
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/CustomerUpdate.java Normal file
View File

@@ -0,0 +1,46 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade;
import java.math.BigDecimal;
/**
* Immutable Value Object representing an update to the customer as stored in the database.
* This object has the customer name, credit ammount, and the operation to be performed
* on them. In the case of an add, a new customer will be entered with the appropriate
* credit. In the case of an update, the customer's credit is considered an absolute update.
* Deletes are currently not supported, but can still be read in from a file.
*
* @author Lucas Ward
* @since 2.0
*/
public class CustomerUpdate {
private final CustomerOperation operation;
private final String customerName;
private final BigDecimal credit;
public CustomerUpdate(CustomerOperation operation, String customerName, BigDecimal credit) {
this.operation = operation;
this.customerName = customerName;
this.credit = credit;
}
public CustomerOperation getOperation() {
return operation;
}
public String getCustomerName() {
return customerName;
}
public BigDecimal getCredit() {
return credit;
}
@Override
public String toString() {
return "Customer Update, name: [" + customerName + "], operation: [" + operation + "], credit: [" + credit + "]";
}
}

52
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/CustomerUpdateFieldSetMapper.java Normal file
View File

@@ -0,0 +1,52 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade;
import java.math.BigDecimal;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
import org.springframework.batch.item.file.mapping.FieldSet;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
/**
* {@link FieldSetMapper} for mapping the
*
* @author Lucas Ward
*
*/
public class CustomerUpdateFieldSetMapper extends StepExecutionListenerSupport implements FieldSetMapper<CustomerUpdate> {
StepExecution stepExecution;
public CustomerUpdate mapLine(FieldSet fs) {
char code = fs.readChar(0);
if(code == 'F'){
long customerUpdateTotal = stepExecution.getReadCount();
long fileUpdateTotal = fs.readLong(1);
if(customerUpdateTotal != fileUpdateTotal){
throw new IllegalStateException("The total number of customer updates in the file footer does not match the " +
"number entered File footer total: [" + fileUpdateTotal + "] Total encountered during processing: [" +
customerUpdateTotal + "]");
}
else{
//return null, because the footer indicates an end of processing.
return null;
}
}
CustomerOperation operation = CustomerOperation.fromCode(code);
String name = fs.readString(1);
BigDecimal credit = fs.readBigDecimal(2);
return new CustomerUpdate(operation, name, credit);
}
@Override
public void beforeStep(StepExecution stepExecution) {
this.stepExecution = stepExecution;
}
}

61
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/CustomerUpdateProcessor.java Normal file
View File

@@ -0,0 +1,61 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade;
import static org.springframework.batch.sample.domain.trade.CustomerOperation.*;
import org.springframework.batch.item.ItemProcessor;
/**
* @author Lucas Ward
*
*/
public class CustomerUpdateProcessor implements ItemProcessor<CustomerUpdate, CustomerUpdate>{
CustomerDao customerDao;
InvalidCustomerLogger invalidCustomerLogger;
public CustomerUpdate process(CustomerUpdate item) throws Exception {
if(item.getOperation() == DELETE){
//delete is not supported
invalidCustomerLogger.log(item);
return null;
}
CustomerCredit customerCredit = customerDao.getCustomerByName(item.getCustomerName());
if(item.getOperation() == ADD && customerCredit == null){
return item;
}
else if(item.getOperation() == ADD && customerCredit != null){
//veto processing
invalidCustomerLogger.log(item);
return null;
}
if(item.getOperation() == UPDATE && customerCredit != null){
return item;
}
else if(item.getOperation() == UPDATE && customerCredit == null){
//veto processing
invalidCustomerLogger.log(item);
return null;
}
//if an item makes it through all these checks it can be assumed to be bad, logged, and skipped
invalidCustomerLogger.log(item);
return null;
}
public void setCustomerDao(CustomerDao customerDao) {
this.customerDao = customerDao;
}
public void setInvalidCustomerLogger(
InvalidCustomerLogger invalidCustomerLogger) {
this.invalidCustomerLogger = invalidCustomerLogger;
}
}

33
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/CustomerUpdateWriter.java Normal file
View File

@@ -0,0 +1,33 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade;
import java.util.List;
import org.springframework.batch.item.ItemWriter;
/**
* @author Lucas Wward
*
*/
public class CustomerUpdateWriter implements ItemWriter<CustomerUpdate> {
private CustomerDao customerDao;
public void write(List<? extends CustomerUpdate> items) throws Exception {
for(CustomerUpdate customerUpdate : items){
if(customerUpdate.getOperation() == CustomerOperation.ADD){
customerDao.insertCustomer(customerUpdate.getCustomerName(), customerUpdate.getCredit());
}
else if(customerUpdate.getOperation() == CustomerOperation.UPDATE){
customerDao.updateCustomer(customerUpdate.getCustomerName(), customerUpdate.getCredit());
}
}
}
public void setCustomerDao(CustomerDao customerDao) {
this.customerDao = customerDao;
}
}

18
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/InvalidCustomerLogger.java Normal file
View File

@@ -0,0 +1,18 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade;
/**
* Interface for logging invalid customers. Customers may need to be logged because
* they already existed when attempted to be added. Or a non existent customer was
* updated.
*
* @author Lucas Ward
*
*/
public interface InvalidCustomerLogger {
void log(CustomerUpdate customerUpdate);
}

28
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/CommonsLoggingInvalidCustomerLogger.java Normal file
View File

@@ -0,0 +1,28 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade.internal;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.launch.support.CommandLineJobRunner;
import org.springframework.batch.sample.domain.trade.CustomerUpdate;
import org.springframework.batch.sample.domain.trade.InvalidCustomerLogger;
/**
* @author Lucas Ward
*
*/
public class CommonsLoggingInvalidCustomerLogger implements
InvalidCustomerLogger {
protected static final Log logger = LogFactory.getLog(CommandLineJobRunner.class);
/* (non-Javadoc)
* @see org.springframework.batch.sample.domain.trade.InvalidCustomerLogger#log(org.springframework.batch.sample.domain.trade.CustomerUpdate)
*/
public void log(CustomerUpdate customerUpdate) {
logger.error("invalid customer encountered: [ " + customerUpdate + "]");
}
}

61
spring-batch-samples/src/main/java/org/springframework/batch/sample/domain/trade/internal/JdbcCustomerDao.java Normal file
View File

@@ -0,0 +1,61 @@
/**
*
*/
package org.springframework.batch.sample.domain.trade.internal;
import java.math.BigDecimal;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.List;
import org.springframework.batch.sample.domain.trade.CustomerCredit;
import org.springframework.batch.sample.domain.trade.CustomerDao;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.support.JdbcDaoSupport;
/**
* @author Lucas Ward
*
*/
public class JdbcCustomerDao extends JdbcDaoSupport implements CustomerDao{
private static final String GET_CUSTOMER_BY_NAME = "SELECT ID, NAME, CREDIT from CUSTOMER where NAME = ?";
private static final String INSERT_CUSTOMER = "INSERT into CUSTOMER(NAME, CREDIT) values(?,?)";
private static final String UPDATE_CUSTOMER = "UPDATE CUSTOMER set CREDIT = ? where NAME = ?";
public CustomerCredit getCustomerByName(String name) {
@SuppressWarnings("unchecked")
List<CustomerCredit> customers = (List<CustomerCredit>) getJdbcTemplate().query(GET_CUSTOMER_BY_NAME, new Object[]{name},
new RowMapper(){
public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
CustomerCredit customer = new CustomerCredit();
customer.setName(rs.getString("NAME"));
customer.setId(rs.getInt("ID"));
customer.setCredit(rs.getBigDecimal("CREDIT"));
return customer;
}
});
if(customers.size() == 0){
return null;
}
else{
return customers.get(0);
}
}
public void insertCustomer(String name, BigDecimal credit) {
getJdbcTemplate().update(INSERT_CUSTOMER, new Object[]{name, credit});
}
public void updateCustomer(String name, BigDecimal credit) {
getJdbcTemplate().update(UPDATE_CUSTOMER, new Object[]{credit, name});
}
}