OPEN - issue BATCH-787: Add write(List) to ItemWriter

Make samples and cli archetype compile again
This commit is contained in:
dsyer
2008-08-19 14:31:41 +00:00
parent 34f7cd8519
commit a5080b772e
31 changed files with 270 additions and 207 deletions

View File

@@ -3,6 +3,7 @@ package org.springframework.batch.sample.common;
import java.io.Serializable;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import org.apache.commons.lang.SerializationUtils;
import org.springframework.batch.core.StepExecution;
@@ -68,22 +69,28 @@ public class StagingItemWriter<T> extends JdbcDaoSupport implements StepExecutio
/**
* Serialize the item to the staging table, and add a NEW processed flag.
*
* @see ItemWriter#write(java.lang.Object)
* @see ItemWriter#write(java.util.List)
*/
public void write(T data) {
final long id = incrementer.nextLongValue();
final long jobId = stepExecution.getJobExecution().getJobId();
final byte[] blob = SerializationUtils.serialize((Serializable) data);
getJdbcTemplate().update("INSERT into BATCH_STAGING (ID, JOB_ID, VALUE, PROCESSED) values (?,?,?,?)",
new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setLong(1, id);
ps.setLong(2, jobId);
lobHandler.getLobCreator().setBlobAsBytes(ps, 3, blob);
ps.setString(4, NEW);
}
public void write(List<? extends T> items) {
for (T data : items) {
final long id = incrementer.nextLongValue();
final long jobId = stepExecution.getJobExecution().getJobId();
final byte[] blob = SerializationUtils.serialize((Serializable) data);
getJdbcTemplate().update("INSERT into BATCH_STAGING (ID, JOB_ID, VALUE, PROCESSED) values (?,?,?,?)",
new PreparedStatementSetter() {
public void setValues(PreparedStatement ps) throws SQLException {
ps.setLong(1, id);
ps.setLong(2, jobId);
lobHandler.getLobCreator().setBlobAsBytes(ps, 3, blob);
ps.setString(4, NEW);
}
});
}
});
}
public void clear() throws ClearFailedException {

View File

@@ -1,5 +1,7 @@
package org.springframework.batch.sample.domain.football.internal;
import java.util.List;
import org.springframework.batch.item.ClearFailedException;
import org.springframework.batch.item.FlushFailedException;
import org.springframework.batch.item.ItemWriter;
@@ -13,35 +15,28 @@ public class JdbcGameDao extends SimpleJdbcDaoSupport implements ItemWriter<Game
private SimpleJdbcInsert insertGame;
protected void initDao() throws Exception {
super.initDao();
insertGame = new SimpleJdbcInsert(getDataSource())
.withTableName("GAMES")
.usingColumns("player_id", "year_no", "team", "week", "opponent", " completes", "attempts",
"passing_yards", "passing_td", "interceptions", "rushes", "rush_yards", "receptions",
"receptions_yards", "total_td");
insertGame = new SimpleJdbcInsert(getDataSource()).withTableName("GAMES").usingColumns("player_id", "year_no",
"team", "week", "opponent", " completes", "attempts", "passing_yards", "passing_td", "interceptions",
"rushes", "rush_yards", "receptions", "receptions_yards", "total_td");
}
public void write(Game game) {
SqlParameterSource values = new MapSqlParameterSource()
.addValue("player_id", game.getId())
.addValue("year_no", game.getYear())
.addValue("team", game.getTeam())
.addValue("week", game.getWeek())
.addValue("opponent", game.getOpponent())
.addValue("completes", game.getCompletes())
.addValue("attempts", game.getAttempts())
.addValue("passing_yards", game.getPassingYards())
.addValue("passing_td", game.getPassingTd())
.addValue("interceptions", game.getInterceptions())
.addValue("rushes", game.getRushes())
.addValue("rush_yards", game.getRushYards())
.addValue("receptions", game.getReceptions())
.addValue("receptions_yards", game.getReceptionYards())
.addValue("total_td", game.getTotalTd());
this.insertGame.execute(values);
public void write(List<? extends Game> games) {
for (Game game : games) {
SqlParameterSource values = new MapSqlParameterSource().addValue("player_id", game.getId()).addValue(
"year_no", game.getYear()).addValue("team", game.getTeam()).addValue("week", game.getWeek())
.addValue("opponent", game.getOpponent()).addValue("completes", game.getCompletes()).addValue(
"attempts", game.getAttempts()).addValue("passing_yards", game.getPassingYards()).addValue(
"passing_td", game.getPassingTd()).addValue("interceptions", game.getInterceptions())
.addValue("rushes", game.getRushes()).addValue("rush_yards", game.getRushYards()).addValue(
"receptions", game.getReceptions()).addValue("receptions_yards", game.getReceptionYards())
.addValue("total_td", game.getTotalTd());
this.insertGame.execute(values);
}
}

View File

@@ -1,5 +1,7 @@
package org.springframework.batch.sample.domain.football.internal;
import java.util.List;
import org.springframework.batch.item.ClearFailedException;
import org.springframework.batch.item.FlushFailedException;
import org.springframework.batch.item.ItemWriter;
@@ -9,30 +11,27 @@ import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
public class JdbcPlayerSummaryDao extends SimpleJdbcDaoSupport implements ItemWriter<PlayerSummary> {
private static final String INSERT_SUMMARY =
"INSERT into PLAYER_SUMMARY(ID, YEAR_NO, COMPLETES, ATTEMPTS, PASSING_YARDS, PASSING_TD, " +
"INTERCEPTIONS, RUSHES, RUSH_YARDS, RECEPTIONS, RECEPTIONS_YARDS, TOTAL_TD) " +
"values(:id, :year, :completes, :attempts, :passingYards, :passingTd, " +
":interceptions, :rushes, :rushYards, :receptions, :receptionYards, :totalTd)";
public void write(PlayerSummary summary) {
MapSqlParameterSource args = new MapSqlParameterSource()
.addValue("id", summary.getId())
.addValue("year", summary.getYear())
.addValue("completes", summary.getCompletes())
.addValue("attempts", summary.getAttempts())
.addValue("passingYards", summary.getPassingYards())
.addValue("passingTd", summary.getPassingTd())
.addValue("interceptions", summary.getInterceptions())
.addValue("rushes", summary.getRushes())
.addValue("rushYards", summary.getRushYards())
.addValue("receptions", summary.getReceptions())
.addValue("receptionYards", summary.getReceptionYards())
.addValue("totalTd", summary.getTotalTd());
private static final String INSERT_SUMMARY = "INSERT into PLAYER_SUMMARY(ID, YEAR_NO, COMPLETES, ATTEMPTS, PASSING_YARDS, PASSING_TD, "
+ "INTERCEPTIONS, RUSHES, RUSH_YARDS, RECEPTIONS, RECEPTIONS_YARDS, TOTAL_TD) "
+ "values(:id, :year, :completes, :attempts, :passingYards, :passingTd, "
+ ":interceptions, :rushes, :rushYards, :receptions, :receptionYards, :totalTd)";
public void write(List<? extends PlayerSummary> summaries) {
for (PlayerSummary summary : summaries) {
MapSqlParameterSource args = new MapSqlParameterSource().addValue("id", summary.getId()).addValue("year",
summary.getYear()).addValue("completes", summary.getCompletes()).addValue("attempts",
summary.getAttempts()).addValue("passingYards", summary.getPassingYards()).addValue("passingTd",
summary.getPassingTd()).addValue("interceptions", summary.getInterceptions()).addValue("rushes",
summary.getRushes()).addValue("rushYards", summary.getRushYards()).addValue("receptions",
summary.getReceptions()).addValue("receptionYards", summary.getReceptionYards()).addValue(
"totalTd", summary.getTotalTd());
getSimpleJdbcTemplate().update(INSERT_SUMMARY, args);
}
getSimpleJdbcTemplate().update(INSERT_SUMMARY, args);
}
public void close() throws Exception {

View File

@@ -1,5 +1,7 @@
package org.springframework.batch.sample.domain.football.internal;
import java.util.List;
import org.springframework.batch.item.support.AbstractItemWriter;
import org.springframework.batch.sample.domain.football.Player;
import org.springframework.batch.sample.domain.football.PlayerDao;
@@ -7,13 +9,15 @@ import org.springframework.batch.sample.domain.football.PlayerDao;
public class PlayerItemWriter extends AbstractItemWriter<Player> {
private PlayerDao playerDao;
public void write(Player player) throws Exception {
playerDao.savePlayer(player);
public void write(List<? extends Player> players) throws Exception {
for (Player player : players) {
playerDao.savePlayer(player);
}
}
public void setPlayerDao(PlayerDao playerDao) {
this.playerDao = playerDao;
}
}

View File

@@ -16,6 +16,8 @@
package org.springframework.batch.sample.domain.person.internal;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.support.AbstractItemWriter;
@@ -26,10 +28,8 @@ import org.springframework.batch.sample.domain.person.Person;
public class PersonWriter extends AbstractItemWriter<Person> {
private static Log log = LogFactory.getLog(PersonWriter.class);
public void write(Person data) {
public void write(List<? extends Person> data) {
log.debug("Processing: " + data);
}
}

View File

@@ -1,6 +1,7 @@
package org.springframework.batch.sample.domain.trade.internal;
import java.math.BigDecimal;
import java.util.List;
import org.springframework.batch.item.support.AbstractItemWriter;
import org.springframework.batch.sample.domain.trade.CustomerCredit;
@@ -14,9 +15,9 @@ import org.springframework.batch.sample.domain.trade.CustomerCreditDao;
public class CustomerCreditIncreaseWriter extends AbstractItemWriter<CustomerCredit> {
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
@@ -24,13 +25,19 @@ public class CustomerCreditIncreaseWriter extends AbstractItemWriter<CustomerCre
public void setCustomerCreditDao(CustomerCreditDao customerCreditDao) {
this.customerCreditDao = customerCreditDao;
}
/* (non-Javadoc)
* @see org.springframework.batch.item.processor.DelegatingItemWriter#doProcess(java.lang.Object)
/*
* (non-Javadoc)
*
* @see
* org.springframework.batch.item.processor.DelegatingItemWriter#doProcess
* (java.lang.Object)
*/
public void write(CustomerCredit customerCredit) throws Exception {
CustomerCredit result = customerCredit.increaseCreditBy(FIXED_AMOUNT);
customerCreditDao.writeCredit(result);
public void write(List<? extends CustomerCredit> customerCredits) throws Exception {
for (CustomerCredit customerCredit : customerCredits) {
CustomerCredit result = customerCredit.increaseCreditBy(FIXED_AMOUNT);
customerCreditDao.writeCredit(result);
}
}
}

View File

@@ -16,29 +16,31 @@
package org.springframework.batch.sample.domain.trade.internal;
import java.util.List;
import org.springframework.batch.item.support.AbstractItemWriter;
import org.springframework.batch.sample.domain.trade.CustomerCredit;
import org.springframework.batch.sample.domain.trade.CustomerCreditDao;
public class CustomerCreditUpdateWriter extends AbstractItemWriter<CustomerCredit> {
private double creditFilter = 800;
private CustomerCreditDao dao;
private double creditFilter = 800;
public void write(CustomerCredit customerCredit) throws Exception {
private CustomerCreditDao dao;
if (customerCredit.getCredit().doubleValue() > creditFilter) {
dao.writeCredit(customerCredit);
}
}
public void write(List<? extends CustomerCredit> customerCredits) throws Exception {
for (CustomerCredit customerCredit : customerCredits) {
if (customerCredit.getCredit().doubleValue() > creditFilter) {
dao.writeCredit(customerCredit);
}
}
}
public void setCreditFilter(double creditFilter) {
this.creditFilter = creditFilter;
}
public void setCreditFilter(double creditFilter) {
this.creditFilter = creditFilter;
}
public void setDao(CustomerCreditDao dao) {
this.dao = dao;
}
public void setDao(CustomerCreditDao dao) {
this.dao = dao;
}
}

View File

@@ -16,14 +16,16 @@
package org.springframework.batch.sample.domain.trade.internal;
import java.util.List;
import org.springframework.batch.item.support.AbstractItemWriter;
import org.springframework.batch.sample.domain.trade.CustomerDebit;
import org.springframework.batch.sample.domain.trade.CustomerDebitDao;
import org.springframework.batch.sample.domain.trade.Trade;
/**
* Transforms Trade to a CustomerDebit and asks DAO delegate to write the result.
* Transforms Trade to a CustomerDebit and asks DAO delegate to write the
* result.
*
* @author Robert Kasanicky
*/
@@ -31,14 +33,16 @@ public class CustomerUpdateWriter extends AbstractItemWriter<Trade> {
private CustomerDebitDao dao;
public void write(Trade trade) {
CustomerDebit customerDebit = new CustomerDebit();
customerDebit.setName(trade.getCustomer());
customerDebit.setDebit(trade.getPrice());
dao.write(customerDebit);
}
public void write(List<? extends Trade> trades) {
for (Trade trade : trades) {
CustomerDebit customerDebit = new CustomerDebit();
customerDebit.setName(trade.getCustomer());
customerDebit.setDebit(trade.getPrice());
dao.write(customerDebit);
}
}
public void setDao(CustomerDebitDao outputSource) {
this.dao = outputSource;
}
public void setDao(CustomerDebitDao outputSource) {
this.dao = outputSource;
}
}

View File

@@ -16,6 +16,8 @@
package org.springframework.batch.sample.domain.trade.internal;
import java.util.Collections;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
@@ -47,7 +49,7 @@ public class FlatFileCustomerCreditDao implements CustomerCreditDao,
String line = "" + customerCredit.getName() + separator
+ customerCredit.getCredit();
itemWriter.write(line);
itemWriter.write(Collections.singletonList(line));
}
public void setSeparator(String separator) {

View File

@@ -16,6 +16,8 @@
package org.springframework.batch.sample.domain.trade.internal;
import java.util.List;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.item.support.AbstractItemWriter;
@@ -44,14 +46,18 @@ public class TradeWriter extends AbstractItemWriter<Trade> {
this.failure = failure;
}
public void write(Trade trade) {
public void write(List<? extends Trade> trades) {
log.debug(trade);
for (Trade trade : trades) {
dao.writeTrade(trade);
log.debug(trade);
dao.writeTrade(trade);
if (index++ == failure) {
throw new RuntimeException("Something unexpected happened!");
}
if (index++ == failure) {
throw new RuntimeException("Something unexpected happened!");
}
}

View File

@@ -15,6 +15,8 @@
*/
package org.springframework.batch.sample.support;
import java.util.List;
import org.springframework.batch.item.support.AbstractItemWriter;
/**
@@ -23,7 +25,7 @@ import org.springframework.batch.item.support.AbstractItemWriter;
*/
public class DummyItemWriter extends AbstractItemWriter<Object> {
public void write(Object item) throws Exception {
public void write(List<? extends Object> item) throws Exception {
// NO-OP
}

View File

@@ -12,15 +12,16 @@ import org.springframework.batch.item.validator.ValidationException;
public class ItemTrackingItemWriter<T> extends AbstractItemWriter<T> {
private List<T> items = new ArrayList<T>();
private int failure = -1;
private int counter = 0;
public void write(T item) throws Exception {
items.add(item);
if (++counter == failure) {
public void write(List<? extends T> item) throws Exception {
items.addAll(item);
int current = counter;
counter += item.size();
if (current < failure && counter >= failure) {
throw new ValidationException("validation failed");
}
}

View File

@@ -1,26 +1,29 @@
package org.springframework.batch.sample.support;
import java.util.List;
import org.springframework.batch.item.support.AbstractItemWriter;
/**
* Simulates temporary output trouble - requires to
* retry 3 times to pass successfully.
* Simulates temporary output trouble - requires to retry 3 times to pass
* successfully.
*
* @author Robert Kasanicky
*/
public class RetrySampleItemWriter<T> extends AbstractItemWriter<T> {
private int counter = 0;
public void write(T data) throws Exception {
counter++;
if (counter == 2 || counter == 3) {
public void write(List<? extends T> items) throws Exception {
int current = counter;
counter += items.size();
if (current < 3 && (counter >= 2 || counter >= 3)) {
throw new RuntimeException("Temporary error");
}
}
/**
* @return number of times {@link #write(Object)} method was called.
* @return number of times {@link #write(List)} method was called.
*/
public int getCounter() {
return counter;