diff --git a/spring-eai-core/src/main/java/org/springframework/integration/annotation/Publisher.java b/spring-eai-core/src/main/java/org/springframework/integration/annotation/Publisher.java new file mode 100644 index 0000000000..804f3da6e8 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/annotation/Publisher.java @@ -0,0 +1,40 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.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; + +/** + * Indicates that the method's return value should be published to the specified + * channel. The value will only be published if non-null. + * + * @author Mark Fisher + */ +@Target(ElementType.METHOD) +@Retention(RetentionPolicy.RUNTIME) +@Inherited +@Documented +public @interface Publisher { + + String channel(); + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/aop/AnnotationAwareMessagePublishingInterceptor.java b/spring-eai-core/src/main/java/org/springframework/integration/aop/AnnotationAwareMessagePublishingInterceptor.java new file mode 100644 index 0000000000..1e73a80f19 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/aop/AnnotationAwareMessagePublishingInterceptor.java @@ -0,0 +1,73 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aop; + +import java.lang.annotation.Annotation; +import java.lang.reflect.Method; + +import org.aopalliance.intercept.MethodInvocation; + +import org.springframework.aop.support.AopUtils; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.util.Assert; + +/** + * {@link MessagePublishingInterceptor} that resolves the channel from the + * publisher annotation of the invoked method. + * + * @author Mark Fisher + */ +public class AnnotationAwareMessagePublishingInterceptor extends MessagePublishingInterceptor { + + private Class publisherAnnotationType; + + private String channelAttributeName; + + private ChannelResolver channelResolver; + + + public AnnotationAwareMessagePublishingInterceptor(Class publisherAnnotationType, + String channelAttributeName, ChannelResolver channelResolver) { + Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null"); + Assert.notNull(channelAttributeName, "channelAttributeName must not be null"); + Assert.notNull(channelResolver, "channelResolver must not be null"); + this.publisherAnnotationType = publisherAnnotationType; + this.channelAttributeName = channelAttributeName; + this.channelResolver = channelResolver; + } + + + @Override + protected MessageChannel resolveChannel(MethodInvocation invocation) { + Class targetClass = AopUtils.getTargetClass(invocation.getThis()); + Method method = AopUtils.getMostSpecificMethod(invocation.getMethod(), targetClass); + Annotation annotation = AnnotationUtils.getAnnotation(method, this.publisherAnnotationType); + if (annotation != null) { + String channelName = (String) AnnotationUtils.getValue(annotation, this.channelAttributeName); + if (channelName != null) { + MessageChannel channel = this.channelResolver.resolve(channelName); + if (channel != null) { + return channel; + } + } + } + return super.resolveChannel(invocation); + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java b/spring-eai-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java index 03bfda6b88..397c9f4ac6 100644 --- a/spring-eai-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java +++ b/spring-eai-core/src/main/java/org/springframework/integration/aop/MessagePublishingInterceptor.java @@ -19,6 +19,9 @@ package org.springframework.integration.aop; import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.MessageMapper; import org.springframework.integration.message.SimplePayloadMessageMapper; @@ -31,17 +34,15 @@ import org.springframework.util.Assert; */ public class MessagePublishingInterceptor implements MethodInterceptor { + protected Log logger = LogFactory.getLog(getClass()); + private MessageMapper mapper = new SimplePayloadMessageMapper(); - private MessageChannel channel; + private MessageChannel defaultChannel; - /** - * Create a publishing interceptor for the given channel. - */ - public MessagePublishingInterceptor(MessageChannel channel) { - Assert.notNull(channel, "channel must not be null"); - this.channel = channel; + public void setDefaultChannel(MessageChannel defaultChannel) { + this.defaultChannel = defaultChannel; } /** @@ -61,9 +62,25 @@ public class MessagePublishingInterceptor implements MethodInterceptor { public Object invoke(MethodInvocation invocation) throws Throwable { Object retval = invocation.proceed(); if (retval != null) { - this.channel.send(mapper.toMessage(retval)); + MessageChannel channel = this.resolveChannel(invocation); + if (channel == null) { + if (logger.isWarnEnabled()) { + logger.warn("unable to resolve channel for intercepted method '" + + invocation.getMethod().getName() + "'"); + } + } + else { + channel.send(mapper.toMessage(retval)); + } } return retval; } + /** + * Subclasses may override this method to provide custom behavior. + */ + protected MessageChannel resolveChannel(MethodInvocation invocation) { + return this.defaultChannel; + } + } diff --git a/spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java b/spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java new file mode 100644 index 0000000000..0a890a588f --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationAdvisor.java @@ -0,0 +1,69 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aop; + +import java.lang.annotation.Annotation; + +import org.aopalliance.aop.Advice; + +import org.springframework.aop.Pointcut; +import org.springframework.aop.support.AbstractPointcutAdvisor; +import org.springframework.aop.support.annotation.AnnotationMatchingPointcut; +import org.springframework.integration.annotation.Publisher; +import org.springframework.integration.channel.ChannelResolver; +import org.springframework.util.Assert; + +/** + * Advisor whose pointcut matches a method annotation and whose advice will + * publish a message to the channel provided by that annotation. + * + * @author Mark Fisher + * @see Publisher + * @see AnnotationAwareMessagePublishingInterceptor + */ +@SuppressWarnings("serial") +public class PublisherAnnotationAdvisor extends AbstractPointcutAdvisor { + + private AnnotationAwareMessagePublishingInterceptor advice; + + private AnnotationMatchingPointcut pointcut; + + + public PublisherAnnotationAdvisor(ChannelResolver channelResolver) { + this(Publisher.class, "channel", channelResolver); + } + + public PublisherAnnotationAdvisor(Class publisherAnnotationType, String channelNameAttribute, + ChannelResolver channelResolver) { + Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null"); + Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null"); + Assert.notNull(channelResolver, "channelResolver must not be null"); + this.pointcut = AnnotationMatchingPointcut.forMethodAnnotation(publisherAnnotationType); + this.advice = new AnnotationAwareMessagePublishingInterceptor(publisherAnnotationType, channelNameAttribute, + channelResolver); + } + + + public Pointcut getPointcut() { + return this.pointcut; + } + + public Advice getAdvice() { + return this.advice; + } + +} diff --git a/spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationPostProcessor.java b/spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationPostProcessor.java new file mode 100644 index 0000000000..467dccfda1 --- /dev/null +++ b/spring-eai-core/src/main/java/org/springframework/integration/aop/PublisherAnnotationPostProcessor.java @@ -0,0 +1,107 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aop; + +import java.lang.annotation.Annotation; + +import org.springframework.aop.Advisor; +import org.springframework.aop.framework.Advised; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.aop.support.AopUtils; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.beans.factory.config.BeanPostProcessor; +import org.springframework.integration.annotation.Publisher; +import org.springframework.integration.channel.ChannelResolver; +import org.springframework.util.Assert; + +/** + * A {@link BeanPostProcessor} that adds a message publishing interceptor when + * it discovers annotated methods. + * + * @author Mark Fisher + */ +public class PublisherAnnotationPostProcessor implements BeanPostProcessor, BeanClassLoaderAware { + + private Class publisherAnnotationType = Publisher.class; + + private String channelNameAttribute = "channel"; + + private ChannelResolver channelResolver; + + private Advisor advisor; + + private ClassLoader beanClassLoader; + + + public void setBeanClassLoader(ClassLoader beanClassLoader) { + this.beanClassLoader = beanClassLoader; + } + + public void setPublisherAnnotationType(Class publisherAnnotationType) { + Assert.notNull(publisherAnnotationType, "publisherAnnotationType must not be null"); + this.publisherAnnotationType = publisherAnnotationType; + } + + public void setChannelNameAttribute(String channelNameAttribute) { + Assert.notNull(channelNameAttribute, "channelNameAttribute must not be null"); + this.channelNameAttribute = channelNameAttribute; + } + + public void setChannelResolver(ChannelResolver channelResolver) { + Assert.notNull(channelResolver, "channelResolver must not be null"); + this.channelResolver = channelResolver; + } + + private void createAdvisor() { + if (this.channelResolver == null) { + throw new IllegalStateException("channelResolver is required"); + } + this.advisor = new PublisherAnnotationAdvisor(this.publisherAnnotationType, this.channelNameAttribute, + this.channelResolver); + } + + public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException { + return bean; + } + + public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { + Class targetClass = bean instanceof Advised ? + ((Advised) bean).getTargetSource().getTargetClass() : bean.getClass(); + if (targetClass == null) { + return bean; + } + if (advisor == null) { + createAdvisor(); + } + if (AopUtils.canApply(this.advisor, targetClass)) { + if (bean instanceof Advised) { + ((Advised) bean).addAdvisor(this.advisor); + return bean; + } + else { + ProxyFactory pf = new ProxyFactory(bean); + pf.addAdvisor(this.advisor); + return pf.getProxy(this.beanClassLoader); + } + } + else { + return bean; + } + } + +} diff --git a/spring-eai-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java b/spring-eai-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java index 0ee18dae1d..f67e434d1e 100644 --- a/spring-eai-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java +++ b/spring-eai-core/src/test/java/org/springframework/integration/aop/MessagePublishingInterceptorTests.java @@ -33,9 +33,10 @@ import org.springframework.integration.message.Message; public class MessagePublishingInterceptorTests { @Test - public void testNonNullReturnValuePublished() { + public void testNonNullReturnValuePublishedWithDefaultChannel() { MessageChannel channel = new PointToPointChannel(); - MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(channel); + MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); + interceptor.setDefaultChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), interceptor); proxy.messageTest(); Message message = channel.receive(0); @@ -46,7 +47,8 @@ public class MessagePublishingInterceptorTests { @Test public void testNullReturnValueNotPublished() { MessageChannel channel = new PointToPointChannel(); - MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(channel); + MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); + interceptor.setDefaultChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl(null), interceptor); proxy.messageTest(); Message message = channel.receive(0); @@ -56,7 +58,8 @@ public class MessagePublishingInterceptorTests { @Test public void testVoidReturnValueNotPublished() { MessageChannel channel = new PointToPointChannel(); - MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(channel); + MessagePublishingInterceptor interceptor = new MessagePublishingInterceptor(); + interceptor.setDefaultChannel(channel); TestService proxy = (TestService) this.createProxy(new TestServiceImpl(null), interceptor); proxy.voidTest(); Message message = channel.receive(0); diff --git a/spring-eai-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java b/spring-eai-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java new file mode 100644 index 0000000000..4e98c87c70 --- /dev/null +++ b/spring-eai-core/src/test/java/org/springframework/integration/aop/PublisherAnnotationAdvisorTests.java @@ -0,0 +1,109 @@ +/* + * Copyright 2002-2007 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.aop; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import org.junit.Test; + +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.integration.annotation.Publisher; +import org.springframework.integration.channel.ChannelResolver; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.PointToPointChannel; +import org.springframework.integration.message.Message; + +/** + * @author Mark Fisher + */ +public class PublisherAnnotationAdvisorTests { + + @Test + public void testPublisherAnnotation() { + final MessageChannel channel = new PointToPointChannel(); + ChannelResolver channelResolver = new ChannelResolver() { + public MessageChannel resolve(String channelName) { + if (channelName.equals("testChannel")) { + return channel; + } + return null; + } + }; + PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelResolver); + TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); + proxy.publisherTest(); + Message message = channel.receive(0); + assertNotNull(message); + assertEquals("hello world", message.getPayload()); + } + + @Test + public void testNoPublisherAnnotation() { + final MessageChannel channel = new PointToPointChannel(); + ChannelResolver channelResolver = new ChannelResolver() { + public MessageChannel resolve(String channelName) { + if (channelName.equals("testChannel")) { + return channel; + } + return null; + } + }; + PublisherAnnotationAdvisor advisor = new PublisherAnnotationAdvisor(channelResolver); + TestService proxy = (TestService) this.createProxy(new TestServiceImpl("hello world"), advisor); + proxy.noPublisherTest(); + Message message = channel.receive(0); + assertNull(message); + } + + + private Object createProxy(Object target, PublisherAnnotationAdvisor advisor) { + ProxyFactory factory = new ProxyFactory(target); + factory.addAdvisor(advisor); + return factory.getProxy(); + } + + + private static interface TestService { + + String publisherTest(); + + String noPublisherTest(); + } + + + private static class TestServiceImpl implements TestService { + + private String message; + + public TestServiceImpl(String message) { + this.message = message; + } + + @Publisher(channel="testChannel") + public String publisherTest() { + return this.message; + } + + public String noPublisherTest() { + return this.message; + } + + } + +}