Spring/spring-batch
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Replace explicit type with diamond operator in integration module

Browse Source
This commit is contained in:
Drummond Dawson
2018-10-09 19:53:42 -04:00
committed by Mahmoud Ben Hassine
parent eb9e0d1466
commit 3fe5fc3427
26 changed files with 71 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemProcessor.java
View File

@@ -93,7 +93,7 @@ public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, In
*/
public Future<O> process(final I item) throws Exception {
final StepExecution stepExecution = getStepExecution();
FutureTask<O> task = new FutureTask<O>(new Callable<O>() {
FutureTask<O> task = new FutureTask<>(new Callable<O>() {
public O call() throws Exception {
if (stepExecution != null) {
StepSynchronizationManager.register(stepExecution);

2
spring-batch-integration/src/main/java/org/springframework/batch/integration/async/AsyncItemWriter.java
View File

@@ -59,7 +59,7 @@ public class AsyncItemWriter<T> implements ItemStreamWriter<Future<T>>, Initiali
* @throws Exception The exception returned by the Future if one was thrown
*/
public void write(List<? extends Future<T>> items) throws Exception {
List<T> list = new ArrayList<T>();
List<T> list = new ArrayList<>();
for (Future<T> future : items) {
try {
T item = future.get();

10
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkMessageChannelItemWriter.java
View File

@@ -104,7 +104,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
if (logger.isDebugEnabled()) {
logger.debug("Dispatching chunk: " + request);
}
messagingGateway.send(new GenericMessage<ChunkRequest<T>>(request));
messagingGateway.send(new GenericMessage<>(request));
localState.incrementExpected();
}
@@ -171,7 +171,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
}
public Collection<StepContribution> getStepContributions() {
List<StepContribution> contributions = new ArrayList<StepContribution>();
List<StepContribution> contributions = new ArrayList<>();
for (ChunkResponse response : localState.pollChunkResponses()) {
StepContribution contribution = response.getStepContribution();
if (logger.isDebugEnabled()) {
@@ -274,14 +274,14 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
private StepExecution stepExecution;
private final Queue<ChunkResponse> contributions = new LinkedBlockingQueue<ChunkResponse>();
private final Queue<ChunkResponse> contributions = new LinkedBlockingQueue<>();
public int getExpecting() {
return expected.get() - actual.get();
}
public <T> ChunkRequest<T> getRequest(List<? extends T> items) {
return new ChunkRequest<T>(current.incrementAndGet(), items, getJobId(), createStepContribution());
return new ChunkRequest<>(current.incrementAndGet(), items, getJobId(), createStepContribution());
}
public void open(int expectedValue, int actualValue) {
@@ -290,7 +290,7 @@ public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSuppo
}
public Collection<ChunkResponse> pollChunkResponses() {
Collection<ChunkResponse> set = new ArrayList<ChunkResponse>();
Collection<ChunkResponse> set = new ArrayList<>();
synchronized (contributions) {
ChunkResponse item = contributions.poll();
while (item != null) {

2
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkProcessorChunkHandler.java
View File

@@ -101,7 +101,7 @@ public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, Initializ
*/
private Throwable process(ChunkRequest<S> chunkRequest, StepContribution stepContribution) throws Exception {
Chunk<S> chunk = new Chunk<S>(chunkRequest.getItems());
Chunk<S> chunk = new Chunk<>(chunkRequest.getItems());
Throwable failure = null;
try {
chunkProcessor.process(stepContribution, chunk);

4
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/RemoteChunkHandlerFactoryBean.java
View File

@@ -139,7 +139,7 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
step.registerStepExecutionListener((StepExecutionListener) chunkWriter);
}
ChunkProcessorChunkHandler<T> handler = new ChunkProcessorChunkHandler<T>();
ChunkProcessorChunkHandler<T> handler = new ChunkProcessorChunkHandler<>();
setNonBuffering(chunkProcessor);
handler.setChunkProcessor(chunkProcessor);
// TODO: create step context for the processor in case it has
@@ -170,7 +170,7 @@ public class RemoteChunkHandlerFactoryBean<T> implements FactoryBean<ChunkHandle
*/
private void replaceChunkProcessor(ChunkOrientedTasklet<?> tasklet, ItemWriter<T> chunkWriter,
final StepContributionSource stepContributionSource) {
setField(tasklet, "chunkProcessor", new SimpleChunkProcessor<T, T>(new PassThroughItemProcessor<T>(),
setField(tasklet, "chunkProcessor", new SimpleChunkProcessor<T, T>(new PassThroughItemProcessor<>(),
chunkWriter) {
@Override
protected void write(StepContribution contribution, Chunk<T> inputs, Chunk<T> outputs) throws Exception {

4
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/MessageChannelPartitionHandler.java
View File

@@ -237,7 +237,7 @@ public class MessageChannelPartitionHandler implements PartitionHandler, Initial
}
private Collection<StepExecution> pollReplies(final StepExecution masterStepExecution, final Set<StepExecution> split) throws Exception {
final Collection<StepExecution> result = new ArrayList<StepExecution>(split.size());
final Collection<StepExecution> result = new ArrayList<>(split.size());
Callable<Collection<StepExecution>> callback = new Callable<Collection<StepExecution>>() {
@Override
@@ -269,7 +269,7 @@ public class MessageChannelPartitionHandler implements PartitionHandler, Initial
}
};
Poller<Collection<StepExecution>> poller = new DirectPoller<Collection<StepExecution>>(pollInterval);
Poller<Collection<StepExecution>> poller = new DirectPoller<>(pollInterval);
Future<Collection<StepExecution>> resultsFuture = poller.poll(callback);
if(timeout >= 0) {