From 2b9add95b6edd0186448bfd05699da248581fedd Mon Sep 17 00:00:00 2001 From: lucasward Date: Thu, 31 Jan 2008 12:37:51 +0000 Subject: [PATCH] BATCH-7:Quick change to allow StreamContext to be used with similar semantics as RestartData. Should be temporary as the StreamManager change is made. --- .../batch/io/cursor/JdbcCursorItemReader.java | 13 ++- .../ColumnMapStreamContextRowMapper.java | 44 ++------ .../batch/item/StreamContext.java | 104 +++++++++++++++++- .../item/stream/GenericStreamContext.java | 19 ++-- .../item/stream/SimpleStreamManager.java | 18 +-- 5 files changed, 131 insertions(+), 67 deletions(-) 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 b94272c8a..b63c67822 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 @@ -406,6 +406,8 @@ 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)); @@ -425,14 +427,13 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource open(); - Properties restartProperties = data.getProperties(); - if (!restartProperties.containsKey(CURRENT_PROCESSED_ROW)) { + // Properties restartProperties = data.getProperties(); + if (!data.containsKey(CURRENT_PROCESSED_ROW)) { return; } try { - this.currentProcessedRow = Integer.parseInt(restartProperties - .getProperty(CURRENT_PROCESSED_ROW)); + this.currentProcessedRow = new Long(data.getLong(CURRENT_PROCESSED_ROW)).intValue(); rs.absolute(currentProcessedRow); } catch (SQLException se) { throw getExceptionTranslator().translate( @@ -440,11 +441,11 @@ public class JdbcCursorItemReader extends AbstractTransactionalIoSource se); } - if (!restartProperties.containsKey(SKIPPED_ROWS)) { + if (!data.containsKey(SKIPPED_ROWS)) { return; } - String[] skipped = StringUtils.commaDelimitedListToStringArray(restartProperties.getProperty(SKIPPED_ROWS)); + String[] skipped = StringUtils.commaDelimitedListToStringArray(data.getString(SKIPPED_ROWS)); for (int i = 0; i < skipped.length; i++) { this.skippedRows.add(new Integer(skipped[i])); } 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 428e480e3..40aada16b 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 @@ -13,6 +13,7 @@ import java.util.Properties; import java.util.Map.Entry; import org.springframework.batch.item.StreamContext; +import org.springframework.batch.item.stream.GenericStreamContext; import org.springframework.core.CollectionFactory; import org.springframework.jdbc.core.ColumnMapRowMapper; import org.springframework.jdbc.core.PreparedStatementSetter; @@ -43,8 +44,9 @@ public class ColumnMapStreamContextRowMapper extends ColumnMapRowMapper implemen ColumnMapRestartData columnData = new ColumnMapRestartData(streamContext.getProperties()); List columns = new ArrayList(); - for (Iterator iterator = columnData.keys.values().iterator(); iterator.hasNext();) { - Object column = (Object) iterator.next(); + for (Iterator iterator = columnData.entrySet().iterator(); iterator.hasNext();) { + Entry entry = (Entry)iterator.next(); + Object column = entry.getValue(); columns.add(column); } @@ -60,42 +62,18 @@ public class ColumnMapStreamContextRowMapper extends ColumnMapRowMapper implemen } - private static class ColumnMapRestartData implements StreamContext{ - - private final Map keys; + private static class ColumnMapRestartData extends GenericStreamContext{ public ColumnMapRestartData(Map keys) { - this.keys = keys; + super(); + for(Iterator it = keys.entrySet().iterator();it.hasNext();){ + Entry entry = (Entry)it.next(); + putString(entry.getKey().toString(), entry.getValue().toString()); + } } public ColumnMapRestartData(Properties props) { - - keys = CollectionFactory.createLinkedCaseInsensitiveMapIfPossible(props.size()); - - - for(int counter = 0; counter < props.size(); counter++){ - String column = props.getProperty(KEY + counter); - - if(column != null){ - keys.put(KEY + counter, 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 + counter, entry.getValue().toString()); - counter++; - } - - return props; + super(props); } } 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 eb072f8c2..1fad0e756 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 @@ -16,12 +16,110 @@ package org.springframework.batch.item; +import java.util.HashMap; +import java.util.Iterator; +import java.util.Map; import java.util.Properties; +import java.util.Set; +import java.util.Map.Entry; + +import org.springframework.util.Assert; + /** - * Interface for representing data necessary to recover state after restart. + * 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 interface StreamContext { +public class StreamContext { - Properties getProperties(); + private boolean dirty = false; + private final Map map; + + public StreamContext(){ + map = new HashMap(); + } + + public void putString(String key, String value){ + + Assert.notNull(value); + put(key, value); + } + + public void putLong(String key, long value){ + + put(key, new Long(value)); + } + + public void putDouble(String key, double value){ + + put(key, new Double(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 long getLong(String key){ + + return ((Long)readAndValidate(key, Long.class)).longValue(); + } + + public Object get(String key){ + + return map.get(key); + } + + 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 + "]"); + } + + return value; + } + + public boolean isEmpty(){ + return map.isEmpty(); + } + + public void clearDirtyFlag(){ + dirty = false; + } + + public Set entrySet(){ + return map.entrySet(); + } + + public boolean containsKey(String key){ + return map.containsKey(key); + } + + public boolean containsValue(Object value){ + return map.containsValue(value); + } + + public Properties getProperties(){ + + Properties props = new Properties(); + 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/stream/GenericStreamContext.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/GenericStreamContext.java index 8e34247a5..82eda4290 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 @@ -16,20 +16,23 @@ package org.springframework.batch.item.stream; +import java.util.Iterator; import java.util.Properties; +import java.util.Map.Entry; import org.springframework.batch.item.StreamContext; -public class GenericStreamContext implements StreamContext { +public class GenericStreamContext extends StreamContext { - private Properties data; + public GenericStreamContext() { + super(); + } public GenericStreamContext(Properties data){ - this.data = data; + super(); + for(Iterator it = data.entrySet().iterator();it.hasNext();){ + Entry entry = (Entry)it.next(); + putString(entry.getKey().toString(), entry.getValue().toString()); + } } - - public Properties getProperties(){ - return data; - } - } diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java index 0e666ad7a..5d8ff30dd 100644 --- a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/stream/SimpleStreamManager.java @@ -54,7 +54,7 @@ public class SimpleStreamManager implements StreamManager { set = new LinkedHashSet(collection); } } - return new SimpleStreamManagerStreamContext(set); + return aggregate(set); } /** @@ -116,20 +116,4 @@ public class SimpleStreamManager implements StreamManager { } } - private class SimpleStreamManagerStreamContext implements StreamContext { - - private StreamContext data; - - public SimpleStreamManagerStreamContext(Set streams) { - this.data = aggregate(streams); - } - - public Properties getProperties() { - if (data != null) { - return data.getProperties(); - } - return new Properties(); - } - } - }