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

Removing the @Poller annotation. Now, any Annotation-based endpoint must have a SubscribableChannel reference for its inputChannel. If necessary, the new <bridge/> element can be used to convert PollableChannels to SubscribableChannels.

Browse Source
This commit is contained in:
Mark Fisher
2008-11-25 19:45:16 +00:00
parent 1f0104d6f8
commit 0e5b4bcb50
18 changed files with 17 additions and 867 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationChannelAdapterAdviceChainTestBean.java
View File

@@ -1,38 +0,0 @@
/*
* Copyright 2002-2008 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.integration.config;
import java.util.concurrent.TimeUnit;
import org.springframework.integration.annotation.ChannelAdapter;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Poller;
/**
* @author Mark Fisher
*/
@MessageEndpoint
public class PollerAnnotationChannelAdapterAdviceChainTestBean {
@ChannelAdapter("output")
@Poller(interval=5, timeUnit=TimeUnit.SECONDS, maxMessagesPerPoll=1,
adviceChain="aroundAdvice,beforeAdvice,afterAdvice")
public String testMethod() {
return "test";
}
}

37
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationChannelAdapterAdviceChainTests-context.xml
View File

@@ -1,37 +0,0 @@
<?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:stream="http://www.springframework.org/schema/integration/stream"
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/integration/stream
http://www.springframework.org/schema/integration/stream/spring-integration-stream-1.0.xsd">
<annotation-config/>
<channel id="output">
<queue capacity="1"/>
</channel>
<beans:bean id="channelAdapter" class="org.springframework.integration.config.PollerAnnotationChannelAdapterAdviceChainTestBean"/>
<beans:bean id="latch" class="java.util.concurrent.CountDownLatch">
<beans:constructor-arg value="4"/>
</beans:bean>
<beans:bean id="beforeAdvice" class="org.springframework.integration.config.TestBeforeAdvice">
<beans:constructor-arg ref="latch"/>
</beans:bean>
<beans:bean id="afterAdvice" class="org.springframework.integration.config.TestAfterAdvice">
<beans:constructor-arg ref="latch"/>
</beans:bean>
<beans:bean id="aroundAdvice" class="org.springframework.integration.config.TestAroundAdvice">
<beans:constructor-arg ref="latch"/>
</beans:bean>
</beans:beans>

69
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationChannelAdapterAdviceChainTests.java
View File

@@ -1,69 +0,0 @@
/*
* Copyright 2002-2008 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.integration.config;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class PollerAnnotationChannelAdapterAdviceChainTests {
@Autowired @Qualifier("output")
private PollableChannel output;
@Autowired
private CountDownLatch latch;
@Autowired
private TestBeforeAdvice beforeAdvice;
@Autowired
private TestAfterAdvice afterAdvice;
@Autowired
private TestAroundAdvice aroundAdvice;
@Test
public void testAdviceChain() throws InterruptedException {
Message<?> reply = output.receive(1000);
assertEquals("test", reply.getPayload());
latch.await(1, TimeUnit.SECONDS);
assertEquals(0, latch.getCount());
assertEquals(4, aroundAdvice.getPreCount());
assertEquals(3, beforeAdvice.getLatchCount());
assertEquals(2, afterAdvice.getLatchCount());
assertEquals(1, aroundAdvice.getPostCount());
}
}

68
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationChannelAdapterTransactionalTestBean.java
View File

@@ -1,68 +0,0 @@
/*
* Copyright 2002-2008 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.integration.config;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import org.springframework.integration.annotation.ChannelAdapter;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Poller;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
/**
* @author Mark Fisher
*/
@MessageEndpoint
public class PollerAnnotationChannelAdapterTransactionalTestBean {
private volatile boolean shouldFail;
private BlockingQueue<String> queue = new ArrayBlockingQueue<String>(1);
public void setShouldFail(boolean shouldFail) {
this.shouldFail = shouldFail;
}
public void setNextValue(String nextValue) {
try {
this.queue.put(nextValue);
}
catch (InterruptedException e) {
}
}
@ChannelAdapter("output")
@Poller(interval=100, maxMessagesPerPoll=1,
transactionManager="testTransactionManager",
transactionAttributes=@Transactional(propagation=Propagation.REQUIRES_NEW))
public String testMethod() {
String result = null;
try {
result = this.queue.take();
}
catch (InterruptedException e) {
}
if (this.shouldFail) {
throw new IllegalArgumentException("intentional test failure");
}
return result;
}
}

23
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationChannelAdapterTransactionalTests-context.xml
View File

@@ -1,23 +0,0 @@
<?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:stream="http://www.springframework.org/schema/integration/stream"
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/integration/stream
http://www.springframework.org/schema/integration/stream/spring-integration-stream-1.0.xsd">
<annotation-config/>
<channel id="output">
<queue capacity="1"/>
</channel>
<beans:bean id="channelAdapter" class="org.springframework.integration.config.PollerAnnotationChannelAdapterTransactionalTestBean"/>
<beans:bean id="testTransactionManager" class="org.springframework.integration.util.TestTransactionManager"/>
</beans:beans>

90
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationChannelAdapterTransactionalTests.java
View File

@@ -1,90 +0,0 @@
/*
* Copyright 2002-2008 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.integration.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.util.TestTransactionManager;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.TransactionDefinition;
/**
* @author Mark Fisher
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class PollerAnnotationChannelAdapterTransactionalTests {
@Autowired @Qualifier("output")
private PollableChannel output;
@Autowired
private PollerAnnotationChannelAdapterTransactionalTestBean adapter;
@Autowired
private TestTransactionManager transactionManager;
@Before
public void resetTransactionManager() {
transactionManager.reset();
}
@Test
public void commit() throws InterruptedException {
adapter.setShouldFail(false);
adapter.setNextValue("commit-test");
transactionManager.waitForCompletion(1000);
Message<?> reply = output.receive(1000);
assertEquals("commit-test", reply.getPayload());
assertEquals(1, transactionManager.getCommitCount());
assertEquals(0, transactionManager.getRollbackCount());
}
@Test
public void rollback() throws InterruptedException {
adapter.setShouldFail(true);
adapter.setNextValue("rollback-test");
transactionManager.waitForCompletion(1000);
assertNull(output.receive(0));
assertEquals(0, transactionManager.getCommitCount());
assertEquals(1, transactionManager.getRollbackCount());
}
@Test
public void verifyPropagationSetting() throws InterruptedException {
adapter.setShouldFail(false);
adapter.setNextValue("propagation-test");
transactionManager.waitForCompletion(1000);
Message<?> reply = output.receive(1000);
assertEquals("propagation-test", reply.getPayload());
assertEquals(TransactionDefinition.PROPAGATION_REQUIRES_NEW,
transactionManager.getLastDefinition().getPropagationBehavior());
}
}

38
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationConsumerAdviceChainTestBean.java
View File

@@ -1,38 +0,0 @@
/*
* Copyright 2002-2008 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.integration.config;
import java.util.concurrent.TimeUnit;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
/**
* @author Mark Fisher
*/
@MessageEndpoint
public class PollerAnnotationConsumerAdviceChainTestBean {
@ServiceActivator(inputChannel="input", outputChannel="output")
@Poller(interval=5, timeUnit=TimeUnit.SECONDS, maxMessagesPerPoll=1,
adviceChain="beforeAdvice ,aroundAdvice, afterAdvice ") // spacing intentional
public String testMethod(String input) {
return input.toUpperCase();
}
}

41
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationConsumerAdviceChainTests-context.xml
View File

@@ -1,41 +0,0 @@
<?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:stream="http://www.springframework.org/schema/integration/stream"
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/integration/stream
http://www.springframework.org/schema/integration/stream/spring-integration-stream-1.0.xsd">
<annotation-config/>
<channel id="input">
<queue capacity="1"/>
</channel>
<channel id="output">
<queue capacity="1"/>
</channel>
<beans:bean id="service" class="org.springframework.integration.config.PollerAnnotationConsumerAdviceChainTestBean"/>
<beans:bean id="latch" class="java.util.concurrent.CountDownLatch">
<beans:constructor-arg value="4"/>
</beans:bean>
<beans:bean id="beforeAdvice" class="org.springframework.integration.config.TestBeforeAdvice">
<beans:constructor-arg ref="latch"/>
</beans:bean>
<beans:bean id="afterAdvice" class="org.springframework.integration.config.TestAfterAdvice">
<beans:constructor-arg ref="latch"/>
</beans:bean>
<beans:bean id="aroundAdvice" class="org.springframework.integration.config.TestAroundAdvice">
<beans:constructor-arg ref="latch"/>
</beans:bean>
</beans:beans>

75
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationConsumerAdviceChainTests.java
View File

@@ -1,75 +0,0 @@
/*
* Copyright 2002-2008 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.integration.config;
import static org.junit.Assert.assertEquals;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.StringMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class PollerAnnotationConsumerAdviceChainTests {
@Autowired @Qualifier("input")
private MessageChannel input;
@Autowired @Qualifier("output")
private PollableChannel output;
@Autowired
private CountDownLatch latch;
@Autowired
private TestBeforeAdvice beforeAdvice;
@Autowired
private TestAfterAdvice afterAdvice;
@Autowired
private TestAroundAdvice aroundAdvice;
@Test
public void testAdviceChain() throws InterruptedException {
input.send(new StringMessage("test"));
Message<?> reply = output.receive(3000);
assertEquals("TEST", reply.getPayload());
latch.await(1, TimeUnit.SECONDS);
assertEquals(0, latch.getCount());
assertEquals(4, beforeAdvice.getLatchCount());
assertEquals(3, aroundAdvice.getPreCount());
assertEquals(2, afterAdvice.getLatchCount());
assertEquals(1, aroundAdvice.getPostCount());
}
}

42
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationConsumerTransactionalTestBean.java
View File

@@ -1,42 +0,0 @@
/*
* Copyright 2002-2008 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.integration.config;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
/**
* @author Mark Fisher
*/
@MessageEndpoint
public class PollerAnnotationConsumerTransactionalTestBean {
@ServiceActivator(inputChannel="input", outputChannel="output")
@Poller(interval=100, maxMessagesPerPoll=1,
transactionManager="testTransactionManager",
transactionAttributes=@Transactional(propagation=Propagation.REQUIRES_NEW))
public String testMethod(String input) {
if ("bad".equals(input)) {
throw new IllegalArgumentException("bad input");
}
return input.toUpperCase();
}
}

27
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationConsumerTransactionalTests-context.xml
View File

@@ -1,27 +0,0 @@
<?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:stream="http://www.springframework.org/schema/integration/stream"
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/integration/stream
http://www.springframework.org/schema/integration/stream/spring-integration-stream-1.0.xsd">
<annotation-config/>
<channel id="input">
<queue capacity="1"/>
</channel>
<channel id="output">
<queue capacity="1"/>
</channel>
<beans:bean id="service" class="org.springframework.integration.config.PollerAnnotationConsumerTransactionalTestBean"/>
<beans:bean id="testTransactionManager" class="org.springframework.integration.util.TestTransactionManager"/>
</beans:beans>

87
org.springframework.integration/src/test/java/org/springframework/integration/config/PollerAnnotationConsumerTransactionalTests.java
View File

@@ -1,87 +0,0 @@
/*
* Copyright 2002-2008 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.integration.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.util.TestTransactionManager;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.transaction.TransactionDefinition;
/**
* @author Mark Fisher
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class PollerAnnotationConsumerTransactionalTests {
@Autowired @Qualifier("input")
private MessageChannel input;
@Autowired @Qualifier("output")
private PollableChannel output;
@Autowired
private TestTransactionManager transactionManager;
@Before
public void resetTransactionManager() {
transactionManager.reset();
}
@Test
public void commit() throws InterruptedException {
input.send(new StringMessage("test"));
transactionManager.waitForCompletion(3000);
Message<?> reply = output.receive(1000);
assertEquals("TEST", reply.getPayload());
assertEquals(1, transactionManager.getCommitCount());
assertEquals(0, transactionManager.getRollbackCount());
}
@Test
public void rollback() throws InterruptedException {
input.send(new StringMessage("bad"));
transactionManager.waitForCompletion(3000);
assertNull(output.receive(0));
assertEquals(0, transactionManager.getCommitCount());
assertEquals(1, transactionManager.getRollbackCount());
}
@Test
public void verifyPropagationSetting() throws InterruptedException {
input.send(new StringMessage("test"));
transactionManager.waitForCompletion(3000);
assertEquals(TransactionDefinition.PROPAGATION_REQUIRES_NEW,
transactionManager.getLastDefinition().getPropagationBehavior());
}
}

4
org.springframework.integration/src/test/java/org/springframework/integration/config/ServiceActivatorAnnotationPostProcessorTests.java
View File

@@ -27,7 +27,7 @@ import org.junit.Test;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.annotation.MessagingAnnotationPostProcessor;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.StringMessage;
@@ -45,7 +45,7 @@ public class ServiceActivatorAnnotationPostProcessorTests {
TestApplicationContext context = TestUtils.createTestApplicationContext();
RootBeanDefinition postProcessorDef = new RootBeanDefinition(MessagingAnnotationPostProcessor.class);
context.registerBeanDefinition("postProcessor", postProcessorDef);
context.registerBeanDefinition("testChannel", new RootBeanDefinition(QueueChannel.class));
context.registerBeanDefinition("testChannel", new RootBeanDefinition(DirectChannel.class));
RootBeanDefinition beanDefinition = new RootBeanDefinition(SimpleServiceActivatorAnnotationTestBean.class);
beanDefinition.getConstructorArgumentValues().addGenericArgumentValue(latch);
context.registerBeanDefinition("testBean", beanDefinition);

88
org.springframework.integration/src/test/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessorTests.java
View File

@@ -22,17 +22,14 @@ import static org.junit.Assert.assertTrue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.support.AbstractApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.annotation.ChannelAdapter;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.channel.BeanFactoryChannelResolver;
@@ -43,12 +40,8 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageHandler;
import org.springframework.integration.message.StringMessage;
import org.springframework.integration.scheduling.IntervalTrigger;
import org.springframework.integration.scheduling.Trigger;
import org.springframework.integration.util.TestUtils;
import org.springframework.integration.util.TestUtils.TestApplicationContext;
@@ -60,7 +53,7 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void serviceActivatorAnnotation() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
QueueChannel inputChannel = new QueueChannel();
DirectChannel inputChannel = new DirectChannel();
context.registerChannel("inputChannel", inputChannel);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
postProcessor.setBeanFactory(context.getBeanFactory());
@@ -172,7 +165,7 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testProxiedMessageEndpointAnnotation() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
QueueChannel inputChannel = new QueueChannel();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
@@ -192,7 +185,7 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInherited() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
QueueChannel inputChannel = new QueueChannel();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
@@ -210,7 +203,7 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInheritedWithProxy() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
QueueChannel inputChannel = new QueueChannel();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
@@ -230,7 +223,7 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInheritedFromInterface() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
QueueChannel inputChannel = new QueueChannel();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
@@ -248,7 +241,7 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInheritedFromInterfaceWithAutoCreatedChannels() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
QueueChannel inputChannel = new QueueChannel();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
@@ -266,7 +259,7 @@ public class MessagingAnnotationPostProcessorTests {
@Test
public void testMessageEndpointAnnotationInheritedFromInterfaceWithProxy() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
QueueChannel inputChannel = new QueueChannel();
DirectChannel inputChannel = new DirectChannel();
QueueChannel outputChannel = new QueueChannel();
context.registerChannel("inputChannel", inputChannel);
context.registerChannel("outputChannel", outputChannel);
@@ -283,51 +276,6 @@ public class MessagingAnnotationPostProcessorTests {
context.stop();
}
@Test
public void testEndpointWithPollerAnnotation() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
QueueChannel testChannel = new QueueChannel();
context.registerChannel("testChannel", testChannel);
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
postProcessor.setBeanFactory(context.getBeanFactory());
postProcessor.afterPropertiesSet();
AnnotatedEndpointWithPolledAnnotation bean = new AnnotatedEndpointWithPolledAnnotation();
postProcessor.postProcessAfterInitialization(bean, "testBean");
PollingConsumer endpoint = (PollingConsumer) context.getBean("testBean.prependFoo.serviceActivator");
Trigger trigger = (Trigger) new DirectFieldAccessor(endpoint).getPropertyValue("trigger");
assertEquals(IntervalTrigger.class, trigger.getClass());
DirectFieldAccessor triggerAccessor = new DirectFieldAccessor(trigger);
assertEquals(new Long(123000), triggerAccessor.getPropertyValue("interval"));
assertEquals(new Long(456000), triggerAccessor.getPropertyValue("initialDelay"));
assertEquals(true, triggerAccessor.getPropertyValue("fixedRate"));
}
@Test
public void testChannelAdapterAnnotation() throws InterruptedException {
TestApplicationContext context = TestUtils.createTestApplicationContext();
MessagingAnnotationPostProcessor postProcessor = new MessagingAnnotationPostProcessor();
postProcessor.setBeanFactory(context.getBeanFactory());
postProcessor.afterPropertiesSet();
ChannelAdapterAnnotationTestBean testBean = new ChannelAdapterAnnotationTestBean();
postProcessor.postProcessAfterInitialization(testBean, "testBean");
ChannelResolver channelResolver = new BeanFactoryChannelResolver(context);
DirectChannel testChannel = (DirectChannel) channelResolver.resolveChannelName("testChannel");
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<Message<?>> receivedMessage = new AtomicReference<Message<?>>();
testChannel.subscribe(new MessageHandler() {
public void handleMessage(Message<?> message) {
receivedMessage.set(message);
latch.countDown();
}
});
context.refresh();
latch.await(3, TimeUnit.SECONDS);
assertEquals(0, latch.getCount());
assertNotNull(receivedMessage.get());
assertEquals("test", receivedMessage.get().getPayload());
context.stop();
}
@Test
public void testTransformer() {
TestApplicationContext context = TestUtils.createTestApplicationContext();
@@ -391,17 +339,6 @@ public class MessagingAnnotationPostProcessorTests {
}
@MessageEndpoint
private static class AnnotatedEndpointWithPolledAnnotation {
@ServiceActivator(inputChannel="testChannel")
@Poller(interval=123, initialDelay=456, fixedRate=true, timeUnit=TimeUnit.SECONDS)
public String prependFoo(String s) {
return "foo" + s;
}
}
@MessageEndpoint
private static class ServiceActivatorAnnotatedBean {
@@ -413,17 +350,6 @@ public class MessagingAnnotationPostProcessorTests {
}
@MessageEndpoint
private static class ChannelAdapterAnnotationTestBean {
@ChannelAdapter("testChannel")
@Poller(interval=1000, initialDelay=0, maxMessagesPerPoll=1)
public String test() {
return "test";
}
}
@MessageEndpoint
private static class TransformerAnnotationTestBean {