BATCHADM-18: remove old job factory and replace with unit test
This commit is contained in:
David Syer
committed by
Michael Minella
Michael Minella
parent
08577466ea
commit
e031518e94
@@ -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));
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -5,6 +5,7 @@ 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.NoSuchStepException;
|
||||
import org.springframework.batch.core.step.StepLocator;
|
||||
import org.springframework.integration.annotation.MessageEndpoint;
|
||||
import org.springframework.integration.annotation.ServiceActivator;
|
||||
@@ -50,15 +51,13 @@ public class StepExecutionRequestHandler {
|
||||
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);
|
||||
throw new NoSuchStepException("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));
|
||||
throw new NoSuchStepException(String.format("No Step with name [%s] could be located.", stepName));
|
||||
}
|
||||
|
||||
try {
|
||||
@@ -66,12 +65,12 @@ public class StepExecutionRequestHandler {
|
||||
}
|
||||
catch (JobInterruptedException e) {
|
||||
stepExecution.setStatus(BatchStatus.STOPPED);
|
||||
// TODO: maybe update stepExecution in repository
|
||||
// The receiver should update the stepExecution in repository
|
||||
}
|
||||
catch (Throwable e) {
|
||||
stepExecution.addFailureException(e);
|
||||
stepExecution.setStatus(BatchStatus.FAILED);
|
||||
// TODO: maybe update stepExecution in repository
|
||||
// The receiver should update the stepExecution in repository
|
||||
}
|
||||
|
||||
return stepExecution;
|
||||
|
||||
Reference in New Issue
Block a user