Introduce SynchronizedItemStreamWriter

Issue #842
This commit is contained in:
Dimitrios Liapis
2018-10-30 02:32:15 +01:00
committed by Mahmoud Ben Hassine
parent 2f8aff2f38
commit 117b7d5f2d
6 changed files with 290 additions and 0 deletions

View File

@@ -3000,6 +3000,7 @@ Spring Batch includes the following decorators:
* <<synchronizedItemStreamReader>>
* <<singleItemPeekableItemReader>>
* <<synchronizedItemStreamWriter>>
* <<multiResourceItemWriter>>
* <<classifierCompositeItemWriter>>
* <<classifierCompositeItemProcessor>>
@@ -3023,6 +3024,13 @@ NOTE: SingleItemPeekableItemReader's peek method is not thread-safe, because it
be possible to honor the peek in multiple threads. Only one of the threads that peeked
would get that item in the next call to read.
[[synchronizedItemStreamWriter]]
===== `SynchronizedItemStreamWriter`
When using an `ItemWriter` that is not thread safe, Spring Batch offers the
`SynchronizedItemStreamWriter` decorator, which can be used to make the `ItemWriter`
thread safe. Spring Batch provides a `SynchronizedItemStreamWriterBuilder` to construct
an instance of the `SynchronizedItemStreamWriter`.
[[multiResourceItemWriter]]
===== `MultiResourceItemWriter`
The `MultiResourceItemWriter` wraps a `ResourceAwareItemWriterItemStream` and creates a new

View File

@@ -0,0 +1,69 @@
/*
* Copyright 2018 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.ItemStreamException;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import java.util.List;
/**
* An {@link ItemStreamWriter} decorator with a synchronized
* {@link SynchronizedItemStreamWriter#write write()} method
*
* @author Dimitrios Liapis
*
* @param <T> type of object being written
*/
public class SynchronizedItemStreamWriter<T> implements ItemStreamWriter<T>, InitializingBean {
private ItemStreamWriter<T> delegate;
public void setDelegate(ItemStreamWriter<T> delegate) {
this.delegate = delegate;
}
/**
* This delegates to the write method of the <code>delegate</code>
*/
@Override
public synchronized void write(List<? extends T> items) throws Exception {
this.delegate.write(items);
}
@Override
public void open(ExecutionContext executionContext) throws ItemStreamException {
this.delegate.open(executionContext);
}
@Override
public void update(ExecutionContext executionContext) throws ItemStreamException {
this.delegate.update(executionContext);
}
@Override
public void close() throws ItemStreamException {
this.delegate.close();
}
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.delegate, "A delegate item writer is required");
}
}

View File

@@ -0,0 +1,44 @@
/*
* Copyright 2018 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.builder;
import org.springframework.batch.item.ItemStreamWriter;
import org.springframework.batch.item.support.SynchronizedItemStreamWriter;
import org.springframework.util.Assert;
/**
* Creates a fully qualified {@link SynchronizedItemStreamWriter}.
*
* @author Dimitrios Liapis
*/
public class SynchronizedItemStreamWriterBuilder<T> {
private ItemStreamWriter<T> delegate;
public SynchronizedItemStreamWriterBuilder<T> delegate(ItemStreamWriter<T> delegate) {
this.delegate = delegate;
return this;
}
public SynchronizedItemStreamWriter<T> build() {
Assert.notNull(this.delegate, "A delegate is required");
SynchronizedItemStreamWriter<T> writer = new SynchronizedItemStreamWriter<>();
writer.setDelegate(this.delegate);
return writer;
}
}

View File

@@ -0,0 +1,84 @@
/*
* Copyright 2018 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.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.mockito.Mock;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ItemStreamWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import static org.mockito.Mockito.verify;
import static org.mockito.MockitoAnnotations.initMocks;
/**
* Common parent class for {@link SynchronizedItemStreamWriterTests} and
* {@link org.springframework.batch.item.support.builder.SynchronizedItemStreamWriterBuilderTests}
*
* @author Dimitrios Liapis
*
*/
public abstract class AbstractSynchronizedItemStreamWriterTests {
@Rule
public ExpectedException expectedException = ExpectedException.none();
@Mock
protected ItemStreamWriter<Object> delegate;
private SynchronizedItemStreamWriter<Object> synchronizedItemStreamWriter;
private final List<Object> testList = Collections.unmodifiableList(new ArrayList<>());
private final ExecutionContext testExecutionContext = new ExecutionContext();
abstract protected SynchronizedItemStreamWriter<Object> createNewSynchronizedItemStreamWriter();
@Before
public void init() {
initMocks(this);
synchronizedItemStreamWriter = createNewSynchronizedItemStreamWriter();
}
@Test
public void testDelegateWriteIsCalled() throws Exception {
synchronizedItemStreamWriter.write(testList);
verify(delegate).write(testList);
}
@Test
public void testDelegateOpenIsCalled() {
synchronizedItemStreamWriter.open(testExecutionContext);
verify(delegate).open(testExecutionContext);
}
@Test
public void testDelegateUpdateIsCalled() {
synchronizedItemStreamWriter.update(testExecutionContext);
verify(delegate).update(testExecutionContext);
}
@Test
public void testDelegateCloseIsClosed() {
synchronizedItemStreamWriter.close();
verify(delegate).close();
}
}

View File

@@ -0,0 +1,42 @@
/*
* Copyright 2018 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.Test;
import org.springframework.beans.factory.InitializingBean;
/**
*
* @author Dimitrios Liapis
*
*/
public class SynchronizedItemStreamWriterTests extends AbstractSynchronizedItemStreamWriterTests {
@Override
protected SynchronizedItemStreamWriter<Object> createNewSynchronizedItemStreamWriter() {
SynchronizedItemStreamWriter<Object> synchronizedItemStreamWriter = new SynchronizedItemStreamWriter<>();
synchronizedItemStreamWriter.setDelegate(delegate);
return synchronizedItemStreamWriter;
}
@Test
public void testDelegateIsNotNullWhenPropertiesSet() throws Exception {
expectedException.expect(IllegalArgumentException.class);
expectedException.expectMessage("A delegate item writer is required");
((InitializingBean) new SynchronizedItemStreamWriter<>()).afterPropertiesSet();
}
}

View File

@@ -0,0 +1,43 @@
/*
* Copyright 2018 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.builder;
import org.junit.Test;
import org.springframework.batch.item.support.AbstractSynchronizedItemStreamWriterTests;
import org.springframework.batch.item.support.SynchronizedItemStreamWriter;
/**
*
* @author Dimitrios Liapis
*
*/
public class SynchronizedItemStreamWriterBuilderTests extends AbstractSynchronizedItemStreamWriterTests {
@Override
protected SynchronizedItemStreamWriter<Object> createNewSynchronizedItemStreamWriter() {
return new SynchronizedItemStreamWriterBuilder<>()
.delegate(delegate)
.build();
}
@Test
public void testBuilderDelegateIsNotNull() {
expectedException.expect(IllegalArgumentException.class);
expectedException.expectMessage("A delegate is required");
new SynchronizedItemStreamWriterBuilder<>().build();
}
}