diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java index 3bdcb1428..fd8f8df09 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java @@ -240,7 +240,8 @@ public class KafkaStreamsBinderSupportAutoConfiguration { } @Bean - @ConditionalOnProperty("spring.cloud.stream.kafka.streams.function.definition") +// @ConditionalOnProperty("spring.cloud.stream.kafka.streams.function.definition") + @ConditionalOnProperty("spring.cloud.stream.function.definition") public KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor(BindingServiceProperties bindingServiceProperties, KafkaStreamsExtendedBindingProperties kafkaStreamsExtendedBindingProperties, KeyValueSerdeResolver keyValueSerdeResolver, diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java index c352bc361..83640ce08 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java @@ -16,11 +16,10 @@ package org.springframework.cloud.stream.binder.kafka.streams.function; -import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsFunctionProcessor; -import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -28,26 +27,23 @@ import org.springframework.context.annotation.Configuration; * @author Soby Chacko */ @Configuration -@ConditionalOnProperty("spring.cloud.stream.kafka.streams.function.definition") -@EnableConfigurationProperties(KafkaStreamsFunctionProperties.class) +@ConditionalOnProperty("spring.cloud.stream.function.definition") +@EnableConfigurationProperties(StreamFunctionProperties.class) public class KafkaStreamsFunctionAutoConfiguration { - @Autowired - ConfigurableApplicationContext context; - @Bean public KafkaStreamsFunctionProcessorInvoker kafkaStreamsFunctionProcessorInvoker( KafkaStreamsFunctionBeanPostProcessor kafkaStreamsFunctionBeanPostProcessor, KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor, - KafkaStreamsFunctionProperties properties) { + StreamFunctionProperties properties) { return new KafkaStreamsFunctionProcessorInvoker(kafkaStreamsFunctionBeanPostProcessor.getResolvableType(), properties.getDefinition(), kafkaStreamsFunctionProcessor); } @Bean public KafkaStreamsFunctionBeanPostProcessor kafkaStreamsFunctionBeanPostProcessor( - ConfigurableApplicationContext context, KafkaStreamsFunctionProperties properties) { - return new KafkaStreamsFunctionBeanPostProcessor(properties, context); + StreamFunctionProperties properties) { + return new KafkaStreamsFunctionBeanPostProcessor(properties); } } 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 1e62f6e09..b824829bd 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 @@ -18,22 +18,30 @@ package org.springframework.cloud.stream.binder.kafka.streams.function; import java.lang.reflect.Method; +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.AnnotatedBeanDefinition; -import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.cloud.stream.function.StreamFunctionProperties; import org.springframework.core.ResolvableType; import org.springframework.util.ClassUtils; -public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean { +/** + * + * @author Soby Chacko + * @since 2.1.0 + * + */ +class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean, BeanFactoryAware { - private final KafkaStreamsFunctionProperties kafkaStreamsFunctionProperties; - private final ConfigurableApplicationContext context; + private final StreamFunctionProperties kafkaStreamsFunctionProperties; + private ConfigurableListableBeanFactory beanFactory; private ResolvableType resolvableType; - public KafkaStreamsFunctionBeanPostProcessor(KafkaStreamsFunctionProperties properties, - ConfigurableApplicationContext context) { + KafkaStreamsFunctionBeanPostProcessor(StreamFunctionProperties properties) { this.kafkaStreamsFunctionProperties = properties; - this.context = context; } public ResolvableType getResolvableType() { @@ -43,16 +51,21 @@ public class KafkaStreamsFunctionBeanPostProcessor implements InitializingBean { @Override public void afterPropertiesSet() throws Exception { final Class classObj = ClassUtils.resolveClassName(((AnnotatedBeanDefinition) - context.getBeanFactory().getBeanDefinition(kafkaStreamsFunctionProperties.getDefinition())) + this.beanFactory.getBeanDefinition(kafkaStreamsFunctionProperties.getDefinition())) .getMetadata().getClassName(), ClassUtils.getDefaultClassLoader()); try { - Method method = classObj.getMethod(this.kafkaStreamsFunctionProperties.getDefinition(), null); + Method method = classObj.getMethod(this.kafkaStreamsFunctionProperties.getDefinition()); this.resolvableType = ResolvableType.forMethodReturnType(method, classObj); } catch (NoSuchMethodException e) { //ignore } } + + @Override + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + this.beanFactory = (ConfigurableListableBeanFactory) beanFactory; + } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java index 8f6a666b1..738f57076 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java @@ -21,13 +21,18 @@ import javax.annotation.PostConstruct; import org.springframework.cloud.stream.binder.kafka.streams.KafkaStreamsFunctionProcessor; import org.springframework.core.ResolvableType; -public class KafkaStreamsFunctionProcessorInvoker { +/** + * + * @author Soby Chacko + * @since 2.1.0 + */ +class KafkaStreamsFunctionProcessorInvoker { private final KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor; private final ResolvableType resolvableType; private final String functionName; - public KafkaStreamsFunctionProcessorInvoker(ResolvableType resolvableType, String functionName, + KafkaStreamsFunctionProcessorInvoker(ResolvableType resolvableType, String functionName, KafkaStreamsFunctionProcessor kafkaStreamsFunctionProcessor) { this.kafkaStreamsFunctionProcessor = kafkaStreamsFunctionProcessor; this.resolvableType = resolvableType; @@ -35,7 +40,7 @@ public class KafkaStreamsFunctionProcessorInvoker { } @PostConstruct - public void invoke() { + void invoke() { this.kafkaStreamsFunctionProcessor.orchestrateStreamListenerSetupMethod(resolvableType, functionName); } } diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProperties.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProperties.java deleted file mode 100644 index 4dbf1e78d..000000000 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProperties.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Copyright 2019-2019 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.binder.kafka.streams.function; - -import org.springframework.boot.context.properties.ConfigurationProperties; - -/** - * @author Soby Chacko - */ -@ConfigurationProperties("spring.cloud.stream.kafka.streams.function") -public class KafkaStreamsFunctionProperties { - - private String definition; - - public String getDefinition() { - return definition; - } - - public void setDefinition(String definition) { - this.definition = definition; - } -} - - diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java index e8e3cb253..9e4894de2 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java @@ -85,7 +85,7 @@ public class KafkaStreamsBinderWordCountBranchesFunctionTests { ConfigurableApplicationContext context = app.run("--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.kafka.streams.function.definition=process", + "--spring.cloud.stream.function.definition=process", "--spring.cloud.stream.bindings.input.destination=words", "--spring.cloud.stream.bindings.output1.destination=counts", "--spring.cloud.stream.bindings.output1.contentType=application/json", diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java index 9dcf0d67e..410ea04bb 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java @@ -85,7 +85,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { try (ConfigurableApplicationContext context = app.run("--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.kafka.streams.function.definition=process", + "--spring.cloud.stream.function.definition=process", "--spring.cloud.stream.bindings.input.destination=words", "--spring.cloud.stream.bindings.output.destination=counts", "--spring.cloud.stream.bindings.output.contentType=application/json", diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java index 83ae82678..c7a2dd671 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java @@ -72,7 +72,7 @@ public class StreamToGlobalKTableFunctionTests { app.setWebApplicationType(WebApplicationType.NONE); try (ConfigurableApplicationContext ignored = app.run("--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.kafka.streams.function.definition=process", + "--spring.cloud.stream.function.definition=process", "--spring.cloud.stream.bindings.input.destination=orders", "--spring.cloud.stream.bindings.input-x.destination=customers", "--spring.cloud.stream.bindings.input-y.destination=products", diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java index 1c4dce870..e37f359ba 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java @@ -84,7 +84,7 @@ public class StreamToTableJoinFunctionTests { try (ConfigurableApplicationContext ignored = app.run("--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.kafka.streams.function.definition=process1", + "--spring.cloud.stream.function.definition=process1", "--spring.cloud.stream.bindings.input-1.destination=user-clicks-1", "--spring.cloud.stream.bindings.input-2.destination=user-regions-1", "--spring.cloud.stream.bindings.output.destination=output-topic-1", @@ -228,10 +228,10 @@ public class StreamToTableJoinFunctionTests { for (KeyValue keyValue : userClicks) { template.sendDefault(keyValue.key, keyValue.value); } - //Thread.sleep(10000L); + try (ConfigurableApplicationContext ignored = app.run("--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.kafka.streams.function.definition=process1", + "--spring.cloud.stream.function.definition=process1", "--spring.cloud.stream.bindings.input-1.destination=user-clicks-2", "--spring.cloud.stream.bindings.input-2.destination=user-regions-2", "--spring.cloud.stream.bindings.output.destination=output-topic-2",