OPEN - issue BATCH-320: Refactor ItemWriter as primary collaborator (with wrapper for processor)

http://jira.springframework.org/browse/BATCH-320

Half way there - tests OK, but some naming and tidying up to do.

spring-batch-samples/src/main/java/org/springframework/batch/sample/dao/CustomerCreditDao.java

@@ -16,7 +16,6 @@
 
 package org.springframework.batch.sample.dao;
 
-import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.sample.domain.CustomerCredit;
 
 /**
@@ -24,8 +23,8 @@ import org.springframework.batch.sample.domain.CustomerCredit;
  *
  * @author Robert Kasanicky
  */
-public interface CustomerCreditWriter extends ItemWriter {
+public interface CustomerCreditDao {
 
-	void writeCredit(CustomerCredit customerCredit);
+	void writeCredit(CustomerCredit customerCredit) throws Exception;
 
 }

spring-batch-samples/src/main/java/org/springframework/batch/sample/dao/FlatFileCustomerCreditWriter.java

@@ -24,10 +24,10 @@ import org.springframework.beans.factory.DisposableBean;
 /**
  * Writes customer's credit information in a file.
  *
- * @see CustomerCreditWriter
+ * @see CustomerCreditDao
  * @author Robert Kasanicky
  */
-public class FlatFileCustomerCreditWriter implements CustomerCreditWriter,
+public class FlatFileCustomerCreditWriter implements CustomerCreditDao,
 		DisposableBean {
 
 	private ItemWriter outputSource;
@@ -36,7 +36,7 @@ public class FlatFileCustomerCreditWriter implements CustomerCreditWriter,
 
 	private volatile boolean opened = false;
 
-	public void writeCredit(CustomerCredit customerCredit) {
+	public void writeCredit(CustomerCredit customerCredit) throws Exception {
 
 		if (!opened) {
 			open();
@@ -78,7 +78,4 @@ public class FlatFileCustomerCreditWriter implements CustomerCreditWriter,
 		close();
 	}
 
-	public void write(Object output) {
-		writeCredit((CustomerCredit)output);
-	}
 }

spring-batch-samples/src/main/java/org/springframework/batch/sample/dao/FlatFileOrderWriter.java

@@ -17,8 +17,8 @@
 package org.springframework.batch.sample.dao;
 
 import org.springframework.batch.io.file.transform.Converter;
-import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.sample.domain.Order;
+import org.springframework.batch.item.processor.DelegatingItemWriter;
+import org.springframework.batch.sample.item.processor.OrderWriter;
 
 
 /**
@@ -28,11 +28,7 @@ import org.springframework.batch.sample.domain.Order;
  * 
  * @author Dave Syer
  */
-public class FlatFileOrderWriter implements OrderWriter {
-    /**
-     * Takes care of writing to a file
-     */
-    private ItemWriter outputSource;
+public class FlatFileOrderWriter extends DelegatingItemWriter {
 
     /**
      * Converter for order
@@ -48,15 +44,11 @@ public class FlatFileOrderWriter implements OrderWriter {
 		this.converter = converter;
 	}
 
-    /**
-     * Writes information from an Order object to a file
-     */
-    public void write(Order data) {
-        outputSource.write(converter.convert(data));
-    }
-    
-	public void setOutputSource(ItemWriter outputSource) {
-        this.outputSource = outputSource;
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.processor.DelegatingItemWriter#doProcess(java.lang.Object)
+	 */
+	protected Object doProcess(Object item) throws Exception {
+        return converter.convert(item);
     }
 
 }

spring-batch-samples/src/main/java/org/springframework/batch/sample/dao/HibernateCreditWriter.java

@@ -30,7 +30,7 @@ import org.springframework.orm.hibernate3.support.HibernateDaoSupport;
  * 
  */
 public class HibernateCreditWriter extends HibernateDaoSupport implements
-		CustomerCreditWriter, RepeatInterceptor {
+		CustomerCreditDao, RepeatInterceptor {
 
 	private int failOnFlush = -1;
 	private List errors = new ArrayList();

spring-batch-samples/src/main/java/org/springframework/batch/sample/dao/IbatisCustomerCreditWriter.java

@@ -15,7 +15,6 @@
  */
 package org.springframework.batch.sample.dao;
 
-import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.sample.domain.CustomerCredit;
 import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport;
 
@@ -24,7 +23,7 @@ import org.springframework.orm.ibatis.support.SqlMapClientDaoSupport;
  *
  */
 public class IbatisCustomerCreditWriter extends SqlMapClientDaoSupport
-		implements CustomerCreditWriter, ItemWriter {
+		implements CustomerCreditDao {
 
 	String statementId;
 
@@ -53,7 +52,4 @@ public class IbatisCustomerCreditWriter extends SqlMapClientDaoSupport
 		this.statementId = statementId;
 	}
 
-	public void write(Object output) {
-		writeCredit((CustomerCredit)output);
-	}
 }

spring-batch-samples/src/main/java/org/springframework/batch/sample/dao/JdbcTradeDao.java

@@ -28,8 +28,8 @@ import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer
  *
  * @author Robert Kasanicky
  */
-public class JdbcTradeWriter implements TradeWriter {
-	private Log log = LogFactory.getLog(JdbcTradeWriter.class);
+public class JdbcTradeDao implements TradeDao {
+	private Log log = LogFactory.getLog(JdbcTradeDao.class);
     /**
      * template for inserting a row
      */
@@ -46,7 +46,7 @@ public class JdbcTradeWriter implements TradeWriter {
     private DataFieldMaxValueIncrementer incrementer;
 
     /**
-     * @see TradeWriter
+     * @see TradeDao
      */
     public void writeTrade(Trade trade) {
         Long id = new Long(incrementer.nextLongValue());
@@ -66,8 +66,4 @@ public class JdbcTradeWriter implements TradeWriter {
         this.incrementer = incrementer;
     }
 
-	public void write(Object output) {
-		this.writeTrade((Trade)output);
-	}
-
 }

spring-batch-samples/src/main/java/org/springframework/batch/sample/dao/OrderDao.java

@@ -21,7 +21,7 @@ import org.springframework.batch.sample.domain.Order;
 /**
  * Interface for writing <code>Order</code> objects.
  */
-public interface OrderWriter {
+public interface OrderDao {
 
     public void write(Order order);
     

spring-batch-samples/src/main/java/org/springframework/batch/sample/dao/TradeDao.java

@@ -16,7 +16,6 @@
 
 package org.springframework.batch.sample.dao;
 
-import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.sample.domain.Trade;
 
 
@@ -26,7 +25,7 @@ import org.springframework.batch.sample.domain.Trade;
  *
  * @author Robert Kasanicky
  */
-public interface TradeWriter extends ItemWriter{
+public interface TradeDao {
     /**
      * Write a trade object to some kind of output,
      * different implementations can write to file, database etc.

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/CustomerCreditIncreaseProcessor.java

@@ -1,30 +0,0 @@
-package org.springframework.batch.sample.item.processor;
-
-import java.math.BigDecimal;
-
-import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.item.ItemWriter;
-import org.springframework.batch.sample.domain.CustomerCredit;
-
-/**
- * Increases customer's credit by fixed amount.
- * 
- * @author Robert Kasanicky
- */
-public class CustomerCreditIncreaseProcessor implements ItemProcessor{
-
-	public static final BigDecimal FIXED_AMOUNT = new BigDecimal(1000);
-	
-	private ItemWriter outputSource;
-	
-	public void process(Object data) throws Exception {
-		CustomerCredit customerCredit = (CustomerCredit) data;
-		customerCredit.increaseCreditBy(FIXED_AMOUNT);
-		outputSource.write(customerCredit);
-	}
-
-	public void setOutputSource(ItemWriter outputSource) {
-		this.outputSource = outputSource;
-	}
-
-}

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/CustomerCreditIncreaseWriter.java

@@ -0,0 +1,37 @@
+package org.springframework.batch.sample.item.processor;
+
+import java.math.BigDecimal;
+
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.sample.dao.CustomerCreditDao;
+import org.springframework.batch.sample.domain.CustomerCredit;
+
+/**
+ * Increases customer's credit by fixed amount.
+ * 
+ * @author Robert Kasanicky
+ */
+public class CustomerCreditIncreaseWriter implements ItemWriter {
+
+	public static final BigDecimal FIXED_AMOUNT = new BigDecimal(1000);
+	
+	private CustomerCreditDao customerCreditDao;
+	
+	/**
+	 * Public setter for the {@link CustomerCreditDao}.
+	 * @param customerCreditDao the {@link CustomerCreditDao} to set
+	 */
+	public void setCustomerCreditDao(CustomerCreditDao customerCreditDao) {
+		this.customerCreditDao = customerCreditDao;
+	}
+	
+	/* (non-Javadoc)
+	 * @see org.springframework.batch.item.processor.DelegatingItemWriter#doProcess(java.lang.Object)
+	 */
+	public void write(Object data) throws Exception {
+		CustomerCredit customerCredit = (CustomerCredit) data;
+		customerCredit.increaseCreditBy(FIXED_AMOUNT);
+		customerCreditDao.writeCredit(customerCredit);
+	}
+
+}

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/CustomerCreditUpdateWriter.java

@@ -16,21 +16,21 @@
 
 package org.springframework.batch.sample.item.processor;
 
-import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.sample.dao.CustomerCreditWriter;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.sample.dao.CustomerCreditDao;
 import org.springframework.batch.sample.domain.CustomerCredit;
 
 
 
-public class CustomerCreditUpdateProcessor implements ItemProcessor {
+public class CustomerCreditUpdateWriter implements ItemWriter {
     private double creditFilter = 800;
-    private CustomerCreditWriter writer;
+    private CustomerCreditDao dao;
 
-    public void process(Object data) {
+    public void write(Object data) throws Exception {
         CustomerCredit customerCredit = (CustomerCredit) data;
 
         if (customerCredit.getCredit().doubleValue() > creditFilter) {
-            writer.writeCredit(customerCredit);
+            dao.writeCredit(customerCredit);
         }
     }
 
@@ -38,8 +38,8 @@ public class CustomerCreditUpdateProcessor implements ItemProcessor {
         this.creditFilter = creditFilter;
     }
 
-    public void setWriter(CustomerCreditWriter writer) {
-        this.writer = writer;
+    public void setWriter(CustomerCreditDao writer) {
+        this.dao = writer;
     }
 
 }

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/CustomerUpdateWriter.java

@@ -16,7 +16,7 @@
 
 package org.springframework.batch.sample.item.processor;
 
-import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.sample.dao.JdbcCustomerDebitWriter;
 import org.springframework.batch.sample.domain.CustomerDebit;
 import org.springframework.batch.sample.domain.Trade;
@@ -27,10 +27,10 @@ import org.springframework.batch.sample.domain.Trade;
  * 
  * @author Robert Kasanicky
  */
-public class CustomerUpdateProcessor implements ItemProcessor {
+public class CustomerUpdateWriter implements ItemWriter {
     private JdbcCustomerDebitWriter dao;
 
-    public void process(Object data) {
+    public void write(Object data) {
         Trade trade = (Trade) data;
         CustomerDebit customerDebit = new CustomerDebit();
         customerDebit.setName(trade.getCustomer());

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/DefaultFlatFileProcessor.java

@@ -1,18 +0,0 @@
-package org.springframework.batch.sample.item.processor;
-
-import org.springframework.batch.io.file.FlatFileItemWriter;
-import org.springframework.batch.item.ItemProcessor;
-
-public class DefaultFlatFileProcessor implements ItemProcessor{
-
-	private FlatFileItemWriter flatFileItemWriter;
-
-	public void process(Object data) throws Exception {
-		flatFileItemWriter.write(""+data);
-	}
-
-	public void setFlatFileOutputSource(FlatFileItemWriter flatFileItemWriter) {
-		this.flatFileItemWriter = flatFileItemWriter;
-	}
-	
-}

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/DummyProcessor.java

@@ -1,41 +0,0 @@
-/*
- * Copyright 2006-2007 the original author or authors.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.springframework.batch.sample.item.processor;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.springframework.batch.item.ItemProcessor;
-
-/**
- * Dummy processor useful for development and testing.
- * 
- * @author Robert Kasanicky
- */
-public class DummyProcessor implements ItemProcessor {
-	
-	private static final Log log = LogFactory.getLog(DummyProcessor.class);
-	
-    public void process(Object object) {
-        log.debug("PROCESSING: " + object);
-    }
-
-	public void close() {
-	}
-
-	public void init() {
-	}
-}

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/OrderWriter.java

@@ -17,27 +17,16 @@
 package org.springframework.batch.sample.item.processor;
 
 import org.springframework.batch.io.exception.BatchCriticalException;
-import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.sample.dao.OrderWriter;
+import org.springframework.batch.item.processor.DelegatingItemWriter;
 import org.springframework.batch.sample.domain.Order;
 
 
 
-public class OrderProcessor implements ItemProcessor {
-    private OrderWriter writer;
-    private Order order;
-
-    public void process(Object data) {
+public class OrderWriter extends DelegatingItemWriter {
+    public Object doProcess(Object data) {
         if (!(data instanceof Order)) {
             throw new BatchCriticalException("OrderProcessor can process only Order objects");
         }
-
-        order = (Order) data;
-        writer.write(order);
+        return data;
     }
-
-    public void setWriter(OrderWriter reportService) {
-        this.writer = reportService;
-    }
-
 }

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/PersonWriter.java

@@ -18,15 +18,15 @@ package org.springframework.batch.sample.item.processor;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.sample.domain.Person;
 
 
 
-public class PersonProcessor implements ItemProcessor {
-    private static Log log = LogFactory.getLog(PersonProcessor.class);
+public class PersonWriter implements ItemWriter {
+    private static Log log = LogFactory.getLog(PersonWriter.class);
  
-    public void process(Object data) {
+    public void write(Object data) {
         if (!(data instanceof Person)) {
             log.warn("PersonProcessor can process only Person objects, skipping record");
 

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/PlayerItemWriter.java

@@ -1,14 +1,14 @@
 package org.springframework.batch.sample.item.processor;
 
-import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemWriter;
 import org.springframework.batch.sample.dao.PlayerDao;
 import org.springframework.batch.sample.domain.Player;
 
-public class PlayerItemProcessor implements ItemProcessor {
+public class PlayerItemWriter implements ItemWriter {
 
 	PlayerDao playerDao;
 	
-	public void process(Object data) throws Exception {		
+	public void write(Object data) throws Exception {		
 		playerDao.savePlayer((Player)data);
 	}
 

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/StagingItemWriter.java

@@ -6,14 +6,14 @@ import java.sql.Types;
 import org.apache.commons.lang.SerializationUtils;
 import org.springframework.batch.execution.scope.StepContext;
 import org.springframework.batch.execution.scope.StepContextAware;
-import org.springframework.batch.item.ItemProcessor;
+import org.springframework.batch.item.ItemWriter;
 import org.springframework.jdbc.core.support.JdbcDaoSupport;
 import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
 
-public class StagingItemProcessor extends JdbcDaoSupport implements
-		StepContextAware, ItemProcessor {
+public class StagingItemWriter extends JdbcDaoSupport implements
+		StepContextAware, ItemWriter {
 
 	public static final String NEW = "N";
 	public static final String DONE = "Y";
@@ -34,7 +34,7 @@ public class StagingItemProcessor extends JdbcDaoSupport implements
 						incrementer,
 						"DataFieldMaxValueIncrementer is required - set the incrementer property in the "
 								+ ClassUtils
-										.getShortName(StagingItemProcessor.class));
+										.getShortName(StagingItemWriter.class));
 	}
 
 	/**
@@ -62,7 +62,7 @@ public class StagingItemProcessor extends JdbcDaoSupport implements
 	 * 
 	 * @see org.springframework.batch.item.ItemProcessor#process(java.lang.Object)
 	 */
-	public void process(Object data) throws Exception {
+	public void write(Object data) {
 		Long id = new Long(incrementer.nextLongValue());
 		Long jobId = stepContext.getStepExecution().getJobExecution().getJobId();
 		byte[] blob = SerializationUtils.serialize((Serializable) data);

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/processor/TradeWriter.java

@@ -18,15 +18,15 @@ package org.springframework.batch.sample.item.processor;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.sample.dao.TradeWriter;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.sample.dao.TradeDao;
 import org.springframework.batch.sample.domain.Trade;
 
 
 
-public class TradeProcessor implements ItemProcessor {
-    private static Log log = LogFactory.getLog(TradeProcessor.class);
-    private TradeWriter writer;
+public class TradeWriter implements ItemWriter {
+    private static Log log = LogFactory.getLog(TradeWriter.class);
+    private TradeDao dao;
     
     private int failure = -1;
     
@@ -41,7 +41,7 @@ public class TradeProcessor implements ItemProcessor {
 		this.failure = failure;
 	}
 
-    public void process(Object data) {
+    public void write(Object data) {
         if (!(data instanceof Trade)) {
             log.warn("TradeProcessor can process only Trade objects, skipping record");
 
@@ -52,7 +52,7 @@ public class TradeProcessor implements ItemProcessor {
         log.debug(data);
 
         //TODO put some processing of the trade object here
-        writer.writeTrade(trade);
+        dao.writeTrade(trade);
         
         if(index++ == failure) {
         	throw new RuntimeException("Something unexpected happened!");
@@ -61,8 +61,8 @@ public class TradeProcessor implements ItemProcessor {
         
     }
 
-    public void setWriter(TradeWriter dao) {
-        this.writer = dao;
+    public void setDao(TradeDao dao) {
+        this.dao = dao;
     }
 
 	public void close() {

spring-batch-samples/src/main/java/org/springframework/batch/sample/item/reader/StagingItemReader.java

@@ -14,7 +14,7 @@ import org.springframework.batch.execution.scope.StepContextAware;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ResourceLifecycle;
 import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager;
-import org.springframework.batch.sample.item.processor.StagingItemProcessor;
+import org.springframework.batch.sample.item.processor.StagingItemWriter;
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.jdbc.core.RowMapper;
@@ -98,7 +98,7 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemReader, Res
 
 			"SELECT ID FROM BATCH_STAGING WHERE JOB_ID=? AND PROCESSED=? ORDER BY ID",
 
-			new Object[] { stepContext.getStepExecution().getJobExecution().getJobId(), StagingItemProcessor.NEW },
+			new Object[] { stepContext.getStepExecution().getJobExecution().getJobId(), StagingItemWriter.NEW },
 
 			new RowMapper() {
 				public Object mapRow(ResultSet rs, int rowNum) throws SQLException {
@@ -131,7 +131,7 @@ public class StagingItemReader extends JdbcDaoSupport implements ItemReader, Res
 				});
 		// Update now - changes will rollback if there is a problem later.
 		int count = getJdbcTemplate().update("UPDATE BATCH_STAGING SET PROCESSED=? WHERE ID=? AND PROCESSED=?",
-				new Object[] { StagingItemProcessor.DONE, id, StagingItemProcessor.NEW });
+				new Object[] { StagingItemWriter.DONE, id, StagingItemWriter.NEW });
 		if (count != 1) {
 			throw new OptimisticLockingFailureException("The staging record with ID=" + id
 					+ " was updated concurrently when trying to mark as complete (updated " + count + " records.");

spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java

@@ -22,7 +22,7 @@ import org.springframework.batch.core.tasklet.Tasklet;
 import org.springframework.batch.execution.tasklet.ItemOrientedTasklet;
 import org.springframework.batch.io.file.DefaultFlatFileItemReader;
 import org.springframework.batch.repeat.ExitStatus;
-import org.springframework.batch.sample.dao.TradeWriter;
+import org.springframework.batch.sample.dao.TradeDao;
 import org.springframework.batch.sample.domain.Trade;
 import org.springframework.batch.statistics.StatisticsProvider;
 
@@ -50,7 +50,7 @@ public class SimpleTradeTasklet implements Tasklet, StatisticsProvider {
 	/*
 	 * writes a Trade object to output
 	 */
-	private TradeWriter tradeWriter;
+	private TradeDao tradeDao;
 
 	/**
 	 * number of trade objects processed
@@ -72,7 +72,7 @@ public class SimpleTradeTasklet implements Tasklet, StatisticsProvider {
 		}
 
 		tradeCount++;
-		tradeWriter.writeTrade(trade);
+		tradeDao.writeTrade(trade);
 		return ExitStatus.CONTINUABLE;
 	}
 
@@ -80,8 +80,8 @@ public class SimpleTradeTasklet implements Tasklet, StatisticsProvider {
 		this.inputSource = inputTemplate;
 	}
 
-	public void setTradeDao(TradeWriter tradeWriter) {
-		this.tradeWriter = tradeWriter;
+	public void setTradeDao(TradeDao tradeDao) {
+		this.tradeDao = tradeDao;
 	}
 
 	public Properties getStatistics() {