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

BATCHADM-81: add step context manipulation for async processors

Browse Source 
- AsyncItemProcessor creates step context for its runnable
 - We also provide StepContextInterceptor for supplying StepExecution to downsteam consumers
This commit is contained in:
Dave Syer
2010-11-27 14:32:53 +00:00
committed by Michael Minella
parent fa3b9f86d3
commit 5b646c4edf
8 changed files with 374 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,9 +1,27 @@
/*
* 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.async;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.SyncTaskExecutor;
@@ -65,13 +83,36 @@ public class AsyncItemProcessor<I, O> implements ItemProcessor<I, Future<O>>, In
* @see ItemProcessor#process(Object)
*/
public Future<O> process(final I item) throws Exception {
final StepExecution stepExecution = getStepExecution();
FutureTask<O> task = new FutureTask<O>(new Callable<O>() {
public O call() throws Exception {
return delegate.process(item);
if (stepExecution != null) {
StepSynchronizationManager.register(stepExecution);
}
try {
return delegate.process(item);
}
finally {
if (stepExecution != null) {
StepSynchronizationManager.close();
}
}
}
});
taskExecutor.execute(task);
return task;
}
/**
* @return the current step execution if there is one
*/
private StepExecution getStepExecution() {
StepContext context = StepSynchronizationManager.getContext();
if (context==null) {
return null;
}
StepExecution stepExecution = context.getStepExecution();
return stepExecution;
}
}

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

@@ -1,3 +1,18 @@
/*
* 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.async;
import java.util.ArrayList;

53
spring-batch-integration/src/main/java/org/springframework/batch/integration/async/StepExecutionInterceptor.java Normal file
View File

@@ -0,0 +1,53 @@
/*
* 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.async;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
import org.springframework.integration.support.MessageBuilder;
/**
* A {@link ChannelInterceptor} that adds the current {@link StepExecution} (if
* there is one) as a header to the message. Downstream asynchronous handlers
* can then take advantage of the step context without needing to be step
* scoped, which is a problem for handlers executing in another thread because
* the scope context is not available.
*
* @author Dave Syer
*
*/
public class StepExecutionInterceptor extends ChannelInterceptorAdapter {
/**
* The name of the header
*/
public static final String STEP_EXECUTION = "stepExecution";
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
StepContext context = StepSynchronizationManager.getContext();
if (context == null) {
return message;
}
return MessageBuilder.fromMessage(message).setHeader(STEP_EXECUTION, context.getStepExecution()).build();
}
}

61
spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests.java
View File

@@ -1,3 +1,18 @@
/*
* 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.async;
import static org.junit.Assert.assertNotNull;
@@ -5,11 +20,20 @@ import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.MethodRule;
import org.junit.runner.RunWith;
import org.junit.runners.model.FrameworkMethod;
import org.junit.runners.model.Statement;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.test.MetaDataInstanceFactory;
import org.springframework.batch.test.StepScopeTestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.annotation.MessageEndpoint;
@@ -23,6 +47,32 @@ public class AsyncItemProcessorMessagingGatewayTests {
private AsyncItemProcessor<String, String> processor = new AsyncItemProcessor<String, String>();
private StepExecution stepExecution = MetaDataInstanceFactory.createStepExecution(new JobParametersBuilder().addLong("factor", 2L).toJobParameters());;
@Rule
public MethodRule rule = new MethodRule() {
public Statement apply(final Statement base, FrameworkMethod method, Object target) {
return new Statement() {
public void evaluate() throws Throwable {
StepScopeTestUtils.doInStepScope(stepExecution, new Callable<Void>() {
public Void call() throws Exception {
try {
base.evaluate();
}
catch (Exception e) {
throw e;
}
catch (Throwable e) {
throw new Error(e);
}
return null;
}
});
};
};
}
};
@Autowired
private ItemProcessor<String, String> delegate;
@@ -50,9 +100,18 @@ public class AsyncItemProcessorMessagingGatewayTests {
@MessageEndpoint
public static class Doubler {
private int factor = 1;
public void setFactor(int factor) {
this.factor = factor;
}
@ServiceActivator
public String cat(String value) {
return value + value;
for (int i=1; i<factor; i++) {
value += value;
}
return value;
}
}

38
spring-batch-integration/src/test/java/org/springframework/batch/integration/async/AsyncItemProcessorTests.java
View File

@@ -1,3 +1,18 @@
/*
* 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.async;
import static org.junit.Assert.assertEquals;
@@ -5,10 +20,15 @@ import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import org.junit.Test;
import org.springframework.batch.core.scope.context.StepContext;
import org.springframework.batch.core.scope.context.StepSynchronizationManager;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.test.MetaDataInstanceFactory;
import org.springframework.batch.test.StepScopeTestUtils;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
public class AsyncItemProcessorTests {
@@ -33,6 +53,24 @@ public class AsyncItemProcessorTests {
assertEquals("foofoo", result.get());
}
@Test
public void testExecutionInStepScope() throws Exception {
delegate = new ItemProcessor<String, String>() {
public String process(String item) throws Exception {
StepContext context = StepSynchronizationManager.getContext();
assertTrue(context != null && context.getStepExecution() != null);
return item + item;
};
};
processor.setDelegate(delegate);
Future<String> result = StepScopeTestUtils.doInStepScope(MetaDataInstanceFactory.createStepExecution(), new Callable<Future<String>>() {
public Future<String> call() throws Exception {
return processor.process("foo");
}
});
assertEquals("foofoo", result.get());
}
@Test
public void testMultiExecution() throws Exception {
processor.setDelegate(delegate);

116
spring-batch-integration/src/test/java/org/springframework/batch/integration/async/PollingAsyncItemProcessorMessagingGatewayTests.java Normal file
View File

@@ -0,0 +1,116 @@
/*
* 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.async;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.MethodRule;
import org.junit.runner.RunWith;
import org.junit.runners.model.FrameworkMethod;
import org.junit.runners.model.Statement;
import org.springframework.batch.core.JobParametersBuilder;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.test.MetaDataInstanceFactory;
import org.springframework.batch.test.StepScopeTestUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.annotation.Header;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
public class PollingAsyncItemProcessorMessagingGatewayTests {
private AsyncItemProcessor<String, String> processor = new AsyncItemProcessor<String, String>();
private StepExecution stepExecution = MetaDataInstanceFactory.createStepExecution(new JobParametersBuilder().addLong("factor", 2L).toJobParameters());;
@Rule
public MethodRule rule = new MethodRule() {
public Statement apply(final Statement base, FrameworkMethod method, Object target) {
return new Statement() {
public void evaluate() throws Throwable {
StepScopeTestUtils.doInStepScope(stepExecution, new Callable<Void>() {
public Void call() throws Exception {
try {
base.evaluate();
}
catch (Exception e) {
throw e;
}
catch (Throwable e) {
throw new Error(e);
}
return null;
}
});
};
};
}
};
@Autowired
private ItemProcessor<String, String> delegate;
@Test
public void testMultiExecution() throws Exception {
processor.setDelegate(delegate);
processor.setTaskExecutor(new SimpleAsyncTaskExecutor());
List<Future<String>> list = new ArrayList<Future<String>>();
for (int count = 0; count < 10; count++) {
list.add(processor.process("foo" + count));
}
for (Future<String> future : list) {
String value = future.get();
/**
* This delegate is a Spring Integration MessagingGateway. It can
* easily return null because of a timeout, but that will be treated
* by Batch as a filtered item, whereas it is really more like a
* skip. So we have to throw an exception in the processor if an
* unexpected null value comes back.
*/
assertNotNull(value);
assertTrue(value.matches("foo.*foo.*"));
}
}
@MessageEndpoint
public static class Doubler {
@ServiceActivator
public String cat(String value, @Header(value="stepExecution.jobExecution.jobInstance.jobParameters.getLong('factor')", required=false) Integer input) {
long factor = input==null ? 1 : input;
for (int i=1; i<factor; i++) {
value += value;
}
return value;
}
}
}

46
spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/AsyncItemProcessorMessagingGatewayTests-context.xml
View File

@@ -1,25 +1,23 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-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/context
http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://www.springframework.org/schema/tx
http://www.springframework.org/schema/tx/spring-tx-2.5.xsd">
<annotation-config />
<channel id="requests" />
<channel id="replies">
<queue />
</channel>
<gateway service-interface="org.springframework.batch.item.ItemProcessor"
id="processor" default-reply-timeout="1000" default-request-channel="requests"
default-reply-channel="replies" />
<service-activator input-channel="requests"
output-channel="replies" ref="doubler" />
<beans:bean id="doubler"
class="org.springframework.batch.integration.async.AsyncItemProcessorMessagingGatewayTests$Doubler" />
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
<annotation-config />
<channel id="requests"/>
<channel id="replies">
<queue />
</channel>
<gateway service-interface="org.springframework.batch.item.ItemProcessor" id="processor" default-reply-timeout="1000"
default-request-channel="requests" default-reply-channel="replies" />
<service-activator input-channel="requests" output-channel="replies" ref="doubler"/>
<beans:bean id="doubler" class="org.springframework.batch.integration.async.AsyncItemProcessorMessagingGatewayTests$Doubler"
scope="step">
<beans:property name="factor" value="#{jobParameters['factor']}" />
</beans:bean>
<beans:bean class="org.springframework.batch.core.scope.StepScope">
<beans:property name="proxyTargetClass" value="true"/>
</beans:bean>
</beans:beans>

28
spring-batch-integration/src/test/resources/org/springframework/batch/integration/async/PollingAsyncItemProcessorMessagingGatewayTests-context.xml Normal file
View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-2.0.xsd">
<annotation-config />
<channel id="requests">
<queue />
<interceptors>
<beans:bean class="org.springframework.batch.integration.async.StepExecutionInterceptor" />
</interceptors>
</channel>
<channel id="replies">
<queue />
</channel>
<gateway service-interface="org.springframework.batch.item.ItemProcessor" id="processor" default-reply-timeout="1000"
default-request-channel="requests" default-reply-channel="replies" />
<service-activator input-channel="requests" output-channel="replies" ref="doubler">
<poller fixed-rate="100" />
</service-activator>
<beans:bean id="doubler"
class="org.springframework.batch.integration.async.PollingAsyncItemProcessorMessagingGatewayTests$Doubler" />
<beans:bean class="org.springframework.batch.core.scope.StepScope">
<beans:property name="proxyTargetClass" value="true" />
</beans:bean>
</beans:beans>