CLOSED - issue BATCH-979: Insert Apache license header in Java sources (where missing)
This commit is contained in:
dsyer
@@ -1,37 +1,53 @@
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import org.springframework.batch.item.ItemWriterException;
|
||||
|
||||
/**
|
||||
* Exception indicating that a failure or early completion condition was
|
||||
* detected in a remote worker.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class AsynchronousFailureException extends ItemWriterException {
|
||||
|
||||
/**
|
||||
* Create a new {@link AsynchronousFailureException} based on a message and
|
||||
* another exception.
|
||||
*
|
||||
* @param message
|
||||
* the message for this exception
|
||||
* @param cause
|
||||
* the other exception
|
||||
*/
|
||||
public AsynchronousFailureException(String message, Throwable cause) {
|
||||
super(message, cause);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new {@link AsynchronousFailureException} based on a message.
|
||||
*
|
||||
* @param message
|
||||
* the message for this exception
|
||||
*/
|
||||
public AsynchronousFailureException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2006-2007 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.integration.chunk;
|
||||
|
||||
import org.springframework.batch.item.ItemWriterException;
|
||||
|
||||
/**
|
||||
* Exception indicating that a failure or early completion condition was
|
||||
* detected in a remote worker.
|
||||
*
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public class AsynchronousFailureException extends ItemWriterException {
|
||||
|
||||
/**
|
||||
* Create a new {@link AsynchronousFailureException} based on a message and
|
||||
* another exception.
|
||||
*
|
||||
* @param message
|
||||
* the message for this exception
|
||||
* @param cause
|
||||
* the other exception
|
||||
*/
|
||||
public AsynchronousFailureException(String message, Throwable cause) {
|
||||
super(message, cause);
|
||||
}
|
||||
|
||||
/**
|
||||
* Create a new {@link AsynchronousFailureException} based on a message.
|
||||
*
|
||||
* @param message
|
||||
* the message for this exception
|
||||
*/
|
||||
public AsynchronousFailureException(String message) {
|
||||
super(message);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,7 +1,23 @@
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
public interface ChunkHandler<T> {
|
||||
|
||||
ChunkResponse handleChunk(ChunkRequest<T> chunk);
|
||||
|
||||
/*
|
||||
* Copyright 2006-2007 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.integration.chunk;
|
||||
|
||||
public interface ChunkHandler<T> {
|
||||
|
||||
ChunkResponse handleChunk(ChunkRequest<T> chunk);
|
||||
|
||||
}
|
||||
@@ -1,179 +1,195 @@
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.ExitStatus;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemStreamException;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.integration.gateway.MessagingGateway;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSupport implements ItemWriter<T>, ItemStream {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class);
|
||||
|
||||
static final String ACTUAL = ChunkMessageChannelItemWriter.class.getName() + ".ACTUAL";
|
||||
|
||||
static final String EXPECTED = ChunkMessageChannelItemWriter.class.getName() + ".EXPECTED";
|
||||
|
||||
private static final long DEFAULT_THROTTLE_LIMIT = 6;
|
||||
|
||||
private MessagingGateway messagingGateway;
|
||||
|
||||
private LocalState localState = new LocalState();
|
||||
|
||||
private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
|
||||
|
||||
/**
|
||||
* Public setter for the throttle limit. This limits the number of pending
|
||||
* requests for chunk processing to avoid overwhelming the receivers.
|
||||
* @param throttleLimit the throttle limit to set
|
||||
*/
|
||||
public void setThrottleLimit(long throttleLimit) {
|
||||
this.throttleLimit = throttleLimit;
|
||||
}
|
||||
|
||||
public void setMessagingGateway(MessagingGateway messagingGateway) {
|
||||
this.messagingGateway = messagingGateway;
|
||||
}
|
||||
|
||||
public void write(List<? extends T> items) throws Exception {
|
||||
|
||||
// Block until expecting <= throttle limit
|
||||
while (localState.getExpecting() > throttleLimit) {
|
||||
getNextResult();
|
||||
}
|
||||
|
||||
if (!items.isEmpty()) {
|
||||
|
||||
logger.debug("Dispatching chunk: " + items);
|
||||
ChunkRequest<T> request = new ChunkRequest<T>(items, localState.getJobId(), localState
|
||||
.createStepContribution());
|
||||
messagingGateway.send(request);
|
||||
localState.expected++;
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void beforeStep(StepExecution stepExecution) {
|
||||
localState.setStepExecution(stepExecution);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ExitStatus afterStep(StepExecution stepExecution) {
|
||||
if (!(stepExecution.getStatus() == BatchStatus.COMPLETED)) {
|
||||
return ExitStatus.EXECUTING;
|
||||
}
|
||||
long expecting = localState.getExpecting();
|
||||
boolean timedOut;
|
||||
try {
|
||||
logger.debug("Waiting for results in step listener...");
|
||||
timedOut = !waitForResults();
|
||||
logger.debug("Finished waiting for results in step listener.");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
logger.debug("Detected failure waiting for results in step listener.", e);
|
||||
stepExecution.setStatus(BatchStatus.FAILED);
|
||||
return ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + e.getMessage());
|
||||
}
|
||||
if (timedOut) {
|
||||
stepExecution.setStatus(BatchStatus.FAILED);
|
||||
throw new ItemStreamException("Timed out waiting for back log at end of step");
|
||||
}
|
||||
return ExitStatus.FINISHED.addExitDescription("Waited for " + expecting + " results.");
|
||||
}
|
||||
|
||||
public void close() throws ItemStreamException {
|
||||
localState.reset();
|
||||
}
|
||||
|
||||
public void open(ExecutionContext executionContext) throws ItemStreamException {
|
||||
if (executionContext.containsKey(EXPECTED)) {
|
||||
localState.expected = executionContext.getLong(EXPECTED);
|
||||
localState.actual = executionContext.getLong(ACTUAL);
|
||||
if (!waitForResults()) {
|
||||
throw new ItemStreamException("Timed out waiting for back log on open");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void update(ExecutionContext executionContext) throws ItemStreamException {
|
||||
executionContext.putLong(EXPECTED, localState.expected);
|
||||
executionContext.putLong(ACTUAL, localState.actual);
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait until all the results that are in the pipeline come back to the
|
||||
* reply channel.
|
||||
*
|
||||
* @return true if successfully received a result, false if timed out
|
||||
*/
|
||||
private boolean waitForResults() {
|
||||
// TODO: cumulative timeout, or throw an exception?
|
||||
int count = 0;
|
||||
int maxCount = 40;
|
||||
while (localState.getExpecting() > 0 && count++ < maxCount) {
|
||||
getNextResult();
|
||||
}
|
||||
return count < maxCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the next result if it is available within the timeout specified,
|
||||
* otherwise return null.
|
||||
*/
|
||||
private void getNextResult() {
|
||||
ChunkResponse payload = (ChunkResponse) messagingGateway.receive();
|
||||
if (payload != null) {
|
||||
Long jobInstanceId = payload.getJobId();
|
||||
Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
|
||||
Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["
|
||||
+ jobInstanceId + "] should have been [" + localState.getJobId() + "].");
|
||||
localState.actual++;
|
||||
// TODO: apply the skip count
|
||||
if (!payload.isSuccessful()) {
|
||||
throw new AsynchronousFailureException("Failure or interrupt detected in handler: "
|
||||
+ payload.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static class LocalState {
|
||||
private long actual;
|
||||
|
||||
private long expected;
|
||||
|
||||
private StepExecution stepExecution;
|
||||
|
||||
public long getExpecting() {
|
||||
return expected - actual;
|
||||
}
|
||||
|
||||
public StepContribution createStepContribution() {
|
||||
return stepExecution.createStepContribution();
|
||||
}
|
||||
|
||||
public Long getJobId() {
|
||||
return stepExecution.getJobExecution().getJobId();
|
||||
}
|
||||
|
||||
public void setStepExecution(StepExecution stepExecution) {
|
||||
this.stepExecution = stepExecution;
|
||||
}
|
||||
|
||||
public void reset() {
|
||||
expected = actual = 0;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2006-2007 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.integration.chunk;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.core.BatchStatus;
|
||||
import org.springframework.batch.core.ExitStatus;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.StepExecution;
|
||||
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
|
||||
import org.springframework.batch.item.ExecutionContext;
|
||||
import org.springframework.batch.item.ItemStream;
|
||||
import org.springframework.batch.item.ItemStreamException;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.integration.gateway.MessagingGateway;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
public class ChunkMessageChannelItemWriter<T> extends StepExecutionListenerSupport implements ItemWriter<T>, ItemStream {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(ChunkMessageChannelItemWriter.class);
|
||||
|
||||
static final String ACTUAL = ChunkMessageChannelItemWriter.class.getName() + ".ACTUAL";
|
||||
|
||||
static final String EXPECTED = ChunkMessageChannelItemWriter.class.getName() + ".EXPECTED";
|
||||
|
||||
private static final long DEFAULT_THROTTLE_LIMIT = 6;
|
||||
|
||||
private MessagingGateway messagingGateway;
|
||||
|
||||
private LocalState localState = new LocalState();
|
||||
|
||||
private long throttleLimit = DEFAULT_THROTTLE_LIMIT;
|
||||
|
||||
/**
|
||||
* Public setter for the throttle limit. This limits the number of pending
|
||||
* requests for chunk processing to avoid overwhelming the receivers.
|
||||
* @param throttleLimit the throttle limit to set
|
||||
*/
|
||||
public void setThrottleLimit(long throttleLimit) {
|
||||
this.throttleLimit = throttleLimit;
|
||||
}
|
||||
|
||||
public void setMessagingGateway(MessagingGateway messagingGateway) {
|
||||
this.messagingGateway = messagingGateway;
|
||||
}
|
||||
|
||||
public void write(List<? extends T> items) throws Exception {
|
||||
|
||||
// Block until expecting <= throttle limit
|
||||
while (localState.getExpecting() > throttleLimit) {
|
||||
getNextResult();
|
||||
}
|
||||
|
||||
if (!items.isEmpty()) {
|
||||
|
||||
logger.debug("Dispatching chunk: " + items);
|
||||
ChunkRequest<T> request = new ChunkRequest<T>(items, localState.getJobId(), localState
|
||||
.createStepContribution());
|
||||
messagingGateway.send(request);
|
||||
localState.expected++;
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void beforeStep(StepExecution stepExecution) {
|
||||
localState.setStepExecution(stepExecution);
|
||||
}
|
||||
|
||||
@Override
|
||||
public ExitStatus afterStep(StepExecution stepExecution) {
|
||||
if (!(stepExecution.getStatus() == BatchStatus.COMPLETED)) {
|
||||
return ExitStatus.EXECUTING;
|
||||
}
|
||||
long expecting = localState.getExpecting();
|
||||
boolean timedOut;
|
||||
try {
|
||||
logger.debug("Waiting for results in step listener...");
|
||||
timedOut = !waitForResults();
|
||||
logger.debug("Finished waiting for results in step listener.");
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
logger.debug("Detected failure waiting for results in step listener.", e);
|
||||
stepExecution.setStatus(BatchStatus.FAILED);
|
||||
return ExitStatus.FAILED.addExitDescription(e.getClass().getName() + ": " + e.getMessage());
|
||||
}
|
||||
if (timedOut) {
|
||||
stepExecution.setStatus(BatchStatus.FAILED);
|
||||
throw new ItemStreamException("Timed out waiting for back log at end of step");
|
||||
}
|
||||
return ExitStatus.FINISHED.addExitDescription("Waited for " + expecting + " results.");
|
||||
}
|
||||
|
||||
public void close() throws ItemStreamException {
|
||||
localState.reset();
|
||||
}
|
||||
|
||||
public void open(ExecutionContext executionContext) throws ItemStreamException {
|
||||
if (executionContext.containsKey(EXPECTED)) {
|
||||
localState.expected = executionContext.getLong(EXPECTED);
|
||||
localState.actual = executionContext.getLong(ACTUAL);
|
||||
if (!waitForResults()) {
|
||||
throw new ItemStreamException("Timed out waiting for back log on open");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void update(ExecutionContext executionContext) throws ItemStreamException {
|
||||
executionContext.putLong(EXPECTED, localState.expected);
|
||||
executionContext.putLong(ACTUAL, localState.actual);
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait until all the results that are in the pipeline come back to the
|
||||
* reply channel.
|
||||
*
|
||||
* @return true if successfully received a result, false if timed out
|
||||
*/
|
||||
private boolean waitForResults() {
|
||||
// TODO: cumulative timeout, or throw an exception?
|
||||
int count = 0;
|
||||
int maxCount = 40;
|
||||
while (localState.getExpecting() > 0 && count++ < maxCount) {
|
||||
getNextResult();
|
||||
}
|
||||
return count < maxCount;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the next result if it is available within the timeout specified,
|
||||
* otherwise return null.
|
||||
*/
|
||||
private void getNextResult() {
|
||||
ChunkResponse payload = (ChunkResponse) messagingGateway.receive();
|
||||
if (payload != null) {
|
||||
Long jobInstanceId = payload.getJobId();
|
||||
Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
|
||||
Assert.state(jobInstanceId.equals(localState.getJobId()), "Message contained wrong job instance id ["
|
||||
+ jobInstanceId + "] should have been [" + localState.getJobId() + "].");
|
||||
localState.actual++;
|
||||
// TODO: apply the skip count
|
||||
if (!payload.isSuccessful()) {
|
||||
throw new AsynchronousFailureException("Failure or interrupt detected in handler: "
|
||||
+ payload.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static class LocalState {
|
||||
private long actual;
|
||||
|
||||
private long expected;
|
||||
|
||||
private StepExecution stepExecution;
|
||||
|
||||
public long getExpecting() {
|
||||
return expected - actual;
|
||||
}
|
||||
|
||||
public StepContribution createStepContribution() {
|
||||
return stepExecution.createStepContribution();
|
||||
}
|
||||
|
||||
public Long getJobId() {
|
||||
return stepExecution.getJobExecution().getJobId();
|
||||
}
|
||||
|
||||
public void setStepExecution(StepExecution stepExecution) {
|
||||
this.stepExecution = stepExecution;
|
||||
}
|
||||
|
||||
public void reset() {
|
||||
expected = actual = 0;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,61 +1,77 @@
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.step.item.ChunkProcessor;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.integration.annotation.MessageEndpoint;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
@MessageEndpoint
|
||||
public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, InitializingBean {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class);
|
||||
|
||||
private ChunkProcessor<S> chunkProcessor;
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
*
|
||||
* @see
|
||||
* org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
|
||||
*/
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
Assert.notNull(chunkProcessor, "A ChunkProcessor must be provided");
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link ChunkProcessor}.
|
||||
*
|
||||
* @param chunkProcessor the chunkProcessor to set
|
||||
*/
|
||||
public void setChunkProcessor(ChunkProcessor<S> chunkProcessor) {
|
||||
this.chunkProcessor = chunkProcessor;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @see ChunkHandler#handleChunk(ChunkRequest)
|
||||
*/
|
||||
@ServiceActivator
|
||||
public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) {
|
||||
|
||||
logger.debug("Handling chunk: " + chunkRequest);
|
||||
|
||||
StepContribution stepContribution = chunkRequest.getStepContribution();
|
||||
try {
|
||||
chunkProcessor.process(stepContribution, chunkRequest.getChunk());
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.debug("Failed chunk", e);
|
||||
return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, e.getClass().getName() + ": "
|
||||
+ e.getMessage());
|
||||
}
|
||||
|
||||
logger.debug("Completed chunk handling with " + stepContribution);
|
||||
return new ChunkResponse(true, chunkRequest.getJobId(), stepContribution);
|
||||
|
||||
}
|
||||
}
|
||||
/*
|
||||
* Copyright 2006-2007 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.integration.chunk;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.step.item.ChunkProcessor;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.integration.annotation.MessageEndpoint;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
import org.springframework.util.Assert;
|
||||
|
||||
@MessageEndpoint
|
||||
public class ChunkProcessorChunkHandler<S> implements ChunkHandler<S>, InitializingBean {
|
||||
|
||||
private static final Log logger = LogFactory.getLog(ChunkProcessorChunkHandler.class);
|
||||
|
||||
private ChunkProcessor<S> chunkProcessor;
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
*
|
||||
* @see
|
||||
* org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
|
||||
*/
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
Assert.notNull(chunkProcessor, "A ChunkProcessor must be provided");
|
||||
}
|
||||
|
||||
/**
|
||||
* Public setter for the {@link ChunkProcessor}.
|
||||
*
|
||||
* @param chunkProcessor the chunkProcessor to set
|
||||
*/
|
||||
public void setChunkProcessor(ChunkProcessor<S> chunkProcessor) {
|
||||
this.chunkProcessor = chunkProcessor;
|
||||
}
|
||||
|
||||
/**
|
||||
*
|
||||
* @see ChunkHandler#handleChunk(ChunkRequest)
|
||||
*/
|
||||
@ServiceActivator
|
||||
public ChunkResponse handleChunk(ChunkRequest<S> chunkRequest) {
|
||||
|
||||
logger.debug("Handling chunk: " + chunkRequest);
|
||||
|
||||
StepContribution stepContribution = chunkRequest.getStepContribution();
|
||||
try {
|
||||
chunkProcessor.process(stepContribution, chunkRequest.getChunk());
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.debug("Failed chunk", e);
|
||||
return new ChunkResponse(false, chunkRequest.getJobId(), stepContribution, e.getClass().getName() + ": "
|
||||
+ e.getMessage());
|
||||
}
|
||||
|
||||
logger.debug("Completed chunk handling with " + stepContribution);
|
||||
return new ChunkResponse(true, chunkRequest.getJobId(), stepContribution);
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,44 +1,60 @@
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Collection;
|
||||
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.step.item.Chunk;
|
||||
|
||||
public class ChunkRequest<T> implements Serializable {
|
||||
|
||||
private final Long jobId;
|
||||
private final Chunk<T> items;
|
||||
private final StepContribution stepContribution;
|
||||
|
||||
public ChunkRequest(Collection<? extends T> items, Long jobId, StepContribution stepContribution) {
|
||||
this.items = new Chunk<T>(items);
|
||||
this.jobId = jobId;
|
||||
this.stepContribution = stepContribution;
|
||||
}
|
||||
|
||||
public Long getJobId() {
|
||||
return jobId;
|
||||
}
|
||||
|
||||
public Chunk<T> getChunk() {
|
||||
return items;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the {@link StepContribution} for this chunk
|
||||
*/
|
||||
public StepContribution getStepContribution() {
|
||||
return stepContribution;
|
||||
}
|
||||
|
||||
/**
|
||||
* @see java.lang.Object#toString()
|
||||
*/
|
||||
@Override
|
||||
public String toString() {
|
||||
return getClass().getSimpleName()+": jobId="+jobId+", contribution="+stepContribution+", item count="+items.size();
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2006-2007 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.integration.chunk;
|
||||
|
||||
import java.io.Serializable;
|
||||
import java.util.Collection;
|
||||
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
import org.springframework.batch.core.step.item.Chunk;
|
||||
|
||||
public class ChunkRequest<T> implements Serializable {
|
||||
|
||||
private final Long jobId;
|
||||
private final Chunk<T> items;
|
||||
private final StepContribution stepContribution;
|
||||
|
||||
public ChunkRequest(Collection<? extends T> items, Long jobId, StepContribution stepContribution) {
|
||||
this.items = new Chunk<T>(items);
|
||||
this.jobId = jobId;
|
||||
this.stepContribution = stepContribution;
|
||||
}
|
||||
|
||||
public Long getJobId() {
|
||||
return jobId;
|
||||
}
|
||||
|
||||
public Chunk<T> getChunk() {
|
||||
return items;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the {@link StepContribution} for this chunk
|
||||
*/
|
||||
public StepContribution getStepContribution() {
|
||||
return stepContribution;
|
||||
}
|
||||
|
||||
/**
|
||||
* @see java.lang.Object#toString()
|
||||
*/
|
||||
@Override
|
||||
public String toString() {
|
||||
return getClass().getSimpleName()+": jobId="+jobId+", contribution="+stepContribution+", item count="+items.size();
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,53 +1,69 @@
|
||||
package org.springframework.batch.integration.chunk;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
|
||||
public class ChunkResponse implements Serializable {
|
||||
|
||||
private final StepContribution stepContribution;
|
||||
private final Long jobId;
|
||||
private final boolean status;
|
||||
private final String message;
|
||||
|
||||
public ChunkResponse(Long jobId, StepContribution stepContribution) {
|
||||
this(true, jobId, stepContribution, null);
|
||||
}
|
||||
|
||||
public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution) {
|
||||
this(status, jobId, stepContribution, null);
|
||||
}
|
||||
|
||||
public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message) {
|
||||
this.status = status;
|
||||
this.jobId = jobId;
|
||||
this.stepContribution = stepContribution;
|
||||
this.message = message;
|
||||
}
|
||||
|
||||
public StepContribution getStepContribution() {
|
||||
return stepContribution;
|
||||
}
|
||||
|
||||
public Long getJobId() {
|
||||
return jobId;
|
||||
}
|
||||
|
||||
public boolean isSuccessful() {
|
||||
return status;
|
||||
}
|
||||
|
||||
public String getMessage() {
|
||||
return message;
|
||||
}
|
||||
|
||||
/**
|
||||
* @see java.lang.Object#toString()
|
||||
*/
|
||||
@Override
|
||||
public String toString() {
|
||||
return getClass().getSimpleName()+": jobId="+jobId+", stepContribution="+stepContribution+", successful="+status;
|
||||
}
|
||||
|
||||
}
|
||||
/*
|
||||
* Copyright 2006-2007 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.integration.chunk;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
import org.springframework.batch.core.StepContribution;
|
||||
|
||||
public class ChunkResponse implements Serializable {
|
||||
|
||||
private final StepContribution stepContribution;
|
||||
private final Long jobId;
|
||||
private final boolean status;
|
||||
private final String message;
|
||||
|
||||
public ChunkResponse(Long jobId, StepContribution stepContribution) {
|
||||
this(true, jobId, stepContribution, null);
|
||||
}
|
||||
|
||||
public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution) {
|
||||
this(status, jobId, stepContribution, null);
|
||||
}
|
||||
|
||||
public ChunkResponse(boolean status, Long jobId, StepContribution stepContribution, String message) {
|
||||
this.status = status;
|
||||
this.jobId = jobId;
|
||||
this.stepContribution = stepContribution;
|
||||
this.message = message;
|
||||
}
|
||||
|
||||
public StepContribution getStepContribution() {
|
||||
return stepContribution;
|
||||
}
|
||||
|
||||
public Long getJobId() {
|
||||
return jobId;
|
||||
}
|
||||
|
||||
public boolean isSuccessful() {
|
||||
return status;
|
||||
}
|
||||
|
||||
public String getMessage() {
|
||||
return message;
|
||||
}
|
||||
|
||||
/**
|
||||
* @see java.lang.Object#toString()
|
||||
*/
|
||||
@Override
|
||||
public String toString() {
|
||||
return getClass().getSimpleName()+": jobId="+jobId+", stepContribution="+stepContribution+", successful="+status;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,3 +1,19 @@
|
||||
/*
|
||||
* Copyright 2006-2007 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.integration.file;
|
||||
|
||||
import org.springframework.batch.core.JobParameters;
|
||||
|
||||
@@ -1,3 +1,19 @@
|
||||
/*
|
||||
* Copyright 2006-2007 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.integration.launch;
|
||||
|
||||
import org.springframework.batch.core.Job;
|
||||
|
||||
Reference in New Issue
Block a user