From 6cc718a78ca42cd974cefa711248f131a8c95a21 Mon Sep 17 00:00:00 2001 From: Mahmoud Ben Hassine Date: Thu, 3 Oct 2024 09:36:27 +0200 Subject: [PATCH] Add composite item reader implementation Resolves #757 --- .../item/support/CompositeItemReader.java | 87 +++++++++++ .../support/CompositeItemReaderTests.java | 110 +++++++++++++ spring-batch-samples/README.md | 11 ++ .../batch/samples/compositereader/README.md | 18 +++ .../samples/compositereader/data/persons1.csv | 2 + .../samples/compositereader/data/persons2.csv | 2 + .../samples/compositereader/sql/data.sql | 2 + .../samples/compositereader/sql/schema.sql | 2 + ...positeItemWriterSampleFunctionalTests.java | 147 ++++++++++++++++++ 9 files changed, 381 insertions(+) create mode 100644 spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemReader.java create mode 100644 spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemReaderTests.java create mode 100644 spring-batch-samples/src/main/java/org/springframework/batch/samples/compositereader/README.md create mode 100644 spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons1.csv create mode 100644 spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons2.csv create mode 100644 spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/data.sql create mode 100644 spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/schema.sql create mode 100644 spring-batch-samples/src/test/java/org/springframework/batch/samples/compositereader/CompositeItemWriterSampleFunctionalTests.java diff --git a/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemReader.java b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemReader.java new file mode 100644 index 000000000..e9b5a72d0 --- /dev/null +++ b/spring-batch-infrastructure/src/main/java/org/springframework/batch/item/support/CompositeItemReader.java @@ -0,0 +1,87 @@ +/* + * 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.support; + +import java.util.Iterator; +import java.util.List; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStreamException; +import org.springframework.batch.item.ItemStreamReader; + +/** + * Composite reader that delegates reading to a list of {@link ItemStreamReader}s. This + * implementation is not thread-safe. + * + * @author Mahmoud Ben Hassine + * @param type of objects to read + * @since 5.2 + */ +public class CompositeItemReader implements ItemStreamReader { + + private final List> delegates; + + private final Iterator> delegatesIterator; + + private ItemStreamReader currentDelegate; + + /** + * Create a new {@link CompositeItemReader}. + * @param delegates the delegate readers to read data + */ + public CompositeItemReader(List> delegates) { + this.delegates = delegates; + this.delegatesIterator = this.delegates.iterator(); + this.currentDelegate = this.delegatesIterator.hasNext() ? this.delegatesIterator.next() : null; + } + + // TODO: check if we need to open/close delegates on the fly in read() to avoid + // opening resources early for a long time + @Override + public void open(ExecutionContext executionContext) throws ItemStreamException { + for (ItemStreamReader delegate : delegates) { + delegate.open(executionContext); + } + } + + @Override + public T read() throws Exception { + if (this.currentDelegate == null) { + return null; + } + T item = currentDelegate.read(); + if (item == null) { + currentDelegate = this.delegatesIterator.hasNext() ? this.delegatesIterator.next() : null; + return read(); + } + return item; + } + + @Override + public void update(ExecutionContext executionContext) throws ItemStreamException { + if (this.currentDelegate != null) { + this.currentDelegate.update(executionContext); + } + } + + @Override + public void close() throws ItemStreamException { + for (ItemStreamReader delegate : delegates) { + delegate.close(); + } + } + +} \ No newline at end of file diff --git a/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemReaderTests.java b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemReaderTests.java new file mode 100644 index 000000000..3775c4299 --- /dev/null +++ b/spring-batch-infrastructure/src/test/java/org/springframework/batch/item/support/CompositeItemReaderTests.java @@ -0,0 +1,110 @@ +/* + * 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.support; + +import java.util.Arrays; + +import org.junit.jupiter.api.Test; + +import org.springframework.batch.item.ExecutionContext; +import org.springframework.batch.item.ItemStreamReader; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +/** + * Test class for {@link CompositeItemReader}. + * + * @author Mahmoud Ben Hassine + */ +public class CompositeItemReaderTests { + + @Test + void testCompositeItemReaderOpen() { + // given + ItemStreamReader reader1 = mock(); + ItemStreamReader reader2 = mock(); + CompositeItemReader compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2)); + ExecutionContext executionContext = new ExecutionContext(); + + // when + compositeItemReader.open(executionContext); + + // then + verify(reader1).open(executionContext); + verify(reader2).open(executionContext); + } + + @Test + void testCompositeItemReaderRead() throws Exception { + // given + ItemStreamReader reader1 = mock(); + ItemStreamReader reader2 = mock(); + CompositeItemReader compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2)); + when(reader1.read()).thenReturn("foo1", "foo2", null); + when(reader2.read()).thenReturn("bar1", "bar2", null); + + // when & then + compositeItemReader.read(); + verify(reader1, times(1)).read(); + compositeItemReader.read(); + verify(reader1, times(2)).read(); + compositeItemReader.read(); + verify(reader1, times(3)).read(); + + compositeItemReader.read(); + verify(reader2, times(2)).read(); + compositeItemReader.read(); + verify(reader2, times(3)).read(); + compositeItemReader.read(); + verify(reader2, times(3)).read(); + } + + @Test + void testCompositeItemReaderUpdate() { + // given + ItemStreamReader reader1 = mock(); + ItemStreamReader reader2 = mock(); + CompositeItemReader compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2)); + ExecutionContext executionContext = new ExecutionContext(); + + // when + compositeItemReader.update(executionContext); + + // then + verify(reader1).update(executionContext); + verifyNoInteractions(reader2); // reader1 is the current delegate in this setup + } + + @Test + void testCompositeItemReaderClose() { + // given + ItemStreamReader reader1 = mock(); + ItemStreamReader reader2 = mock(); + CompositeItemReader compositeItemReader = new CompositeItemReader<>(Arrays.asList(reader1, reader2)); + + // when + compositeItemReader.close(); + + // then + verify(reader1).close(); + verify(reader2).close(); + } + +} \ No newline at end of file diff --git a/spring-batch-samples/README.md b/spring-batch-samples/README.md index 046ca1ade..eb26857fc 100644 --- a/spring-batch-samples/README.md +++ b/spring-batch-samples/README.md @@ -26,6 +26,7 @@ Here is a list of samples with checks to indicate which features each one demons | [Hello world Job Sample](#hello-world-job-sample) | | | | | | | | | | X | | | [Amqp Job Sample](#amqp-job-sample) | | | | | | | | | | X | | | [BeanWrapperMapper Sample](#beanwrappermapper-sample) | | | | X | | | | | | | | +| [Composite ItemReader Sample](#composite-itemreader-sample) | | | | | | | X | | | | | | [Composite ItemWriter Sample](#composite-itemwriter-sample) | | | | | | | X | | | | | | [Customer Filter Sample](#customer-filter-sample) | | | | | | | | | | | X | | [Reader Writer Adapter Sample](#reader-writer-adapter-sample) | | | | | | | X | | | | | @@ -121,6 +122,16 @@ prototype according to the field names in the file. [BeanWrapperMapper Sample](src/main/java/org/springframework/batch/samples/beanwrapper/README.md) +### Composite ItemReader Sample + +This sample shows how to use a composite item reader to read data with +the same format from different data sources. + +In this sample, data items of type `Person` are read from two flat files +and a relational database table. + +[Composite reader Sample](src/main/java/org/springframework/batch/samples/compositereader/README.md) + ### Composite ItemWriter Sample This shows a common use case using a composite pattern, composing diff --git a/spring-batch-samples/src/main/java/org/springframework/batch/samples/compositereader/README.md b/spring-batch-samples/src/main/java/org/springframework/batch/samples/compositereader/README.md new file mode 100644 index 000000000..4342fbf05 --- /dev/null +++ b/spring-batch-samples/src/main/java/org/springframework/batch/samples/compositereader/README.md @@ -0,0 +1,18 @@ +## Composite ItemReader Sample + +### About + +This sample shows how to use a composite item reader to read data with +the same format from different data sources. + +In this sample, data items of type `Person` are read from two flat files +and a relational database table. + +### Run the sample + +You can run the sample from the command line as following: + +``` +$>cd spring-batch-samples +$>../mvnw -Dtest=CompositeItemReaderSampleFunctionalTests#testJobLaunch test +``` \ No newline at end of file diff --git a/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons1.csv b/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons1.csv new file mode 100644 index 000000000..839754d23 --- /dev/null +++ b/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons1.csv @@ -0,0 +1,2 @@ +1,foo1 +2,foo2 \ No newline at end of file diff --git a/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons2.csv b/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons2.csv new file mode 100644 index 000000000..e5a88e340 --- /dev/null +++ b/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/data/persons2.csv @@ -0,0 +1,2 @@ +3,bar1 +4,bar2 \ No newline at end of file diff --git a/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/data.sql b/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/data.sql new file mode 100644 index 000000000..6b99ba0b4 --- /dev/null +++ b/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/data.sql @@ -0,0 +1,2 @@ +insert into person_source values (5, 'baz1'); +insert into person_source values (6, 'baz2'); \ No newline at end of file diff --git a/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/schema.sql b/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/schema.sql new file mode 100644 index 000000000..1ab4a1366 --- /dev/null +++ b/spring-batch-samples/src/main/resources/org/springframework/batch/samples/compositereader/sql/schema.sql @@ -0,0 +1,2 @@ +create table person_source (id int primary key, name varchar(20)); +create table person_target (id int primary key, name varchar(20)); \ No newline at end of file diff --git a/spring-batch-samples/src/test/java/org/springframework/batch/samples/compositereader/CompositeItemWriterSampleFunctionalTests.java b/spring-batch-samples/src/test/java/org/springframework/batch/samples/compositereader/CompositeItemWriterSampleFunctionalTests.java new file mode 100644 index 000000000..8c90257b6 --- /dev/null +++ b/spring-batch-samples/src/test/java/org/springframework/batch/samples/compositereader/CompositeItemWriterSampleFunctionalTests.java @@ -0,0 +1,147 @@ +/* + * 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.samples.compositereader; + +import java.util.Arrays; + +import javax.sql.DataSource; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.springframework.batch.core.ExitStatus; +import org.springframework.batch.core.Job; +import org.springframework.batch.core.JobExecution; +import org.springframework.batch.core.JobParameters; +import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; +import org.springframework.batch.core.job.builder.JobBuilder; +import org.springframework.batch.core.launch.JobLauncher; +import org.springframework.batch.core.repository.JobRepository; +import org.springframework.batch.core.step.builder.StepBuilder; +import org.springframework.batch.item.database.JdbcBatchItemWriter; +import org.springframework.batch.item.database.JdbcCursorItemReader; +import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder; +import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder; +import org.springframework.batch.item.file.FlatFileItemReader; +import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder; +import org.springframework.batch.item.support.CompositeItemReader; +import org.springframework.context.ApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.core.io.ClassPathResource; +import org.springframework.jdbc.core.DataClassRowMapper; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; +import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; +import org.springframework.jdbc.support.JdbcTransactionManager; +import org.springframework.test.jdbc.JdbcTestUtils; + +public class CompositeItemWriterSampleFunctionalTests { + + record Person(int id, String name) { + } + + @Test + void testJobLaunch() throws Exception { + // given + ApplicationContext context = new AnnotationConfigApplicationContext(JobConfiguration.class); + JobLauncher jobLauncher = context.getBean(JobLauncher.class); + Job job = context.getBean(Job.class); + + // when + JobExecution jobExecution = jobLauncher.run(job, new JobParameters()); + + // then + Assertions.assertEquals(ExitStatus.COMPLETED, jobExecution.getExitStatus()); + JdbcTemplate jdbcTemplate = new JdbcTemplate(context.getBean(DataSource.class)); + int personsCount = JdbcTestUtils.countRowsInTable(jdbcTemplate, "person_target"); + Assertions.assertEquals(6, personsCount); + } + + @Configuration + @EnableBatchProcessing + static class JobConfiguration { + + @Bean + public FlatFileItemReader itemReader1() { + return new FlatFileItemReaderBuilder().name("personItemReader1") + .resource(new ClassPathResource("org/springframework/batch/samples/compositereader/data/persons1.csv")) + .delimited() + .names("id", "name") + .targetType(Person.class) + .build(); + } + + @Bean + public FlatFileItemReader itemReader2() { + return new FlatFileItemReaderBuilder().name("personItemReader2") + .resource(new ClassPathResource("org/springframework/batch/samples/compositereader/data/persons2.csv")) + .delimited() + .names("id", "name") + .targetType(Person.class) + .build(); + } + + @Bean + public JdbcCursorItemReader itemReader3() { + String sql = "select * from person_source"; + return new JdbcCursorItemReaderBuilder().name("personItemReader3") + .dataSource(dataSource()) + .sql(sql) + .rowMapper(new DataClassRowMapper<>(Person.class)) + .build(); + } + + @Bean + public CompositeItemReader itemReader() { + return new CompositeItemReader<>(Arrays.asList(itemReader1(), itemReader2(), itemReader3())); + } + + @Bean + public JdbcBatchItemWriter itemWriter() { + String sql = "insert into person_target (id, name) values (:id, :name)"; + return new JdbcBatchItemWriterBuilder().dataSource(dataSource()).sql(sql).beanMapped().build(); + } + + @Bean + public Job job(JobRepository jobRepository, JdbcTransactionManager transactionManager) { + return new JobBuilder("job", jobRepository) + .start(new StepBuilder("step", jobRepository).chunk(5, transactionManager) + .reader(itemReader()) + .writer(itemWriter()) + .build()) + .build(); + } + + @Bean + public DataSource dataSource() { + return new EmbeddedDatabaseBuilder().setType(EmbeddedDatabaseType.HSQL) + .addScript("/org/springframework/batch/core/schema-drop-hsqldb.sql") + .addScript("/org/springframework/batch/core/schema-hsqldb.sql") + .addScript("/org/springframework/batch/samples/compositereader/sql/schema.sql") + .addScript("/org/springframework/batch/samples/compositereader/sql/data.sql") + .build(); + } + + @Bean + public JdbcTransactionManager transactionManager(DataSource dataSource) { + return new JdbcTransactionManager(dataSource); + } + + } + +} \ No newline at end of file