From 7c871e584c1cd888b3b459fb24518ffb419f5687 Mon Sep 17 00:00:00 2001 From: Mahmoud Ben Hassine Date: Fri, 11 Oct 2024 09:07:05 +0200 Subject: [PATCH] Add blocking queue item reader and writer Resolves #2350 Resolves #2044 --- .../item/queue/BlockingQueueItemReader.java | 63 ++++++++++++++++ .../item/queue/BlockingQueueItemWriter.java | 49 +++++++++++++ .../BlockingQueueItemReaderBuilder.java | 71 +++++++++++++++++++ .../BlockingQueueItemWriterBuilder.java | 53 ++++++++++++++ .../queue/BlockingQueueItemReaderTests.java | 48 +++++++++++++ .../queue/BlockingQueueItemWriterTests.java | 49 +++++++++++++ .../BlockingQueueItemReaderBuilderTests.java | 55 ++++++++++++++ .../BlockingQueueItemWriterBuilderTests.java | 56 +++++++++++++++ 8 files changed, 444 insertions(+) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/BlockingQueueItemReader.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/BlockingQueueItemWriter.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/builder/BlockingQueueItemReaderBuilder.java create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/builder/BlockingQueueItemWriterBuilder.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/BlockingQueueItemReaderTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/BlockingQueueItemWriterTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/builder/BlockingQueueItemReaderBuilderTests.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/builder/BlockingQueueItemWriterBuilderTests.java diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/BlockingQueueItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/BlockingQueueItemReader.java new file mode 100644 index 000000000..e5e411045 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/BlockingQueueItemReader.java @@ -0,0 +1,63 @@ +/* + * Copyright 2024 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.queue; + +import org.springframework.batch.item.ItemReader; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +/** + * This is an {@link ItemReader} that reads items from a {@link BlockingQueue}. It stops + * reading (ie returns {@code null}) if no items are available in the queue after a + * configurable timeout. + * + * @param type of items to read. + * @author Mahmoud Ben Hassine + * @since 5.2.0 + */ +public class BlockingQueueItemReader implements ItemReader { + + private final BlockingQueue queue; + + private long timeout = 1L; + + private TimeUnit timeUnit = TimeUnit.SECONDS; + + /** + * Create a new {@link BlockingQueueItemReader}. + * @param queue the queue to read items from + */ + public BlockingQueueItemReader(BlockingQueue queue) { + this.queue = queue; + } + + /** + * Set the reading timeout and time unit. Defaults to 1 second. + * @param timeout the timeout after which the reader stops reading + * @param timeUnit the unit of the timeout + */ + public void setTimeout(long timeout, TimeUnit timeUnit) { + this.timeout = timeout; + this.timeUnit = timeUnit; + } + + @Override + public T read() throws Exception { + return this.queue.poll(this.timeout, this.timeUnit); + } + +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/BlockingQueueItemWriter.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/BlockingQueueItemWriter.java new file mode 100644 index 000000000..68a667b00 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/BlockingQueueItemWriter.java @@ -0,0 +1,49 @@ +/* + * Copyright 2024 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.queue; + +import org.springframework.batch.item.Chunk; +import org.springframework.batch.item.ItemWriter; + +import java.util.concurrent.BlockingQueue; + +/** + * This is an {@link ItemWriter} that writes items to a {@link BlockingQueue}. + * + * @param type of items to write + * @since 5.2.0 + * @author Mahmoud Ben Hassine + */ +public class BlockingQueueItemWriter implements ItemWriter { + + private final BlockingQueue queue; + + /** + * Create a new {@link BlockingQueueItemWriter}. + * @param queue the queue to write items to + */ + public BlockingQueueItemWriter(BlockingQueue queue) { + this.queue = queue; + } + + @Override + public void write(Chunk items) throws Exception { + for (T item : items) { + this.queue.put(item); + } + } + +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/builder/BlockingQueueItemReaderBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/builder/BlockingQueueItemReaderBuilder.java new file mode 100644 index 000000000..9c305ca04 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/builder/BlockingQueueItemReaderBuilder.java @@ -0,0 +1,71 @@ +/* + * Copyright 2024 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.queue.builder; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.springframework.batch.item.queue.BlockingQueueItemReader; +import org.springframework.util.Assert; + +/** + * Builder for {@link BlockingQueueItemReader}. + * + * @param type of items to read + * @since 5.2.0 + * @author Mahmoud Ben Hassine + */ +public class BlockingQueueItemReaderBuilder { + + private BlockingQueue queue; + + private long timeout = 1L; + + private TimeUnit timeUnit = TimeUnit.SECONDS; + + /** + * Set the queue to read items from. + * @param queue the queue to read items from. + * @return this instance of the builder + */ + public BlockingQueueItemReaderBuilder queue(BlockingQueue queue) { + this.queue = queue; + return this; + } + + /** + * Set the reading timeout. Defaults to 1 second. + * @param timeout the reading timeout. + * @return this instance of the builder + */ + public BlockingQueueItemReaderBuilder timeout(long timeout, TimeUnit timeUnit) { + this.timeout = timeout; + this.timeUnit = timeUnit; + return this; + } + + /** + * Create a configured {@link BlockingQueueItemReader}. + * @return a configured {@link BlockingQueueItemReader}. + */ + public BlockingQueueItemReader build() { + Assert.state(this.queue != null, "The blocking queue is required."); + BlockingQueueItemReader blockingQueueItemReader = new BlockingQueueItemReader<>(this.queue); + blockingQueueItemReader.setTimeout(this.timeout, this.timeUnit); + return blockingQueueItemReader; + } + +} diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/builder/BlockingQueueItemWriterBuilder.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/builder/BlockingQueueItemWriterBuilder.java new file mode 100644 index 000000000..6e7fe772b --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/queue/builder/BlockingQueueItemWriterBuilder.java @@ -0,0 +1,53 @@ +/* + * Copyright 2024 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.queue.builder; + +import java.util.concurrent.BlockingQueue; + +import org.springframework.batch.item.queue.BlockingQueueItemWriter; +import org.springframework.util.Assert; + +/** + * Builder for a {@link BlockingQueueItemWriter}. + * + * @param type of items to write + * @since 5.2.0 + * @author Mahmoud Ben Hassine + */ +public class BlockingQueueItemWriterBuilder { + + private BlockingQueue queue; + + /** + * Create a new {@link BlockingQueueItemWriterBuilder} + * @param queue the queue to write items to + * @return this instance of the builder + */ + public BlockingQueueItemWriterBuilder queue(BlockingQueue queue) { + this.queue = queue; + return this; + } + + /** + * Create a configured {@link BlockingQueueItemWriter}. + * @return a configured {@link BlockingQueueItemWriter}. + */ + public BlockingQueueItemWriter build() { + Assert.state(this.queue != null, "The blocking queue is required."); + return new BlockingQueueItemWriter<>(this.queue); + } + +} diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/BlockingQueueItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/BlockingQueueItemReaderTests.java new file mode 100644 index 000000000..5806e576e --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/BlockingQueueItemReaderTests.java @@ -0,0 +1,48 @@ +/* + * Copyright 2024 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.queue; + +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.springframework.batch.item.queue.builder.BlockingQueueItemReaderBuilder; + +/** + * Test class for {@link BlockingQueueItemReader}. + * + * @author Mahmoud Ben Hassine + */ +class BlockingQueueItemReaderTests { + + @Test + void testRead() throws Exception { + // given + BlockingQueue queue = new ArrayBlockingQueue<>(10); + queue.put("foo"); + BlockingQueueItemReader reader = new BlockingQueueItemReaderBuilder().queue(queue) + .timeout(10, TimeUnit.MILLISECONDS) + .build(); + + // when & then + Assertions.assertEquals("foo", reader.read()); + Assertions.assertNull(reader.read()); + } + +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/BlockingQueueItemWriterTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/BlockingQueueItemWriterTests.java new file mode 100644 index 000000000..cfd47b26f --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/BlockingQueueItemWriterTests.java @@ -0,0 +1,49 @@ +/* + * Copyright 2024 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.queue; + +import java.util.List; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +import org.junit.jupiter.api.Test; + +import org.springframework.batch.item.Chunk; +import org.springframework.batch.item.queue.builder.BlockingQueueItemWriterBuilder; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Test class for {@link BlockingQueueItemWriter}. + * + * @author Mahmoud Ben Hassine + */ +class BlockingQueueItemWriterTests { + + @Test + void testWrite() throws Exception { + // given + BlockingQueue queue = new ArrayBlockingQueue<>(10); + BlockingQueueItemWriter writer = new BlockingQueueItemWriterBuilder().queue(queue).build(); + + // when + writer.write(Chunk.of("foo", "bar")); + + // then + assertTrue(queue.containsAll(List.of("foo", "bar"))); + } + +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/builder/BlockingQueueItemReaderBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/builder/BlockingQueueItemReaderBuilderTests.java new file mode 100644 index 000000000..1676e5051 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/builder/BlockingQueueItemReaderBuilderTests.java @@ -0,0 +1,55 @@ +/* + * Copyright 2024 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.queue.builder; + +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +import org.junit.jupiter.api.Test; + +import org.springframework.batch.item.queue.BlockingQueueItemReader; +import org.springframework.test.util.ReflectionTestUtils; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Test class for {@link BlockingQueueItemReaderBuilder}. + * + * @author Mahmoud Ben Hassine + */ +class BlockingQueueItemReaderBuilderTests { + + @Test + void testMandatoryQueue() { + assertThrows(IllegalStateException.class, () -> new BlockingQueueItemReaderBuilder().build()); + } + + @Test + void testBuildReader() { + // given + BlockingQueue queue = new ArrayBlockingQueue<>(5); + + // when + BlockingQueueItemReader reader = new BlockingQueueItemReaderBuilder().queue(queue).build(); + + // then + assertNotNull(reader); + assertEquals(queue, ReflectionTestUtils.getField(reader, "queue")); + } + +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/builder/BlockingQueueItemWriterBuilderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/builder/BlockingQueueItemWriterBuilderTests.java new file mode 100644 index 000000000..6a8eec4cd --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/queue/builder/BlockingQueueItemWriterBuilderTests.java @@ -0,0 +1,56 @@ +/* + * Copyright 2024 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.queue.builder; + +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; + +import org.junit.jupiter.api.Test; + +import org.springframework.batch.item.queue.BlockingQueueItemReader; +import org.springframework.batch.item.queue.BlockingQueueItemWriter; +import org.springframework.test.util.ReflectionTestUtils; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +/** + * Test class for {@link BlockingQueueItemWriterBuilder}. + * + * @author Mahmoud Ben Hassine + */ +class BlockingQueueItemWriterBuilderTests { + + @Test + void testMandatoryQueue() { + assertThrows(IllegalStateException.class, () -> new BlockingQueueItemWriterBuilder().build()); + } + + @Test + void testBuildWriter() { + // given + BlockingQueue queue = new ArrayBlockingQueue<>(5); + + // when + BlockingQueueItemWriter writer = new BlockingQueueItemWriterBuilder().queue(queue).build(); + + // then + assertNotNull(writer); + assertEquals(queue, ReflectionTestUtils.getField(writer, "queue")); + } + +} \ No newline at end of file