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

BATCH-1466: removed integration module

Browse Source
This commit is contained in:
dsyer
2009-12-29 09:26:07 +00:00
parent 6f5f372375
commit ff42839b01
88 changed files with 0 additions and 6282 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-batch-integration/src/main/java/META-INF/MANIFEST.MF
View File

@@ -1,3 +0,0 @@
Manifest-Version: 1.0
Class-Path:

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

@@ -1,77 +0,0 @@
package org.springframework.batch.integration.async;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.SyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;
/**
* An {@link ItemProcessor} that delegates to a nested processor and in the
* background. To allow for background processing the return value from the
* processor is a {@link Future} which needs to be unpacked before the item can
* be used by a client.
*
* @author Dave Syer
*
* @param <I> the input object type
* @param <O> the output object type (will be wrapped in a Future)
*/
public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, InitializingBean {
private ItemProcessor<I, O> delegate;
private TaskExecutor taskExecutor = new SyncTaskExecutor();
/**
* Check mandatory properties (the {@link #setDelegate(ItemProcessor)}).
*
* @see InitializingBean#afterPropertiesSet()
*/
public void afterPropertiesSet() throws Exception {
Assert.notNull(delegate, "The delegate must be set.");
}
/**
* The {@link ItemProcessor} to use to delegate processing to in a
* background thread.
*
* @param delegate the {@link ItemProcessor} to use as a delegate
*/
public void setDelegate(ItemProcessor<I, O> delegate) {
this.delegate = delegate;
}
/**
* The {@link TaskExecutor} to use to allow the item processing to proceed
* in the background. Defaults to a {@link SyncTaskExecutor} so no threads
* are created unless this is overridden.
*
* @param taskExecutor a {@link TaskExecutor}
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
}
/**
* Transform the input by delegating to the provided item processor. The
* return value is wrapped in a {@link Future} so that clients can unpack it
* later.
*
* @see ItemProcessor#process(Object)
*/
public Future<O> process(final I item) throws Exception {
FutureTask<O> task = new FutureTask<O>(new Callable<O>() {
public O call() throws Exception {
return delegate.process(item);
}
});
taskExecutor.execute(task);
return task;
}
}

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

@@ -1,34 +0,0 @@
package org.springframework.batch.integration.async;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
public class AsyncItemWriter<T> implements ItemWriter<Future<T>>, InitializingBean {
private ItemWriter<T> delegate;
public void afterPropertiesSet() throws Exception {
Assert.notNull(delegate, "A delegate ItemWriter must be provided.");
}
/**
* @param delegate
*/
public void setDelegate(ItemWriter<T> delegate) {
this.delegate = delegate;
}
public void write(List<? extends Future<T>> items) throws Exception {
List<T> list = new ArrayList<T>();
for (Future<T> future : items) {
list.add(future.get());
}
delegate.write(list);
}
}

53
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/AsynchronousFailureException.java
View File

@@ -1,53 +0,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 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);
}
}

23
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkHandler.java
View File

@@ -1,23 +0,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;
public interface ChunkHandler<T> {
ChunkResponse handleChunk(ChunkRequest<T> chunk);
}

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

@@ -1,202 +0,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.COMPLETED.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 in
* the gateway), otherwise do nothing.
*
* @throws AsynchronousFailureException If there is a response and it
* contains a failed chunk response.
*
* @throws IllegalStateException if the result contains the wrong job
* instance id (maybe we are sharing a channel and we shouldn't be)
*/
private void getNextResult() {
// TODO: make sure this is transactional (should be if single threaded)
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;
}
}
}

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

@@ -1,78 +0,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 org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.step.item.Chunk;
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, new Chunk<S>(chunkRequest.getItems()));
}
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);
}
}

59
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkRequest.java
View File

@@ -1,59 +0,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.io.Serializable;
import java.util.Collection;
import org.springframework.batch.core.StepContribution;
public class ChunkRequest<T> implements Serializable {
private final Long jobId;
private final Collection<? extends T> items;
private final StepContribution stepContribution;
public ChunkRequest(Collection<? extends T> items, Long jobId, StepContribution stepContribution) {
this.items = items;
this.jobId = jobId;
this.stepContribution = stepContribution;
}
public Long getJobId() {
return jobId;
}
public Collection<? extends T> getItems() {
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();
}
}

69
spring-batch-integration/src/main/java/org/springframework/batch/integration/chunk/ChunkResponse.java
View File

@@ -1,69 +0,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.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;
}
}

389
spring-batch-integration/src/main/java/org/springframework/batch/integration/file/FileToMessagesJobFactoryBean.java
View File

@@ -1,389 +0,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.file;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URL;
import java.util.List;
import java.util.Properties;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.converter.DefaultJobParametersConverter;
import org.springframework.batch.core.converter.JobParametersConverter;
import org.springframework.batch.core.job.SimpleJob;
import org.springframework.batch.core.listener.StepExecutionListenerSupport;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.item.SimpleStepFactoryBean;
import org.springframework.batch.integration.launch.JobLaunchingMessageHandler;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.xml.StaxEventItemReader;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.FileSystemResourceLoader;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceLoader;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.util.Assert;
/**
* A FactoryBean for a {@link Job} with a single step which just pumps messages
* from a file into a channel. The channel has to be a {@link DirectChannel} to
* ensure that failures propagate up to the step and fail the job execution.
* Normally this job will be used in conjunction with a
* {@link JobLaunchingMessageHandler} and a
* {@link ResourcePayloadAsJobParameterStrategy}, so that the user can just send
* a message to a request channel listing the files to be processed, and
* everything else just happens by magic. After a failure the job will be
* restarted just by sending it the same message.
*
* @author Dave Syer
*
*/
public class FileToMessagesJobFactoryBean<T> implements FactoryBean, BeanNameAware {
private String name = "fileToMessageJob";
private ItemReader<? extends T> itemReader;
private MessageChannel channel;
private PlatformTransactionManager transactionManager;
private JobRepository jobRepository;
/*
* (non-Javadoc)
*
* @see
* org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang
* .String)
*/
public void setBeanName(String name) {
this.name = name;
}
/**
* Public setter for the {@link ItemReader}. Must be either a
* {@link FlatFileItemReader} or a {@link StaxEventItemReader}. In either
* case there is no need to set the resource property as it will be set by
* this factory.
*
* @param itemReader the itemReader to set
*/
@Required
public void setItemReader(ItemReader<? extends T> itemReader) {
this.itemReader = itemReader;
}
/**
* Public setter for the channel. Each item from the item reader will be
* sent to this channel.
*
* @param channel the channel to set
*/
@Required
public void setChannel(MessageChannel channel) {
this.channel = channel;
}
/**
* Public setter for the {@link JobRepository}.
*
* @param jobRepository the job repository to set
*/
@Required
public void setJobRepository(JobRepository jobRepository) {
this.jobRepository = jobRepository;
}
/**
* Public setter for the {@link PlatformTransactionManager}.
*
* @param transactionManager the transaction manager to set
*/
@Required
public void setTransactionManager(PlatformTransactionManager transactionManager) {
this.transactionManager = transactionManager;
}
/**
* Creates a {@link Job} that can process a flat file or XML file into
* messages. To launch the job will require only a {@link JobParameters}
* instance with a resource location as a URL.
*
* @see org.springframework.beans.factory.FactoryBean#getObject()
*/
public Object getObject() throws Exception {
SimpleJob job = new SimpleJob();
job.setName(name);
job.setJobRepository(jobRepository);
SimpleStepFactoryBean<T, T> stepFactory = new SimpleStepFactoryBean<T, T>();
stepFactory.setBeanName("step");
Assert.state((itemReader instanceof FlatFileItemReader) || (itemReader instanceof StaxEventItemReader),
"ItemReader must be either a FlatFileItemReader or a StaxEventItemReader");
JobParameterResourceProxy resourceProxy = new JobParameterResourceProxy();
resourceProxy.setKey(ResourcePayloadAsJobParameterStrategy.FILE_INPUT_PATH);
stepFactory.setListeners(new StepExecutionListener[] { resourceProxy });
setResource(itemReader, resourceProxy);
stepFactory.setItemReader(itemReader);
Assert.notNull(channel, "A channel must be provided");
Assert.state(channel instanceof DirectChannel,
"The channel must be a DirectChannel (otherwise failures can not be recovered from)");
MessageChannelItemWriter<? super T> itemWriter = new MessageChannelItemWriter<T>(channel);
stepFactory.setItemWriter(itemWriter);
Assert.notNull(transactionManager, "A transaction manager must be provided");
stepFactory.setTransactionManager(transactionManager);
Assert.notNull(jobRepository, "A job repository must be provided");
stepFactory.setJobRepository(jobRepository);
job.addStep((Step) stepFactory.getObject());
return job;
}
/**
* @param itemReader
* @param resource
*/
private void setResource(ItemReader<? extends T> itemReader, Resource resource) {
if (itemReader instanceof FlatFileItemReader) {
((FlatFileItemReader<? extends T>) itemReader).setResource(resource);
}
else {
((StaxEventItemReader<? extends T>) itemReader).setResource(resource);
}
}
/**
* Always returns {@link Job}.
*
* @see org.springframework.beans.factory.FactoryBean#getObjectType()
*/
public Class<Job> getObjectType() {
return Job.class;
}
/**
* Always true. TODO: should it be false?
*
* @see org.springframework.beans.factory.FactoryBean#isSingleton()
*/
public boolean isSingleton() {
return true;
}
/**
* Strategy for resolving a filename just prior to step execution. The proxy
* is given a key that will correspond to a key in the job parameters. Just
* before the step is executed, the resource will be created with its
* filename as the value found in the job parameters.
*
* To use this resource it must be initialised with a {@link StepExecution}.
* The best way to do that is to register it as a listener in the step that
* is going to need it. For this reason the resource implements
* {@link StepExecutionListener}.
*
* @see Resource
*/
private class JobParameterResourceProxy extends StepExecutionListenerSupport implements Resource,
ResourceLoaderAware, StepExecutionListener {
private JobParametersConverter jobParametersConverter = new DefaultJobParametersConverter();
private ResourceLoader resourceLoader = new FileSystemResourceLoader();
private Resource delegate;
private String key = null;
private static final String NOT_INITIALISED = "The delegate resource has not been initialised. "
+ "Remember to register this object as a StepListener.";
/**
* @param relativePath
* @throws IOException
* @see org.springframework.core.io.Resource#createRelative(java.lang.String)
*/
public Resource createRelative(String relativePath) throws IOException {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.createRelative(relativePath);
}
/**
* @see org.springframework.core.io.Resource#exists()
*/
public boolean exists() {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.exists();
}
/**
* @see org.springframework.core.io.Resource#getDescription()
*/
public String getDescription() {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.getDescription();
}
/**
* @throws IOException
* @see org.springframework.core.io.Resource#getFile()
*/
public File getFile() throws IOException {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.getFile();
}
/**
* @see org.springframework.core.io.Resource#getFilename()
*/
public String getFilename() {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.getFilename();
}
/**
* @throws IOException
* @see org.springframework.core.io.InputStreamSource#getInputStream()
*/
public InputStream getInputStream() throws IOException {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.getInputStream();
}
/**
* @throws IOException
* @see org.springframework.core.io.Resource#getURI()
*/
public URI getURI() throws IOException {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.getURI();
}
/**
* @throws IOException
* @see org.springframework.core.io.Resource#getURL()
*/
public URL getURL() throws IOException {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.getURL();
}
/**
* @see org.springframework.core.io.Resource#isOpen()
*/
public boolean isOpen() {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.isOpen();
}
/**
* @see org.springframework.core.io.Resource#isReadable()
*/
public boolean isReadable() {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.isReadable();
}
/**
* @see org.springframework.core.io.Resource#lastModified()
*/
public long lastModified() throws IOException {
Assert.state(delegate != null, NOT_INITIALISED);
return delegate.lastModified();
}
/**
* Public setter for the {@link JobParametersConverter} used to
* translate {@link JobParameters} into {@link Properties}. Defaults to
* a {@link DefaultJobParametersConverter}.
*
* @param jobParametersConverter the {@link JobParametersConverter} to
* set
*/
public void setJobParametersFactory(JobParametersConverter jobParametersConverter) {
this.jobParametersConverter = jobParametersConverter;
}
/**
* Always false because we are expecting to be step scoped.
*
* @see org.springframework.beans.factory.config.AbstractFactoryBean#isSingleton()
*/
public boolean isSingleton() {
return false;
}
/**
* @see org.springframework.context.ResourceLoaderAware#setResourceLoader(org.springframework.core.io.ResourceLoader)
*/
public void setResourceLoader(ResourceLoader resourceLoader) {
this.resourceLoader = resourceLoader;
}
public void setKey(String key) {
this.key = key;
}
/**
* Collect the properties of the enclosing {@link StepExecution} that
* will be needed to create a file name.
*
* @see org.springframework.batch.core.StepExecutionListener#beforeStep(org.springframework.batch.core.StepExecution)
*/
public void beforeStep(StepExecution execution) {
Properties properties = jobParametersConverter.getProperties(execution.getJobExecution().getJobInstance()
.getJobParameters());
delegate = resourceLoader.getResource(properties.getProperty(this.key));
}
}
private static class MessageChannelItemWriter<T> implements ItemWriter<T> {
private MessageChannel channel;
public MessageChannelItemWriter(MessageChannel channel) {
super();
this.channel = channel;
}
public void write(List<? extends T> items) throws Exception {
for (T item : items) {
channel.send(new GenericMessage<T>(item));
}
}
}
}

32
spring-batch-integration/src/main/java/org/springframework/batch/integration/file/MessageToJobParametersStrategy.java
View File

@@ -1,32 +0,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.file;
import org.springframework.batch.core.JobParameters;
import org.springframework.integration.core.Message;
/**
*
* @author Jonas Partner
*
*/
public interface MessageToJobParametersStrategy {
public JobParameters getJobParameters(Message<?> message);
}

55
spring-batch-integration/src/main/java/org/springframework/batch/integration/file/ResourcePayloadAsJobParameterStrategy.java
View File

@@ -1,55 +0,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.file;
import java.io.IOException;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.core.io.Resource;
import org.springframework.integration.core.Message;
/**
* @author Dave Syer
*
*/
public class ResourcePayloadAsJobParameterStrategy implements MessageToJobParametersStrategy {
/**
* The key name for the job parameter that will be a URL for the input file
*/
public static final String FILE_INPUT_PATH = "input.file.path";
/**
* Convert a message payload which is a {@link Resource} to its URL
* representation and load that into a job parameter.
*
* @see MessageToJobParametersStrategy#getJobParameters(Message)
*/
public JobParameters getJobParameters(Message<?> message) {
JobParametersBuilder builder = new JobParametersBuilder();
Resource resource = (Resource) message.getPayload();
try {
builder.addString(FILE_INPUT_PATH, resource.getURL().toExternalForm());
}
catch (IOException e) {
throw new ItemStreamException("Could not create URL for resource: [" + resource + "]", e);
}
return builder.toJobParameters();
}
}

24
spring-batch-integration/src/main/java/org/springframework/batch/integration/job/JobExecutionReply.java
View File

@@ -1,24 +0,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.job;
/**
* @author dsyer
*
*/
public class JobExecutionReply {
}

114
spring-batch-integration/src/main/java/org/springframework/batch/integration/job/JobExecutionRequest.java
View File

@@ -1,114 +0,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.job;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
/**
* Encapsulation of a request to execute a job execution through a message flow
* consisting of step handlers. A handler should pass the message on as it is,
* modifying the request properties as necessary. Generally a handler will
* execute a step as part of the {@link JobExecution} passed in, and should
* change the status to {@link BatchStatus#COMPLETED} if the step is successful
* (generally a handler cannot determine if the whole job execution is complete,
* so this is just information about the step).<br/>
*
* If the incoming status is {@link BatchStatus#FAILED},
* {@link BatchStatus#ABANDONED} or {@link BatchStatus#STOPPING} the request
* should be ignored by handlers (passed on without modification).
*
* @author Dave Syer
*
*/
public class JobExecutionRequest {
private JobExecution jobExecution;
private BatchStatus status;
private Throwable throwable;
/**
* @param jobExecution
*/
public JobExecutionRequest(JobExecution jobExecution) {
this.jobExecution = jobExecution;
status = jobExecution.getStatus();
}
/**
* @return the current job execution id
*/
public Long getJobId() {
return this.jobExecution.getJobId();
}
/**
* @return the current {@link BatchStatus}
*/
public BatchStatus getStatus() {
return status;
}
/**
* Public setter for the status.
* @param status the status to set
*/
public void setStatus(BatchStatus status) {
this.status = status;
}
/**
* @return true if there are errors
*/
public boolean hasErrors() {
return throwable != null;
}
/**
* Public getter for the throwable.
* @return the throwable
*/
public Throwable getLastThrowable() {
return throwable;
}
/**
* Public setter for the throwable.
* @param throwable the throwable to set
*/
public void registerThrowable(Throwable throwable) {
this.throwable = throwable;
}
/**
* Public getter for the jobExecution.
* @return the jobExecution
*/
public JobExecution getJobExecution() {
return jobExecution;
}
/* (non-Javadoc)
* @see java.lang.Object#toString()
*/
@Override
public String toString() {
return getClass().getSimpleName()+": "+jobExecution;
}
}

179
spring-batch-integration/src/main/java/org/springframework/batch/integration/job/MessageOrientedStep.java
View File

@@ -1,179 +0,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.job;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.batch.core.step.AbstractStep;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.util.Assert;
/**
* @author Dave Syer
*
*/
public class MessageOrientedStep extends AbstractStep {
/**
* Key in execution context for flag to say we are waiting.
*/
public static final String WAITING = MessageOrientedStep.class.getName() + ".WAITING";
private MessageChannel outputChannel;
private PollableChannel source;
private static long MINUTE = 1000 * 60;
private long executionTimeout = 30*MINUTE ;
private long pollingInterval = 5;
/**
* Public setter for the execution timeout in minutes. Defaults to 30.
* @param executionTimeoutMinutes the timeout to set
*/
public void setExecutionTimeoutMinutes(int executionTimeoutMinutes) {
this.executionTimeout = executionTimeoutMinutes * MINUTE;
}
/**
* Public setter for the execution timeout in milliseconds. Defaults to 30 minutes.
* @param executionTimeout
*/
public void setExecutionTimeout(long executionTimeout) {
this.executionTimeout = executionTimeout;
}
/**
* Public setter for the polling interval in milliseconds while waiting for
* replies signalling the end of the step. Defaults to 5.
* @param pollingInterval the polling interval to set
*/
public void setPollingInterval(long pollingInterval) {
this.pollingInterval = pollingInterval;
}
/**
* Public setter for the target.
* @param outputChannel the target to set
*/
@Required
public void setOutputChannel(MessageChannel outputChannel) {
this.outputChannel = outputChannel;
}
/**
* Public setter for the source.
* @param source the source to set
*/
@Required
public void setInputChannel(PollableChannel source) {
this.source = source;
}
/**
* @see AbstractStep#execute(StepExecution)
*/
@Override
protected void doExecute(StepExecution stepExecution) throws JobInterruptedException,
UnexpectedJobExecutionException {
JobExecutionRequest request = new JobExecutionRequest(stepExecution.getJobExecution());
ExecutionContext executionContext = stepExecution.getExecutionContext();
if (executionContext.containsKey(WAITING)) {
// restart scenario: we are still waiting for a response
waitForReply(request.getJobId());
}
else {
executionContext.putString(WAITING, "true");
// TODO: need these two lines to be atomic
getJobRepository().update(stepExecution);
outputChannel.send(new GenericMessage<JobExecutionRequest>(request));
waitForReply(request.getJobId());
}
stepExecution.setExitStatus(ExitStatus.COMPLETED);
}
/**
* @param expectedJobId
*/
private void waitForReply(Long expectedJobId) {
long timeout = pollingInterval;
long maxCount = executionTimeout / timeout;
long count = 0;
while (count++ < maxCount) {
// TODO: timeout?
@SuppressWarnings("unchecked")
Message<JobExecutionRequest> message = (Message<JobExecutionRequest>) source.receive(timeout);
if (message != null) {
JobExecutionRequest payload = message.getPayload();
Long jobInstanceId = payload.getJobId();
Assert.state(jobInstanceId != null, "Message did not contain job instance id.");
Assert.state(jobInstanceId.equals(expectedJobId), "Message contained wrong job instance id ["
+ jobInstanceId + "] should have been [" + expectedJobId + "].");
if (payload.getStatus() == BatchStatus.COMPLETED) {
// One of the steps decided we were finished
// TODO: wait for all the other steps that might be
// executing concurrently?
// TODO: maybe *any* reply on this channel should
// mean the end of the step?
break;
}
if (payload.hasErrors()) {
rethrow(payload.getLastThrowable());
}
}
}
if (count >= maxCount) {
throw new StepExecutionTimeoutException("Timed out waiting for steps to execute.");
}
}
/**
* @param lastThrowable
*/
private static void rethrow(Throwable t) throws RuntimeException {
if (t instanceof RuntimeException) {
throw (RuntimeException) t;
}
if (t instanceof Exception) {
throw new UnexpectedJobExecutionException("Unexpected checked exception thrown by step.", t);
}
throw (Error) t;
}
}

193
spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionMessageHandler.java
View File

@@ -1,193 +0,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.job;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.StartLimitExceededException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
/**
* @author Dave Syer
*
*/
@MessageEndpoint
public class StepExecutionMessageHandler {
private Step step;
private JobRepository jobRepository;
/**
* Public setter for the {@link Step}.
* @param step the step to set
*/
@Required
public void setStep(Step step) {
this.step = step;
}
/**
* Public setter for the {@link JobRepository} that is needed to manage the
* state of the batch meta domain (jobs, steps, executions) during the life
* of a job.
*
* @param jobRepository
*/
@Required
public void setJobRepository(JobRepository jobRepository) {
this.jobRepository = jobRepository;
}
@ServiceActivator
public JobExecutionRequest handle(JobExecutionRequest request) {
// Hand off immediately if the job has already failed
if (isComplete(request)) {
return request;
}
JobExecution jobExecution = request.getJobExecution();
JobInstance jobInstance = jobExecution.getJobInstance();
StepExecution stepExecution = jobExecution.createStepExecution(step.getName());
try {
StepExecution lastStepExecution = jobRepository.getLastStepExecution(jobInstance, step.getName());
// Even if it completed successfully we want to pass on the output
// attributes, so set up the execution context here if it is
// available.
if (lastStepExecution != null) {
stepExecution.setExecutionContext(lastStepExecution.getExecutionContext());
}
// If it is already complete and not restartable it will simply be
// skipped
if (shouldStart(lastStepExecution, step)) {
if (!isRestart(jobInstance, lastStepExecution)) {
stepExecution.setExecutionContext(new ExecutionContext());
}
jobRepository.add(stepExecution);
step.execute(stepExecution);
}
else if (lastStepExecution != null) {
/*
* We only set these if the step is not going to execute. They
* might be needed by the next step to receive the request, but
* they won't be persisted because the step is not executed.
*/
stepExecution.setStatus(lastStepExecution.getStatus());
stepExecution.setExitStatus(lastStepExecution.getExitStatus());
}
// (the job might actually not be complete, but the stage is).
request.setStatus(BatchStatus.COMPLETED);
}
catch (Exception e) {
handleFailure(request, e);
}
catch (Error e) {
handleFailure(request, e);
throw e;
}
return request;
}
/**
* @param jobInstance
* @param lastStepExecution
* @return
*/
private boolean isRestart(JobInstance jobInstance, StepExecution lastStepExecution) {
return (lastStepExecution != null && !lastStepExecution.getStatus().equals(BatchStatus.COMPLETED));
}
/**
* @param request
* @return
*/
private boolean isComplete(JobExecutionRequest request) {
return request.getStatus() == BatchStatus.FAILED || request.getStatus() == BatchStatus.ABANDONED
|| request.getStatus() == BatchStatus.STOPPING;
}
/**
* @param request
* @param e
*/
private void handleFailure(JobExecutionRequest request, Throwable e) {
request.registerThrowable(e);
request.setStatus(BatchStatus.FAILED);
}
/*
* TODO: merge this with SimpleJob implementation.
*
* Given a step and configuration, return true if the step should start,
* false if it should not, and throw an exception if the job should finish.
*/
private boolean shouldStart(StepExecution lastStepExecution, Step step) throws JobExecutionException {
BatchStatus stepStatus;
// if the last execution is null, the step has never been executed.
if (lastStepExecution == null) {
return true;
}
else {
stepStatus = lastStepExecution.getStatus();
}
if (stepStatus == BatchStatus.UNKNOWN) {
throw new JobExecutionException("Cannot restart step from UNKNOWN status. "
+ "The last execution may have ended with a failure that could not be rolled back, "
+ "so it may be dangerous to proceed. " + "Manual intervention is probably necessary.");
}
if (stepStatus == BatchStatus.COMPLETED && step.isAllowStartIfComplete() == false) {
// step is complete, false should be returned, indicating that the
// step should not be started
return false;
}
if (jobRepository.getStepExecutionCount(lastStepExecution.getJobExecution().getJobInstance(), step.getName()) < step
.getStartLimit()) {
// step start count is less than start max, return true
return true;
}
else {
// start max has been exceeded, throw an exception.
throw new StartLimitExceededException("Maximum start limit exceeded for step: " + step.getName()
+ "StartMax: " + step.getStartLimit());
}
}
}

36
spring-batch-integration/src/main/java/org/springframework/batch/integration/job/StepExecutionTimeoutException.java
View File

@@ -1,36 +0,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.job;
import org.springframework.batch.core.UnexpectedJobExecutionException;
/**
* @author Dave Syer
*
*/
public class StepExecutionTimeoutException extends UnexpectedJobExecutionException {
/**
* Constructs a new instance.
*
* @param msg the exception message.
*
*/
public StepExecutionTimeoutException(String msg) {
super(msg);
}
}

57
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchRequest.java
View File

@@ -1,57 +0,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.launch;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobParameters;
/**
* Encapsulation of a {@link Job} and its {@link JobParameters} forming a
* request for a job to be launched.
*
* @author Dave Syer
*
*/
public class JobLaunchRequest {
private final Job job;
private final JobParameters jobParameters;
/**
* @param job
* @param jobParameters
*/
public JobLaunchRequest(Job job, JobParameters jobParameters) {
super();
this.job = job;
this.jobParameters = jobParameters;
}
/**
* @return the {@link Job} to be executed
*/
public Job getJob() {
return this.job;
}
/**
* @return the {@link JobParameters} for this request
*/
public JobParameters getJobParameters() {
return this.jobParameters;
}
}

60
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandler.java
View File

@@ -1,60 +0,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.launch;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.UnexpectedJobExecutionException;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.integration.annotation.ServiceActivator;
/**
* Message handler which uses strategies to convert a Message into a job and a
* set of job parameters
* @author Jonas Partner
* @author Dave Syer
*
*/
public class JobLaunchingMessageHandler {
private final JobLauncher jobLauncher;
/**
* @param jobLauncher
*/
public JobLaunchingMessageHandler(JobLauncher jobLauncher) {
super();
this.jobLauncher = jobLauncher;
}
@ServiceActivator
public JobExecution launch(JobLaunchRequest request) {
Job job = request.getJob();
JobParameters jobParameters = request.getJobParameters();
try {
JobExecution execution = jobLauncher.run(job, jobParameters);
return execution;
}
catch (JobExecutionException e) {
throw new UnexpectedJobExecutionException("Exception executing job: ["+request+"]", e);
}
}
}

47
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/BeanFactoryStepLocator.java
View File

@@ -1,47 +0,0 @@
package org.springframework.batch.integration.partition;
import java.util.Arrays;
import java.util.Collection;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.util.Assert;
/**
* A {@link StepLocator} implementation that just looks in its enclosing bean
* factory for components of type {@link Step}.
*
* @author Dave Syer
*
*/
public class BeanFactoryStepLocator implements StepLocator, BeanFactoryAware {
private BeanFactory beanFactory;
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
this.beanFactory = beanFactory;
}
/**
* Look up a bean with the provided name of type {@link Step}.
* @see StepLocator#getStep(String)
*/
public Step getStep(String stepName) {
return (Step) beanFactory.getBean(stepName, Step.class);
}
/**
* Look in the bean factory for all beans of type {@link Step}.
* @throws IllegalStateException if the {@link BeanFactory} is not listable
* @see StepLocator#getStepNames()
*/
public Collection<String> getStepNames() {
Assert.state(beanFactory instanceof ListableBeanFactory, "BeanFactory is not listable.");
return Arrays.asList(((ListableBeanFactory) beanFactory).getBeanNamesForType(Step.class));
}
}

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

@@ -1,138 +0,0 @@
package org.springframework.batch.integration.partition;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.partition.PartitionHandler;
import org.springframework.batch.core.partition.StepExecutionSplitter;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.integration.annotation.Aggregator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.gateway.MessagingGateway;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.util.Assert;
/**
* A {@link PartitionHandler} that uses {@link MessageChannel} instances to send
* instructions to remote workers and receive their responses. The
* {@link MessageChannel} provides a nice abstraction so that the location of
* the workers and the transport used to communicate with them can be changed at
* run time. The communication with the remote workers does not need to be
* transactional or have guaranteed delivery, so a local thread pool based
* implementation works as well as a remote web service or JMS implementation.
* If a remote worker fails or doesn't send a reply message, the job will fail
* and can be restarted to pick up missing messages and processing. The remote
* workers need access to the Spring Batch {@link JobRepository} so that the
* shared state across those restarts can be managed centrally.
*
* @author Dave Syer
*
*/
@MessageEndpoint
public class MessageChannelPartitionHandler implements PartitionHandler {
private int gridSize = 1;
private MessagingGateway messagingGateway;
private String stepName;
public void afterPropertiesSet() throws Exception {
Assert.notNull(stepName, "A step name must be provided for the remote workers.");
Assert.state(messagingGateway != null, "The MessagingGateway must be set");
}
/**
* A pre-configured gateway for sending and receiving messages to the remote
* workers. Using this property allows a large degree of control over the
* timeouts and other properties of the send. It should have channels set up
* internally:
* <ul>
* <li>request channel capable of accepting {@link StepExecutionRequest}
* payloads</li>
* <li>reply channel that returns a list of {@link StepExecution} results</li>
* </ul>
* The timeout for the repoy should be set sufficiently long that the remote
* steps have time to complete.
*
* @param messagingGateway the {@link MessagingGateway} to set
*/
public void setMessagingGateway(MessagingGateway messagingGateway) {
this.messagingGateway = messagingGateway;
}
/**
* Passed to the {@link StepExecutionSplitter} in the
* {@link #handle(StepExecutionSplitter, StepExecution)} method, instructing
* it how many {@link StepExecution} instances are required, ideally. The
* {@link StepExecutionSplitter} is allowed to ignore the grid size in the
* case of a restart, since the input data partitions must be preserved.
*
* @param gridSize the number of step executions that will be created
*/
public void setGridSize(int gridSize) {
this.gridSize = gridSize;
}
/**
* The name of the {@link Step} that will be used to execute the partitioned
* {@link StepExecution}. This is a regular Spring Batch step, with all the
* business logic required to complete an execution based on the input
* parameters in its {@link StepExecution} context. The name will be
* translated into a {@link Step} instance by the remote worker.
*
* @param stepName the name of the {@link Step} instance to execute business
* logic
*/
public void setStepName(String stepName) {
this.stepName = stepName;
}
/**
* @param messages the messages to be aggregated
* @return the list as it was passed in
*/
@Aggregator(sendPartialResultsOnTimeout = true)
public List<?> aggregate(List<?> messages) {
return messages;
}
/**
* Sends {@link StepExecutionRequest} objects to the request channel of the
* {@link MessagingGateway}, and then receives the result back as a list of
* {@link StepExecution} on a reply channel. Use the
* {@link #aggregate(List)} method as an aggregator of the individual remote
* replies. The receive timeout needs to be set realistically in the
* {@link MessagingGateway} <b>and</b> the aggregator, so that there is a
* good chance of all work being done.
*
* @see PartitionHandler#handle(StepExecutionSplitter, StepExecution)
*/
public Collection<StepExecution> handle(StepExecutionSplitter stepExecutionSplitter,
StepExecution masterStepExecution) throws Exception {
Set<StepExecution> split = stepExecutionSplitter.split(masterStepExecution, gridSize);
int count = 0;
for (StepExecution stepExecution : split) {
messagingGateway.send(createMessage(count++, split.size(), new StepExecutionRequest(stepName, stepExecution
.getJobExecutionId(), stepExecution.getId())));
}
@SuppressWarnings("unchecked")
Collection<StepExecution> result = (Collection<StepExecution>) messagingGateway.receive();
return result;
}
private Message<StepExecutionRequest> createMessage(int sequenceNumber, int sequenceSize,
StepExecutionRequest stepExecutionRequest) {
return MessageBuilder.withPayload(stepExecutionRequest).setSequenceNumber(sequenceNumber).setSequenceSize(
sequenceSize).setCorrelationId(
stepExecutionRequest.getJobExecutionId() + ":" + stepExecutionRequest.getStepName()).build();
}
}

37
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequest.java
View File

@@ -1,37 +0,0 @@
package org.springframework.batch.integration.partition;
import java.io.Serializable;
public class StepExecutionRequest implements Serializable {
private final Long stepExecutionId;
private final String stepName;
private final Long jobExecutionId;
public StepExecutionRequest(String stepName, Long jobExecutionId, Long stepExecutionId) {
this.stepName = stepName;
this.jobExecutionId = jobExecutionId;
this.stepExecutionId = stepExecutionId;
}
public Long getJobExecutionId() {
return jobExecutionId;
}
public Long getStepExecutionId() {
return stepExecutionId;
}
public String getStepName() {
return stepName;
}
@Override
public String toString() {
return String.format("StepExecutionRequest: [jobExecutionId=%d, stepExecutionId=%d, stepName=%s]",
jobExecutionId, stepExecutionId, stepName);
}
}

81
spring-batch-integration/src/main/java/org/springframework/batch/integration/partition/StepExecutionRequestHandler.java
View File

@@ -1,81 +0,0 @@
package org.springframework.batch.integration.partition;
import org.springframework.batch.core.BatchStatus;
import org.springframework.batch.core.JobInterruptedException;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.explore.JobExplorer;
import org.springframework.batch.core.step.StepLocator;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
/**
* A {@link MessageEndpoint} that can handle a {@link StepExecutionRequest} and
* return a {@link StepExecution} as the result. Typically these need to be
* aggregated into a response to a partition handler.
*
* @author Dave Syer
*
*/
@MessageEndpoint
public class StepExecutionRequestHandler {
private JobExplorer jobExplorer;
private StepLocator stepLocator;
/**
* Used to locate a {@link Step} to execute for each request.
* @param stepLocator a {@link StepLocator}
*/
public void setStepLocator(StepLocator stepLocator) {
this.stepLocator = stepLocator;
}
/**
* An explorer that should be used to check for {@link StepExecution}
* completion.
*
* @param jobExplorer a {@link JobExplorer} that is linked to the shared
* repository used by all remote workers.
*/
public void setJobExplorer(JobExplorer jobExplorer) {
this.jobExplorer = jobExplorer;
}
@ServiceActivator
public StepExecution handle(StepExecutionRequest request) {
Long jobExecutionId = request.getJobExecutionId();
Long stepExecutionId = request.getStepExecutionId();
StepExecution stepExecution = jobExplorer.getStepExecution(jobExecutionId, stepExecutionId);
if (stepExecution == null) {
// TODO: create new Exception
throw new IllegalStateException("No StepExecution could be located for this request: " + request);
}
String stepName = request.getStepName();
Step step = stepLocator.getStep(stepName);
if (step == null) {
// TODO: create new Exception
throw new IllegalStateException(String.format("No Step with name [%s] could be located.", stepName));
}
try {
step.execute(stepExecution);
}
catch (JobInterruptedException e) {
stepExecution.setStatus(BatchStatus.STOPPED);
// TODO: maybe update stepExecution in repository
}
catch (Throwable e) {
stepExecution.addFailureException(e);
stepExecution.setStatus(BatchStatus.FAILED);
// TODO: maybe update stepExecution in repository
}
return stepExecution;
}
}