diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java index 7998cb12a..b99e20971 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/SimpleStepContext.java @@ -21,7 +21,6 @@ import java.util.HashSet; import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.Properties; import java.util.Set; import org.springframework.batch.core.domain.StepExecution; @@ -30,8 +29,6 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.context.SynchronizedAttributeAccessor; -import org.springframework.batch.statistics.StatisticsProvider; -import org.springframework.batch.statistics.StatisticsService; /** * Simple implementation of {@link StepContext}. @@ -47,8 +44,6 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements private StepExecution stepExecution; - private StatisticsService statisticsService; - private StreamManager streamManager; private StreamContext streamContext; @@ -57,24 +52,22 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements * Default constructor. */ public SimpleStepContext(StepExecution stepExecution) { - this(stepExecution, null, null, null); + this(stepExecution, null, null); } /** * Default constructor. */ public SimpleStepContext(StepExecution stepExecution, StepContext parent) { - this(stepExecution, parent, null, null); + this(stepExecution, parent, null); } /** * @param object */ - public SimpleStepContext(StepExecution stepExecution, StepContext parent, StatisticsService statisticsService, - StreamManager streamManager) { + public SimpleStepContext(StepExecution stepExecution, StepContext parent, StreamManager streamManager) { super(); this.parent = parent; - this.statisticsService = statisticsService; this.streamManager = streamManager; this.stepExecution = stepExecution; } @@ -86,9 +79,6 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements */ public void setAttribute(String name, Object value) { super.setAttribute(name, value); - if (statisticsService != null && (value instanceof StatisticsProvider)) { - statisticsService.register(this, (StatisticsProvider) value); - } if (streamManager != null && (value instanceof ItemStream)) { ItemStream stream = (ItemStream) value; streamManager.register(this, stream); @@ -99,17 +89,6 @@ public class SimpleStepContext extends SynchronizedAttributeAccessor implements } } - /* - * (non-Javadoc) - * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics() - */ - public Properties getStatistics() { - if (statisticsService == null) { - return new Properties(); - } - return statisticsService.getStatistics(this); - } - /* * (non-Javadoc) * diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java index a4a8f3ee0..457729b3c 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/scope/StepContext.java @@ -19,7 +19,6 @@ import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamContextProvider; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.core.AttributeAccessor; /** @@ -28,7 +27,7 @@ import org.springframework.core.AttributeAccessor; * @author Dave Syer * */ -public interface StepContext extends AttributeAccessor, StreamContextProvider, StatisticsProvider { +public interface StepContext extends AttributeAccessor, StreamContextProvider { /** * Accessor for the {@link StepExecution} associated with the currently diff --git a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java index f6e1d0a7a..a7b3c9023 100644 --- a/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java +++ b/spring-batch-execution/src/main/java/org/springframework/batch/execution/step/simple/SimpleStepExecutor.java @@ -17,7 +17,6 @@ package org.springframework.batch.execution.step.simple; import java.util.Date; -import java.util.Properties; import org.springframework.batch.core.domain.BatchStatus; import org.springframework.batch.core.domain.Step; @@ -34,6 +33,7 @@ import org.springframework.batch.execution.scope.StepScope; import org.springframework.batch.execution.scope.StepSynchronizationManager; import org.springframework.batch.io.Skippable; import org.springframework.batch.io.exception.BatchCriticalException; +import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.stream.StreamManager; import org.springframework.batch.repeat.ExitStatus; @@ -45,9 +45,6 @@ import org.springframework.batch.repeat.exception.handler.SimpleLimitExceptionHa import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; -import org.springframework.batch.statistics.SimpleStatisticsService; -import org.springframework.batch.statistics.StatisticsProvider; -import org.springframework.batch.statistics.StatisticsService; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.transaction.TransactionStatus; @@ -91,8 +88,6 @@ public class SimpleStepExecutor { // Not for production use... protected PlatformTransactionManager transactionManager = new ResourcelessTransactionManager(); - private StatisticsService statisticsService = new SimpleStatisticsService(); - private Tasklet tasklet; private AbstractStep step; @@ -106,20 +101,6 @@ public class SimpleStepExecutor { this.step = abstractStep; } - /** - * Public setter for the {@link StatisticsService}. This will be used to - * create the {@link StepContext}, and hence any component that is a - * {@link StatisticsProvider} and in step scope will be registered with the - * service. The {@link StepContext} is then a source of aggregate statistics - * for the step. - * - * @param statisticsService the {@link StatisticsService} to set. Default is - * a {@link SimpleStatisticsService}. - */ - public void setStatisticsService(StatisticsService statisticsService) { - this.statisticsService = statisticsService; - } - /** * Public setter for the {@link StreamManager}. This will be used to create * the {@link StepContext}, and hence any component that is a @@ -198,8 +179,7 @@ public class SimpleStepExecutor { ExitStatus status = ExitStatus.FAILED; StepContext parentStepContext = StepSynchronizationManager.getContext(); - final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, statisticsService, - streamManager); + final StepContext stepContext = new SimpleStepContext(stepExecution, parentStepContext, streamManager); StepSynchronizationManager.register(stepContext); // Add the job identifier so that it can be used to identify // the conversation in StepScope @@ -246,8 +226,8 @@ public class SimpleStepExecutor { // TODO: check that stepExecution can // aggregate these contributions if they // come in asynchronously. - Properties statistics = stepContext.getStatistics(); - contribution.setStatistics(statistics); + StreamContext statistics = stepContext.getStreamContext(); + contribution.setStatistics(statistics.getProperties()); contribution.incrementCommitCount(); // Apply the contribution to the step // only if chunk was successful diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java index 80afe4825..9c5fc2635 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/scope/SimpleStepContextTests.java @@ -19,7 +19,6 @@ import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Properties; import junit.framework.TestCase; @@ -27,9 +26,8 @@ import org.springframework.batch.core.domain.StepExecution; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; +import org.springframework.batch.item.stream.ItemStreamAdapter; import org.springframework.batch.item.stream.StreamManager; -import org.springframework.batch.statistics.StatisticsProvider; -import org.springframework.batch.statistics.StatisticsService; import org.springframework.batch.support.PropertiesConverter; /** @@ -136,47 +134,23 @@ public class SimpleStepContextTests extends TestCase { assertTrue(list.contains("spam")); } - public void testStatisticsWithNullService() throws Exception { - assertEquals(0, context.getStatistics().size()); - } - public void testStatisticsWithNotNullService() throws Exception { Map map = new HashMap(); - context = new SimpleStepContext(null, null, new StubStatisticsService(map), new StubStreamManager(map)); - assertEquals(1, context.getStatistics().size()); - assertEquals("bar", context.getStatistics().getProperty("foo")); + context = new SimpleStepContext(null, null, new StubStreamManager(map)); + assertEquals(1, context.getStreamContext().getProperties().size()); + assertEquals("bar", context.getStreamContext().getProperties().getProperty("foo")); } public void testStatisticsServiceRegistration() throws Exception { Map map = new HashMap(); - context = new SimpleStepContext(null, null, new StubStatisticsService(map), new StubStreamManager(map)); - StubStatisticsProvider provider = new StubStatisticsProvider(); + context = new SimpleStepContext(null, null, new StubStreamManager(map)); + ItemStreamAdapter provider = new ItemStreamAdapter(); context.setAttribute("foo", provider); assertEquals(1, map.size()); assertEquals(context, map.keySet().iterator().next()); assertEquals(provider, map.values().iterator().next()); } - /** - * @author Dave Syer - * - */ - private class StubStatisticsService implements StatisticsService { - private final Map map; - - private StubStatisticsService(Map map) { - this.map = map; - } - - public Properties getStatistics(Object key) { - return PropertiesConverter.stringToProperties("foo=bar"); - } - - public void register(Object key, StatisticsProvider provider) { - map.put(key, provider); - } - } - /** * @author Dave Syer * @@ -206,14 +180,4 @@ public class SimpleStepContextTests extends TestCase { } } - /** - * @author Dave Syer - * - */ - private class StubStatisticsProvider implements StatisticsProvider { - public Properties getStatistics() { - return null; - } - } - } diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java index e555b9e08..6a4b3cf2b 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/step/simple/SimpleStepExecutorTests.java @@ -21,7 +21,6 @@ import java.util.Arrays; import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.Properties; import junit.framework.TestCase; @@ -46,6 +45,7 @@ import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.StreamException; import org.springframework.batch.item.reader.ListItemReader; import org.springframework.batch.item.stream.GenericStreamContext; +import org.springframework.batch.item.stream.SimpleStreamManager; import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.repeat.RepeatContext; @@ -54,8 +54,6 @@ import org.springframework.batch.repeat.exception.handler.ExceptionHandler; import org.springframework.batch.repeat.interceptor.RepeatInterceptorAdapter; import org.springframework.batch.repeat.policy.SimpleCompletionPolicy; import org.springframework.batch.repeat.support.RepeatTemplate; -import org.springframework.batch.statistics.StatisticsProvider; -import org.springframework.batch.statistics.StatisticsService; import org.springframework.batch.support.PropertiesConverter; import org.springframework.batch.support.transaction.ResourcelessTransactionManager; @@ -399,7 +397,7 @@ public class SimpleStepExecutorTests extends TestCase { assertEquals(1, list.size()); } - public void testStatisticsService() throws Exception { + public void testStreamManager() throws Exception { StepInstance step = new StepInstance(new Long(1)); step.setStepExecutionCount(1); stepConfiguration.setTasklet(new Tasklet() { @@ -414,22 +412,14 @@ public class SimpleStepExecutorTests extends TestCase { assertEquals(null, stepExecution.getStatistics().getProperty("foo")); final Map map = new HashMap(); - stepExecutor.setStatisticsService(new StatisticsService() { - public Properties getStatistics(Object key) { - return PropertiesConverter.stringToProperties("foo=bar"); - } - - public void register(Object key, StatisticsProvider provider) { - map.put(key, provider); + stepExecutor.setStreamManager(new SimpleStreamManager() { + public StreamContext getStreamContext(Object key) { + // TODO Auto-generated method stub + return new GenericStreamContext(PropertiesConverter.stringToProperties("foo=bar")); } }); - try { - stepExecutor.execute(stepExecution); - } - catch (Throwable t) { - fail(); - } + stepExecutor.execute(stepExecution); // At least once in that process the statistics service was asked for // statistics... diff --git a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java index 1ac0bbeb1..2712ffbfe 100644 --- a/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java +++ b/spring-batch-execution/src/test/java/org/springframework/batch/execution/tasklet/ItemOrientedTaskletTests.java @@ -34,7 +34,6 @@ import org.springframework.batch.item.writer.AbstractItemWriter; import org.springframework.batch.repeat.context.RepeatContextSupport; import org.springframework.batch.repeat.synch.RepeatSynchronizationManager; import org.springframework.batch.retry.policy.SimpleRetryPolicy; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.batch.support.PropertiesConverter; /** @@ -135,7 +134,8 @@ public class ItemOrientedTaskletTests extends TestCase { try { module.execute(); fail("RuntimeException was expected"); - } catch (RuntimeException bce) { + } + catch (RuntimeException bce) { // expected assertEquals("foo", bce.getMessage()); } @@ -144,7 +144,8 @@ public class ItemOrientedTaskletTests extends TestCase { public void testNotSkippable() throws Exception { try { module.skip(); - } catch (Exception e) { + } + catch (Exception e) { // Unexpected throw e; } @@ -183,7 +184,7 @@ public class ItemOrientedTaskletTests extends TestCase { return "bar"; } - public void close() throws StreamException { + public void close() throws StreamException { } }); @@ -198,7 +199,8 @@ public class ItemOrientedTaskletTests extends TestCase { try { module.execute(); fail("Expected RuntimeException"); - } catch (RuntimeException e) { + } + catch (RuntimeException e) { assertEquals("FOO", e.getMessage()); } @@ -206,8 +208,7 @@ public class ItemOrientedTaskletTests extends TestCase { // verify method calls assertEquals(1, list.size()); - assertEquals("The item was not passed in to recover method", "bar", - list.get(0)); + assertEquals("The item was not passed in to recover method", "bar", list.get(0)); } public void testRetryPolicy() throws Exception { @@ -215,7 +216,7 @@ public class ItemOrientedTaskletTests extends TestCase { module.setItemRecoverer(new ItemRecoverer() { public boolean recover(Object item, Throwable cause) { assertEquals("FOO", cause.getMessage()); - list.add(item+"_recovered"); + list.add(item + "_recovered"); return true; } }); @@ -232,7 +233,8 @@ public class ItemOrientedTaskletTests extends TestCase { try { module.execute(); fail("Expected RuntimeException"); - } catch (RuntimeException e) { + } + catch (RuntimeException e) { assertEquals("FOO", e.getMessage()); } @@ -243,15 +245,15 @@ public class ItemOrientedTaskletTests extends TestCase { // verify method calls assertEquals(1, list.size()); - assertEquals("The item was not passed in to recover method", - "foo_recovered", list.get(0)); + assertEquals("The item was not passed in to recover method", "foo_recovered", list.get(0)); } public void testInitialisationWithNullProvider() throws Exception { module.setItemReader(null); try { module.afterPropertiesSet(); - } catch (IllegalArgumentException e) { + } + catch (IllegalArgumentException e) { assertTrue(e.getMessage().toLowerCase().indexOf("reader") >= 0); } } @@ -260,8 +262,10 @@ public class ItemOrientedTaskletTests extends TestCase { module.setItemWriter(null); try { module.afterPropertiesSet(); - } catch (IllegalArgumentException e) { - assertTrue("Message did not contain writer: "+e.getMessage(), e.getMessage().toLowerCase().indexOf("writer") >= 0); + } + catch (IllegalArgumentException e) { + assertTrue("Message did not contain writer: " + e.getMessage(), e.getMessage().toLowerCase().indexOf( + "writer") >= 0); } } @@ -269,31 +273,34 @@ public class ItemOrientedTaskletTests extends TestCase { public Object read() throws Exception { return "foo"; } + public Object getKey(Object item) { return item; } } - private class SkippableItemReader implements KeyedItemReader, - Skippable, StatisticsProvider { + private class SkippableItemReader implements KeyedItemReader, Skippable { public Object read() throws Exception { return itemProvider.read(); } + public Object getKey(Object item) { return item; } + public void skip() { list.add("provider"); } + public Properties getStatistics() { return PropertiesConverter.stringToProperties("foo=bar"); } - public void close() throws StreamException { + + public void close() throws StreamException { } } - private class SkippableItemWriter implements ItemWriter, Skippable, - StatisticsProvider { + private class SkippableItemWriter implements ItemWriter, Skippable { String props = "foo=bar"; public SkippableItemWriter() { diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java index b63c67822..3ff704665 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/cursor/JdbcCursorItemReader.java @@ -23,7 +23,6 @@ import java.sql.SQLWarning; import java.sql.Statement; import java.util.ArrayList; import java.util.List; -import java.util.Properties; import javax.sql.DataSource; @@ -35,7 +34,6 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.KeyedItemReader; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.dao.DataAccessException; @@ -115,9 +113,8 @@ import org.springframework.util.StringUtils; * @author Lucas Ward * @author Peter Zozom */ -public class JdbcCursorItemReader extends AbstractTransactionalIoSource - implements KeyedItemReader, DisposableBean, - InitializingBean, ItemStream, StatisticsProvider, Skippable { +public class JdbcCursorItemReader extends AbstractTransactionalIoSource implements KeyedItemReader, DisposableBean, + InitializingBean, ItemStream, Skippable { private static Log log = LogFactory.getLog(JdbcCursorItemReader.class); @@ -156,9 +153,9 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource private SQLExceptionTranslator exceptionTranslator; /* Current count of processed records. */ - private int currentProcessedRow = 0; + private long currentProcessedRow = 0; - private int lastCommittedRow = 0; + private long lastCommittedRow = 0; private RowMapper mapper; @@ -167,8 +164,8 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource /** * Assert that mandatory properties are set. * - * @throws IllegalArgumentException - * if either data source or sql properties not set. + * @throws IllegalArgumentException if either data source or sql properties + * not set. */ public void afterPropertiesSet() throws Exception { Assert.notNull(dataSource, "DataSOurce must be provided"); @@ -193,8 +190,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * * @returns Object returned by RowMapper * @throws DataAccessException - * @throws IllegalStateExceptino - * if mapper is null. + * @throws IllegalStateExceptino if mapper is null. */ public Object read() { @@ -207,12 +203,12 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource try { if (!rs.next()) { return null; - } else { + } + else { currentProcessedRow++; if (!skippedRows.isEmpty()) { // while is necessary to handle successive skips. - while (skippedRows - .contains(new Integer(currentProcessedRow))) { + while (skippedRows.contains(new Long(currentProcessedRow))) { if (!rs.next()) { return null; } @@ -220,20 +216,20 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource } } - Object mappedResult = mapper.mapRow(rs, currentProcessedRow); + Object mappedResult = mapper.mapRow(rs, (int)currentProcessedRow); verifyCursorPosition(currentProcessedRow); return mappedResult; } - } catch (SQLException se) { - throw getExceptionTranslator().translate( - "Trying to process next row", sql, se); + } + catch (SQLException se) { + throw getExceptionTranslator().translate("Trying to process next row", sql, se); } } - public int getCurrentProcessedRow() { + public long getCurrentProcessedRow() { return currentProcessedRow; } @@ -255,15 +251,15 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource try { currentProcessedRow = lastCommittedRow; if (currentProcessedRow > 0) { - rs.absolute(currentProcessedRow); - } else { + rs.absolute((int)currentProcessedRow); + } + else { rs.beforeFirst(); } - } catch (SQLException se) { - throw getExceptionTranslator().translate( - "Attempted to move ResultSet to last committed row", sql, - se); + } + catch (SQLException se) { + throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", sql, se); } } @@ -297,12 +293,10 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource // Check the result set is in synch with the currentRow attribute. This is // important // to ensure that the user hasn't modified the current row. - private void verifyCursorPosition(int expectedCurrentRow) - throws SQLException { + private void verifyCursorPosition(long expectedCurrentRow) throws SQLException { if (verifyCursorPosition) { if (expectedCurrentRow != this.rs.getRow()) { - throw new InvalidDataAccessResourceUsageException( - "Unexpected cursor position change."); + throw new InvalidDataAccessResourceUsageException("Unexpected cursor position change."); } } } @@ -320,17 +314,15 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource try { this.con = dataSource.getConnection(); - this.stmt = this.con.createStatement( - ResultSet.TYPE_SCROLL_INSENSITIVE, - ResultSet.CONCUR_READ_ONLY, + this.stmt = this.con.createStatement(ResultSet.TYPE_SCROLL_INSENSITIVE, ResultSet.CONCUR_READ_ONLY, ResultSet.HOLD_CURSORS_OVER_COMMIT); applyStatementSettings(this.stmt); this.rs = this.stmt.executeQuery(sql); handleWarnings(this.stmt.getWarnings()); - } catch (SQLException se) { + } + catch (SQLException se) { close(); - throw getExceptionTranslator() - .translate("Executing query", sql, se); + throw getExceptionTranslator().translate("Executing query", sql, se); } super.registerSynchronization(); @@ -367,9 +359,9 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource protected SQLExceptionTranslator getExceptionTranslator() { if (exceptionTranslator == null) { if (dataSource != null) { - exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator( - dataSource); - } else { + exceptionTranslator = new SQLErrorCodeSQLExceptionTranslator(dataSource); + } + else { exceptionTranslator = new SQLStateSQLExceptionTranslator(); } } @@ -389,13 +381,12 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource if (ignoreWarnings) { SQLWarning warningToLog = warnings; while (warningToLog != null) { - log.debug("SQLWarning ignored: SQL state '" - + warningToLog.getSQLState() + "', error code '" - + warningToLog.getErrorCode() + "', message [" - + warningToLog.getMessage() + "]"); + log.debug("SQLWarning ignored: SQL state '" + warningToLog.getSQLState() + "', error code '" + + warningToLog.getErrorCode() + "', message [" + warningToLog.getMessage() + "]"); warningToLog = warningToLog.getNextWarning(); } - } else if (warnings != null) { + } + else if (warnings != null) { throw new SQLWarningException("Warning not ignored", warnings); } } @@ -406,12 +397,12 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * @see org.springframework.batch.restart.Restartable#getRestartData() */ public StreamContext getStreamContext() { - StreamContext streamContext = new StreamContext(); - String skipped = skippedRows.toString(); - Properties statistics = getStatistics(); - statistics.setProperty(SKIPPED_ROWS, skipped.substring(1,skipped.length()-1)); - return new GenericStreamContext(statistics); + StreamContext context = new GenericStreamContext(); + context.putString(SKIPPED_ROWS, skipped.substring(1, skipped.length() - 1)); + context.putLong(CURRENT_PROCESSED_ROW, currentProcessedRow); + context.putLong(SKIP_COUNT, skipCount); + return context; } /* @@ -427,18 +418,17 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource open(); - // Properties restartProperties = data.getProperties(); + // Properties restartProperties = data.getProperties(); if (!data.containsKey(CURRENT_PROCESSED_ROW)) { return; } try { - this.currentProcessedRow = new Long(data.getLong(CURRENT_PROCESSED_ROW)).intValue(); - rs.absolute(currentProcessedRow); - } catch (SQLException se) { - throw getExceptionTranslator().translate( - "Attempted to move ResultSet to last committed row", sql, - se); + this.currentProcessedRow = data.getLong(CURRENT_PROCESSED_ROW); + rs.absolute((int)currentProcessedRow); + } + catch (SQLException se) { + throw getExceptionTranslator().translate("Attempted to move ResultSet to last committed row", sql, se); } if (!data.containsKey(SKIPPED_ROWS)) { @@ -447,24 +437,10 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource String[] skipped = StringUtils.commaDelimitedListToStringArray(data.getString(SKIPPED_ROWS)); for (int i = 0; i < skipped.length; i++) { - this.skippedRows.add(new Integer(skipped[i])); + this.skippedRows.add(new Long(skipped[i])); } } - /* - * (non-Javadoc) - * - * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics() - */ - public Properties getStatistics() { - - Properties props = new Properties(); - props.setProperty(CURRENT_PROCESSED_ROW, new Integer( - currentProcessedRow).toString()); - props.setProperty(SKIP_COUNT, new Integer(skipCount).toString()); - return props; - } - /** * Skip the current row. If the transaction is rolled back, this row will * not be represented to the RowMapper when read() is called. For example, @@ -472,7 +448,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * continue processing and find */ public void skip() { - skippedRows.add(new Integer(currentProcessedRow)); + skippedRows.add(new Long(currentProcessedRow)); skipCount++; } @@ -482,8 +458,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * ResultSet object. If the fetch size specified is zero, the * JDBC driver ignores the value. * - * @param fetchSize - * the number of rows to fetch + * @param fetchSize the number of rows to fetch * @see ResultSet#setFetchSize(int) */ public void setFetchSize(int fetchSize) { @@ -494,8 +469,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * Sets the limit for the maximum number of rows that any * ResultSet object can contain to the given number. * - * @param maxRows - * the new max rows limit; zero means there is no limit + * @param maxRows the new max rows limit; zero means there is no limit * @see Statement#setMaxRows(int) */ public void setMaxRows(int maxRows) { @@ -508,9 +482,8 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * seconds. If the limit is exceeded, an SQLException is * thrown. * - * @param queryTimeout - * seconds the new query timeout limit in seconds; zero means - * there is no limit + * @param queryTimeout seconds the new query timeout limit in seconds; zero + * means there is no limit * @see Statement#setQueryTimeout(int) */ public void setQueryTimeout(int queryTimeout) { @@ -521,8 +494,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * Set whether SQLWarnings should be ignored (only logged) or exception * should be thrown. * - * @param ignoreWarnings - * if TRUE, warnings are ignored + * @param ignoreWarnings if TRUE, warnings are ignored */ public void setIgnoreWarnings(boolean ignoreWarnings) { this.ignoreWarnings = ignoreWarnings; @@ -532,8 +504,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource * Allow verification of cursor position after current row is processed by * RowMapper or RowCallbackHandler. Default value is TRUE. * - * @param verifyCursorPosition - * if true, cursor position is verified + * @param verifyCursorPosition if true, cursor position is verified */ public void setVerifyCursorPosition(boolean verifyCursorPosition) { this.verifyCursorPosition = verifyCursorPosition; @@ -565,7 +536,7 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource initialized = true; } - + /** * Return the item itself (which is already a key). * @see org.springframework.batch.item.ItemReader#getKey(java.lang.Object) @@ -573,5 +544,5 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource public Object getKey(Object item) { return item; } - + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapStreamContextRowMapper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapStreamContextRowMapper.java index 40aada16b..40a398331 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapStreamContextRowMapper.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/ColumnMapStreamContextRowMapper.java @@ -37,14 +37,14 @@ import org.springframework.util.ClassUtils; */ public class ColumnMapStreamContextRowMapper extends ColumnMapRowMapper implements RestartDataRowMapper{ - static final String KEY = ClassUtils.getQualifiedName(ColumnMapStreamContextRowMapper.class) + ".KEY."; + public static final String KEY_PREFIX = ClassUtils.getQualifiedName(ColumnMapStreamContextRowMapper.class) + ".KEY."; public PreparedStatementSetter createSetter(StreamContext streamContext) { ColumnMapRestartData columnData = new ColumnMapRestartData(streamContext.getProperties()); List columns = new ArrayList(); - for (Iterator iterator = columnData.entrySet().iterator(); iterator.hasNext();) { + for (Iterator iterator = columnData.keys.entrySet().iterator(); iterator.hasNext();) { Entry entry = (Entry)iterator.next(); Object column = entry.getValue(); columns.add(column); @@ -53,27 +53,51 @@ public class ColumnMapStreamContextRowMapper extends ColumnMapRowMapper implemen return new ArgPreparedStatementSetter(columns.toArray()); } - public StreamContext createRestartData(Object key) { - - Assert.isInstanceOf(Map.class, key, "Key must be of type Map."); + public StreamContext createStreamContext(Object key) { + Assert.isInstanceOf(Map.class, key, "Input to create StreamContext must be of type Map."); Map keys = (Map)key; - return new ColumnMapRestartData(keys); } - private static class ColumnMapRestartData extends GenericStreamContext{ + private static class ColumnMapRestartData extends GenericStreamContext { + private final Map keys; + public ColumnMapRestartData(Map keys) { - super(); - for(Iterator it = keys.entrySet().iterator();it.hasNext();){ - Entry entry = (Entry)it.next(); - putString(entry.getKey().toString(), entry.getValue().toString()); - } + this.keys = keys; } public ColumnMapRestartData(Properties props) { - super(props); + + keys = CollectionFactory.createLinkedCaseInsensitiveMapIfPossible(props.size()); + + for(int counter = 0; counter < props.size(); counter++){ + + String key = KEY_PREFIX + counter; + String column = props.getProperty(key); + + if(column != null){ + keys.put(key, column); + } + else{ + break; + } + + } + } + + public Properties getProperties() { + Properties props = new Properties(); + + int counter = 0; + for (Iterator iterator = keys.entrySet().iterator(); iterator.hasNext();) { + Entry entry = (Entry) iterator.next(); + props.setProperty(KEY_PREFIX + counter, entry.getValue().toString()); + counter++; + } + + return props; } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java index c720b4cb3..a271fb5a8 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGenerator.java @@ -98,7 +98,7 @@ public class MultipleColumnJdbcKeyGenerator implements */ public StreamContext getKeyAsStreamContext(Object key) { Assert.state(keyMapper != null, "RestartDataConverter must not be null."); - return keyMapper.createRestartData(key); + return keyMapper.createStreamContext(key); } /** diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/RestartDataRowMapper.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/RestartDataRowMapper.java index a47f58358..091e7b997 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/RestartDataRowMapper.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/driving/support/RestartDataRowMapper.java @@ -40,7 +40,7 @@ public interface RestartDataRowMapper extends RowMapper { * @return ResartData representing the composite key. * @throws IllegalArgumentException if key is null or of an unsupported type. */ - public StreamContext createRestartData(Object key); + public StreamContext createStreamContext(Object key); /** * Given the provided restart data, return a PreparedStatementSeter that can diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java index c9267c50b..2813709f6 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/DefaultFlatFileItemReader.java @@ -26,9 +26,9 @@ import org.springframework.batch.io.Skippable; import org.springframework.batch.io.file.separator.LineReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamException; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationAdapter; @@ -42,8 +42,7 @@ import org.springframework.transaction.support.TransactionSynchronizationAdapter * @author Tomas Slanina * @author Robert Kasanicky */ -public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implements Skippable, ItemStream, - StatisticsProvider { +public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implements Skippable, ItemStream { private static Log log = LogFactory.getLog(DefaultFlatFileItemReader.class); @@ -113,8 +112,10 @@ public class DefaultFlatFileItemReader extends SimpleFlatFileItemReader implemen * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics() */ public Properties getStatistics() { - LineReader is = getReader(); - statistics.setProperty(READ_STATISTICS_NAME, String.valueOf(is.getCurrentLineCount())); + if (reader==null) { + throw new StreamException("ItemStream not open or already closed."); + } + statistics.setProperty(READ_STATISTICS_NAME, String.valueOf(reader.getCurrentLineCount())); return statistics; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java index 17f46dfe4..d5ae418c2 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/FlatFileItemWriter.java @@ -36,7 +36,6 @@ import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.item.writer.ItemTransformer; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; @@ -59,9 +58,8 @@ import org.springframework.util.Assert; * @author Robert Kasanicky * @author Dave Syer */ -public class FlatFileItemWriter extends AbstractTransactionalIoSource implements - ItemWriter, ItemStream, StatisticsProvider, InitializingBean, - DisposableBean { +public class FlatFileItemWriter extends AbstractTransactionalIoSource implements ItemWriter, ItemStream, + InitializingBean, DisposableBean { private static final String LINE_SEPARATOR = System.getProperty("line.separator"); @@ -88,7 +86,7 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements private static class BooleanHolder { public boolean value; } - + /** * Assert that mandatory properties (resource) are set. * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet() @@ -119,7 +117,7 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements } /** - * Commit the transaction. + * Commit the transaction. */ protected void transactionCommitted() { getOutputState().mark(); @@ -159,7 +157,7 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements * Convert the date to a format that can be output and then write it out. * @param data * @param converted - * @throws Exception + * @throws Exception */ private void transformAndWrite(Object data, BooleanHolder converted) throws Exception { @@ -262,7 +260,7 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements public StreamContext getStreamContext() { final OutputState os = getOutputState(); - streamContext.getProperties().setProperty(RESTART_DATA_NAME, String.valueOf(os.position())); + streamContext.putString(RESTART_DATA_NAME, String.valueOf(os.position())); return streamContext; } @@ -453,7 +451,7 @@ public class FlatFileItemWriter extends AbstractTransactionalIoSource implements } } String parent = file.getParent(); - if (parent!=null) { + if (parent != null) { new File(parent).mkdirs(); } file.createNewFile(); diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java index 9f0ec041e..8ef2fef8d 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/file/SimpleFlatFileItemReader.java @@ -68,7 +68,7 @@ public class SimpleFlatFileItemReader extends AbstractItemReader implements Item * Encapsulates the state of the input source. If it is null then we are * uninitialized. */ - private LineReader reader; + protected LineReader reader; private RecordSeparatorPolicy recordSeparatorPolicy; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java index 1710cecd7..4c0703697 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemReader.java @@ -23,7 +23,6 @@ import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.reader.AbstractItemReader; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; @@ -43,7 +42,7 @@ import org.springframework.util.Assert; * @author Robert Kasanicky */ public class StaxEventItemReader extends AbstractItemReader implements ItemReader, - Skippable, ItemStream, StatisticsProvider, InitializingBean, DisposableBean { + Skippable, ItemStream, InitializingBean, DisposableBean { public static final String READ_COUNT_STATISTICS_NAME = "StaxEventReaderItemReader.readCount"; diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java index b19abb083..dcffd7b7b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/io/xml/StaxEventItemWriter.java @@ -21,7 +21,6 @@ import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.repeat.synch.BatchTransactionSynchronizationManager; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.Resource; @@ -32,17 +31,16 @@ import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; /** - * An implementation of {@link ItemWriter} which uses - * StAX and {@link EventWriterSerializer} for serializing object to XML. - * - * This output source also provides restart, statistics and transaction - * features by implementing corresponding interfaces. - * + * An implementation of {@link ItemWriter} which uses StAX and + * {@link EventWriterSerializer} for serializing object to XML. + * + * This output source also provides restart, statistics and transaction features + * by implementing corresponding interfaces. + * * @author Peter Zozom - * + * */ -public class StaxEventItemWriter implements ItemWriter, ItemStream, - StatisticsProvider, InitializingBean, DisposableBean { +public class StaxEventItemWriter implements ItemWriter, ItemStream, InitializingBean, DisposableBean { // default encoding private static final String DEFAULT_ENCODING = "UTF-8"; @@ -83,13 +81,15 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, // signalizes that marshalling was restarted private boolean restarted = false; - // TRUE means, that output file will be overwritten if exists - default is TRUE + // TRUE means, that output file will be overwritten if exists - default is + // TRUE private boolean overwriteOutput = true; // file channel private FileChannel channel; - // wrapper for XML event writer that swallows StartDocument and EndDocument events + // wrapper for XML event writer that swallows StartDocument and EndDocument + // events private XMLEventWriter eventWriter; // XML event writer @@ -107,10 +107,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, // current count of processed records private long currentRecordCount = 0; - /** * Set output file. - * + * * @param resource the output file */ public void setResource(Resource resource) { @@ -119,7 +118,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Set Object to XML serializer. - * + * * @param serializer the Object to XML serializer */ public void setSerializer(EventWriterSerializer serializer) { @@ -128,7 +127,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Get used encoding. - * + * * @return the encoding used */ public String getEncoding() { @@ -137,7 +136,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Set encoding to be used for output file. - * + * * @param encoding the encoding to be used */ public void setEncoding(String encoding) { @@ -146,7 +145,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Get XML version. - * + * * @return the XML version used */ public String getVersion() { @@ -155,7 +154,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Set XML version to be used for output XML. - * + * * @param version the XML version to be used */ public void setVersion(String version) { @@ -164,7 +163,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Get the tag name of the root element. - * + * * @return the root element tag name */ public String getRootTagName() { @@ -172,8 +171,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, } /** - * Set the tag name of the root element. If not set, default name is used ("root"). - * + * Set the tag name of the root element. If not set, default name is used + * ("root"). + * * @param rootTagName the tag name to be used for the root element */ public void setRootTagName(String rootTagName) { @@ -182,7 +182,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Get attributes of the root element. - * + * * @return attributes of the root element */ public Map getRootElementAttributes() { @@ -191,7 +191,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Set the root element attributes to be written. - * + * * @param rootElementAttributes attributes of the root element */ public void setRootElementAttributes(Map rootElementAttributes) { @@ -199,8 +199,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, } /** - * Set "overwrite" flag for the output file. Flag is ignored when output file processing is restarted. - * + * Set "overwrite" flag for the output file. Flag is ignored when output + * file processing is restarted. + * * @param shouldDeleteIfExists */ public void setOverwriteOutput(boolean overwriteOutput) { @@ -237,7 +238,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Open the output source - * + * * @see org.springframework.batch.item.ResourceLifecycle#open() */ public void open() { @@ -253,16 +254,16 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, File file; FileOutputStream os = null; - + try { file = resource.getFile(); FileUtils.setUpOutputFile(file, restarted, overwriteOutput); os = new FileOutputStream(file, true); channel = os.getChannel(); setPosition(position); - } catch (IOException ioe) { - throw new DataAccessResourceFailureException( - "Unable to write to file resource: [" + resource + "]", ioe); + } + catch (IOException ioe) { + throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", ioe); } XMLOutputFactory outputFactory = XMLOutputFactory.newInstance(); @@ -273,9 +274,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, if (!restarted) { startDocument(delegateEventWriter); } - } catch (XMLStreamException xse) { - throw new DataAccessResourceFailureException( - "Unable to write to file resource: [" + resource + "]", xse); + } + catch (XMLStreamException xse) { + throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", xse); } initialized = true; @@ -289,27 +290,26 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, * * If this is not sufficient for you, simply override this method. Encoding, * version and root tag name can be retrieved with corresponding getters. - * - * @param writer - * XML event writer + * + * @param writer XML event writer * @throws XMLStreamException */ protected void startDocument(XMLEventWriter writer) throws XMLStreamException { XMLEventFactory factory = XMLEventFactory.newInstance(); - //write start document + // write start document writer.add(factory.createStartDocument(getEncoding(), getVersion())); - //write root tag + // write root tag writer.add(factory.createStartElement("", "", getRootTagName())); - //write root tag attributes + // write root tag attributes if (!CollectionUtils.isEmpty(getRootElementAttributes())) { for (Iterator i = getRootElementAttributes().entrySet().iterator(); i.hasNext();) { - Map.Entry entry = (Map.Entry)i.next(); - writer.add(factory.createAttribute((String)entry.getKey(), (String)entry.getValue())); + Map.Entry entry = (Map.Entry) i.next(); + writer.add(factory.createAttribute((String) entry.getKey(), (String) entry.getValue())); } } @@ -319,29 +319,27 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /** * Finishes the XML document. It closes any start tag and writes * corresponding end tags. - * - * @param writer - * XML event writer + * + * @param writer XML event writer * @throws XMLStreamException */ - protected void endDocument(XMLEventWriter writer) - throws XMLStreamException { + protected void endDocument(XMLEventWriter writer) throws XMLStreamException { - //writer.writeEndDocument(); <- this doesn't work after restart - //we need to write end tag of the root element manually + // writer.writeEndDocument(); <- this doesn't work after restart + // we need to write end tag of the root element manually writer.flush(); ByteBuffer bbuf = ByteBuffer.wrap(("").getBytes()); try { getChannel().write(bbuf); - } catch (IOException ioe) { - throw new DataAccessResourceFailureException( - "Unable to close file resource: [" + resource + "]", ioe); + } + catch (IOException ioe) { + throw new DataAccessResourceFailureException("Unable to close file resource: [" + resource + "]", ioe); } } /** * Close the output source. - * + * * @see org.springframework.batch.item.ResourceLifecycle#close() */ public void close() { @@ -350,19 +348,18 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, endDocument(delegateEventWriter); eventWriter.close(); channel.close(); - } catch (XMLStreamException xse) { - throw new DataAccessResourceFailureException( - "Unable to close file resource: [" + resource + "]", xse); + } + catch (XMLStreamException xse) { + throw new DataAccessResourceFailureException("Unable to close file resource: [" + resource + "]", xse); } catch (IOException ioe) { - throw new DataAccessResourceFailureException( - "Unable to close file resource: [" + resource + "]", ioe); + throw new DataAccessResourceFailureException("Unable to close file resource: [" + resource + "]", ioe); } } /** * Write the value object to XML stream. - * + * * @param output the value object * @see org.springframework.batch.item.ItemWriter#write(java.lang.Object) */ @@ -399,12 +396,10 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, long startAtPosition = 0; - //if restart data is provided, restart from provided offset - //otherwise start from beginning - if (data != null && data.getProperties() != null - && data.getProperties().getProperty(RESTART_DATA_NAME) != null) { - startAtPosition = Long.parseLong(data.getProperties().getProperty( - RESTART_DATA_NAME)); + // if restart data is provided, restart from provided offset + // otherwise start from beginning + if (data != null && data.getProperties() != null && data.getProperties().getProperty(RESTART_DATA_NAME) != null) { + startAtPosition = Long.parseLong(data.getProperties().getProperty(RESTART_DATA_NAME)); restarted = true; } @@ -426,9 +421,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, } /* - * Get the actual position in file channel. - * This method flushes any buffered data before position is read. - * + * Get the actual position in file channel. This method flushes any buffered + * data before position is read. + * * @return byte offset in file channel */ private long getPosition() { @@ -438,9 +433,9 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, try { eventWriter.flush(); position = channel.position(); - } catch (Exception e) { - throw new DataAccessResourceFailureException( - "Unable to write to file resource: [" + resource + "]", e); + } + catch (Exception e) { + throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", e); } return position; @@ -448,7 +443,7 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, /* * Set the file channel position. - * + * * @param newPosition new file channel position */ private void setPosition(long newPosition) { @@ -458,24 +453,23 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, "Current file size is smaller than size at last commit"); channel.truncate(newPosition); channel.position(newPosition); - } catch (IOException e) { - throw new DataAccessResourceFailureException( - "Unable to write to file resource: [" + resource + "]", e); } - + catch (IOException e) { + throw new DataAccessResourceFailureException("Unable to write to file resource: [" + resource + "]", e); + } } /** * Encapsulates transaction events for the StaxEventWriterOutputSource. */ - private class StaxEventWriterItemWriterTransactionSychronization extends - TransactionSynchronizationAdapter { + private class StaxEventWriterItemWriterTransactionSychronization extends TransactionSynchronizationAdapter { public void afterCompletion(int status) { if (status == TransactionSynchronization.STATUS_COMMITTED) { transactionComitted(); - } else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { + } + else if (status == TransactionSynchronization.STATUS_ROLLED_BACK) { transactionRolledback(); } } @@ -488,10 +482,11 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, private void transactionRolledback() { currentRecordCount = lastCommitPointRecordCount; - //close output + // close output close(); - //and reopen it - we do this because we need to reopen stream - //reader at specified position - calling setPosition() is not enough! + // and reopen it - we do this because we need to reopen stream + // reader at specified position - calling setPosition() is not + // enough! restarted = true; open(lastCommitPointPosition); } @@ -502,5 +497,4 @@ public class StaxEventItemWriter implements ItemWriter, ItemStream, return synchronization; } - } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java index fd9cb0fb5..a799c3076 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/ItemStream.java @@ -26,8 +26,8 @@ package org.springframework.batch.item; * The state that is stored is represented as {@link StreamContext} which * enforces a requirement that any restart data can be represented by a * Properties object. In general, the contract is that {@link StreamContext} - * that is returned via the {@link #getStreamContext()} method will be given back - * to the {@link #restoreFrom(StreamContext)} method, exactly as it was + * that is returned via the {@link #getStreamContext()} method will be given + * back to the {@link #restoreFrom(StreamContext)} method, exactly as it was * provided. *

* @@ -51,7 +51,8 @@ public interface ItemStream extends StreamContextProvider { /** * If any resources are needed for the stream to operate they need to be - * destroyed here. + * destroyed here. Once this method has been called all other methods + * (except open) may throw an exception. */ void close() throws StreamException; } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContext.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContext.java index 1fad0e756..3ca0affd7 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContext.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamContext.java @@ -24,102 +24,104 @@ import java.util.Set; import java.util.Map.Entry; import org.springframework.util.Assert; - +import org.springframework.util.ClassUtils; /** - * Value object representing a context for an {@link ItemStream}. It is - * essentially a thin wrapper for a map that allows for type safety - * on reads. It also allows for dirty checking by setting a 'dirty' - * flag whenever any put is called. + * Value object representing a context for an {@link ItemStream}. It is + * essentially a thin wrapper for a map that allows for type safety on reads. It + * also allows for dirty checking by setting a 'dirty' flag whenever any put is + * called. */ public class StreamContext { private boolean dirty = false; + private final Map map; - - public StreamContext(){ + + public StreamContext() { map = new HashMap(); } - - public void putString(String key, String value){ - + + public void putString(String key, String value) { + Assert.notNull(value); put(key, value); } - - public void putLong(String key, long value){ - + + public void putLong(String key, long value) { + put(key, new Long(value)); } - - public void putDouble(String key, double value){ - + + public void putDouble(String key, double value) { + put(key, new Double(value)); } - - public void put(String key, Object value){ + + public void put(String key, Object value) { dirty = true; map.put(key, value); } - + public boolean isDirty() { return dirty; } - - public String getString(String key){ - - return (String)readAndValidate(key, String.class); + + public String getString(String key) { + + return (String) readAndValidate(key, String.class); } - - public long getLong(String key){ - - return ((Long)readAndValidate(key, Long.class)).longValue(); + + public long getLong(String key) { + + return ((Long) readAndValidate(key, Long.class)).longValue(); } - - public Object get(String key){ - + + public Object get(String key) { + return map.get(key); } - - private Object readAndValidate(String key, Class type){ - + + private Object readAndValidate(String key, Class type) { + Object value = map.get(key); - - if(!type.isInstance(key)){ - throw new ClassCastException("Value is not of type: [" + type + "]"); - } - + +// if (!type.isInstance(key)) { +// throw new ClassCastException("Value for key=[" + key + "] is not of type: [" + ClassUtils.getShortName(type) +// + "], it is [" + (value == null ? null : ClassUtils.getShortName(value.getClass())) + "]"); +// } + return value; } - - public boolean isEmpty(){ + + public boolean isEmpty() { return map.isEmpty(); } - - public void clearDirtyFlag(){ + + public void clearDirtyFlag() { dirty = false; } - - public Set entrySet(){ + + public Set entrySet() { return map.entrySet(); } - - public boolean containsKey(String key){ + + public boolean containsKey(String key) { return map.containsKey(key); } - - public boolean containsValue(Object value){ + + public boolean containsValue(Object value) { return map.containsValue(value); } - - public Properties getProperties(){ - + + public Properties getProperties() { + Properties props = new Properties(); - for(Iterator it = map.entrySet().iterator();it.hasNext();){ - Entry entry = (Entry)it.next(); + for (Iterator it = map.entrySet().iterator(); it.hasNext();) { + Entry entry = (Entry) it.next(); props.setProperty(entry.getKey().toString(), entry.getValue().toString()); } - + return props; } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java index 8396a1f84..f046b857b 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/StreamException.java @@ -23,4 +23,11 @@ import org.springframework.batch.io.exception.BatchCriticalException; */ public class StreamException extends BatchCriticalException { + /** + * @param message + */ + public StreamException(String message) { + super(message); + } + } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/GenericStreamContext.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/GenericStreamContext.java index 82eda4290..cd221e5eb 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/GenericStreamContext.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/GenericStreamContext.java @@ -28,11 +28,13 @@ public class GenericStreamContext extends StreamContext { super(); } - public GenericStreamContext(Properties data){ + public GenericStreamContext(Properties data) { super(); - for(Iterator it = data.entrySet().iterator();it.hasNext();){ - Entry entry = (Entry)it.next(); - putString(entry.getKey().toString(), entry.getValue().toString()); + if (data != null) { + for (Iterator it = data.entrySet().iterator(); it.hasNext();) { + Entry entry = (Entry) it.next(); + putString(entry.getKey().toString(), entry.getValue().toString()); + } } } } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/SimpleStatisticsService.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/SimpleStatisticsService.java deleted file mode 100644 index 2bdb29ab9..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/SimpleStatisticsService.java +++ /dev/null @@ -1,89 +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.statistics; - -import java.util.Collection; -import java.util.HashMap; -import java.util.Iterator; -import java.util.LinkedHashSet; -import java.util.Map; -import java.util.Properties; -import java.util.Set; - -/** - * Simple {@link StatisticsService} that makes no attempt to aggregate or - * resolve conflicts between key names. All the contributions registered are - * simply polled and added "as is" to the aggregate properties. - * - * @author Dave Syer - * - */ -public class SimpleStatisticsService implements StatisticsService { - - private Map registry = new HashMap(); - - /** - * Simple aggregate statistics provider for the contributions registered - * under the given key. - * - * @see org.springframework.batch.statistics.StatisticsService#getStatistics(java.lang.Object) - */ - public Properties getStatistics(Object key) { - Set set = new LinkedHashSet(); - synchronized (registry) { - Collection collection = (Collection) registry.get(key); - if (collection != null) { - set = new LinkedHashSet(collection); - } - } - return aggregate(set); - } - - /** - * @param list a list of {@link StatisticsProvider}s - * @return aggregated statistics - */ - private Properties aggregate(Collection list) { - Properties result = new Properties(); - for (Iterator iterator = list.iterator(); iterator.hasNext();) { - StatisticsProvider provider = (StatisticsProvider) iterator.next(); - Properties properties = provider.getStatistics(); - if (properties != null) { - result.putAll(properties); - } - } - return result; - } - - /** - * Register a {@link StatisticsProvider} as one of the interesting providers - * under the provided key. - * - * @see org.springframework.batch.statistics.StatisticsService#register(java.lang.Object, - * org.springframework.batch.statistics.StatisticsProvider) - */ - public void register(Object key, StatisticsProvider provider) { - synchronized (registry) { - Set set = (Set) registry.get(key); - if (set == null) { - set = new LinkedHashSet(); - registry.put(key, set); - } - set.add(provider); - } - } - -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/StatisticsProvider.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/StatisticsProvider.java deleted file mode 100644 index e50bcf194..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/StatisticsProvider.java +++ /dev/null @@ -1,31 +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.statistics; - -import java.util.Properties; - -/** - * Provides statistics for a given module run. Any class that implements - * this interface is guaranteeing that it will provide Statistics. - * - * @author Lucas Ward - * - */ -public interface StatisticsProvider { - - Properties getStatistics(); -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/StatisticsService.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/StatisticsService.java deleted file mode 100644 index 2d93838e9..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/StatisticsService.java +++ /dev/null @@ -1,50 +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.statistics; - -import java.util.Properties; - -/** - * Generalised statistics aggregation strategy. Clients register - * {@link StatisticsProvider} instances under a well-known key, and then when - * they ask for statistics by that key, they receive an aggregate of all the - * values given by registered providers. - * - * @author Dave Syer - * - */ -public interface StatisticsService { - - /** - * Register the {@link StatisticsProvider} instance as one of possibly - * several that are associated with the given key. - * - * @param key the key under which to add the provider - * @param provider a {@link StatisticsProvider} - */ - void register(Object key, StatisticsProvider provider); - - /** - * Extract and aggregate the statistics from all providers under this key. - * - * @param key the key under which {@link StatisticsProvider} instances might - * have been registered. - * @return {@link Properties} summarising the statistics of all providers - * registered under this key, or empty otherwise. - */ - Properties getStatistics(Object key); - -} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/package.html b/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/package.html deleted file mode 100644 index bb5fbbea8..000000000 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/statistics/package.html +++ /dev/null @@ -1,7 +0,0 @@ - - -

-Infrastructure implementations of statistics concerns. -

- - diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapperTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapperTests.java index 4fe509c0d..996fadb95 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapperTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/ColumnMapRestartDataRowMapperTests.java @@ -15,21 +15,20 @@ import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.core.CollectionFactory; import org.springframework.jdbc.core.PreparedStatementSetter; -import org.springframework.util.ClassUtils; /** * @author Lucas Ward */ public class ColumnMapRestartDataRowMapperTests extends TestCase { - private static final String KEY = ClassUtils.getQualifiedName(ColumnMapStreamContextRowMapper.class) + ".KEY."; + private static final String KEY = ColumnMapStreamContextRowMapper.KEY_PREFIX; - ColumnMapStreamContextRowMapper mapper; + private ColumnMapStreamContextRowMapper mapper; - Map key; + private Map key; - MockControl psControl = MockControl.createControl(PreparedStatement.class); - PreparedStatement ps; + private MockControl psControl = MockControl.createControl(PreparedStatement.class); + private PreparedStatement ps; protected void setUp() throws Exception { super.setUp(); @@ -44,7 +43,7 @@ public class ColumnMapRestartDataRowMapperTests extends TestCase { public void testCreateRestartDataWithInvalidType() throws Exception { try{ - mapper.createRestartData(new Object()); + mapper.createStreamContext(new Object()); fail(); }catch(IllegalArgumentException ex){ //expected @@ -54,7 +53,7 @@ public class ColumnMapRestartDataRowMapperTests extends TestCase { public void testCreateRestartDataWithNull(){ try{ - mapper.createRestartData(null); + mapper.createStreamContext(null); fail(); }catch(IllegalArgumentException ex){ //expected @@ -62,8 +61,7 @@ public class ColumnMapRestartDataRowMapperTests extends TestCase { } public void testCreateRestartData() throws Exception { - - StreamContext streamContext = mapper.createRestartData(key); + StreamContext streamContext = mapper.createStreamContext(key); Properties props = streamContext.getProperties(); assertEquals("1", props.getProperty(KEY + "0")); assertEquals("2", props.getProperty(KEY + "1")); @@ -71,7 +69,7 @@ public class ColumnMapRestartDataRowMapperTests extends TestCase { public void testCreateRestartDataFromEmptyKeys() throws Exception { - StreamContext streamContext = mapper.createRestartData(new HashMap()); + StreamContext streamContext = mapper.createStreamContext(new HashMap()); assertEquals(0, streamContext.getProperties().size()); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java index 64354d3e4..f2bb312a2 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/driving/support/MultipleColumnJdbcKeyGeneratorIntegrationTests.java @@ -47,8 +47,8 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran public void testRestoreKeys(){ Properties props = new Properties(); - props.setProperty(ColumnMapStreamContextRowMapper.KEY + "0", "3"); - props.setProperty(ColumnMapStreamContextRowMapper.KEY + "1", "3"); + props.setProperty(ColumnMapStreamContextRowMapper.KEY_PREFIX + "0", "3"); + props.setProperty(ColumnMapStreamContextRowMapper.KEY_PREFIX + "1", "3"); StreamContext streamContext = new GenericStreamContext(props); List keys = keyStrategy.restoreKeys(streamContext); @@ -72,8 +72,8 @@ public class MultipleColumnJdbcKeyGeneratorIntegrationTests extends AbstractTran Properties props = streamContext.getProperties(); assertEquals(2, props.size()); - assertEquals("3", props.get(ColumnMapStreamContextRowMapper.KEY + "0")); - assertEquals("3", props.get(ColumnMapStreamContextRowMapper.KEY + "1")); + assertEquals("3", props.get(ColumnMapStreamContextRowMapper.KEY_PREFIX + "0")); + assertEquals("3", props.get(ColumnMapStreamContextRowMapper.KEY_PREFIX + "1")); } public void testGetNullKeyAsStreamContext(){ diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java index a172f2af6..4f8e1fe0e 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/io/file/DefaultFlatFileItemReaderTests.java @@ -27,6 +27,7 @@ import org.springframework.batch.io.file.mapping.FieldSetMapper; import org.springframework.batch.io.file.mapping.FieldSet; import org.springframework.batch.io.file.transform.LineTokenizer; import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamException; import org.springframework.core.io.ByteArrayResource; import org.springframework.core.io.Resource; import org.springframework.transaction.support.TransactionSynchronization; @@ -179,12 +180,16 @@ public class DefaultFlatFileItemReaderTests extends TestCase { assertEquals("[FlatFileInputTemplate-TestData]", inputSource.read().toString()); } - public void testRestartWithNullReader() throws Exception { + public void testRestartBeforeOpen() throws Exception { inputSource = new DefaultFlatFileItemReader(); inputSource.setResource(getInputResource(TEST_STRING)); inputSource.setFieldSetMapper(fieldSetMapper); // do not open the template... - inputSource.restoreFrom(inputSource.getStreamContext()); + try { + inputSource.restoreFrom(inputSource.getStreamContext()); + } catch (StreamException e) { + assertTrue("Message does not contain open: "+e.getMessage(), e.getMessage().contains("open")); + } assertEquals("[FlatFileInputTemplate-TestData]", inputSource.read().toString()); } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java index e795dfa3a..c6cf2f9a8 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/reader/DelegatingItemReaderTests.java @@ -24,15 +24,12 @@ import org.springframework.batch.io.Skippable; import org.springframework.batch.item.ItemReader; import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.StreamContext; -import org.springframework.batch.item.reader.AbstractItemReader; -import org.springframework.batch.item.reader.DelegatingItemReader; import org.springframework.batch.item.stream.GenericStreamContext; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.batch.support.PropertiesConverter; /** * Unit test for {@link DelegatingItemReader} - * + * * @author Robert Kasanicky */ public class DelegatingItemReaderTests extends TestCase { @@ -48,24 +45,25 @@ public class DelegatingItemReaderTests extends TestCase { itemProvider.setItemReader(source); } - public void testAfterPropertiesSet()throws Exception{ - //shouldn't throw an exception since the input source is set + public void testAfterPropertiesSet() throws Exception { + // shouldn't throw an exception since the input source is set itemProvider.afterPropertiesSet(); } - public void testNullItemReader(){ - try{ + public void testNullItemReader() { + try { itemProvider.setItemReader(null); itemProvider.afterPropertiesSet(); fail(); - }catch(Exception ex){ + } + catch (Exception ex) { assertTrue(ex instanceof IllegalArgumentException); } } /** * Uses input template to provide the domain object. - * @throws Exception + * @throws Exception */ public void testNext() throws Exception { Object result = itemProvider.read(); @@ -82,7 +80,7 @@ public class DelegatingItemReaderTests extends TestCase { /** * Forwared restart data to input template - * @throws Exception + * @throws Exception */ public void testRestoreFrom() throws Exception { itemProvider.restoreFrom(new GenericStreamContext(PropertiesConverter.stringToProperties("value=bar"))); @@ -94,7 +92,7 @@ public class DelegatingItemReaderTests extends TestCase { assertEquals("after skip", itemProvider.read()); } - private static class MockItemReader extends AbstractItemReader implements ItemReader, StatisticsProvider, ItemStream, Skippable { + private static class MockItemReader extends AbstractItemReader implements ItemReader, ItemStream, Skippable { private Object value; diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java index 07402aa7f..0589fb385 100644 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/writer/ItemWriterItemProcessorTests.java @@ -26,8 +26,6 @@ import org.springframework.batch.item.ItemStream; import org.springframework.batch.item.ItemWriter; import org.springframework.batch.item.StreamContext; import org.springframework.batch.item.stream.GenericStreamContext; -import org.springframework.batch.item.writer.DelegatingItemWriter; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.batch.support.PropertiesConverter; /** @@ -55,7 +53,7 @@ public class ItemWriterItemProcessorTests extends TestCase { assertEquals(1, list.size()); assertEquals("test:foo", list.get(0)); } - + /** * Gets restart data from the input template */ @@ -66,7 +64,7 @@ public class ItemWriterItemProcessorTests extends TestCase { /** * Forward restart data to input template - * @throws Exception + * @throws Exception */ public void testRestoreFrom() throws Exception { processor.restoreFrom(new GenericStreamContext(PropertiesConverter.stringToProperties("value=bar"))); @@ -76,7 +74,7 @@ public class ItemWriterItemProcessorTests extends TestCase { /** * Forward restart data to input template - * @throws Exception + * @throws Exception */ public void testGetStreamContextWithoutItemStream() throws Exception { processor.setDelegate(null); @@ -91,7 +89,7 @@ public class ItemWriterItemProcessorTests extends TestCase { /** * Forward restart data to input template - * @throws Exception + * @throws Exception */ public void testRestoreFromWithoutRestartable() throws Exception { processor.setDelegate(null); @@ -103,7 +101,7 @@ public class ItemWriterItemProcessorTests extends TestCase { // expected } } - + public void testSkip() { processor.skip(); assertEquals(1, list.size()); @@ -123,14 +121,14 @@ public class ItemWriterItemProcessorTests extends TestCase { // expected } } - - private List list = new ArrayList(); + + private List list = new ArrayList(); /** * @author Dave Syer * */ - public class MockOutputSource implements ItemWriter, StatisticsProvider, ItemStream, Skippable { + public class MockOutputSource implements ItemWriter, ItemStream, Skippable { private String value; @@ -139,7 +137,7 @@ public class ItemWriterItemProcessorTests extends TestCase { } public void write(Object output) { - list.add(value+":"+output); + list.add(value + ":" + output); } public void close() { @@ -161,7 +159,7 @@ public class ItemWriterItemProcessorTests extends TestCase { } public void skip() { - list.add("after skip"); + list.add("after skip"); } } diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/statistics/SimpleStatisticsServiceTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/statistics/SimpleStatisticsServiceTests.java deleted file mode 100644 index d1a4b7d67..000000000 --- a/spring-batch-infrastructure/src/test/java/org/springframework/batch/statistics/SimpleStatisticsServiceTests.java +++ /dev/null @@ -1,72 +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.statistics; - -import java.util.Properties; - -import org.springframework.batch.support.PropertiesConverter; - -import junit.framework.TestCase; - -/** - * @author Dave Syer - * - */ -public class SimpleStatisticsServiceTests extends TestCase { - - private SimpleStatisticsService service = new SimpleStatisticsService(); - - public void testRegistration() throws Exception { - service.register("FOO", new StubStatisticsProvider()); - assertEquals("bar", service.getStatistics("FOO").getProperty("foo")); - } - - public void testAggregation() throws Exception { - service.register("FOO", new StubStatisticsProvider()); - service.register("FOO", new StubStatisticsProvider("spam=bucket")); - assertEquals("bar", service.getStatistics("FOO").getProperty("foo")); - assertEquals("bucket", service.getStatistics("FOO").getProperty("spam")); - } - - public void testDoubleAggregationOrder() throws Exception { - StubStatisticsProvider provider = new StubStatisticsProvider(); - service.register("FOO", provider); - service.register("FOO", provider); - assertEquals(1, service.getStatistics("FOO").size()); - } - - /** - * @author Dave Syer - * - */ - private class StubStatisticsProvider implements StatisticsProvider { - String values = "foo=bar"; - - public StubStatisticsProvider(String values) { - super(); - this.values = values; - } - - public StubStatisticsProvider() { - super(); - } - - public Properties getStatistics() { - return PropertiesConverter.stringToProperties(values); - } - } - -} diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/InfiniteLoopTasklet.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/InfiniteLoopTasklet.java index e77c4157d..6b1fc71b2 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/InfiniteLoopTasklet.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/InfiniteLoopTasklet.java @@ -16,43 +16,44 @@ package org.springframework.batch.sample.tasklet; -import java.util.Properties; - import org.springframework.batch.core.tasklet.Tasklet; +import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.StreamContextProvider; +import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.repeat.ExitStatus; -import org.springframework.batch.statistics.StatisticsProvider; import org.springframework.batch.support.PropertiesConverter; /** - * Simple module implementation that will always return true to indicate - * that processing should continue. This is useful for testing graceful - * shutdown of jobs. + * Simple module implementation that will always return true to indicate that + * processing should continue. This is useful for testing graceful shutdown of + * jobs. * * @author Lucas Ward - * + * */ -public class InfiniteLoopTasklet implements Tasklet, StatisticsProvider { - +public class InfiniteLoopTasklet implements Tasklet, StreamContextProvider { + private int count = 0; - + /** * */ public InfiniteLoopTasklet() { super(); } - + public ExitStatus execute() throws Exception { Thread.sleep(500); count++; return ExitStatus.CONTINUABLE; } - - /* (non-Javadoc) - * @see org.springframework.batch.statistics.StatisticsProvider#getStatistics() + + /* + * (non-Javadoc) + * @see org.springframework.batch.item.stream.ItemStreamAdapter#getStreamContext() */ - public Properties getStatistics() { - return PropertiesConverter.stringToProperties("count="+count); + public StreamContext getStreamContext() { + return new GenericStreamContext(PropertiesConverter.stringToProperties("count=" + count)); } } diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java index ab7aeed32..51caf3f3c 100644 --- a/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java +++ b/spring-batch-samples/src/main/java/org/springframework/batch/sample/tasklet/SimpleTradeTasklet.java @@ -21,10 +21,12 @@ import java.util.Properties; 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.item.StreamContext; +import org.springframework.batch.item.StreamContextProvider; +import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.batch.repeat.ExitStatus; import org.springframework.batch.sample.dao.TradeDao; import org.springframework.batch.sample.domain.Trade; -import org.springframework.batch.statistics.StatisticsProvider; /** * Simple implementation of a {@link Tasklet}, which illustrates the reading @@ -40,7 +42,7 @@ import org.springframework.batch.statistics.StatisticsProvider; * @author Lucas Ward * @author Dave Syer */ -public class SimpleTradeTasklet implements Tasklet, StatisticsProvider { +public class SimpleTradeTasklet implements Tasklet, StreamContextProvider { /* * reads the data from input file @@ -84,11 +86,14 @@ public class SimpleTradeTasklet implements Tasklet, StatisticsProvider { this.tradeDao = tradeDao; } - public Properties getStatistics() { + /* (non-Javadoc) + * @see org.springframework.batch.item.StreamContextProvider#getStreamContext() + */ + public StreamContext getStreamContext() { Properties statistics = new Properties(); statistics.setProperty("trade.count", String.valueOf(tradeCount)); statistics.putAll(inputSource.getStatistics()); - return statistics; + return new GenericStreamContext(statistics); } }