From 6fc3b85481f9af8a217740a7518555e85ce2025a Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Tue, 11 Feb 2020 14:09:59 +0100 Subject: [PATCH] GH-1876 Some cleanup on functional and annotation mix Some plishing and cleanup as well as re-enabling few tests,. but as explained in GH issue there is really nothing to fix here since the behav ior is by design. --- .../config/BindingServiceConfiguration.java | 3 +-- .../function/FunctionConfiguration.java | 13 ++++++------ .../DirectWithAttributesChannel.java | 8 ++++++- .../ProcessorToFunctionsSupportTests.java | 21 +++++++++++-------- 4 files changed, 26 insertions(+), 19 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java index 59b7849d4..7ec5e5510 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java @@ -79,6 +79,7 @@ import org.springframework.util.ObjectUtils; * @author Oleg Zhurakousky * @author Soby Chacko */ +@SuppressWarnings("deprecation") @Configuration @EnableConfigurationProperties({ BindingServiceProperties.class, SpringIntegrationProperties.class, StreamFunctionProperties.class }) @@ -88,13 +89,11 @@ import org.springframework.util.ObjectUtils; @ConditionalOnBean(value = BinderTypeRegistry.class, search = SearchStrategy.CURRENT) public class BindingServiceConfiguration { - // @checkstyle:off /** * Name of the Spring Cloud Stream stream listener annotation bean post processor. */ public static final String STREAM_LISTENER_ANNOTATION_BEAN_POST_PROCESSOR_NAME = "streamListenerAnnotationBeanPostProcessor"; - // @checkstyle:on @Autowired(required = false) private Collection binderFactoryListeners; diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index 9fb35aa53..1efbb8bfb 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -59,7 +59,6 @@ import org.springframework.cloud.function.context.config.FunctionContextUtils; import org.springframework.cloud.function.context.config.RoutingFunction; import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.binder.BinderHeaders; -import org.springframework.cloud.stream.binder.BinderTypeRegistry; import org.springframework.cloud.stream.binder.BindingCreatedEvent; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.PartitionHandler; @@ -121,8 +120,8 @@ public class FunctionConfiguration { @Bean public InitializingBean functionBindingRegistrar(Environment environment, FunctionCatalog functionCatalog, - StreamFunctionProperties streamFunctionProperties, BinderTypeRegistry binderTypeRegistry) { - return new FunctionBindingRegistrar(binderTypeRegistry, functionCatalog, streamFunctionProperties); + StreamFunctionProperties streamFunctionProperties) { + return new FunctionBindingRegistrar(functionCatalog, streamFunctionProperties); } @Bean @@ -599,13 +598,14 @@ public class FunctionConfiguration { /** * Creates and registers instances of BindableFunctionProxyFactory for each user defined function * thus triggering destination bindings between function arguments and destinations. + * + * In other words this class is responsible to do the same work as EnableBinding except that it derives the input/output names + * from the names of the function (e.g., function-in-0). */ private static class FunctionBindingRegistrar implements InitializingBean, ApplicationContextAware, EnvironmentAware { protected final Log logger = LogFactory.getLog(getClass()); - private final BinderTypeRegistry binderTypeRegistry; - private final FunctionCatalog functionCatalog; private final StreamFunctionProperties streamFunctionProperties; @@ -618,8 +618,7 @@ public class FunctionConfiguration { private int outputCount; - FunctionBindingRegistrar(BinderTypeRegistry binderTypeRegistry, FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties) { - this.binderTypeRegistry = binderTypeRegistry; + FunctionBindingRegistrar(FunctionCatalog functionCatalog, StreamFunctionProperties streamFunctionProperties) { this.functionCatalog = functionCatalog; this.streamFunctionProperties = streamFunctionProperties; } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/DirectWithAttributesChannel.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/DirectWithAttributesChannel.java index cc6ebeaac..71c82b5f0 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/DirectWithAttributesChannel.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/messaging/DirectWithAttributesChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -20,6 +20,7 @@ import java.util.HashMap; import java.util.Map; import org.springframework.integration.channel.DirectChannel; +import org.springframework.messaging.MessageHandler; /** * @author Oleg Zhurakousky @@ -41,4 +42,9 @@ public class DirectWithAttributesChannel extends DirectChannel { public String getBeanName() { return this.getComponentName(); } + + @Override + public boolean subscribe(MessageHandler handler) { + return this.getDispatcher().getHandlerCount() == 1 ? false : super.subscribe(handler); + } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java index 86c707f8b..f3cff5517 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ProcessorToFunctionsSupportTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2019 the original author or authors. + * Copyright 2019-2020 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. @@ -17,11 +17,11 @@ package org.springframework.cloud.stream.function; import java.nio.charset.StandardCharsets; +import java.util.List; import java.util.function.Consumer; import java.util.function.Function; import org.junit.After; -import org.junit.Ignore; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; @@ -59,7 +59,6 @@ public class ProcessorToFunctionsSupportTests { } @Test - @Ignore public void testPathThrough() { this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration .getCompleteConfiguration(FunctionsConfiguration.class)) @@ -72,7 +71,6 @@ public class ProcessorToFunctionsSupportTests { } @Test - @Ignore public void testSingleFunction() { this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration .getCompleteConfiguration(FunctionsConfiguration.class)) @@ -85,10 +83,13 @@ public class ProcessorToFunctionsSupportTests { source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); assertThat(target.receive(1000).getPayload()) .isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8)); + //to ensure there is no possibility of load balancing to the EnableBinding + source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); + assertThat(target.receive(1000).getPayload()) + .isEqualTo("HELLO".getBytes(StandardCharsets.UTF_8)); } @Test - @Ignore public void testComposedFunction() { this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration .getCompleteConfiguration(FunctionsConfiguration.class)) @@ -101,12 +102,10 @@ public class ProcessorToFunctionsSupportTests { OutputDestination target = this.context.getBean(OutputDestination.class); source.send(new GenericMessage("hello".getBytes(StandardCharsets.UTF_8))); String result = new String(target.receive(1000).getPayload()); - System.out.println(result); assertThat(result).isEqualTo("HELLO:HELLO"); } @Test - @Ignore public void testConsumer() { this.context = new SpringApplicationBuilder(TestChannelBinderConfiguration .getCompleteConfiguration(ConsumerConfiguration.class)) @@ -135,7 +134,10 @@ public class ProcessorToFunctionsSupportTests { @Bean public Function toUpperCase() { - return String::toUpperCase; + return v -> { + System.out.println(); + return v.toUpperCase(); + }; } @Bean @@ -149,11 +151,12 @@ public class ProcessorToFunctionsSupportTests { @Import(BaseProcessorConfiguration.class) public static class ConsumerConfiguration { + @SuppressWarnings("unchecked") @Bean public Consumer log(OutputDestination out) { return x -> { DirectFieldAccessor dfa = new DirectFieldAccessor(out); - MessageChannel channel = (MessageChannel) dfa.getPropertyValue("channel"); + MessageChannel channel = ((List) dfa.getPropertyValue("channels")).get(0); channel.send(new GenericMessage(x.getBytes())); }; }