Add synchronized decorators for non-stream item readers/writers

Resolves #4368

Author: fmbenhassine

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

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2023 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
+ *
+ *      https://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 java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.springframework.batch.item.ItemReader;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * This is an {@link ItemReader} decorator with a synchronized {@link ItemReader#read}
+ * method. This decorator is useful when using a non thread-safe item reader in a
+ * multi-threaded step.
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 5.1.0
+ * @param <T> type of objects to read
+ */
+public class SynchronizedItemReader<T> implements ItemReader<T> {
+
+	private final ItemReader<T> delegate;
+
+	private final Lock lock = new ReentrantLock();
+
+	public SynchronizedItemReader(ItemReader<T> delegate) {
+		Assert.notNull(delegate, "The delegate must not be null");
+		this.delegate = delegate;
+	}
+
+	/**
+	 * This method delegates to the {@code read} method of the delegate and is
+	 * synchronized with a lock.
+	 */
+	@Nullable
+	public T read() throws Exception {
+		this.lock.lock();
+		try {
+			return this.delegate.read();
+		}
+		finally {
+			this.lock.unlock();
+		}
+	}
+
+}

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

@@ -0,0 +1,60 @@
+/*
+ * Copyright 2023 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
+ *
+ *      https://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 java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.springframework.batch.item.Chunk;
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.util.Assert;
+
+/**
+ * This is an {@link ItemWriter} decorator with a synchronized {@link ItemWriter#write}
+ * method. This decorator is useful when using a non thread-safe item writer in a
+ * multi-threaded step.
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 5.1.0
+ * @param <T> type of objects to write
+ */
+public class SynchronizedItemWriter<T> implements ItemWriter<T> {
+
+	private final ItemWriter<T> delegate;
+
+	private final Lock lock = new ReentrantLock();
+
+	public SynchronizedItemWriter(ItemWriter<T> delegate) {
+		Assert.notNull(delegate, "The delegate must not be null");
+		this.delegate = delegate;
+	}
+
+	/**
+	 * This method delegates to the {@code write} method of the delegate and is
+	 * synchronized with a lock.
+	 */
+	@Override
+	public void write(Chunk<? extends T> items) throws Exception {
+		this.lock.lock();
+		try {
+			this.delegate.write(items);
+		}
+		finally {
+			this.lock.unlock();
+		}
+	}
+
+}

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

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2023 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
+ *
+ *          https://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.builder;
+
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.support.SynchronizedItemReader;
+import org.springframework.util.Assert;
+
+/**
+ * Builder for {@link SynchronizedItemReader}.
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 5.1.0
+ */
+public class SynchronizedItemReaderBuilder<T> {
+
+	private ItemReader<T> delegate;
+
+	/**
+	 * The item reader to use as a delegate.
+	 * @param delegate the delegate reader to set
+	 * @return this instance for method chaining
+	 */
+	public SynchronizedItemReaderBuilder<T> delegate(ItemReader<T> delegate) {
+		this.delegate = delegate;
+
+		return this;
+	}
+
+	/**
+	 * Returns a new {@link SynchronizedItemReader}.
+	 * @return a new {@link SynchronizedItemReader}
+	 */
+	public SynchronizedItemReader<T> build() {
+		Assert.notNull(this.delegate, "A delegate is required");
+
+		return new SynchronizedItemReader<>(this.delegate);
+	}
+
+}

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

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2023 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
+ *
+ *          https://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.builder;
+
+import org.springframework.batch.item.ItemWriter;
+import org.springframework.batch.item.support.SynchronizedItemWriter;
+import org.springframework.util.Assert;
+
+/**
+ * Builder for {@link SynchronizedItemWriter}.
+ *
+ * @author Mahmoud Ben Hassine
+ * @since 5.1.0
+ */
+public class SynchronizedItemWriterBuilder<T> {
+
+	private ItemWriter<T> delegate;
+
+	/**
+	 * The item writer to use as a delegate.
+	 * @param delegate the delegate writer to set
+	 * @return this instance for method chaining
+	 */
+	public SynchronizedItemWriterBuilder<T> delegate(ItemWriter<T> delegate) {
+		this.delegate = delegate;
+
+		return this;
+	}
+
+	/**
+	 * Returns a new {@link SynchronizedItemWriter}.
+	 * @return a new {@link SynchronizedItemWriter}
+	 */
+	public SynchronizedItemWriter<T> build() {
+		Assert.notNull(this.delegate, "A delegate is required");
+
+		return new SynchronizedItemWriter<>(this.delegate);
+	}
+
+}

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

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2023 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.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.springframework.batch.item.ItemReader;
+
+import static org.mockito.Mockito.verify;
+
+/**
+ * Test class for {@link SynchronizedItemReader}.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@ExtendWith(MockitoExtension.class)
+public class SynchronizedItemReaderTests {
+
+	@Mock
+	private ItemReader<Object> delegate;
+
+	@Test
+	void testDelegateReadIsCalled() throws Exception {
+		// given
+		SynchronizedItemReader<Object> synchronizedItemReader = new SynchronizedItemReader<>(this.delegate);
+
+		// when
+		synchronizedItemReader.read();
+
+		// then
+		verify(this.delegate).read();
+	}
+
+	@Test
+	void testNullDelegate() {
+		// when
+		IllegalArgumentException exception = Assertions.assertThrows(IllegalArgumentException.class,
+				() -> new SynchronizedItemReader<>(null));
+
+		// then
+		Assertions.assertEquals("The delegate must not be null", exception.getMessage());
+	}
+
+}

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

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2023 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.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import org.springframework.batch.item.Chunk;
+import org.springframework.batch.item.ItemWriter;
+
+import static org.mockito.Mockito.verify;
+
+/**
+ * Test class for {@link SynchronizedItemWriter}.
+ *
+ * @author Mahmoud Ben Hassine
+ */
+@ExtendWith(MockitoExtension.class)
+public class SynchronizedItemWriterTests {
+
+	@Mock
+	private ItemWriter<Object> delegate;
+
+	@Test
+	void testDelegateWriteIsCalled() throws Exception {
+		// given
+		Chunk<Object> chunk = new Chunk<>();
+		SynchronizedItemWriter<Object> synchronizedItemWriter = new SynchronizedItemWriter<>(this.delegate);
+
+		// when
+		synchronizedItemWriter.write(chunk);
+
+		// then
+		verify(this.delegate).write(chunk);
+	}
+
+	@Test
+	void testNullDelegate() {
+		// when
+		IllegalArgumentException exception = Assertions.assertThrows(IllegalArgumentException.class,
+				() -> new SynchronizedItemWriter<>(null));
+
+		// then
+		Assertions.assertEquals("The delegate must not be null", exception.getMessage());
+	}
+
+}