Skip to content

[BATCH-2762] Introducing SynchronizedItemStreamWriter #665

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 5 commits into from
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions spring-batch-docs/asciidoc/readersAndWriters.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -3000,6 +3000,7 @@ Spring Batch includes the following decorators:

* <<synchronizedItemStreamReader>>
* <<singleItemPeekableItemReader>>
* <<synchronizedItemStreamWriter>>
* <<multiResourceItemWriter>>
* <<classifierCompositeItemWriter>>
* <<classifierCompositeItemProcessor>>
Expand All @@ -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
Expand Down
Original file line number Diff line number Diff line change
@@ -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");
}
}
Original file line number Diff line number Diff line change
@@ -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;
}
}
Original file line number Diff line number Diff line change
@@ -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();
}

}
Original file line number Diff line number Diff line change
@@ -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();
}
}
Original file line number Diff line number Diff line change
@@ -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();
}
}