diff --git a/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodSetupOrchestratorTests.java b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodSetupOrchestratorTests.java new file mode 100644 index 000000000..9e9198282 --- /dev/null +++ b/spring-cloud-stream-integration-tests/src/test/java/org/springframework/cloud/stream/config/StreamListenerMethodSetupOrchestratorTests.java @@ -0,0 +1,147 @@ +/* + * Copyright 2018 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.cloud.stream.config; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.ArrayList; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Set; + +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.test.mock.mockito.SpyBean; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor; +import org.springframework.cloud.stream.binding.StreamListenerSetupMethodOrchestrator; +import org.springframework.cloud.stream.messaging.Sink; +import org.springframework.cloud.stream.messaging.Source; +import org.springframework.context.annotation.Bean; +import org.springframework.core.annotation.AnnotatedElementUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.handler.annotation.SendTo; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.util.ReflectionUtils; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +/** + * @author Soby Chacko + */ +@RunWith(SpringJUnit4ClassRunner.class) +@SpringBootTest +public class StreamListenerMethodSetupOrchestratorTests { + + @SpyBean + CustomOrchestrator customOrchestrator; + + @SpyBean + MultipleStreamListenerProcessor multipleStreamListenerProcessor; + + @SpyBean + StreamListenerAnnotationBeanPostProcessor streamListenerAnnotationBeanPostProcessor; + + @Test + @SuppressWarnings("unchecked") + public void testCustomStreamListenerOrchestratorAndDefaultTogetherInSameContext() throws Exception { + + //Two StreamListener methods, so 2 invocations + verify(customOrchestrator, times(2)).supports(any()); + + Method method = multipleStreamListenerProcessor.getClass().getMethod("handleMessage"); + StreamListener streamListener = AnnotatedElementUtils.findMergedAnnotation(method, StreamListener.class); + //verify that the invocation happened on the custom Orchestrator + verify(customOrchestrator).orchestrateStreamListenerSetupMethod(streamListener, method, multipleStreamListenerProcessor); + + Method method1 = multipleStreamListenerProcessor.getClass().getMethod("produceString"); + StreamListener streamListener1 = AnnotatedElementUtils.findMergedAnnotation(method, StreamListener.class); + + //Verify that the invocation did not happen on the custom orchestrator + verify(customOrchestrator, never()).orchestrateStreamListenerSetupMethod(streamListener1, method1, multipleStreamListenerProcessor); + + Field field = ReflectionUtils.findField(streamListenerAnnotationBeanPostProcessor.getClass(), "streamListenerSetupMethodOrchestrators"); + ReflectionUtils.makeAccessible(field); + + Set field1 = + (LinkedHashSet)ReflectionUtils.getField(field, streamListenerAnnotationBeanPostProcessor); + List list = new ArrayList<>(field1); + + //Ensure that the custom orchestrator did not support this request + assertThat(list.get(0).supports(method1)).isEqualTo(false); + //Ensure that we are using the default Orchestrator in StreamListenerAnnoatationBeanPostProcessor + assertThat(list.get(1).supports(method1)).isEqualTo(true); + } + + public interface SomeProcessor { + + @Input(Sink.INPUT) + SubscribableChannel channel1(); + + @Input("foobar") + SubscribableChannel channel2(); + + @Output(Source.OUTPUT) + MessageChannel channel3(); + + } + + @EnableBinding(SomeProcessor.class) + @EnableAutoConfiguration + public static class MultipleStreamListenerProcessor { + + @StreamListener(Sink.INPUT) + public void handleMessage() { + } + + @StreamListener("foobar") + @SendTo("output") + public String produceString(){ + return "foobar"; + } + + @Bean + public CustomOrchestrator myOrchestrator() { + return new CustomOrchestrator(); + } + + } + + static class CustomOrchestrator implements StreamListenerSetupMethodOrchestrator { + + @Override + public boolean supports(Method method) { + return method.getReturnType() != String.class; + } + + @Override + public void orchestrateStreamListenerSetupMethod(StreamListener streamListener, Method method, Object bean) { + //stub method + } + } +} diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java index 6839a4337..4cd6085f5 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerAnnotationBeanPostProcessor.java @@ -20,12 +20,12 @@ import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collection; import java.util.HashSet; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; -import reactor.core.publisher.Flux; - import org.springframework.aop.framework.Advised; import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; @@ -76,12 +76,6 @@ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProces private final MultiValueMap mappedListenerMethods = new LinkedMultiValueMap<>(); // == dependencies that are injected in 'afterSingletonsInstantiated' to avoid early initialization - @SuppressWarnings("rawtypes") - private Collection streamListenerParameterAdapters; - - @SuppressWarnings("rawtypes") - private Collection streamListenerResultAdapters; - private DestinationResolver binderAwareChannelResolver; private MessageHandlerMethodFactory messageHandlerMethodFactory; @@ -93,12 +87,12 @@ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProces private ConfigurableApplicationContext applicationContext; - private EvaluationContext evaluationContext; - private BeanExpressionResolver resolver; private BeanExpressionContext expressionContext; + private Set streamListenerSetupMethodOrchestrators = new LinkedHashSet<>(); + @Override public final void setApplicationContext(ApplicationContext applicationContext) throws BeansException { this.applicationContext = (ConfigurableApplicationContext) applicationContext; @@ -109,7 +103,7 @@ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProces @Override public final void afterSingletonsInstantiated() { this.injectAndPostProcessDependencies(); - this.evaluationContext = IntegrationContextUtils.getEvaluationContext(this.applicationContext.getBeanFactory()); + EvaluationContext evaluationContext = IntegrationContextUtils.getEvaluationContext(this.applicationContext.getBeanFactory()); for (Map.Entry> mappedBindingEntry : mappedListenerMethods .entrySet()) { ArrayList handlers = new ArrayList<>(); @@ -147,7 +141,7 @@ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProces AbstractReplyProducingMessageHandler handler; if (handlers.size() > 1 || handlers.get(0).getCondition() != null) { - handler = new DispatchingStreamListenerMessageHandler(handlers, this.evaluationContext); + handler = new DispatchingStreamListenerMessageHandler(handlers, evaluationContext); } else { handler = handlers.get(0).getStreamListenerMessageHandler(); @@ -177,26 +171,6 @@ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProces return bean; } - protected final void registerHandlerMethodOnListenedChannel(Method method, StreamListener streamListener, Object bean) { - Assert.hasText(streamListener.value(), "The binding name cannot be null"); - if (!StringUtils.hasText(streamListener.value())) { - throw new BeanInitializationException("A bound component name must be specified"); - } - final String defaultOutputChannel = StreamListenerMethodUtils.getOutboundBindingTargetName(method); - if (Void.TYPE.equals(method.getReturnType())) { - Assert.isTrue(StringUtils.isEmpty(defaultOutputChannel), - "An output channel cannot be specified for a method that does not return a value"); - } - else { - Assert.isTrue(!StringUtils.isEmpty(defaultOutputChannel), - "An output channel must be specified for a method that can return a value"); - } - StreamListenerMethodUtils.validateStreamListenerMessageHandler(method); - mappedListenerMethods.add(streamListener.value(), - new StreamListenerHandlerMethodMapping(bean, method, streamListener.condition(), defaultOutputChannel, - streamListener.copyHeaders())); - } - /** * Extension point, allowing subclasses to customize the {@link StreamListener} * annotation detected by the postprocessor. @@ -209,145 +183,16 @@ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProces return originalAnnotation; } - private boolean checkDeclarativeMethod(Method method, String methodAnnotatedInboundName, String methodAnnotatedOutboundName) { - int methodArgumentsLength = method.getParameterTypes().length; - for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) { - MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); - if (methodParameter.hasParameterAnnotation(Input.class)) { - String inboundName = (String) AnnotationUtils - .getValue(methodParameter.getParameterAnnotation(Input.class)); - Assert.isTrue(StringUtils.hasText(inboundName), StreamListenerErrorMessages.INVALID_INBOUND_NAME); - Assert.isTrue(isDeclarativeMethodParameter(inboundName, methodParameter), - StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS); - return true; - } - else if (methodParameter.hasParameterAnnotation(Output.class)) { - String outboundName = (String) AnnotationUtils - .getValue(methodParameter.getParameterAnnotation(Output.class)); - Assert.isTrue(StringUtils.hasText(outboundName), StreamListenerErrorMessages.INVALID_OUTBOUND_NAME); - Assert.isTrue(isDeclarativeMethodParameter(outboundName, methodParameter), - StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS); - return true; - } - else if (StringUtils.hasText(methodAnnotatedOutboundName)) { - return isDeclarativeMethodParameter(methodAnnotatedOutboundName, methodParameter); - } - else if (StringUtils.hasText(methodAnnotatedInboundName)) { - return isDeclarativeMethodParameter(methodAnnotatedInboundName, methodParameter); - } - } - return false; - } - - /** - * Determines if method parameters signify an imperative or declarative listener definition. - *
- * Imperative - where handler method is invoked on each message by the handler infrastructure provided - * by the framework - *
- * Declarative - where handler is provided by the method itself. - *
- * Declarative method parameter could either be {@link MessageChannel} or any other Object for which - * there is a {@link StreamListenerParameterAdapter} (i.e., {@link Flux}). Declarative method is invoked only - * once during initialization phase. - */ - @SuppressWarnings("unchecked") - private boolean isDeclarativeMethodParameter(String targetBeanName, MethodParameter methodParameter) { - boolean declarative = false; - if (!methodParameter.getParameterType().isAssignableFrom(Object.class) && this.applicationContext.containsBean(targetBeanName)) { - declarative = MessageChannel.class.isAssignableFrom(methodParameter.getParameterType()); - if (!declarative) { - Class targetBeanClass = this.applicationContext.getType(targetBeanName); - declarative = this.streamListenerParameterAdapters.stream() - .filter(slpa -> slpa.supports(targetBeanClass, methodParameter)).findFirst().isPresent(); - } - } - return declarative; - } - - @SuppressWarnings({ "rawtypes", "unchecked" }) - private void invokeSetupMethodOnListenedChannel(Method method, Object bean, String inboundName, String outboundName) { - Object[] arguments = new Object[method.getParameterTypes().length]; - for (int parameterIndex = 0; parameterIndex < arguments.length; parameterIndex++) { - MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); - Class parameterType = methodParameter.getParameterType(); - Object targetReferenceValue = null; - if (methodParameter.hasParameterAnnotation(Input.class)) { - targetReferenceValue = AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Input.class)); - } - else if (methodParameter.hasParameterAnnotation(Output.class)) { - targetReferenceValue = AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Output.class)); - } - else if (arguments.length == 1 && StringUtils.hasText(inboundName)) { - targetReferenceValue = inboundName; - } - if (targetReferenceValue != null) { - Assert.isInstanceOf(String.class, targetReferenceValue, "Annotation value must be a String"); - Object targetBean = this.applicationContext.getBean((String) targetReferenceValue); - // Iterate existing parameter adapters first - for (StreamListenerParameterAdapter streamListenerParameterAdapter : this.streamListenerParameterAdapters) { - if (streamListenerParameterAdapter.supports(targetBean.getClass(), methodParameter)) { - arguments[parameterIndex] = streamListenerParameterAdapter.adapt(targetBean, methodParameter); - break; - } - } - if (arguments[parameterIndex] == null && parameterType.isAssignableFrom(targetBean.getClass())) { - arguments[parameterIndex] = targetBean; - } - Assert.notNull(arguments[parameterIndex], "Cannot convert argument " + parameterIndex + " of " + method - + "from " + targetBean.getClass() + " to " + parameterType); - } - else { - throw new IllegalStateException(StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS); - } - } - try { - if (Void.TYPE.equals(method.getReturnType())) { - method.invoke(bean, arguments); - } - else { - Object result = method.invoke(bean, arguments); - if (!StringUtils.hasText(outboundName)) { - for (int parameterIndex = 0; parameterIndex < method.getParameterTypes().length; parameterIndex++) { - MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); - if (methodParameter.hasParameterAnnotation(Output.class)) { - outboundName = methodParameter.getParameterAnnotation(Output.class).value(); - } - } - } - Object targetBean = this.applicationContext.getBean(outboundName); - for (StreamListenerResultAdapter streamListenerResultAdapter : this.streamListenerResultAdapters) { - if (streamListenerResultAdapter.supports(result.getClass(), targetBean.getClass())) { - streamListenerResultAdapter.adapt(result, targetBean); - break; - } - } - } - } - catch (Exception e) { - throw new BeanInitializationException("Cannot setup StreamListener for " + method, e); - } - } - private void doPostProcess(StreamListener streamListener, Method method, Object bean) { streamListener = postProcessAnnotation(streamListener, method); - - String methodAnnotatedInboundName = streamListener.value(); - String methodAnnotatedOutboundName = StreamListenerMethodUtils.getOutboundBindingTargetName(method); - - int inputAnnotationCount = StreamListenerMethodUtils.inputAnnotationCount(method); - int outputAnnotationCount = StreamListenerMethodUtils.outputAnnotationCount(method); - boolean isDeclarative = checkDeclarativeMethod(method, methodAnnotatedInboundName, methodAnnotatedOutboundName); - StreamListenerMethodUtils.validateStreamListenerMethod(method, - inputAnnotationCount, outputAnnotationCount, - methodAnnotatedInboundName, methodAnnotatedOutboundName, - isDeclarative, streamListener.condition()); - if (isDeclarative) { - invokeSetupMethodOnListenedChannel(method, bean, methodAnnotatedInboundName, methodAnnotatedOutboundName); - } - else { - registerHandlerMethodOnListenedChannel(method, streamListener, bean); - } + Optional streamListenerSetupMethodOrchestratorAvailable = + streamListenerSetupMethodOrchestrators.stream() + .filter(t -> t.supports(method)) + .findFirst(); + Assert.isTrue(streamListenerSetupMethodOrchestratorAvailable.isPresent(), + "A matching StreamListenerSetupMethodOrchestrator must be present"); + StreamListenerSetupMethodOrchestrator streamListenerSetupMethodOrchestrator = streamListenerSetupMethodOrchestratorAvailable.get(); + streamListenerSetupMethodOrchestrator.orchestrateStreamListenerSetupMethod(streamListener, method, bean); } private Method checkProxy(Method methodArg, Object bean) { @@ -420,15 +265,24 @@ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProces } /** - * This operations ensures that required dependencies are not accidently injected early given that this bean is BPP. + * This operations ensures that required dependencies are not accidentally injected early given that this bean is BPP. */ + @SuppressWarnings("unchecked") private void injectAndPostProcessDependencies() { - this.streamListenerParameterAdapters = this.applicationContext.getBeansOfType(StreamListenerParameterAdapter.class).values(); - this.streamListenerResultAdapters = this.applicationContext.getBeansOfType(StreamListenerResultAdapter.class).values(); + Collection streamListenerParameterAdapters = this.applicationContext.getBeansOfType(StreamListenerParameterAdapter.class).values(); + Collection streamListenerResultAdapters = this.applicationContext.getBeansOfType(StreamListenerResultAdapter.class).values(); this.binderAwareChannelResolver = this.applicationContext.getBean(DestinationResolver.class); this.messageHandlerMethodFactory = this.applicationContext.getBean(MessageHandlerMethodFactory.class); this.springIntegrationProperties = this.applicationContext.getBean(SpringIntegrationProperties.class); - this.streamListenerCallbacks.forEach(r -> r.run()); + + this.streamListenerSetupMethodOrchestrators.addAll( + this.applicationContext.getBeansOfType(StreamListenerSetupMethodOrchestrator.class).values()); + + //Default orchestrator for StreamListener method invocation is added last into the LinkedHashSet. + this.streamListenerSetupMethodOrchestrators.add(new DefaultStreamListenerSetupMethodOrchestrator(this.applicationContext, + streamListenerParameterAdapters, streamListenerResultAdapters)); + + this.streamListenerCallbacks.forEach(Runnable::run); } private class StreamListenerHandlerMethodMapping { @@ -473,4 +327,152 @@ public class StreamListenerAnnotationBeanPostProcessor implements BeanPostProces } } + private class DefaultStreamListenerSetupMethodOrchestrator implements StreamListenerSetupMethodOrchestrator { + + private final ConfigurableApplicationContext applicationContext; + + private final Collection streamListenerParameterAdapters; + + private final Collection streamListenerResultAdapters; + + private DefaultStreamListenerSetupMethodOrchestrator(ConfigurableApplicationContext applicationContext, Collection streamListenerParameterAdapters, Collection streamListenerResultAdapters) { + this.applicationContext = applicationContext; + this.streamListenerParameterAdapters = streamListenerParameterAdapters; + this.streamListenerResultAdapters = streamListenerResultAdapters; + } + + @Override + public void orchestrateStreamListenerSetupMethod(StreamListener streamListener, Method method, Object bean) { + String methodAnnotatedInboundName = streamListener.value(); + + String methodAnnotatedOutboundName = StreamListenerMethodUtils.getOutboundBindingTargetName(method); + int inputAnnotationCount = StreamListenerMethodUtils.inputAnnotationCount(method); + int outputAnnotationCount = StreamListenerMethodUtils.outputAnnotationCount(method); + boolean isDeclarative = checkDeclarativeMethod(method, methodAnnotatedInboundName, methodAnnotatedOutboundName); + StreamListenerMethodUtils.validateStreamListenerMethod(method, + inputAnnotationCount, outputAnnotationCount, + methodAnnotatedInboundName, methodAnnotatedOutboundName, + isDeclarative, streamListener.condition()); + if (isDeclarative) { + StreamListenerParameterAdapter[] toSlpaArray = new StreamListenerParameterAdapter[this.streamListenerParameterAdapters.size()]; + Object[] adaptedInboundArguments = adaptAndRetrieveInboundArguments(method, methodAnnotatedInboundName, + this.applicationContext, + this.streamListenerParameterAdapters.toArray(toSlpaArray)); + invokeStreamListenerResultAdapter(method, bean, methodAnnotatedOutboundName, adaptedInboundArguments); + } else { + registerHandlerMethodOnListenedChannel(method, streamListener, bean); + } + } + + @Override + public boolean supports(Method method) { + //default catch all orchestrator + return true; + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private void invokeStreamListenerResultAdapter(Method method, Object bean, String outboundName, Object... arguments) { + try { + if (Void.TYPE.equals(method.getReturnType())) { + method.invoke(bean, arguments); + } else { + Object result = method.invoke(bean, arguments); + if (!StringUtils.hasText(outboundName)) { + for (int parameterIndex = 0; parameterIndex < method.getParameterTypes().length; parameterIndex++) { + MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); + if (methodParameter.hasParameterAnnotation(Output.class)) { + outboundName = methodParameter.getParameterAnnotation(Output.class).value(); + } + } + } + Object targetBean = this.applicationContext.getBean(outboundName); + for (StreamListenerResultAdapter streamListenerResultAdapter : this.streamListenerResultAdapters) { + if (streamListenerResultAdapter.supports(result.getClass(), targetBean.getClass())) { + streamListenerResultAdapter.adapt(result, targetBean); + break; + } + } + } + } + catch (Exception e) { + throw new BeanInitializationException("Cannot setup StreamListener for " + method, e); + } + } + + private void registerHandlerMethodOnListenedChannel(Method method, StreamListener streamListener, Object bean) { + Assert.hasText(streamListener.value(), "The binding name cannot be null"); + if (!StringUtils.hasText(streamListener.value())) { + throw new BeanInitializationException("A bound component name must be specified"); + } + final String defaultOutputChannel = StreamListenerMethodUtils.getOutboundBindingTargetName(method); + if (Void.TYPE.equals(method.getReturnType())) { + Assert.isTrue(StringUtils.isEmpty(defaultOutputChannel), + "An output channel cannot be specified for a method that does not return a value"); + } + else { + Assert.isTrue(!StringUtils.isEmpty(defaultOutputChannel), + "An output channel must be specified for a method that can return a value"); + } + StreamListenerMethodUtils.validateStreamListenerMessageHandler(method); + mappedListenerMethods.add(streamListener.value(), + new StreamListenerHandlerMethodMapping(bean, method, streamListener.condition(), defaultOutputChannel, + streamListener.copyHeaders())); + } + + private boolean checkDeclarativeMethod(Method method, String methodAnnotatedInboundName, String methodAnnotatedOutboundName) { + int methodArgumentsLength = method.getParameterTypes().length; + for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) { + MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); + if (methodParameter.hasParameterAnnotation(Input.class)) { + String inboundName = (String) AnnotationUtils + .getValue(methodParameter.getParameterAnnotation(Input.class)); + Assert.isTrue(StringUtils.hasText(inboundName), StreamListenerErrorMessages.INVALID_INBOUND_NAME); + Assert.isTrue(isDeclarativeMethodParameter(inboundName, methodParameter), + StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS); + return true; + } + else if (methodParameter.hasParameterAnnotation(Output.class)) { + String outboundName = (String) AnnotationUtils + .getValue(methodParameter.getParameterAnnotation(Output.class)); + Assert.isTrue(StringUtils.hasText(outboundName), StreamListenerErrorMessages.INVALID_OUTBOUND_NAME); + Assert.isTrue(isDeclarativeMethodParameter(outboundName, methodParameter), + StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS); + return true; + } + else if (StringUtils.hasText(methodAnnotatedOutboundName)) { + return isDeclarativeMethodParameter(methodAnnotatedOutboundName, methodParameter); + } + else if (StringUtils.hasText(methodAnnotatedInboundName)) { + return isDeclarativeMethodParameter(methodAnnotatedInboundName, methodParameter); + } + } + return false; + } + + /** + * Determines if method parameters signify an imperative or declarative listener definition. + *
+ * Imperative - where handler method is invoked on each message by the handler infrastructure provided + * by the framework + *
+ * Declarative - where handler is provided by the method itself. + *
+ * Declarative method parameter could either be {@link MessageChannel} or any other Object for which + * there is a {@link StreamListenerParameterAdapter} (i.e., {@link reactor.core.publisher.Flux}). Declarative method is invoked only + * once during initialization phase. + */ + @SuppressWarnings("unchecked") + private boolean isDeclarativeMethodParameter(String targetBeanName, MethodParameter methodParameter) { + boolean declarative = false; + if (!methodParameter.getParameterType().isAssignableFrom(Object.class) && this.applicationContext.containsBean(targetBeanName)) { + declarative = MessageChannel.class.isAssignableFrom(methodParameter.getParameterType()); + if (!declarative) { + Class targetBeanClass = this.applicationContext.getType(targetBeanName); + declarative = this.streamListenerParameterAdapters.stream() + .anyMatch(slpa -> slpa.supports(targetBeanClass, methodParameter)); + } + } + return declarative; + } + } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerSetupMethodOrchestrator.java new file mode 100644 index 000000000..e2bf272b5 --- /dev/null +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/StreamListenerSetupMethodOrchestrator.java @@ -0,0 +1,123 @@ +/* + * Copyright 2018 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.cloud.stream.binding; + +import java.lang.reflect.Method; + +import org.springframework.cloud.stream.annotation.Input; +import org.springframework.cloud.stream.annotation.Output; +import org.springframework.cloud.stream.annotation.StreamListener; +import org.springframework.context.ApplicationContext; +import org.springframework.core.MethodParameter; +import org.springframework.core.annotation.AnnotationUtils; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Orchestrator used for invoking the {@link StreamListener} setup method. + * + * By default {@link StreamListenerAnnotationBeanPostProcessor} will use an internal implementation + * of this interface to invoke {@link StreamListenerParameterAdapter}s and {@link StreamListenerResultAdapter}s + * or handler mappings on the method annotated with {@link StreamListener}. + * + * By providing a different implementation of this interface and registering it as a Spring Bean in the + * context, one can override the default invocation strategies used by the {@link StreamListenerAnnotationBeanPostProcessor}. + * A typical usecase for such overriding can happen when a downstream {@link org.springframework.cloud.stream.binder.Binder} + * implementation wants to change the way in which any of the default StreamListener handling needs to be changed in a + * custom manner. + * + * When beans of this interface are present in the context, they get priority in the {@link StreamListenerAnnotationBeanPostProcessor} + * before falling back to the default implementation. + * + * {@see StreamListener} + * {@see StreamListenerAnnotationBeanPostProcessor} + * + * @author Soby Chacko + */ +public interface StreamListenerSetupMethodOrchestrator { + + /** + * Checks the method annotated with {@link StreamListener} to see if this implementation + * can successfully orchestrate this method. + * + * @param method annotated with {@link StreamListener} + * @return true if this implementation can orchestrate this method, false otherwise + */ + boolean supports(Method method); + + /** + * Method that allows custom orchestration on the {@link StreamListener} setup method. + * + * @param streamListener reference to the {@link StreamListener} annotation on the method + * @param method annotated with {@link StreamListener} + * @param bean that contains the StreamListener method + * + */ + void orchestrateStreamListenerSetupMethod(StreamListener streamListener, Method method, Object bean); + + /** + * Default implementation for adapting each of the incoming method arguments using an available + * {@link StreamListenerParameterAdapter} and provide the adapted collection of arguments back to the caller. + * + * @param method annotated with {@link StreamListener} + * @param inboundName inbound binding + * @param applicationContext spring application context + * @param streamListenerParameterAdapters used for adapting the method arguments + * @return adapted incoming arguments + */ + @SuppressWarnings({ "rawtypes", "unchecked" }) + default Object[] adaptAndRetrieveInboundArguments(Method method, String inboundName, + ApplicationContext applicationContext, + StreamListenerParameterAdapter... streamListenerParameterAdapters) { + Object[] arguments = new Object[method.getParameterTypes().length]; + for (int parameterIndex = 0; parameterIndex < arguments.length; parameterIndex++) { + MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex); + Class parameterType = methodParameter.getParameterType(); + Object targetReferenceValue = null; + if (methodParameter.hasParameterAnnotation(Input.class)) { + targetReferenceValue = AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Input.class)); + } + else if (methodParameter.hasParameterAnnotation(Output.class)) { + targetReferenceValue = AnnotationUtils.getValue(methodParameter.getParameterAnnotation(Output.class)); + } + else if (arguments.length == 1 && StringUtils.hasText(inboundName)) { + targetReferenceValue = inboundName; + } + if (targetReferenceValue != null) { + Assert.isInstanceOf(String.class, targetReferenceValue, "Annotation value must be a String"); + Object targetBean = applicationContext.getBean((String) targetReferenceValue); + // Iterate existing parameter adapters first + for (StreamListenerParameterAdapter streamListenerParameterAdapter : streamListenerParameterAdapters) { + if (streamListenerParameterAdapter.supports(targetBean.getClass(), methodParameter)) { + arguments[parameterIndex] = streamListenerParameterAdapter.adapt(targetBean, methodParameter); + break; + } + } + if (arguments[parameterIndex] == null && parameterType.isAssignableFrom(targetBean.getClass())) { + arguments[parameterIndex] = targetBean; + } + Assert.notNull(arguments[parameterIndex], "Cannot convert argument " + parameterIndex + " of " + method + + "from " + targetBean.getClass() + " to " + parameterType); + } + else { + throw new IllegalStateException(StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS); + } + } + return arguments; + } + +}