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

Upgrade Spring Integration to M5 (actually CI-SNAPSHOT).

Browse Source
This commit is contained in:
dsyer
2008-07-11 08:04:17 +00:00
parent 28f388d329
commit b079affa09
9 changed files with 125 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
spring-batch-integration/pom.xml
View File

@@ -12,7 +12,7 @@
<relativePath>..</relativePath>
</parent>
<properties>
<spring.integration.version>1.0.0.M4</spring.integration.version>
<spring.integration.version>1.0.0.CI-SNAPSHOT</spring.integration.version>
</properties>
<profiles>
<profile>
@@ -83,6 +83,10 @@
<groupId>org.springframework</groupId>
<artifactId>org.springframework.aop</artifactId>
</exclusion>
<exclusion>
<groupId>org.springframework</groupId>
<artifactId>org.springframework.transaction</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>
@@ -115,6 +119,14 @@
<groupId>org.springframework</groupId>
<artifactId>org.springframework.transaction</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.commons</groupId>
<artifactId>com.springsource.org.apache.commons.net</artifactId>
</exclusion>
<exclusion>
<groupId>org.apache.oro</groupId>
<artifactId>com.springsource.org.apache.oro</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>

2
spring-batch-integration/src/test/java/org/springframework/batch/integration/file/ResourcePayloadAsJobParameterStrategyTests.java
View File

@@ -56,7 +56,7 @@ public class ResourcePayloadAsJobParameterStrategyTests {
fail("Expected ClassCastException");
} catch (ClassCastException e) {
String message = e.getMessage();
assertTrue("Wrong message: "+message, message.contains("String cannot be cast"));
assertTrue("Wrong message: "+message, message.contains("String"));
}
}

16
spring-batch-integration/src/test/java/org/springframework/batch/integration/item/MessageChannelItemWriterTests.java
View File

@@ -23,7 +23,6 @@ import java.lang.annotation.Annotation;
import java.lang.reflect.Method;
import org.junit.Test;
import org.springframework.batch.integration.item.MessageChannelItemWriter;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.channel.MessageChannel;
@@ -32,8 +31,7 @@ import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Target;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.ReflectionUtils;
/**
@@ -78,7 +76,7 @@ public class MessageChannelItemWriterTests {
@Test
public void testWriteWithRollback() throws Exception {
DirectChannel channel = new DirectChannel();
channel.subscribe(new Target() {
channel.subscribe(new MessageTarget() {
public boolean send(Message<?> message) {
throw new RuntimeException("Planned failure");
}
@@ -107,11 +105,11 @@ public class MessageChannelItemWriterTests {
}
});
// INT-184: this shouldn't be necessary?
endpoint.setErrorHandler(new ErrorHandler() {
public void handle(Throwable t) {
throw (RuntimeException)t;
}
});
// endpoint.setErrorHandler(new ErrorHandler() {
// public void handle(Throwable t) {
// throw (RuntimeException)t;
// }
// });
channel.subscribe(endpoint);
endpoint.start();
MessageChannelItemWriter writer = new MessageChannelItemWriter();

9
spring-batch-integration/src/test/java/org/springframework/batch/integration/job/MessageOrientedStepTests.java
View File

@@ -31,9 +31,6 @@ import org.springframework.batch.core.JobInstance;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.integration.JobRepositorySupport;
import org.springframework.batch.integration.job.JobExecutionRequest;
import org.springframework.batch.integration.job.MessageOrientedStep;
import org.springframework.batch.integration.job.StepExecutionTimeoutException;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.integration.channel.MessageChannel;
@@ -41,7 +38,7 @@ import org.springframework.integration.channel.ThreadLocalChannel;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Target;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.ReflectionUtils;
/**
@@ -121,7 +118,7 @@ public class MessageOrientedStepTests {
@Test
public void testVanillaExecute() throws Exception {
requestChannel.subscribe(new Target() {
requestChannel.subscribe(new MessageTarget() {
public boolean send(Message<?> message) {
JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload();
jobExecution.setStatus(BatchStatus.COMPLETED);
@@ -133,7 +130,7 @@ public class MessageOrientedStepTests {
@Test
public void testExecuteWithFailure() throws Exception {
requestChannel.subscribe(new Target() {
requestChannel.subscribe(new MessageTarget() {
public boolean send(Message<?> message) {
JobExecutionRequest jobExecution = (JobExecutionRequest) message.getPayload();
jobExecution.registerThrowable(new RuntimeException("Planned failure"));

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

@@ -2,6 +2,7 @@ package org.springframework.batch.integration.launch;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import org.junit.Before;
import org.junit.Test;
@@ -15,6 +16,7 @@ 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.MessageHandlingException;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -44,7 +46,12 @@ public class JobLaunchingMessageHandlerIntegrationTests {
@SuppressWarnings("unchecked")
public void testNoReply() {
GenericMessage<JobLaunchRequest> trigger = new GenericMessage<JobLaunchRequest>(new JobLaunchRequest(job, new JobParameters()));
requestChannel.send(trigger);
try {
requestChannel.send(trigger);
} catch (MessageHandlingException e) {
String message = e.getMessage();
assertTrue("Wrong message: "+message, message.contains("reply channel"));
}
Message<JobExecution> executionMessage = (Message<JobExecution>) responseChannel.receive(1000);
assertNull("JobExecution message received when no return address set", executionMessage);

142
spring-batch-integration/src/test/java/org/springframework/batch/integration/retry/PollableSourceRetryTests.java
View File

@@ -12,6 +12,9 @@ import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.junit.Test;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.DefaultPointcutAdvisor;
import org.springframework.aop.support.NameMatchMethodPointcut;
import org.springframework.batch.item.ItemKeyGenerator;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemRecoverer;
@@ -26,14 +29,15 @@ import org.springframework.batch.support.transaction.TransactionAwareProxyFactor
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.dispatcher.DirectChannel;
import org.springframework.integration.endpoint.EndpointTrigger;
import org.springframework.integration.endpoint.SourceEndpoint;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.Source;
import org.springframework.integration.message.Target;
import org.springframework.integration.scheduling.MessagingTaskScheduler;
import org.springframework.integration.message.MessageSource;
import org.springframework.integration.message.MessageTarget;
import org.springframework.integration.scheduling.PollingSchedule;
import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
import org.springframework.integration.scheduling.SimpleTaskScheduler;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.util.ErrorHandler;
import org.springframework.transaction.interceptor.TransactionInterceptor;
import org.springframework.util.StringUtils;
@@ -72,19 +76,21 @@ public class PollableSourceRetryTests {
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k")));
int beforeCount = list.size();
Target handler = new Target() {
MessageTarget handler = new MessageTarget() {
public boolean send(Message<?> message) {
Object payload = message.getPayload();
logger.debug("Handling: " + payload);
return processed.add((String) payload);
}
};
Source<Object> source = getPollableSource(list);
MessageSource<Object> source = getPollableSource(list);
DirectChannel channel = getChannel(handler, source);
SourceEndpoint endpoint = getSourceEndpoint(source, channel);
endpoint.setDispatchAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor() }));
endpoint.initializeTask();
MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint);
MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getTransactionInterceptor()}, "poll");
endpoint.afterPropertiesSet();
EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule());
trigger.addTarget(target);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 2, 40);
@@ -103,7 +109,7 @@ public class PollableSourceRetryTests {
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k")));
int beforeCount = list.size();
Target handler = new Target() {
MessageTarget handler = new MessageTarget() {
public boolean send(Message<?> message) {
Object payload = message.getPayload();
logger.debug("Handling: " + payload);
@@ -111,10 +117,13 @@ public class PollableSourceRetryTests {
throw new RuntimeException("Planned failure: " + payload);
}
};
Source<Object> source = getPollableSource(list);
MessageSource<Object> source = getPollableSource(list);
DirectChannel channel = getChannel(handler, source);
SourceEndpoint endpoint = getSourceEndpoint(source, channel);
MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint);
endpoint.afterPropertiesSet();
EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule());
trigger.addTarget(endpoint);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 2, 20);
@@ -135,7 +144,7 @@ public class PollableSourceRetryTests {
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,c,d,e,f,g,h,j,k")));
int beforeCount = list.size();
Target handler = new Target() {
MessageTarget handler = new MessageTarget() {
public boolean send(Message<?> message) {
Object payload = message.getPayload();
logger.debug("Handling: " + payload);
@@ -143,12 +152,14 @@ public class PollableSourceRetryTests {
throw new RuntimeException("Planned failure: " + payload);
}
};
Source<Object> source = getPollableSource(list);
MessageSource<Object> source = getPollableSource(list);
DirectChannel channel = getChannel(handler, source);
SourceEndpoint endpoint = getSourceEndpoint(source, channel);
endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor() }));
endpoint.initializeTask();
MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint);
MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getTransactionInterceptor()}, "poll");
endpoint.afterPropertiesSet();
EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule());
trigger.addTarget(target);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 2, 20);
@@ -171,7 +182,7 @@ public class PollableSourceRetryTests {
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k")));
int beforeCount = list.size();
Target handler = new Target() {
MessageTarget handler = new MessageTarget() {
public boolean send(Message<?> message) {
Object payload = message.getPayload();
logger.debug("Handling: " + payload);
@@ -183,12 +194,14 @@ public class PollableSourceRetryTests {
}
};
Source<Object> source = getPollableSource(list);
MessageSource<Object> source = getPollableSource(list);
DirectChannel channel = getChannel(handler, source);
SourceEndpoint endpoint = getSourceEndpoint(source, channel);
endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor() }));
endpoint.initializeTask();
MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint);
MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getTransactionInterceptor()}, "poll");
endpoint.afterPropertiesSet();
EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule());
trigger.addTarget(target);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 5, 20);
@@ -214,7 +227,7 @@ public class PollableSourceRetryTests {
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k")));
int beforeCount = list.size();
Target handler = new Target() {
MessageTarget handler = new MessageTarget() {
public boolean send(Message<?> message) {
Object payload = message.getPayload();
logger.debug("Handling: " + payload);
@@ -226,13 +239,15 @@ public class PollableSourceRetryTests {
}
};
Source<Object> source = getPollableSource(list);
MessageSource<Object> source = getPollableSource(list);
DirectChannel channel = getChannel(handler, source);
SourceEndpoint endpoint = getSourceEndpoint(source, channel);
endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor(),
getRepeatOperationsInterceptor(3) }));
endpoint.initializeTask();
MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint);
// endpoint.addInterceptor(getTransactionInterceptor());
MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getRepeatOperationsInterceptor(3), getTransactionInterceptor()}, "poll");
endpoint.afterPropertiesSet();
EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule());
trigger.addTarget(target);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 6, 100);
@@ -258,7 +273,7 @@ public class PollableSourceRetryTests {
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,b,fail,d,e,f,g,h,j,k")));
int beforeCount = list.size();
Target handler = new Target() {
MessageTarget handler = new MessageTarget() {
public boolean send(Message<?> message) {
if (message == null) {
return false;
@@ -275,14 +290,17 @@ public class PollableSourceRetryTests {
}
};
Source<Object> source = getPollableSource(list);
MessageSource<Object> source = getPollableSource(list);
MessageChannel channel = getChannel(handler, source);
SourceEndpoint endpoint = getSourceEndpoint(source, channel);
endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor() }));
endpoint
.setDispatchAdviceChain(Arrays.asList(new Advice[] { getRetryOperationsInterceptor(itemKeyGenerator) }));
endpoint.initializeTask();
MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint);
MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getTransactionInterceptor()}, "poll");
// this was the old dispatch advice chain
channel = (MessageChannel) getProxy(channel, MessageChannel.class,
new Advice[] { getRetryOperationsInterceptor(itemKeyGenerator) }, "send");
endpoint.afterPropertiesSet();
EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule());
trigger.addTarget(target);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 4, 20);
@@ -308,7 +326,7 @@ public class PollableSourceRetryTests {
list.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray("a,fail,c,d,e,f,g,h,j,k")));
int beforeCount = list.size();
Target handler = new Target() {
MessageTarget handler = new MessageTarget() {
public boolean send(Message<?> message) {
Object payload = message.getPayload();
logger.debug("Handling: " + payload);
@@ -320,15 +338,17 @@ public class PollableSourceRetryTests {
}
};
Source<Object> source = getPollableSource(list);
MessageSource<Object> source = getPollableSource(list);
MessageChannel channel = getChannel(handler, source);
SourceEndpoint endpoint = getSourceEndpoint(source, channel);
endpoint.setTaskAdviceChain(Arrays.asList(new Advice[] { getTransactionInterceptor(),
getRepeatOperationsInterceptor(3) }));
endpoint
.setDispatchAdviceChain(Arrays.asList(new Advice[] { getRetryOperationsInterceptor(itemKeyGenerator) }));
endpoint.initializeTask();
MessagingTaskScheduler scheduler = getSchedulerWithErrorHandler(endpoint);
MessageTarget target = (MessageTarget) getProxy(endpoint, MessageTarget.class, new Advice[] {getRepeatOperationsInterceptor(3), getTransactionInterceptor()}, "poll");
// this was the old dispatch advice chain
channel = (MessageChannel) getProxy(channel, MessageChannel.class,
new Advice[] { getRetryOperationsInterceptor(itemKeyGenerator) }, "send");
endpoint.afterPropertiesSet();
EndpointTrigger trigger = new EndpointTrigger(endpoint.getSchedule());
trigger.addTarget(target);
TaskScheduler scheduler = getSchedulerWithErrorHandler(trigger);
waitForResults(scheduler, 6, 100);
System.err.println(processed);
@@ -352,10 +372,13 @@ public class PollableSourceRetryTests {
* @param channel
* @return
*/
private SourceEndpoint getSourceEndpoint(Source<Object> source, MessageChannel channel) {
private SourceEndpoint getSourceEndpoint(MessageSource<Object> source, MessageChannel channel) {
PollingSchedule schedule = new PollingSchedule(50);
schedule.setFixedRate(true); // used to be the default
return new SourceEndpoint(source, channel, schedule);
SourceEndpoint endpoint = new SourceEndpoint(source);
endpoint.setOutputChannel(channel);
endpoint.setSchedule(schedule);
return endpoint;
}
/**
@@ -363,7 +386,7 @@ public class PollableSourceRetryTests {
* @param source
* @return
*/
private DirectChannel getChannel(Target handler, Source<Object> source) {
private DirectChannel getChannel(MessageTarget handler, MessageSource<Object> source) {
DirectChannel channel = new DirectChannel(source);
channel.setName("input");
channel.subscribe(handler);
@@ -379,7 +402,7 @@ public class PollableSourceRetryTests {
lifecycle.stop();
}
private Source<Object> getPollableSource(List<String> list) {
private MessageSource<Object> getPollableSource(List<String> list) {
final ItemReader reader = new ListItemReader(list) {
public Object read() {
Object item = super.read();
@@ -387,7 +410,7 @@ public class PollableSourceRetryTests {
return item;
}
};
Source<Object> source = new Source<Object>() {
MessageSource<Object> source = new MessageSource<Object>() {
public Message<Object> receive() {
try {
return new GenericMessage<Object>(reader.read());
@@ -404,9 +427,8 @@ public class PollableSourceRetryTests {
}
// Workaround for INT-182
private MessagingTaskScheduler getSchedulerWithErrorHandler(Runnable task) {
SimpleMessagingTaskScheduler scheduler = new SimpleMessagingTaskScheduler(Executors
.newSingleThreadScheduledExecutor());
private TaskScheduler getSchedulerWithErrorHandler(Runnable task) {
SimpleTaskScheduler scheduler = new SimpleTaskScheduler(Executors.newSingleThreadScheduledExecutor());
scheduler.setErrorHandler(new ErrorHandler() {
public void handle(Throwable t) {
logger.error("Exception in scheduler", t);
@@ -444,8 +466,7 @@ public class PollableSourceRetryTests {
* @return
*/
private TransactionInterceptor getTransactionInterceptor() {
return new TransactionInterceptor(new ResourcelessTransactionManager(), PropertiesConverter
.stringToProperties("*=PROPAGATION_REQUIRED"));
return new TransactionInterceptor(new ResourcelessTransactionManager(), PropertiesConverter.stringToProperties("*=PROPAGATION_REQUIRED"));
}
/**
@@ -459,4 +480,19 @@ public class PollableSourceRetryTests {
advice.setRepeatOperations(repeatTemplate);
return advice;
}
private Object getProxy(Object target, Class<?> intf, Advice[] advices, String methodName) {
ProxyFactory factory = new ProxyFactory(target);
for (int i = 0; i < advices.length; i++) {
DefaultPointcutAdvisor advisor = new DefaultPointcutAdvisor(advices[i]);
NameMatchMethodPointcut pointcut = new NameMatchMethodPointcut();
pointcut.addMethodName(methodName);
advisor.setPointcut(pointcut);
factory.addAdvisor(advisor);
}
factory.setProxyTargetClass(false);
factory.addInterface(intf);
return factory.getProxy();
}
}

2
spring-batch-integration/src/test/resources/org/springframework/batch/integration/chunk/ChunkMessageItemWriterIntegrationTests-context.xml
View File

@@ -28,6 +28,6 @@
</beans:property>
</beans:bean>
<handler-endpoint input-channel="requests" output-channel="replies" handler="chunkHandler" method="handleChunk" />
<service-activator input-channel="requests" output-channel="replies" ref="chunkHandler" method="handleChunk" />
</beans:beans>

2
spring-batch-integration/src/test/resources/org/springframework/batch/integration/job/MessageOrientedStepIntegrationTests-context.xml
View File

@@ -42,6 +42,6 @@
</property>
<property name="jobRepository" ref="jobRepository" />
</bean>
<integration:handler-endpoint handler="step" method="handle" input-channel="requests"
<integration:service-activator ref="step" method="handle" input-channel="requests"
output-channel="replies" />
</beans>

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

@@ -22,10 +22,8 @@
<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" />
<integration:service-activator input-channel="requests" ref="jobLaunchingHandler" return-address-overrides="true" />
<integration:service-activator input-channel="none" output-channel="jobs" ref="jobRequestConverter" />
<bean id="jobRequestConverter" class="org.springframework.batch.integration.launch.JobRequestConverter" />