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

INT-1451, more refactoring

Browse Source
This commit is contained in:
Oleg Zhurakousky
2010-09-25 16:16:34 -04:00
parent 176115af57
commit 47b1f7ae8e
6 changed files with 54 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractPollingEndpoint.java
View File

@@ -97,23 +97,27 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
@SuppressWarnings("unchecked")
private Runnable createPoller() throws Exception{
ProxyFactory proxyFactory = new ProxyFactory(this);
// Add Transaction advice first
Callable<Boolean> pollingTask = this;
Advisor transactionAdvice = this.pollerMetadata.getTransactionAdvisor();
if (transactionAdvice != null){
proxyFactory.addAdvisor(transactionAdvice);
}
// . . .then add the rest of the advises
List<Advice> adviceChain = this.pollerMetadata.getAdviceChain();
if (!CollectionUtils.isEmpty(adviceChain)){
for (Advice advice : adviceChain) {
proxyFactory.addAdvice(advice);
if (transactionAdvice != null || !CollectionUtils.isEmpty(adviceChain)){
ProxyFactory proxyFactory = new ProxyFactory(this);
// Add Transaction advice first
if (transactionAdvice != null){
proxyFactory.addAdvisor(transactionAdvice);
}
// . . .then add the rest of the advises
if (!CollectionUtils.isEmpty(adviceChain)){
for (Advice advice : adviceChain) {
proxyFactory.addAdvice(advice);
}
}
pollingTask = (Callable<Boolean>) proxyFactory.getProxy(this.beanClassLoader);
}
return new Poller((Callable<Boolean>) proxyFactory.getProxy(this.beanClassLoader));
return new Poller(pollingTask);
}
// LifecycleSupport implementation
@@ -134,6 +138,7 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
this.runningTask.cancel(true);
}
this.runningTask = null;
this.initialized = false;
}
public void setPollerMetadata(PollerMetadata pollerMetadata) {
@@ -167,7 +172,8 @@ public abstract class AbstractPollingEndpoint extends AbstractEndpoint implement
int count = 0;
while (maxMessagesPerPoll <= 0 || count < maxMessagesPerPoll) {
try {
if (!pollingTask.call()){
boolean b = pollingTask.call();
if (!b){
break;
}
count++;

6
spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java
View File

@@ -32,6 +32,7 @@ import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.Message;
@@ -45,6 +46,7 @@ import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.integration.monitor.LifecycleMessageHandlerMonitor;
import org.springframework.integration.monitor.QueueChannelMonitor;
import org.springframework.integration.monitor.DirectChannelMonitor;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.jmx.support.MBeanServerFactoryBean;
import org.springframework.jmx.support.ObjectNameManager;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
@@ -146,7 +148,9 @@ public class ControlBusTests {
RootBeanDefinition endpointDef = new RootBeanDefinition(PollingConsumer.class);
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("testChannel"));
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RootBeanDefinition(BridgeHandler.class));
endpointDef.getPropertyValues().add("trigger", new PeriodicTrigger(10000));
RootBeanDefinition pollerMetaDefinition = new RootBeanDefinition(PollerMetadata.class);
pollerMetaDefinition.getPropertyValues().add("trigger", new PeriodicTrigger(10000));
endpointDef.getPropertyValues().add("pollerMetadata", pollerMetaDefinition);
context.registerBeanDefinition("pollingConsumer", endpointDef);
context.registerBeanDefinition("taskScheduler", new RootBeanDefinition(ThreadPoolTaskScheduler.class));
registerControlBus(context, "domain.test4");

4
spring-integration-stream/pom.xml
View File

@@ -29,5 +29,9 @@
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
</dependency>
<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
</dependency>
</dependencies>
</project>

10
spring-integration-stream/src/test/java/org/springframework/integration/stream/ByteStreamWritingMessageHandlerTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.stream;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
@@ -28,6 +29,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
@@ -67,6 +69,7 @@ public class ByteStreamWritingMessageHandlerTests {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setBeanFactory(mock(BeanFactory.class));
}
@After
@@ -96,6 +99,7 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskSameAsMessageCount() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(3);
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
@@ -113,6 +117,7 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskLessThanMessageCount() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
@@ -129,6 +134,7 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskExceedsMessageCount() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(5);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
@@ -146,6 +152,7 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void testMaxMessagesLessThanMessageCountWithMultipleDispatches() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
@@ -171,6 +178,7 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void testMaxMessagesExceedsMessageCountWithMultipleDispatches() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(5);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
@@ -196,6 +204,7 @@ public class ByteStreamWritingMessageHandlerTests {
public void testStreamResetBetweenDispatches() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
channel.send(new GenericMessage<byte[]>(new byte[] {1,2,3}), 0);
@@ -219,6 +228,7 @@ public class ByteStreamWritingMessageHandlerTests {
@Test
public void testStreamWriteBetweenDispatches() throws IOException {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);

10
spring-integration-stream/src/test/java/org/springframework/integration/stream/CharacterStreamWritingMessageHandlerTests.java
View File

@@ -17,6 +17,7 @@
package org.springframework.integration.stream;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock;
import java.io.StringWriter;
import java.util.Date;
@@ -27,6 +28,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.GenericMessage;
@@ -67,6 +69,7 @@ public class CharacterStreamWritingMessageHandlerTests {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setBeanFactory(mock(BeanFactory.class));
}
@After
@@ -85,6 +88,7 @@ public class CharacterStreamWritingMessageHandlerTests {
public void twoStringsAndNoNewLinesByDefault() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(1);
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<String>("foo"), 0);
channel.send(new GenericMessage<String>("bar"), 0);
@@ -103,6 +107,7 @@ public class CharacterStreamWritingMessageHandlerTests {
public void twoStringsWithNewLines() {
handler.setShouldAppendNewLine(true);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<String>("foo"), 0);
@@ -122,6 +127,7 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskSameAsMessageCount() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerMetadata(pollerMetadata);
channel.send(new GenericMessage<String>("foo"), 0);
@@ -135,6 +141,7 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void maxMessagesPerTaskExceedsMessageCountWithAppendedNewLines() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(10);
endpoint.setPollerMetadata(pollerMetadata);
endpoint.setReceiveTimeout(0);
@@ -151,6 +158,7 @@ public class CharacterStreamWritingMessageHandlerTests {
@Test
public void singleNonStringObject() {
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(1);
endpoint.setPollerMetadata(pollerMetadata);
TestObject testObject = new TestObject("foo");
@@ -165,6 +173,7 @@ public class CharacterStreamWritingMessageHandlerTests {
public void twoNonStringObjectWithOutNewLines() {
endpoint.setReceiveTimeout(0);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setTrigger(trigger);
pollerMetadata.setMaxMessagesPerPoll(2);
endpoint.setPollerMetadata(pollerMetadata);
TestObject testObject1 = new TestObject("foo");
@@ -183,6 +192,7 @@ public class CharacterStreamWritingMessageHandlerTests {
endpoint.setReceiveTimeout(0);
PollerMetadata pollerMetadata = new PollerMetadata();
pollerMetadata.setMaxMessagesPerPoll(2);
pollerMetadata.setTrigger(trigger);
endpoint.setPollerMetadata(pollerMetadata);
TestObject testObject1 = new TestObject("foo");
TestObject testObject2 = new TestObject("bar");

9
spring-integration-ws/src/test/java/org/springframework/integration/ws/config/WebServiceOutboundGatewayParserTests.java
View File

@@ -27,6 +27,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.ws.MarshallingWebServiceOutboundGateway;
import org.springframework.integration.ws.SimpleWebServiceOutboundGateway;
import org.springframework.oxm.Marshaller;
@@ -228,9 +229,11 @@ public class WebServiceOutboundGatewayParserTests {
"simpleWebServiceOutboundGatewayParserTests.xml", this.getClass());
AbstractEndpoint endpoint = (AbstractEndpoint) context.getBean("gatewayWithPoller");
assertEquals(PollingConsumer.class, endpoint.getClass());
Object obj = new DirectFieldAccessor(endpoint).getPropertyValue("trigger");
assertEquals(PeriodicTrigger.class, obj.getClass());
PeriodicTrigger trigger = (PeriodicTrigger) obj;
Object pollerMetadata = new DirectFieldAccessor(endpoint).getPropertyValue("pollerMetadata");
assertEquals(PollerMetadata.class, pollerMetadata.getClass());
Object triggerObject = new DirectFieldAccessor(pollerMetadata).getPropertyValue("trigger");
assertEquals(PeriodicTrigger.class, triggerObject.getClass());
PeriodicTrigger trigger = (PeriodicTrigger) triggerObject;
DirectFieldAccessor accessor = new DirectFieldAccessor(trigger);
accessor = new DirectFieldAccessor(trigger);
assertEquals("PeriodicTrigger had wrong period",