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

Added FunctionCatalogWrapper for better error handling

Browse Source 
Resolves #1449
Resolves #1447
This commit is contained in:
David Turanski
2018-08-20 12:19:29 -04:00
committed by Oleg Zhurakousky
parent 6b1d584c6c
commit 0fc9d09baf
5 changed files with 116 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -0,0 +1,45 @@
/*
* 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.cloud.function.context.FunctionCatalog;
import org.springframework.util.Assert;
/**
* @author David Turanski
**/
public class FunctionCatalogWrapper {
private final FunctionCatalog catalog;
FunctionCatalogWrapper(FunctionCatalog catalog) {
this.catalog = catalog;
}
<T> T lookup(Class<T> functionType, String name) {
T function = catalog.lookup(functionType, name);
Assert.notNull(function, functionType == null ?
String.format("User provided Function '%s' cannot be located.", name) :
String.format("User provided %s '%s' cannot be located.", functionType.getSimpleName(), name));
return function;
}
<T> T lookup(String name) {
return lookup(null, name);
}
}

7
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -34,7 +34,7 @@ import org.springframework.context.annotation.Configuration;
public class FunctionConfiguration {
@Bean
public IntegrationFlowFunctionSupport functionSupport(FunctionCatalog functionCatalog,
public IntegrationFlowFunctionSupport functionSupport(FunctionCatalogWrapper functionCatalog,
FunctionInspector functionInspector, CompositeMessageConverterFactory messageConverterFactory,
FunctionProperties functionProperties) {
@@ -42,4 +42,9 @@ public class FunctionConfiguration {
functionProperties);
}
@Bean
public FunctionCatalogWrapper functionCatalogWrapper(FunctionCatalog catalog) {
return new FunctionCatalogWrapper(catalog);
}
}

5
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
View File

@@ -24,7 +24,6 @@ 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;
@@ -35,6 +34,7 @@ import org.springframework.util.Assert;
/**
*
* @author Oleg Zhurakousky
* @author David Turanski
*
* @param <I> the payload type of the input Message
* @param <O> the payload type of the output Message
@@ -51,11 +51,10 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
private final CompositeMessageConverter messageConverter;
FunctionInvoker(String functionName, FunctionCatalog functionCatalog, FunctionInspector functionInspector,
FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory compositeMessageConverterFactory) {
this.userFunction = functionCatalog.lookup(functionName);
Assert.isInstanceOf(Function.class, this.userFunction);
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();

6
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
View File

@@ -25,7 +25,6 @@ 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.function.core.FluxSupplier;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
@@ -39,12 +38,13 @@ import org.springframework.util.StringUtils;
/**
*
* @author Oleg Zhurakousky
* @author David Turanski
*
* @since 2.1
*/
public class IntegrationFlowFunctionSupport {
private final FunctionCatalog functionCatalog;
private final FunctionCatalogWrapper functionCatalog;
private final FunctionInspector functionInspector;
@@ -58,7 +58,7 @@ public class IntegrationFlowFunctionSupport {
* @param messageConverterFactory
* @param functionProperties
*/
public IntegrationFlowFunctionSupport(FunctionCatalog functionCatalog, FunctionInspector functionInspector,
public IntegrationFlowFunctionSupport(FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
CompositeMessageConverterFactory messageConverterFactory, FunctionProperties functionProperties) {
Assert.notNull(functionCatalog, "'functionCatalog' must not be null");

118
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
View File

@@ -21,10 +21,13 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Supplier;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import reactor.core.publisher.Flux;
import org.springframework.beans.BeanInstantiationException;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
@@ -45,23 +48,27 @@ import org.springframework.util.Assert;
import org.springframework.util.MimeTypeUtils;
import static org.assertj.core.api.Assertions.assertThat;
import static org.hamcrest.CoreMatchers.allOf;
import static org.hamcrest.CoreMatchers.endsWith;
import static org.hamcrest.CoreMatchers.isA;
import static org.hamcrest.Matchers.hasProperty;
/**
*
* @author Oleg Zhurakousky
*
* @author David Turanski
* @since 2.1
*
*/
public class SourceToFunctionsSupportTests {
@Rule
public ExpectedException expectedException = ExpectedException.none();
@Test
public void testFunctionIsAppliedToExistingMessageSource() {
try (ConfigurableApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=toUpperCase", "--spring.jmx.enabled=false")) {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.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));
@@ -70,78 +77,75 @@ public class SourceToFunctionsSupportTests {
@Test
public void testComposedFunctionIsAppliedToExistingMessageSource() {
try (ConfigurableApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf",
"--spring.jmx.enabled=false")) {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfiguration.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));
assertThat(target.receive(1000).getPayload()).isEqualTo(
"HELLO FUNCTION:HELLO FUNCTION".getBytes(StandardCharsets.UTF_8));
}
}
@Test
public void testMessageSourceIsCreatedFromProvidedSupplier() {
try (ConfigurableApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=number",
"--spring.jmx.enabled=false")) {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(
WebApplicationType.NONE).run("--spring.cloud.stream.function.name=number", "--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(10000).getPayload())
.isEqualTo("1".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("2".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("3".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("1".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("2".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("3".getBytes(StandardCharsets.UTF_8));
//etc
}
}
@Test
public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithSingleFunction() {
try (ConfigurableApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=number|concatWithSelf",
"--spring.jmx.enabled=false")) {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(
WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=number|concatWithSelf", "--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(10000).getPayload())
.isEqualTo("11".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("22".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("33".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("11".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("22".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("33".getBytes(StandardCharsets.UTF_8));
//etc
}
}
@Test
public void testMessageSourceIsCreatedFromProvidedSupplierComposedWithMultipleFunctions() {
try (ConfigurableApplicationContext context =
new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class))
.web(WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=number|concatWithSelf|multiplyByTwo",
"--spring.jmx.enabled=false")) {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(
WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=number|concatWithSelf|multiplyByTwo",
"--spring.jmx.enabled=false")) {
OutputDestination target = context.getBean(OutputDestination.class);
assertThat(target.receive(10000).getPayload())
.isEqualTo("22".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("44".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload())
.isEqualTo("66".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("22".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("44".getBytes(StandardCharsets.UTF_8));
assertThat(target.receive(10000).getPayload()).isEqualTo("66".getBytes(StandardCharsets.UTF_8));
//etc
}
}
@Test
public void testFunctionDoesNotExist() {
expectedException.expect(BeanCreationException.class);
expectedException.expectCause(
allOf(isA(BeanInstantiationException.class), hasProperty("cause", isA(IllegalArgumentException.class)),
hasProperty("message", endsWith("'doesNotExist' cannot be located."))));
ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(
WebApplicationType.NONE)
.run("--spring.cloud.stream.function.name=doesNotExist", "--spring.jmx.enabled=false");
}
@EnableAutoConfiguration
@Import(ProvidedMessageSourceConfiguration.class)
public static class SupplierConfiguration {
@@ -195,7 +199,8 @@ public class SourceToFunctionsSupportTests {
@Bean
public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) {
Supplier<Message<String>> messageSource = () -> MessageBuilder.withPayload("hello function")
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN).build();
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
.build();
IntegrationFlowBuilder flowBuilder = functionSupport.integrationFlowFromProvidedSupplier(messageSource);
@@ -221,10 +226,7 @@ public class SourceToFunctionsSupportTests {
public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) {
Assert.hasText(this.functionProperties.getName(), "Supplier name must be provided");
return functionSupport
.integrationFlowFromNamedSupplier()
.channel(this.source.output())
.get();
return functionSupport.integrationFlowFromNamedSupplier().channel(this.source.output()).get();
}
}