diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/BridgeFrom.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/BridgeFrom.java new file mode 100644 index 0000000000..2dd3517121 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/BridgeFrom.java @@ -0,0 +1,68 @@ +/* + * Copyright 2014 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.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Messaging Annotation to mark a {@link org.springframework.context.annotation.Bean} + * method for a {@link org.springframework.messaging.MessageChannel} to produce a + * {@link org.springframework.integration.handler.BridgeHandler} and Consumer Endpoint. + *

+ * The {@code inputChannel} for the {@link org.springframework.integration.endpoint.AbstractEndpoint} + * is the {@link #value()} of this annotation and determines the type of endpoint - + * {@link org.springframework.integration.endpoint.EventDrivenConsumer} or + * {@link org.springframework.integration.endpoint.PollingConsumer}. + *

+ * The {@link org.springframework.messaging.MessageChannel} {@link org.springframework.context.annotation.Bean} + * is used as the {@code outputChannel} of the {@link org.springframework.integration.handler.BridgeHandler}. + * + * @author Artem Bilan + * @since 4.0 + */ +@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +public @interface BridgeFrom { + /** + * @return the inbound channel name to receive message for the + * {@link org.springframework.integration.handler.BridgeHandler} + */ + String value(); + + /* + {@code SmartLifecycle} options. + Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + */ + String autoStartup() default "true"; + + String phase() default "0"; + + /** + * @return the {@link Poller} options for a polled endpoint + * ({@link org.springframework.integration.scheduling.PollerMetadata}). + * This attribute is an {@code array} just to allow an empty default (no poller). + * Only one {@link Poller} element is allowed. + */ + Poller[] poller() default {}; +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/BridgeTo.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/BridgeTo.java new file mode 100644 index 0000000000..04b1d3cb8d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/BridgeTo.java @@ -0,0 +1,75 @@ +/* + * Copyright 2014 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.annotation; + +import java.lang.annotation.Documented; +import java.lang.annotation.ElementType; +import java.lang.annotation.Inherited; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Messaging Annotation to mark a {@link org.springframework.context.annotation.Bean} + * method for a {@link org.springframework.messaging.MessageChannel} to produce a + * {@link org.springframework.integration.handler.BridgeHandler} and Consumer Endpoint. + *

+ * The {@link org.springframework.messaging.MessageChannel} {@link org.springframework.context.annotation.Bean} + * marked with this annotation is used as the {@code inputChannel} for the + * {@link org.springframework.integration.endpoint.AbstractEndpoint} + * and determines the type of endpoint - + * {@link org.springframework.integration.endpoint.EventDrivenConsumer} or + * {@link org.springframework.integration.endpoint.PollingConsumer}. + *

+ * The {@link #value()} of this annotation is the {@code outputChannel} for the + * {@link org.springframework.integration.handler.BridgeHandler}. + * If it isn't present, the {@link org.springframework.integration.handler.BridgeHandler} + * sends the message to the {@code reply-channel} in its message headers, if present. + * If no output channel is provided and no reply-channel exists, an exception is thrown. + * + * @author Artem Bilan + * @since 4.0 + */ +@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE}) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +public @interface BridgeTo { + /** + * @return the outbound channel name to send the message to + * {@link org.springframework.integration.handler.BridgeHandler}. + * Optional: when omitted the message is sent to the {@code reply-channel} + * in its headers (if present - an exception is thrown otherwise). + */ + String value() default ""; + + /* + {@code SmartLifecycle} options. + Can be specified as 'property placeholder', e.g. {@code ${foo.autoStartup}}. + */ + String autoStartup() default "true"; + + String phase() default "0"; + + /** + * @return the {@link org.springframework.integration.annotation.Poller} options for a polled endpoint + * ({@link org.springframework.integration.scheduling.PollerMetadata}). + * This attribute is an {@code array} just to allow an empty default (no poller). + * Only one {@link org.springframework.integration.annotation.Poller} element is allowed. + */ + Poller[] poller() default {}; +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java index 9da21e50ab..7cc30d06fb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/AbstractMethodAnnotationPostProcessor.java @@ -93,8 +93,8 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations) { - MessageHandler handler = this.createHandler(bean, method, annotations); - this.setAdviceChainIfPresent(beanName, annotations, handler); + MessageHandler handler = createHandler(bean, method, annotations); + setAdviceChainIfPresent(beanName, annotations, handler); if (handler instanceof Orderable) { Order orderAnnotation = AnnotationUtils.findAnnotation(method, Order.class); if (orderAnnotation != null) { @@ -102,18 +102,28 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations) { + String inputChannel = MessagingAnnotationUtils.resolveAttribute(annotations, getInputChannelAttribute(), + String.class); + return StringUtils.hasText(inputChannel); + } + + protected String getInputChannelAttribute() { + return INPUT_CHANNEL_ATTRIBUTE; + } protected final void setAdviceChainIfPresent(String beanName, List annotations, MessageHandler handler) { String[] adviceChainNames = MessagingAnnotationUtils.resolveAttribute(annotations, ADVICE_CHAIN_ATTRIBUTE, @@ -152,9 +162,9 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations) { + protected AbstractEndpoint createEndpoint(MessageHandler handler, Method method, List annotations) { AbstractEndpoint endpoint = null; - String inputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, INPUT_CHANNEL_ATTRIBUTE, + String inputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, getInputChannelAttribute(), String.class); if (StringUtils.hasText(inputChannelName)) { MessageChannel inputChannel; @@ -171,17 +181,24 @@ public abstract class AbstractMethodAnnotationPostProcessor annotations) { + AbstractEndpoint endpoint; + if (inputChannel instanceof PollableChannel) { + PollingConsumer pollingConsumer = new PollingConsumer((PollableChannel) inputChannel, handler); + this.configurePollingEndpoint(pollingConsumer, annotations); + endpoint = pollingConsumer; + } + else { + Poller[] pollers = MessagingAnnotationUtils.resolveAttribute(annotations, "poller", Poller[].class); + Assert.state(ObjectUtils.isEmpty(pollers), "A '@Poller' should not be specified for Annotation-based " + + "endpoint, since '" + inputChannel + "' is a SubscribableChannel (not pollable)."); + endpoint = new EventDrivenConsumer((SubscribableChannel) inputChannel, handler); } return endpoint; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/BridgeFromAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/BridgeFromAnnotationPostProcessor.java new file mode 100644 index 0000000000..dc5d346995 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/BridgeFromAnnotationPostProcessor.java @@ -0,0 +1,89 @@ +/* + * Copyright 2014 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.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.List; + +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.core.annotation.AnnotatedElementUtils; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.core.env.Environment; +import org.springframework.integration.annotation.BridgeFrom; +import org.springframework.integration.annotation.BridgeTo; +import org.springframework.integration.handler.BridgeHandler; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; + +/** + * Post-processor for the {@link BridgeFrom @BridgeFrom} annotation. + * + * @author Artem Bilan + * @since 4.0 + */ +public class BridgeFromAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { + + public BridgeFromAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { + super(beanFactory, environment); + } + + @Override + public boolean shouldCreateEndpoint(Method method, List annotations) { + boolean isBean = AnnotatedElementUtils.isAnnotated(method, Bean.class.getName()); + Assert.isTrue(isBean, "'@BridgeFrom' is eligible only for '@Bean' methods"); + + boolean isMessageChannelBean = MessageChannel.class.isAssignableFrom(method.getReturnType()); + Assert.isTrue(isMessageChannelBean, "'@BridgeFrom' is eligible only for 'MessageChannel' '@Bean' methods"); + + String channel = MessagingAnnotationUtils.resolveAttribute(annotations, "value", String.class); + Assert.isTrue(StringUtils.hasText(channel), "'@BridgeFrom.value()' (inputChannelName) must not be empty"); + + boolean hasBridgeTo = AnnotatedElementUtils.isAnnotated(method, BridgeTo.class.getName()); + + Assert.isTrue(!hasBridgeTo, "'@BridgeFrom' and '@BridgeTo' are mutually exclusive 'MessageChannel' " + + "'@Bean' method annotations"); + + + return true; + } + + @Override + protected String getInputChannelAttribute() { + return AnnotationUtils.VALUE; + } + + @Override + protected MessageHandler createHandler(Object bean, Method method, List annotations) { + BridgeHandler handler = new BridgeHandler(); + String outputChannelName = null; + String[] names = AnnotationUtils.getAnnotation(method, Bean.class).name(); + if (!ObjectUtils.isEmpty(names)) { + outputChannelName = names[0]; + } + if (!StringUtils.hasText(outputChannelName)) { + outputChannelName = method.getName(); + } + handler.setOutputChannelName(outputChannelName); + return handler; + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/BridgeToAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/BridgeToAnnotationPostProcessor.java new file mode 100644 index 0000000000..2b44046377 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/BridgeToAnnotationPostProcessor.java @@ -0,0 +1,90 @@ +/* + * Copyright 2014 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.annotation; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; +import java.util.List; + +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.context.annotation.Bean; +import org.springframework.core.annotation.AnnotatedElementUtils; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.core.env.Environment; +import org.springframework.integration.annotation.BridgeFrom; +import org.springframework.integration.annotation.BridgeTo; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.integration.handler.BridgeHandler; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; +import org.springframework.util.StringUtils; + +/** + * Post-processor for the {@link BridgeTo @BridgeTo} annotation. + * + * @author Artem Bilan + * @since 4.0 + */ +public class BridgeToAnnotationPostProcessor extends AbstractMethodAnnotationPostProcessor { + + public BridgeToAnnotationPostProcessor(ListableBeanFactory beanFactory, Environment environment) { + super(beanFactory, environment); + } + + @Override + public boolean shouldCreateEndpoint(Method method, List annotations) { + boolean isBean = AnnotatedElementUtils.isAnnotated(method, Bean.class.getName()); + Assert.isTrue(isBean, "'@BridgeTo' is eligible only for '@Bean' methods"); + + boolean isMessageChannelBean = MessageChannel.class.isAssignableFrom(method.getReturnType()); + Assert.isTrue(isMessageChannelBean, "'@BridgeTo' is eligible only for 'MessageChannel' '@Bean' methods"); + + boolean hasBridgeFrom = AnnotatedElementUtils.isAnnotated(method, BridgeFrom.class.getName()); + + Assert.isTrue(!hasBridgeFrom, "'@BridgeFrom' and '@BridgeTo' are mutually exclusive 'MessageChannel' " + + "'@Bean' method annotations"); + + return true; + } + + @Override + protected MessageHandler createHandler(Object bean, Method method, List annotations) { + BridgeHandler handler = new BridgeHandler(); + String outputChannelName = MessagingAnnotationUtils.resolveAttribute(annotations, "value", String.class); + handler.setOutputChannelName(outputChannelName); + return handler; + } + + @Override + protected AbstractEndpoint createEndpoint(MessageHandler handler, Method method, List annotations) { + String inputChannelName = null; + String[] names = AnnotationUtils.getAnnotation(method, Bean.class).name(); + if (!ObjectUtils.isEmpty(names)) { + inputChannelName = names[0]; + } + if (!StringUtils.hasText(inputChannelName)) { + inputChannelName = method.getName(); + } + + MessageChannel inputChannel = this.beanFactory.getBean(inputChannelName, MessageChannel.class); + + return doCreateEndpoint(handler, inputChannel, annotations); + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java index 007591116e..f010ae3e8e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java @@ -46,6 +46,11 @@ public class InboundChannelAdapterAnnotationPostProcessor extends super(beanFactory, environment); } + @Override + protected String getInputChannelAttribute() { + return AnnotationUtils.VALUE; + } + @Override public Object postProcess(Object bean, String beanName, Method method, List annotations) { Assert.isTrue(!Void.class.isAssignableFrom(method.getReturnType()), "The method '" + method diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java index 786a31257a..5f1fabd6db 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MessagingAnnotationPostProcessor.java @@ -45,6 +45,8 @@ import org.springframework.core.annotation.AnnotatedElementUtils; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.core.env.Environment; import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.BridgeFrom; +import org.springframework.integration.annotation.BridgeTo; import org.springframework.integration.annotation.Filter; import org.springframework.integration.annotation.InboundChannelAdapter; import org.springframework.integration.annotation.Router; @@ -108,6 +110,8 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean postProcessors.put(Splitter.class, new SplitterAnnotationPostProcessor(this.beanFactory, this.environment)); postProcessors.put(Aggregator.class, new AggregatorAnnotationPostProcessor(this.beanFactory, this.environment)); postProcessors.put(InboundChannelAdapter.class, new InboundChannelAdapterAnnotationPostProcessor(this.beanFactory, this.environment)); + postProcessors.put(BridgeFrom.class, new BridgeFromAnnotationPostProcessor(this.beanFactory, this.environment)); + postProcessors.put(BridgeTo.class, new BridgeToAnnotationPostProcessor(this.beanFactory, this.environment)); } @Override @@ -142,7 +146,7 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean Class annotationType = entry.getKey(); List annotations = entry.getValue(); MethodAnnotationPostProcessor postProcessor = postProcessors.get(annotationType); - if (postProcessor != null && shouldCreateEndpoint(annotations)) { + if (postProcessor != null && postProcessor.shouldCreateEndpoint(method, annotations)) { Object result = postProcessor.postProcess(bean, beanName, method, annotations); if (result != null && result instanceof AbstractEndpoint) { AbstractEndpoint endpoint = (AbstractEndpoint) result; @@ -231,22 +235,6 @@ public class MessagingAnnotationPostProcessor implements BeanPostProcessor, Bean return false; } - private boolean shouldCreateEndpoint(List annotations) { - for (Annotation annotation : annotations) { - Object inputChannel = AnnotationUtils.getValue(annotation, "inputChannel"); - if (inputChannel == null && - (annotation instanceof InboundChannelAdapter || - AnnotationUtils.findAnnotation(annotation.annotationType(), InboundChannelAdapter.class) != null)) { - inputChannel = AnnotationUtils.getValue(annotation); - } - if (inputChannel != null && inputChannel instanceof String - && StringUtils.hasText((String) inputChannel)) { - return true; - } - } - return false; - } - private Class getBeanClass(Object bean) { Class targetClass = AopUtils.getTargetClass(bean); return (targetClass != null) ? targetClass : bean.getClass(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MethodAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MethodAnnotationPostProcessor.java index 82aed8d58c..72280892a4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MethodAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/MethodAnnotationPostProcessor.java @@ -25,9 +25,21 @@ import java.util.List; * * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ public interface MethodAnnotationPostProcessor { Object postProcess(Object bean, String beanName, Method method, List annotations); + /** + * Determine if the provided {@code method} and its {@code annotations} are eligible + * to create an {@link org.springframework.integration.endpoint.AbstractEndpoint}. + * @param method the method to check if it is eligible to create an Endpoint + * @param annotations the List of annotations to process + * @return the {@code boolean} flag to determine whether or not to create an + * {@link org.springframework.integration.endpoint.AbstractEndpoint} + * @since 4.0 + */ + boolean shouldCreateEndpoint(Method method, List annotations); + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java index cb3e2c2a00..20bbe80280 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java @@ -16,9 +16,16 @@ package org.springframework.integration.configuration; -import static org.hamcrest.Matchers.*; -import static org.junit.Assert.*; -import static org.mockito.Mockito.*; +import static org.hamcrest.Matchers.containsString; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; +import static org.mockito.Mockito.mock; import java.lang.annotation.ElementType; import java.lang.annotation.Retention; @@ -39,6 +46,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.AbstractFactoryBean; import org.springframework.context.ApplicationContext; +import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; @@ -51,6 +59,8 @@ import org.springframework.core.convert.converter.Converter; import org.springframework.core.serializer.support.SerializingConverter; import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler; import org.springframework.integration.annotation.Aggregator; +import org.springframework.integration.annotation.BridgeFrom; +import org.springframework.integration.annotation.BridgeTo; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.annotation.GatewayHeader; import org.springframework.integration.annotation.InboundChannelAdapter; @@ -83,6 +93,7 @@ import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.messaging.support.ChannelInterceptorAdapter; @@ -175,11 +186,42 @@ public class EnableIntegrationTests { @Autowired private PollableChannel counterChannel; + @Autowired + private PollableChannel messageChannel; + @Autowired private PollableChannel fooChannel; @Autowired - private PollableChannel messageChannel; + private MessageChannel bridgeInput; + + @Autowired + private PollableChannel bridgeOutput; + + @Autowired + private MessageChannel pollableBridgeInput; + + @Autowired + private PollableChannel pollableBridgeOutput; + + @Autowired + private MessageChannel metaBridgeInput; + + @Autowired + private PollableChannel metaBridgeOutput; + + @Autowired + private MessageChannel bridgeToInput; + + @Autowired + private PollableChannel bridgeToOutput; + + @Autowired + private PollableChannel pollableBridgeToInput; + + @Autowired + private MessageChannel myBridgeToInput; + @Test public void testAnnotatedServiceActivator() { @@ -415,6 +457,71 @@ public class EnableIntegrationTests { assertTrue(TestUtils.getPropertyValue(consumer, "handler.sendPartialResultOnExpiry", Boolean.class)); } + @Test + public void testBridgeAnnotations() { + GenericMessage testMessage = new GenericMessage("foo"); + this.bridgeInput.send(testMessage); + Message receive = this.bridgeOutput.receive(2000); + assertNotNull(receive); + assertSame(receive, testMessage); + assertNull(this.bridgeOutput.receive(10)); + + this.pollableBridgeInput.send(testMessage); + receive = this.pollableBridgeOutput.receive(2000); + assertNotNull(receive); + assertSame(receive, testMessage); + assertNull(this.pollableBridgeOutput.receive(10)); + + try { + this.metaBridgeInput.send(testMessage); + fail("MessageDeliveryException expected"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(MessageDeliveryException.class)); + assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers")); + } + + this.context.getBean("enableIntegrationTests.ContextConfiguration.metaBridgeOutput.bridgeFrom", + Lifecycle.class).start(); + + this.metaBridgeInput.send(testMessage); + receive = this.metaBridgeOutput.receive(2000); + assertNotNull(receive); + assertSame(receive, testMessage); + assertNull(this.metaBridgeOutput.receive(10)); + + this.bridgeToInput.send(testMessage); + receive = this.bridgeToOutput.receive(2000); + assertNotNull(receive); + assertSame(receive, testMessage); + assertNull(this.bridgeToOutput.receive(10)); + + PollableChannel replyChannel = new QueueChannel(); + Message bridgeMessage = MessageBuilder.fromMessage(testMessage).setReplyChannel(replyChannel).build(); + this.pollableBridgeToInput.send(bridgeMessage); + receive = replyChannel.receive(2000); + assertNotNull(receive); + assertSame(receive, bridgeMessage); + assertNull(replyChannel.receive(10)); + + try { + this.myBridgeToInput.send(testMessage); + fail("MessageDeliveryException expected"); + } + catch (Exception e) { + assertThat(e, Matchers.instanceOf(MessageDeliveryException.class)); + assertThat(e.getMessage(), Matchers.containsString("Dispatcher has no subscribers")); + } + + this.context.getBean("enableIntegrationTests.ContextConfiguration.myBridgeToInput.bridgeTo", + Lifecycle.class).start(); + + this.myBridgeToInput.send(bridgeMessage); + receive = replyChannel.receive(2000); + assertNotNull(receive); + assertSame(receive, bridgeMessage); + assertNull(replyChannel.receive(10)); + } @Configuration @ComponentScan @@ -513,6 +620,68 @@ public class EnableIntegrationTests { }; } + @Bean + @BridgeFrom("bridgeInput") + public QueueChannel bridgeOutput() { + return new QueueChannel(); + } + + @Bean + public PollableChannel pollableBridgeInput() { + return new QueueChannel(); + } + + + @Bean + @BridgeFrom(value = "pollableBridgeInput", poller = @Poller(fixedDelay = "1000")) + public QueueChannel pollableBridgeOutput() { + return new QueueChannel(); + } + + @Bean + @MyBridgeFrom + public QueueChannel metaBridgeOutput() { + return new QueueChannel(); + } + + @Bean + public QueueChannel bridgeToOutput() { + return new QueueChannel(); + } + + @Bean + @BridgeTo("bridgeToOutput") + public MessageChannel bridgeToInput() { + return new DirectChannel(); + } + + @Bean + @BridgeTo(poller = @Poller(fixedDelay = "500")) + public QueueChannel pollableBridgeToInput() { + return new QueueChannel(); + } + + @Bean + @MyBridgeTo + public MessageChannel myBridgeToInput() { + return new DirectChannel(); + } + + // Error because @Bridge* annotations are only for MessageChannel beans. + /*@Bean + @BridgeTo + public String invalidBridgeAnnotation() { + return "invalidBridgeAnnotation"; + }*/ + + // Error because @Bridge* annotations are mutually exclusive. + /*@Bean + @BridgeTo + @BridgeFrom("foo") + public MessageChannel invalidBridgeAnnotation2() { + return new DirectChannel(); + }*/ + // beans for metaAnnotation tests @Bean @@ -819,6 +988,10 @@ public class EnableIntegrationTests { return 42; } + // Error because @Bridge* annotations are only for @Bean methods. + /*@BridgeFrom("") + public void invalidBridgeAnnotationMethod(Object payload) {}*/ + } @TestMessagingGateway @@ -1041,6 +1214,20 @@ public class EnableIntegrationTests { } + @Target(ElementType.METHOD) + @Retention(RetentionPolicy.RUNTIME) + @BridgeFrom(value = "metaBridgeInput", autoStartup = "false") + public static @interface MyBridgeFrom { + + String value() default ""; + } + + @Target(ElementType.METHOD) + @Retention(RetentionPolicy.RUNTIME) + @BridgeTo(autoStartup = "false") + public static @interface MyBridgeTo { + } + // Error because the annotation is on a class; it must be on an interface // @MessagingGateway(defaultRequestChannel = "gatewayChannel", defaultHeaders = @GatewayHeader(name = "foo", value = "FOO")) // public static class TestGateway2 { }