New abstraction for StreamListener method setup invoker.
StreamListenerMethodSetupOrchestrator is an API hook that allows downstram binder implementations or applications to inject custom strategies to alter the default StreamListener adapter method invocations. This PR primarily focuses on allowing customizations on the oubound side. For the inbound, StreamListenerMethodSetupOrchestrator interface provides a default implementation. Corresponding refactoring in StreamListenerAnnotationBeanPostProcessor. By default StreamListenerAnnotationBeanPostProcessor will use the current strategies used for invoking the StreamListener adapters. If beans are provided for orchestration, they take precedence however. Test changes. Further polishing in StreamListenerAnnotationBeanPostProcessor. Resolves #1177 Resolves #1179
This commit is contained in:
Soby Chacko
committed by
Oleg Zhurakousky
Oleg Zhurakousky
parent
9ef1181f7a
commit
c1749ca866
@@ -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<StreamListenerSetupMethodOrchestrator> field1 =
|
||||
(LinkedHashSet<StreamListenerSetupMethodOrchestrator>)ReflectionUtils.getField(field, streamListenerAnnotationBeanPostProcessor);
|
||||
List<StreamListenerSetupMethodOrchestrator> 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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<String, StreamListenerHandlerMethodMapping> mappedListenerMethods = new LinkedMultiValueMap<>();
|
||||
|
||||
// == dependencies that are injected in 'afterSingletonsInstantiated' to avoid early initialization
|
||||
@SuppressWarnings("rawtypes")
|
||||
private Collection<StreamListenerParameterAdapter> streamListenerParameterAdapters;
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
private Collection<StreamListenerResultAdapter> streamListenerResultAdapters;
|
||||
|
||||
private DestinationResolver<MessageChannel> 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<StreamListenerSetupMethodOrchestrator> 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<String, List<StreamListenerHandlerMethodMapping>> mappedBindingEntry : mappedListenerMethods
|
||||
.entrySet()) {
|
||||
ArrayList<DispatchingStreamListenerMessageHandler.ConditionalStreamListenerMessageHandlerWrapper> 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.
|
||||
* <br>
|
||||
* Imperative - where handler method is invoked on each message by the handler infrastructure provided
|
||||
* by the framework
|
||||
* <br>
|
||||
* Declarative - where handler is provided by the method itself.
|
||||
* <br>
|
||||
* 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<StreamListenerSetupMethodOrchestrator> 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<StreamListenerParameterAdapter> streamListenerParameterAdapters = this.applicationContext.getBeansOfType(StreamListenerParameterAdapter.class).values();
|
||||
Collection<StreamListenerResultAdapter> 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<StreamListenerParameterAdapter> streamListenerParameterAdapters;
|
||||
|
||||
private final Collection<StreamListenerResultAdapter> streamListenerResultAdapters;
|
||||
|
||||
private DefaultStreamListenerSetupMethodOrchestrator(ConfigurableApplicationContext applicationContext, Collection<StreamListenerParameterAdapter> streamListenerParameterAdapters, Collection<StreamListenerResultAdapter> 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.
|
||||
* <br>
|
||||
* Imperative - where handler method is invoked on each message by the handler infrastructure provided
|
||||
* by the framework
|
||||
* <br>
|
||||
* Declarative - where handler is provided by the method itself.
|
||||
* <br>
|
||||
* 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;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user