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

GH-1437 Added initial components to support SCF

Browse Source 
Resolves spring-cloud/spring-cloud-stream#1437
Resolves spring-cloud/spring-cloud-stream#1438

Added initial set of supporting classes, tests and configuration
to support integration of Spring Cloud Function with the current
set of streaming binders.
This commit primarily addresses the Source aspect of such integration

addressed PR comments

* Polishing some code style
* Fix typos
This commit is contained in:
Oleg Zhurakousky
2018-08-14 15:24:14 +02:00
committed by Artem Bilan
parent 15f9da86d4
commit 7a9628506d
6 changed files with 431 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

28
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceConfiguration.java
View File

@@ -44,6 +44,8 @@ import org.springframework.cloud.stream.binding.InputBindingLifecycle;
import org.springframework.cloud.stream.binding.MessageChannelStreamListenerResultAdapter;
import org.springframework.cloud.stream.binding.OutputBindingLifecycle;
import org.springframework.cloud.stream.binding.StreamListenerAnnotationBeanPostProcessor;
import org.springframework.cloud.stream.function.FunctionConfiguration;
import org.springframework.cloud.stream.function.FunctionProperties;
import org.springframework.cloud.stream.micrometer.DestinationPublishingMetricsAutoConfiguration;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Bean;
@@ -76,8 +78,8 @@ import org.springframework.util.Assert;
* @author Soby Chacko
*/
@Configuration
@EnableConfigurationProperties({ BindingServiceProperties.class, SpringIntegrationProperties.class })
@Import({DestinationPublishingMetricsAutoConfiguration.class, SpelExpressionConverterConfiguration.class})
@EnableConfigurationProperties({ BindingServiceProperties.class, SpringIntegrationProperties.class, FunctionProperties.class })
@Import({ DestinationPublishingMetricsAutoConfiguration.class, SpelExpressionConverterConfiguration.class, FunctionConfiguration.class })
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
@ConditionalOnBean(value = BinderTypeRegistry.class, search = SearchStrategy.CURRENT)
public class BindingServiceConfiguration {
@@ -91,7 +93,8 @@ public class BindingServiceConfiguration {
@Bean
@ConditionalOnMissingBean(BinderFactory.class)
public BinderFactory binderFactory(BinderTypeRegistry binderTypeRegistry,
BindingServiceProperties bindingServiceProperties) {
BindingServiceProperties bindingServiceProperties) {
DefaultBinderFactory binderFactory = new DefaultBinderFactory(
getBinderConfigurations(binderTypeRegistry, bindingServiceProperties), binderTypeRegistry);
binderFactory.setDefaultBinder(bindingServiceProperties.getDefaultBinder());
@@ -100,7 +103,8 @@ public class BindingServiceConfiguration {
}
private static Map<String, BinderConfiguration> getBinderConfigurations(BinderTypeRegistry binderTypeRegistry,
BindingServiceProperties bindingServiceProperties) {
BindingServiceProperties bindingServiceProperties) {
Map<String, BinderConfiguration> binderConfigurations = new HashMap<>();
Map<String, BinderProperties> declaredBinders = bindingServiceProperties.getBinders();
boolean defaultCandidatesExist = false;
@@ -161,12 +165,15 @@ public class BindingServiceConfiguration {
@ConditionalOnMissingBean(search = SearchStrategy.CURRENT)
public BindingService bindingService(BindingServiceProperties bindingServiceProperties,
BinderFactory binderFactory, TaskScheduler taskScheduler) {
return new BindingService(bindingServiceProperties, binderFactory, taskScheduler);
}
@Bean
@DependsOn("bindingService")
public OutputBindingLifecycle outputBindingLifecycle(BindingService bindingService, Map<String, Bindable> bindables) {
public OutputBindingLifecycle outputBindingLifecycle(BindingService bindingService,
Map<String, Bindable> bindables) {
return new OutputBindingLifecycle(bindingService, bindables);
}
@@ -189,6 +196,7 @@ public class BindingServiceConfiguration {
DynamicDestinationsBindable dynamicDestinationsBindable,
@Nullable BinderAwareChannelResolver.NewDestinationBindingCallback callback,
@Nullable GlobalChannelInterceptorProcessor globalChannelInterceptorProcessor) {
return new BinderAwareChannelResolver(bindingService, bindingTargetFactory, dynamicDestinationsBindable,
callback, globalChannelInterceptorProcessor);
}
@@ -202,8 +210,8 @@ public class BindingServiceConfiguration {
@Bean
@ConditionalOnMissingBean
public org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor binderAwareRouterBeanPostProcessor(
@Autowired(required=false) AbstractMappingMessageRouter[] routers,
@Autowired(required=false)DestinationResolver<MessageChannel> channelResolver) {
@Autowired(required = false) AbstractMappingMessageRouter[] routers,
@Autowired(required = false) DestinationResolver<MessageChannel> channelResolver) {
return new org.springframework.cloud.stream.binding.BinderAwareRouterBeanPostProcessor(routers, channelResolver);
}
@@ -211,11 +219,15 @@ public class BindingServiceConfiguration {
@Bean
public ApplicationListener<ContextRefreshedEvent> appListener(SpringIntegrationProperties springIntegrationProperties) {
return new ApplicationListener<ContextRefreshedEvent>() {
@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
event.getApplicationContext().getBeansOfType(AbstractReplyProducingMessageHandler.class).values()
.forEach(mh -> mh.addNotPropagatedHeaders(springIntegrationProperties.getMessageHandlerNotPropagatedHeaders()));
.forEach(mh -> mh.addNotPropagatedHeaders(springIntegrationProperties
.getMessageHandlerNotPropagatedHeaders()));
}
};
}
}

43
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* 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.function;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
*
* @author Oleg Zhurakousky
*
* @since 2.1
*/
@Configuration
@ConditionalOnProperty("spring.cloud.stream.function.name")
public class FunctionConfiguration {
@Bean
public FunctionSupport functionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory messageConverterFactory) {
return new FunctionSupport(functionCatalog, functionInspector, messageConverterFactory);
}
}

101
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java Normal file
View File

@@ -0,0 +1,101 @@
/*
* 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.function;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import reactor.core.publisher.Flux;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.FunctionType;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.CompositeMessageConverter;
import org.springframework.util.Assert;
/**
*
* @author Oleg Zhurakousky
*
* @param <I> the payload type of the input Message
* @param <O> the payload type of the output Message
*
* @since 2.1
*/
class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O>>> {
private static final Log logger = LogFactory.getLog(FunctionInvoker.class);
private final Class<?> inputClass;
private final Function<Flux<?>, Flux<?>> userFunction;
private final CompositeMessageConverter messageConverter;
FunctionInvoker(String functionName, FunctionCatalog functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory compositeMessageConverterFactory) {
this.userFunction = functionCatalog.lookup(Function.class, functionName);
Assert.notNull(this.userFunction, "userFunction: " + functionName + " can not be located.");
this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
FunctionType functionType = functionInspector.getRegistration(this.userFunction).getType();
this.inputClass = functionType.getInputType();
}
@Override
public Flux<Message<O>> apply(Flux<Message<I>> input) {
AtomicReference<Message<I>> originalMessage = new AtomicReference<>();
return input
.doOnNext(originalMessage::set) // to preserve the original message
.map(this::resolveArgument) // resolves argument type before invocation of user function
.transform(this.userFunction::apply) // invoke user function
.map(resultMessage -> toMessage(resultMessage, originalMessage.get())); // create output message
}
@SuppressWarnings("unchecked")
private <T> Message<O> toMessage(T value, Message<I> originalMessage) {
if (logger.isDebugEnabled()) {
logger.debug("Converting result back to message using the original message: " + originalMessage);
}
return (Message<O>)
(value instanceof Message
? value
: this.messageConverter.toMessage(value, originalMessage.getHeaders()));
}
@SuppressWarnings("unchecked")
private <T> T resolveArgument(Message<I> message) {
if (logger.isDebugEnabled()) {
logger.debug("Resolving input argument from message: " + message);
}
return (T) (shouldConvertFromMessage(message)
? this.messageConverter.fromMessage(message, this.inputClass)
: message);
}
private boolean shouldConvertFromMessage(Message<?> message) {
return !this.inputClass.isAssignableFrom(byte[].class) &&
!this.inputClass.isAssignableFrom(Object.class);
}
}

44
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionProperties.java Normal file
View File

@@ -0,0 +1,44 @@
/*
* 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.function;
import org.springframework.boot.context.properties.ConfigurationProperties;
/**
*
* @author Oleg Zhurakousky
*
* @since 2.1
*/
@ConfigurationProperties("spring.cloud.stream.function")
public class FunctionProperties {
/**
* Name of functions to bind. If several functions need to be composed into one, use pipes (e.g., 'fooFunc|barFunc')
*/
private String name;
public String getName() {
return this.name;
}
public void setName(String name) {
this.name = name;
}
}

86
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionSupport.java Normal file
View File

@@ -0,0 +1,86 @@
/*
* 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.function;
import java.util.function.Consumer;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import org.springframework.cloud.function.context.FunctionCatalog;
import org.springframework.cloud.function.context.catalog.FunctionInspector;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
*
* @author Oleg Zhurakousky
*
* @since 2.1
*/
public class FunctionSupport {
private final FunctionCatalog functionCatalog;
private final FunctionInspector functionInspector;
private final CompositeMessageConverterFactory messageConverterFactory;
public FunctionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory messageConverterFactory) {
Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
Assert.notNull(functionInspector, "'functionInspector' must not be null");
Assert.notNull(messageConverterFactory, "'messageConverterFactory' must not be null");
this.functionCatalog = functionCatalog;
this.functionInspector = functionInspector;
this.messageConverterFactory = messageConverterFactory;
}
public <I,O> void applyFunctionToIntegrationFlow(String functionName, IntegrationFlowBuilder flowBuilder,
Consumer<Message<O>> outputProcessor) {
FunctionInvoker<I,O> functionInvoker =
new FunctionInvoker<>(functionName, this.functionCatalog, this.functionInspector,
this.messageConverterFactory);
subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputProcessor);
}
private <O> Mono<Void> subscribeToOutput(Consumer<Message<O>> outputProcessor,
Publisher<Message<O>> outputPublisher) {
Flux<Message<O>> output = outputProcessor == null
? Flux.from(outputPublisher)
: Flux.from(outputPublisher).doOnNext(outputProcessor);
return output.then();
}
private <I,O> void subscribeToInput(FunctionInvoker<I,O> functionInvoker, Publisher<Message<I>> publisher,
Consumer<Message<O>> outputProcessor) {
Flux<Message<I>> inputPublisher = Flux.from(publisher);
subscribeToOutput(outputProcessor, functionInvoker.apply(inputPublisher)).subscribe();
}
}

137
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionSupportTests.java Normal file
View File

@@ -0,0 +1,137 @@
/*
* 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.function;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.function.Supplier;
import org.junit.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.binder.test.OutputDestination;
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.StringUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
*
* @author Oleg Zhurakousky
*
* @since 2.1
*
*/
public class FunctionSupportTests {
@Test
public void testFunctionIsAppliedToMessageSource() {
ApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=toUpperCase", "--spring.jmx.enabled=false");
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(1000).getPayload()).isEqualTo("HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
}
@Test
public void testComposedFunctionIsAppliedToMessageSource() {
ApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(MyFunctionsConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false");
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(1000).getPayload())
.isEqualTo("HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
}
@SpringBootApplication
@Import(SourceConfiguration.class)
public static class MyFunctionsConfiguration {
@Bean
public Function<String, String> toUpperCase() {
return String::toUpperCase;
}
@Bean
public Function<String, String> concatWithSelf() {
return x -> x + ":" + x;
}
}
/**
* This configuration essentially emulates our existing app-starters for Sources
* and essentially demonstrates how a function(s) could be applied to an existing
* source via {@link FunctionSupport} class.
*/
@EnableBinding(Source.class)
public static class SourceConfiguration {
@Autowired
private Source source;
@Autowired
private FunctionProperties functionProperties;
@Bean
public IntegrationFlow messageSourceFlow(FunctionSupport functionSupport) {
Supplier<Message<String>> messageSource = () -> MessageBuilder.withPayload("hello function")
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build();
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(messageSource);
if (StringUtils.hasText(functionProperties.getName())) {
functionSupport
.applyFunctionToIntegrationFlow(functionProperties.getName(), flowBuilder,
message -> source.output().send(message));
}
else {
flowBuilder = flowBuilder.channel(source.output());
}
return flowBuilder.get();
}
}
}