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

Refactor job launching message handler to use special request object.

Browse Source
This commit is contained in:
dsyer
2008-06-16 16:45:52 +00:00
parent 66b7a4ef11
commit cb9f35c1b4
13 changed files with 182 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/MessageToJobParametersStrategy.java → spring-batch-integration/src/main/java/org/springframework/batch/integration/file/MessageToJobParametersStrategy.java
View File

@@ -1,4 +1,4 @@
package org.springframework.batch.integration.launch;
package org.springframework.batch.integration.file;
import org.springframework.batch.core.JobParameters;
import org.springframework.integration.message.Message;

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

@@ -19,7 +19,6 @@ import java.io.IOException;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.integration.launch.MessageToJobParametersStrategy;
import org.springframework.batch.item.ItemStreamException;
import org.springframework.core.io.Resource;
import org.springframework.integration.message.Message;
@@ -39,7 +38,7 @@ public class ResourcePayloadAsJobParameterStrategy implements MessageToJobParame
* Convert a message payload which is a {@link Resource} to its URL
* representation and load that into a job parameter.
*
* @see org.springframework.batch.integration.launch.MessageToJobParametersStrategy#getJobParameters(org.springframework.integration.message.Message)
* @see org.springframework.batch.integration.file.MessageToJobParametersStrategy#getJobParameters(org.springframework.integration.message.Message)
*/
public JobParameters getJobParameters(Message<?> message) {
JobParametersBuilder builder = new JobParametersBuilder();

57
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/JobExecutionRequest.java Normal file
View File

@@ -0,0 +1,57 @@
/*
* 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 JobExecutionRequest {
private final Job job;
private final JobParameters jobParameters;
/**
* @param job
* @param jobParameters
*/
public JobExecutionRequest(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;
}
}

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

@@ -4,10 +4,9 @@ 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.Handler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
/**
* Message handler which uses strategies to convert a Message into a job and a
@@ -18,37 +17,28 @@ import org.springframework.integration.message.MessageHandlingException;
*/
public class JobLaunchingMessageHandler {
private MessageToJobStrategy messageToJobStrategy;
private final JobLauncher jobLauncher;
private MessageToJobParametersStrategy messageToJobParametersStrategy = new MessagePropertiesToJobParametersStrategy();
private JobLauncher jobLauncher;
public JobLaunchingMessageHandler(JobLauncher jobLauncher, MessageToJobStrategy messageToJobStrategy) {
/**
* @param jobLauncher
*/
public JobLaunchingMessageHandler(JobLauncher jobLauncher) {
super();
this.jobLauncher = jobLauncher;
this.messageToJobStrategy = messageToJobStrategy;
}
@Handler
public JobExecution handle(Message<?> message) {
Job job = messageToJobStrategy.getJob(message);
JobParameters jobParameters = messageToJobParametersStrategy.getJobParameters(message);
public JobExecution launch(JobExecutionRequest request) {
Job job = request.getJob();
JobParameters jobParameters = request.getJobParameters();
try {
JobExecution execution = jobLauncher.run(job, jobParameters);
if (message.getHeader().getReturnAddress() != null) {
return execution;
}
return null;
return execution;
}
catch (JobExecutionException e) {
throw new MessageHandlingException(message, "Exception executing job ");
throw new UnexpectedJobExecutionException("Exception executing job: ["+request+"]", e);
}
}
public void setMessageToJobParametersStrategy(MessageToJobParametersStrategy messageToJobParametersStrategy) {
this.messageToJobParametersStrategy = messageToJobParametersStrategy;
}
}

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

@@ -1,57 +0,0 @@
package org.springframework.batch.integration.launch;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecutionException;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
/**
* Channel interceptor which launches the configured job after the message has
* been received
* @author Jonas Partner
*
*/
public class JobLaunchingPostReceiveChannelInterceptor extends ChannelInterceptorAdapter {
private JobLauncher jobLauncher;
private Job job;
private MessageToJobParametersStrategy messageToJobParametersStrategy = new MessagePropertiesToJobParametersStrategy();
/**
*
* @param job The job to launch
* @param jobLauncher
*/
public JobLaunchingPostReceiveChannelInterceptor(Job job, JobLauncher jobLauncher) {
super();
this.job = job;
this.jobLauncher = jobLauncher;
}
public MessageToJobParametersStrategy getMessageToJobParametersStrategy() {
return messageToJobParametersStrategy;
}
public void setMessageToJobParametersStrategy(MessageToJobParametersStrategy messageToJobParametersStrategy) {
this.messageToJobParametersStrategy = messageToJobParametersStrategy;
}
@Override
public void postReceive(Message<?> message, MessageChannel channel) {
JobParameters parameters = messageToJobParametersStrategy.getJobParameters(message);
try {
jobLauncher.run(job, parameters);
}
catch (JobExecutionException e) {
throw new MessageHandlingException(message, "Excpetion executing job ", e);
}
}
}

25
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/MessagePropertiesToJobParametersStrategy.java
View File

@@ -1,25 +0,0 @@
package org.springframework.batch.integration.launch;
import java.util.Set;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.integration.message.Message;
/**
* Builds an instance of JobParameters from the properties in the message header
* @author Jonas Partner
*
*/
public class MessagePropertiesToJobParametersStrategy implements MessageToJobParametersStrategy {
public JobParameters getJobParameters(Message<?> message) {
JobParametersBuilder parametersBuilder = new JobParametersBuilder();
Set<String> propertyNames = message.getHeader().getPropertyNames();
for (String key : propertyNames) {
parametersBuilder.addString(key, message.getHeader().getProperty(key));
}
return parametersBuilder.toJobParameters();
}
}

16
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/MessageToJobStrategy.java
View File

@@ -1,16 +0,0 @@
package org.springframework.batch.integration.launch;
import org.springframework.batch.core.Job;
import org.springframework.integration.message.Message;
/**
* Interface for strategy implementations which convert from a Message to a Spring batch Job
* @author Jonas Partner
*
*/
public interface MessageToJobStrategy{
public Job getJob(Message<?> message);
}

32
spring-batch-integration/src/main/java/org/springframework/batch/integration/launch/StringPayloadAsJobNameStrategy.java
View File

@@ -1,32 +0,0 @@
package org.springframework.batch.integration.launch;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.configuration.JobLocator;
import org.springframework.batch.core.repository.NoSuchJobException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
/**
* Takes the string payload of a message and delegates to a JobLocator
* @author Jonas Partner
*
*/
public class StringPayloadAsJobNameStrategy implements MessageToJobStrategy{
private JobLocator jobLocator;
public StringPayloadAsJobNameStrategy(JobLocator jobLocator){
this.jobLocator = jobLocator;
}
public Job getJob(Message<?> message) {
String name = (String)message.getPayload();
try {
return jobLocator.getJob(name);
}
catch (NoSuchJobException e) {
throw new MessageHandlingException(message, "Could not find job with name " + name, e);
}
}
}

61
spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests.java
View File

@@ -1,58 +1,69 @@
package org.springframework.batch.integration.launch;
import static org.junit.Assert.*;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.integration.JobSupport;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@ContextConfiguration()
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class JobLaunchingMessageHandlerIntegrationTests {
@Autowired @Qualifier("requests")
public class JobLaunchingMessageHandlerIntegrationTests {
@Autowired
@Qualifier("requests")
private MessageChannel requestChannel;
@Autowired @Qualifier("response")
@Autowired
@Qualifier("response")
private MessageChannel responseChannel;
private JobSupport job = new JobSupport("testJob");
@Before
public void setUp(){
public void setUp() {
requestChannel.purge(null);
responseChannel.purge(null);
}
@Test @DirtiesContext @SuppressWarnings("unchecked")
public void testNoReply(){
requestChannel.send(new StringMessage("testJob"));
Message<JobExecution> executionMessage = (Message<JobExecution>)responseChannel.receive(1000);
assertNull("JobExecution message received when no return address set", executionMessage);
}
@Test
@DirtiesContext
@SuppressWarnings("unchecked")
@Test @DirtiesContext
public void testReply(){
StringMessage trigger = new StringMessage("testJob");
trigger.getHeader().setProperty("dontclash", "12");
public void testNoReply() {
GenericMessage<JobExecutionRequest> trigger = new GenericMessage<JobExecutionRequest>(new JobExecutionRequest(job, new JobParameters()));
requestChannel.send(trigger);
Message<JobExecution> executionMessage = (Message<JobExecution>) responseChannel.receive(1000);
assertNull("JobExecution message received when no return address set", executionMessage);
}
@SuppressWarnings("unchecked")
@Test
@DirtiesContext
public void testReply() {
JobParametersBuilder builder = new JobParametersBuilder();
builder.addString("dontclash", "12");
GenericMessage<JobExecutionRequest> trigger = new GenericMessage<JobExecutionRequest>(new JobExecutionRequest(job, builder.toJobParameters()));
trigger.getHeader().setReturnAddress("response");
requestChannel.send(trigger);
Message<JobExecution> executionMessage = (Message<JobExecution>)responseChannel.receive(1000);
Message<JobExecution> executionMessage = (Message<JobExecution>) responseChannel.receive(1000);
assertNotNull("No response received", executionMessage);
JobExecution execution = executionMessage.getPayload();
assertNotNull("JobExectuion not returned", execution);
}
}
}

19
spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerTests.java
View File

@@ -14,13 +14,9 @@ import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.integration.JobSupport;
import org.springframework.batch.integration.launch.JobLaunchingMessageHandler;
import org.springframework.batch.integration.launch.MessageToJobStrategy;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.StringMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
@@ -30,8 +26,6 @@ public class JobLaunchingMessageHandlerTests extends AbstractJUnit4SpringContext
JobLaunchingMessageHandler messageHandler;
StubJobLauncher jobLauncher;
// @Autowired
// @Qualifier("jobs") TODO: Qualifier seems to be broken here why ?????
@@ -43,13 +37,13 @@ public class JobLaunchingMessageHandlerTests extends AbstractJUnit4SpringContext
@Before
public void setUp() {
jobLauncher = new StubJobLauncher();
messageHandler = new JobLaunchingMessageHandler(jobLauncher, new StubMessageToJobStrategy());
messageHandler = new JobLaunchingMessageHandler(jobLauncher);
jobsChannel = (AbstractMessageChannel) applicationContext.getBean("jobs");
}
@Test
public void testSimpleDelivery() throws Exception{
messageHandler.handle(new StringMessage("testjob"));
messageHandler.launch(new JobExecutionRequest(new JobSupport("testjob"), null));
assertEquals("Wrong job count", 1, jobLauncher.jobs.size());
assertEquals("Wrong job name", jobLauncher.jobs.get(0).getName(), "testjob");
@@ -71,14 +65,5 @@ public class JobLaunchingMessageHandlerTests extends AbstractJUnit4SpringContext
}
}
private static class StubMessageToJobStrategy implements MessageToJobStrategy {
public Job getJob(Message<?> message) {
String name = (String) message.getPayload();
return new JobSupport(name);
}
}
}

84
spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobLaunchingPostReceiveChannelAdapterTests.java
View File

@@ -1,84 +0,0 @@
package org.springframework.batch.integration.launch;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.launch.JobLauncher;
import org.springframework.batch.integration.JobSupport;
import org.springframework.batch.integration.launch.JobLaunchingPostReceiveChannelInterceptor;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.message.StringMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@ContextConfiguration(locations = { "/job-execution-context.xml" })
@RunWith(SpringJUnit4ClassRunner.class)
public class JobLaunchingPostReceiveChannelAdapterTests extends AbstractJUnit4SpringContextTests {
JobLaunchingPostReceiveChannelInterceptor interceptor;
StubJobLauncher jobLauncher;
JobSupport job;
// @Autowired
// @Qualifier("jobs") TODO: Qualifier seems to be broken here why ?????
public AbstractMessageChannel jobsChannel;
@Before
public void setUp() {
job = new JobSupport(getClass().getName());
jobLauncher = new StubJobLauncher();
interceptor = new JobLaunchingPostReceiveChannelInterceptor(job, jobLauncher);
jobsChannel = (AbstractMessageChannel) applicationContext.getBean("jobs");
jobsChannel.addInterceptor(interceptor);
}
@Test
public void testJobPassedToLauncherCalled() {
StringMessage message = new StringMessage("test payload");
jobsChannel.send(message);
assertTrue("Job launcher called before recevie", (jobLauncher.jobs.size() == 0));
jobsChannel.receive();
assertEquals(job, jobLauncher.jobs.get(0));
}
@Test
public void testMessagePropertiesPassedAsJobParameters() {
StringMessage message = new StringMessage("test payload");
message.getHeader().setProperty("testOne", "a");
message.getHeader().setProperty("testTwo", "b");
jobsChannel.send(message);
jobsChannel.receive();
JobParameters parameters = jobLauncher.parameters.get(0);
assertEquals("a", parameters.getString("testOne"));
assertEquals("b", parameters.getString("testTwo"));
}
private static class StubJobLauncher implements JobLauncher {
List<Job> jobs = new ArrayList<Job>();
List<JobParameters> parameters = new ArrayList<JobParameters>();
public JobExecution run(Job job, JobParameters jobParameters){
jobs.add(job);
parameters.add(jobParameters);
return null;
}
}
}

37
spring-batch-integration/src/test/java/org/springframework/batch/integration/launch/JobRequestConverter.java Normal file
View File

@@ -0,0 +1,37 @@
/*
* 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 java.util.Properties;
import org.springframework.batch.core.converter.DefaultJobParametersConverter;
import org.springframework.batch.integration.JobSupport;
import org.springframework.integration.annotation.Handler;
/**
* @author Dave Syer
*
*/
public class JobRequestConverter {
@Handler
public JobExecutionRequest convert(String jobName) {
// TODO: get these from message header
Properties properties = new Properties();
return new JobExecutionRequest(new JobSupport(jobName), new DefaultJobParametersConverter().getJobParameters(properties));
}
}

71
spring-batch-integration/src/test/resources/org/springframework/batch/integration/launch/JobLaunchingMessageHandlerIntegrationTests-context.xml
View File

@@ -1,44 +1,47 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:integration="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
xmlns:p="http://www.springframework.org/schema/p" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
<import resource="classpath:simple-job-launcher-context.xml"/>
<integration:message-bus auto-create-channels="true" />
<integration:annotation-driven />
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:integration="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context" xmlns:aop="http://www.springframework.org/schema/aop"
xmlns:tx="http://www.springframework.org/schema/tx" xmlns:p="http://www.springframework.org/schema/p"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-2.5.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
<import resource="classpath:simple-job-launcher-context.xml" />
<integration:message-bus auto-create-channels="true" channel-factory="channelFactory" />
<integration:annotation-driven />
<integration:channel id="requests" />
<integration:channel id="response" />
<integration:handler-endpoint input-channel="requests" handler="jobLaunchingHandler" method="handle"/>
<integration:channel id="jobs" />
<integration:thread-local-channel id="response" />
<bean id="channelFactory" class="org.springframework.integration.dispatcher.DirectChannelFactory" />
<integration:handler-endpoint input-channel="requests" handler="jobLaunchingHandler" method="launch"
output-channel="null" return-address-overrides="true" />
<integration:handler-endpoint input-channel="none" output-channel="jobs" handler="jobRequestConverter"
method="convert" />
<bean id="jobRequestConverter" class="org.springframework.batch.integration.launch.JobRequestConverter" />
<bean id="jobLaunchingHandler" class="org.springframework.batch.integration.launch.JobLaunchingMessageHandler">
<constructor-arg ref="jobLauncher"/>
<constructor-arg ref="messageToJobStrategy"/>
<constructor-arg ref="jobLauncher" />
</bean>
<bean id="messageToJobStrategy" class="org.springframework.batch.integration.launch.StringPayloadAsJobNameStrategy">
<constructor-arg ref="jobRegistry" />
</bean>
<bean id="testJob" parent="simpleJob">
<property name="steps" ref="step"/>
<property name="steps" ref="step" />
</bean>
<bean id="step" class="org.springframework.batch.core.step.tasklet.TaskletStep">
<property name="tasklet">
<bean class="org.springframework.batch.integration.job.TestTasklet"/>
<bean class="org.springframework.batch.integration.job.TestTasklet" />
</property>
<property name="jobRepository" ref="jobRepository"/>
<property name="jobRepository" ref="jobRepository" />
</bean>
</beans>