From dec9ff696ff3a3f1eb107626a810fcfe293bee3c Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Wed, 1 May 2019 20:25:55 +0200 Subject: [PATCH] polishing Resolves #637 --- .../streams/function/FunctionDetectorCondition.java | 12 +++++------- .../KafkaStreamsFunctionBeanPostProcessor.java | 8 ++++---- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java index f1f16b9c5..18f0fd535 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/FunctionDetectorCondition.java @@ -43,17 +43,15 @@ import org.springframework.util.ClassUtils; */ public class FunctionDetectorCondition extends SpringBootCondition { + @SuppressWarnings({ "unchecked", "rawtypes" }) @Override public ConditionOutcome getMatchOutcome(ConditionContext context, AnnotatedTypeMetadata metadata) { if (context != null && context.getBeanFactory() != null) { + Map functionTypes = context.getBeanFactory().getBeansOfType(Function.class); + functionTypes.putAll(context.getBeanFactory().getBeansOfType(Consumer.class)); + final Map kstreamFunctions = pruneFunctionBeansForKafkaStreams(functionTypes, context); - final Map functionTypes = context.getBeanFactory().getBeansOfType(Function.class); - final Map consumerTypes = context.getBeanFactory().getBeansOfType(Consumer.class); - - final Map prunedFunctionMap = pruneFunctionBeansForKafkaStreams(functionTypes, context); - final Map prunedConsumerMap = pruneFunctionBeansForKafkaStreams(consumerTypes, context); - - if (!prunedFunctionMap.isEmpty() || !prunedConsumerMap.isEmpty()) { + if (!kstreamFunctions.isEmpty()) { return ConditionOutcome.match("Matched. Function/Consumer beans found"); } else { diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java index 914983b83..d13f4c6c2 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java @@ -21,6 +21,7 @@ import java.util.Map; import java.util.TreeMap; import java.util.function.Consumer; import java.util.function.Function; +import java.util.stream.Stream; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -49,11 +50,10 @@ class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, BeanFac @Override public void afterPropertiesSet() { - final Map functionTypes = this.beanFactory.getBeansOfType(Function.class); - final Map consumerTypes = this.beanFactory.getBeansOfType(Consumer.class); + String[] functionNames = this.beanFactory.getBeanNamesForType(Function.class); + String[] consumerNames = this.beanFactory.getBeanNamesForType(Consumer.class); - functionTypes.keySet().forEach(this::extractResolvableTypes); - consumerTypes.keySet().forEach(this::extractResolvableTypes); + Stream.concat(Stream.of(functionNames), Stream.of(consumerNames)).forEach(this::extractResolvableTypes); } private void extractResolvableTypes(String key) {