add new SynchronizedItemStreamReader

Decorator for an ItemStreamReader

spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/SynchronizedItemStreamReader.java

@@ -0,0 +1,70 @@
+/*
+ * Copyright 2015 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.item.support;
+
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.ItemStream;
+import org.springframework.batch.item.ItemStreamReader;
+import org.springframework.batch.item.NonTransientResourceException;
+import org.springframework.batch.item.ParseException;
+import org.springframework.batch.item.UnexpectedInputException;
+
+/**
+ * 
+ * This is a simple ItemStreamReader decorator with a synchronized ItemReader.read() 
+ * method - which makes a non-thread-safe ItemReader thread-safe.
+ * 
+ * However, if reprocessing an item is problematic then using this will make a job not 
+ * restartable.  If a restartable job is desired in that case, then further co-ordination 
+ * between the read and close methods needs to be implemented.
+ * 
+ * Here are some links about the motivation behind this class:
+ * - http://projects.spring.io/spring-batch/faq.html#threading-reader}
+ * - http://stackoverflow.com/a/20002493/2910265}
+ * 
+ * @author Matthew Ouyang
+ * @since 3.0
+ *
+ * @param <T>
+ */
+public class SynchronizedItemStreamReader<T> implements ItemStream, ItemReader<T> {
+
+	ItemStreamReader<T> itemStreamReader;
+
+	public void setItemStreamReader(ItemStreamReader<T> itemStreamReader) {
+		this.itemStreamReader = itemStreamReader;
+	}
+
+	/**
+	 * This delegates to the read method of the <code>itemStreamReader</code>
+	 */
+	public synchronized T read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
+		return this.itemStreamReader.read();
+	}
+
+	public void close() {
+		this.itemStreamReader.close();
+	}
+
+	public void open(ExecutionContext executionContext) {
+		this.itemStreamReader.open(executionContext);
+	}
+
+	public void update(ExecutionContext executionContext) {
+		this.itemStreamReader.update(executionContext);
+	}
+}

spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/SynchronizedItemStreamReaderTests.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2015 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.item.support;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import org.junit.Test;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemStreamReader;
+import org.springframework.batch.item.NonTransientResourceException;
+import org.springframework.batch.item.ParseException;
+import org.springframework.batch.item.UnexpectedInputException;
+
+/**
+ * 
+ * @author Matthew Ouyang
+ *
+ */
+public class SynchronizedItemStreamReaderTests {
+
+	/**
+	 * A simple class used to test the SynchronizedItemStreamReader.  It simply returns 
+	 * the number of times the read method has been called, manages some state variables 
+	 * and updates an ExecutionContext.
+	 * 
+	 * @author Matthew Ouyang
+	 *
+	 */
+	private class TestItemReader extends AbstractItemStreamItemReader<Integer> implements ItemStreamReader<Integer> {
+
+		private int cursor = 0;
+		private boolean isClosed = false;
+
+		public static final String HAS_BEEN_OPENED = "hasBeenOpened";
+		public static final String UPDATE_COUNT_KEY = "updateCount";
+
+		public Integer read() throws Exception, UnexpectedInputException, ParseException, 
+			NonTransientResourceException {
+			cursor = cursor + 1;
+			return cursor;
+		}
+
+		public void close() {
+			this.isClosed = true;
+		}
+
+		public void open(ExecutionContext executionContext) {
+			this.isClosed = false;
+			executionContext.put(HAS_BEEN_OPENED, true);
+			executionContext.remove(UPDATE_COUNT_KEY);
+		}
+
+		public void update(ExecutionContext executionContext) {
+
+			if (!executionContext.containsKey(UPDATE_COUNT_KEY)) {
+				executionContext.putInt(UPDATE_COUNT_KEY, 0);
+			}
+
+			executionContext.putInt(UPDATE_COUNT_KEY
+				, executionContext.getInt(UPDATE_COUNT_KEY) + 1
+			);
+		}
+
+		public boolean isClosed() {
+			return this.isClosed;
+		}
+	}
+
+	@Test
+	public void testMultipleThreads() throws Exception {
+
+		// Initialized an ExecutionContext and a SynchronizedItemStreamReader to test.
+		final ExecutionContext executionContext = new ExecutionContext();
+
+		final TestItemReader testItemReader = new TestItemReader();
+		final SynchronizedItemStreamReader<Integer> synchronizedItemStreamReader = new SynchronizedItemStreamReader<Integer>();
+		synchronizedItemStreamReader.setItemStreamReader(testItemReader);
+
+		// Open the ItemReader and make sure it's initialized properly.
+		synchronizedItemStreamReader.open(executionContext);
+		assertEquals(true, executionContext.get(TestItemReader.HAS_BEEN_OPENED));
+		assertFalse(testItemReader.isClosed());
+
+		/* Set up SIZE threads that read from the reader and updates the execution 
+		 * context.
+		 */
+		final Set<Integer> ecSet = new HashSet<Integer>();
+		final int SIZE = 20;
+		Thread[] threads = new Thread[SIZE];
+		for (int i = 0; i < SIZE; i++) {
+			threads[i] = new Thread() {
+				public void run() {
+					try {
+						ecSet.add(synchronizedItemStreamReader.read());
+						synchronizedItemStreamReader.update(executionContext);
+					} catch (Exception e) {
+					} finally {
+					}
+				}
+			};
+		}
+
+		// Start the threads and block until all threads are done.
+		for (Thread thread : threads) {
+			thread.run();
+		}
+		for (Thread thread : threads) {
+			thread.join();
+		}
+		testItemReader.close();
+
+		/* Ensure cleanup happens as expected: status variable is set correctly and 
+		 * ExecutionContext variable is set properly.  Lastly, the Set<Integer> should 
+		 * have 1 to 20 which may not always be the case if the read is not synchronized.
+		 */
+		for (int i = 1; i <= SIZE; i++) {
+			assertTrue(ecSet.contains(i));
+		}
+		assertTrue(testItemReader.isClosed());
+		assertEquals(SIZE, executionContext.getInt(TestItemReader.UPDATE_COUNT_KEY));
+	}
+}