SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Minore cleanup and refactorings after previous commit

Browse Source 
Removed KafkaStreamsFunctionProperties in favor of StreamFunctionProperties provided by core module

Resolves #537
This commit is contained in:
Oleg Zhurakousky
2019-03-21 17:06:18 +01:00
parent 792705d304
commit 73d3d79651
9 changed files with 44 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsBinderSupportAutoConfiguration.java
View File

@@ -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,

16
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionAutoConfiguration.java
View File

@@ -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);
}
}

31
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionBeanPostProcessor.java
View File

@@ -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;
}
}

11
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProcessorInvoker.java
View File

@@ -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);
}
}

38
spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsFunctionProperties.java
View File

@@ -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;
}
}

2
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountBranchesFunctionTests.java
View File

@@ -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",

2
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java
View File

@@ -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",

2
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToGlobalKTableFunctionTests.java
View File

@@ -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",

6
spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/StreamToTableJoinFunctionTests.java
View File

@@ -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<String, Long> 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",