GH-1742 Adjusted previous commit to new function infrustructure
Resolves #1738 Resolves #1742
This commit is contained in:
Oleg Zhurakousky
@@ -20,12 +20,15 @@ import java.io.IOException;
|
||||
import java.lang.reflect.ParameterizedType;
|
||||
import java.lang.reflect.Type;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Collection;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import com.fasterxml.jackson.databind.JavaType;
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
|
||||
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
|
||||
import org.springframework.core.MethodParameter;
|
||||
import org.springframework.core.ParameterizedTypeReference;
|
||||
import org.springframework.lang.Nullable;
|
||||
@@ -118,6 +121,9 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
|
||||
try {
|
||||
JavaType type = this.typeCache.get(conversionHint);
|
||||
if (type == null) {
|
||||
conversionHint = FunctionTypeUtils.isMessage(conversionHint)
|
||||
? FunctionTypeUtils.getImmediateGenericType(conversionHint, 0)
|
||||
: conversionHint;
|
||||
type = objectMapper.getTypeFactory()
|
||||
.constructType(conversionHint);
|
||||
this.typeCache.put(conversionHint, type);
|
||||
@@ -129,6 +135,26 @@ class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageC
|
||||
return objectMapper.readValue((String) payload, type);
|
||||
}
|
||||
else {
|
||||
final JavaType typeToUse = type;
|
||||
if (payload instanceof Collection) {
|
||||
Collection<?> collection = (Collection<?>) ((Collection<?>) payload).stream()
|
||||
.map(value -> {
|
||||
try {
|
||||
if (value instanceof byte[]) {
|
||||
return objectMapper.readValue((byte[]) value, typeToUse.getContentType());
|
||||
}
|
||||
else if (value instanceof String) {
|
||||
return objectMapper.readValue((String) value, typeToUse.getContentType());
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to convert payload " + value, e);
|
||||
}
|
||||
return null;
|
||||
}).collect(Collectors.toList());
|
||||
|
||||
return collection;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,331 +0,0 @@
|
||||
/*
|
||||
* Copyright 2018-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
|
||||
*
|
||||
* https://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.lang.reflect.Field;
|
||||
import java.lang.reflect.ParameterizedType;
|
||||
import java.lang.reflect.Type;
|
||||
import java.time.Duration;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
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.binder.ConsumerProperties;
|
||||
import org.springframework.cloud.stream.binder.ProducerProperties;
|
||||
import org.springframework.cloud.stream.config.BindingServiceProperties;
|
||||
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.converter.CompositeMessageConverter;
|
||||
import org.springframework.messaging.support.ErrorMessage;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.MimeType;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
|
||||
/**
|
||||
* @param <I> the payload type of the input Message
|
||||
* @param <O> the payload type of the output Message
|
||||
* @author Oleg Zhurakousky
|
||||
* @author David Turanski
|
||||
* @author Tolga Kavukcu
|
||||
* @author Gary Russell
|
||||
* @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 static final Field MESSAGE_HEADERS_FIELD;
|
||||
|
||||
static {
|
||||
MESSAGE_HEADERS_FIELD = ReflectionUtils.findField(MessageHeaders.class,
|
||||
"headers");
|
||||
MESSAGE_HEADERS_FIELD.setAccessible(true);
|
||||
}
|
||||
|
||||
private final Class<?> inputClass;
|
||||
|
||||
private final ParameterizedType inputParameterizedType;
|
||||
|
||||
private final Class<?> outputClass;
|
||||
|
||||
private final Function<Flux<?>, Flux<?>> userFunction;
|
||||
|
||||
private final CompositeMessageConverter messageConverter;
|
||||
|
||||
private final MessageChannel errorChannel;
|
||||
|
||||
private final boolean isInputArgumentMessage;
|
||||
|
||||
private final Class<?> messagePayloadClass;
|
||||
|
||||
private final Type messagePayloadType;
|
||||
|
||||
private final ConsumerProperties consumerProperties;
|
||||
|
||||
private final ProducerProperties producerProperties;
|
||||
|
||||
private final BindingServiceProperties bindingServiceProperties;
|
||||
|
||||
private final StreamFunctionProperties functionProperties;
|
||||
|
||||
private final boolean batchMode;
|
||||
|
||||
private final Type listContentParameterizedType;
|
||||
|
||||
private final Class<?> listContentClass;
|
||||
|
||||
FunctionInvoker(StreamFunctionProperties functionProperties,
|
||||
FunctionCatalog functionCatalog, FunctionInspector functionInspector,
|
||||
CompositeMessageConverterFactory compositeMessageConverterFactory) {
|
||||
this(functionProperties, functionCatalog, functionInspector,
|
||||
compositeMessageConverterFactory, null);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
FunctionInvoker(StreamFunctionProperties functionProperties,
|
||||
FunctionCatalog functionCatalog, FunctionInspector functionInspector,
|
||||
CompositeMessageConverterFactory compositeMessageConverterFactory,
|
||||
MessageChannel errorChannel) {
|
||||
|
||||
this.functionProperties = functionProperties;
|
||||
Object originalUserFunction = functionCatalog
|
||||
.lookup(functionProperties.getDefinition());
|
||||
|
||||
this.userFunction = (Function<Flux<?>, Flux<?>>) originalUserFunction;
|
||||
|
||||
Assert.isInstanceOf(Function.class, this.userFunction);
|
||||
this.messageConverter = compositeMessageConverterFactory
|
||||
.getMessageConverterForAllRegistered();
|
||||
FunctionType functionType = functionInspector
|
||||
.getRegistration(originalUserFunction).getType();
|
||||
this.isInputArgumentMessage = functionType.isMessage();
|
||||
this.inputClass = functionType.getInputType();
|
||||
this.outputClass = functionType.getOutputType();
|
||||
this.errorChannel = errorChannel;
|
||||
this.bindingServiceProperties = functionProperties.getBindingServiceProperties();
|
||||
this.consumerProperties = this.bindingServiceProperties
|
||||
.getConsumerProperties(functionProperties.getInputDestinationName());
|
||||
this.producerProperties = this.bindingServiceProperties
|
||||
.getProducerProperties(functionProperties.getOutputDestinationName());
|
||||
this.batchMode = this.consumerProperties.isBatchMode();
|
||||
Type type = functionType.getType();
|
||||
ParameterizedType functionInputParameterizedType = null;
|
||||
Type listContainsType = null;
|
||||
Type payloadType = null;
|
||||
if (type instanceof ParameterizedType) {
|
||||
Type functionInputType = ((ParameterizedType) type).getActualTypeArguments()[0];
|
||||
if (functionInputType instanceof ParameterizedType) {
|
||||
functionInputParameterizedType = (ParameterizedType) functionInputType;
|
||||
Type rawType = ((ParameterizedType) functionInputType).getRawType();
|
||||
if (rawType.equals(List.class)) {
|
||||
listContainsType = ((ParameterizedType) functionInputType).getActualTypeArguments()[0];
|
||||
}
|
||||
else if (rawType.equals(Message.class)) {
|
||||
payloadType = determinePayloadType(functionInputType);
|
||||
}
|
||||
}
|
||||
}
|
||||
if (listContainsType instanceof Class) {
|
||||
this.listContentClass = (Class<?>) listContainsType;
|
||||
this.listContentParameterizedType = null;
|
||||
}
|
||||
else {
|
||||
this.listContentClass = Object.class;
|
||||
this.listContentParameterizedType = listContainsType;
|
||||
}
|
||||
if ((functionInputParameterizedType != null && functionInputParameterizedType.getRawType().equals(Flux.class))
|
||||
|| payloadType != null) {
|
||||
functionInputParameterizedType = null;
|
||||
}
|
||||
this.inputParameterizedType = functionInputParameterizedType;
|
||||
if (payloadType instanceof Class) {
|
||||
this.messagePayloadClass = (Class<?>) payloadType;
|
||||
this.messagePayloadType = null;
|
||||
}
|
||||
else {
|
||||
this.messagePayloadClass = Object.class;
|
||||
this.messagePayloadType = payloadType;
|
||||
}
|
||||
}
|
||||
|
||||
private Type determinePayloadType(Type functionInputType) {
|
||||
Type payloadType;
|
||||
payloadType = ((ParameterizedType) functionInputType).getActualTypeArguments()[0];
|
||||
if (payloadType instanceof ParameterizedType) {
|
||||
Type payloadRawType = ((ParameterizedType) payloadType).getRawType();
|
||||
if (payloadRawType.equals(List.class)) {
|
||||
payloadType = ((ParameterizedType) payloadType).getActualTypeArguments()[0];
|
||||
}
|
||||
}
|
||||
return payloadType;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<Message<O>> apply(Flux<Message<I>> input) {
|
||||
AtomicReference<Message<I>> originalMessageRef = new AtomicReference<>();
|
||||
|
||||
return input.concatMap(message -> {
|
||||
return Flux.just(message).doOnNext(originalMessageRef::set)
|
||||
.map(this::resolveArgument)
|
||||
.transform(this.userFunction::apply)
|
||||
.retryBackoff(this.consumerProperties.getMaxAttempts(),
|
||||
Duration.ofMillis(
|
||||
this.consumerProperties.getBackOffInitialInterval()),
|
||||
Duration.ofMillis(
|
||||
this.consumerProperties.getBackOffMaxInterval()))
|
||||
.onErrorResume(e -> {
|
||||
onError(e, originalMessageRef.get());
|
||||
return Mono.empty();
|
||||
});
|
||||
}).map(resultMessage -> toMessage(resultMessage, originalMessageRef.get())); // create
|
||||
// output
|
||||
// message
|
||||
}
|
||||
|
||||
private void onError(Throwable t, Message<I> originalMessage) {
|
||||
if (this.errorChannel != null) {
|
||||
ErrorMessage em = new ErrorMessage(t, originalMessage);
|
||||
logger.error(em);
|
||||
this.errorChannel.send(em);
|
||||
}
|
||||
else {
|
||||
logger.error(t);
|
||||
}
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
|
||||
Message<O> returnMessage;
|
||||
if (this.producerProperties.isUseNativeEncoding()) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug(
|
||||
"Native encoding enabled wrapping result to message using the original message: "
|
||||
+ originalMessage);
|
||||
}
|
||||
returnMessage = wrapOutputToMessage(value, originalMessage);
|
||||
}
|
||||
else {
|
||||
returnMessage = (Message<O>) (value instanceof Message ? value
|
||||
: this.messageConverter.toMessage(value,
|
||||
originalMessage.getHeaders()));
|
||||
if (returnMessage == null
|
||||
&& value.getClass().isAssignableFrom(this.outputClass)) {
|
||||
returnMessage = wrapOutputToMessage(value, originalMessage);
|
||||
}
|
||||
else if (this.bindingServiceProperties != null
|
||||
&& this.bindingServiceProperties.getBindingProperties(
|
||||
this.functionProperties.getOutputDestinationName()) != null
|
||||
&& !returnMessage.getHeaders()
|
||||
.containsKey(MessageHeaders.CONTENT_TYPE)) {
|
||||
|
||||
((Map<String, Object>) ReflectionUtils.getField(MESSAGE_HEADERS_FIELD,
|
||||
returnMessage.getHeaders())).put(
|
||||
MessageHeaders.CONTENT_TYPE,
|
||||
MimeType.valueOf(this.bindingServiceProperties
|
||||
.getBindingProperties("output")
|
||||
.getContentType()));
|
||||
|
||||
}
|
||||
Assert.notNull(returnMessage,
|
||||
"Failed to convert result value '" + value + "' to message.");
|
||||
}
|
||||
return returnMessage;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private <T> Message<O> wrapOutputToMessage(T value, Message<I> originalMessage) {
|
||||
Message<O> returnMessage = (Message<O>) MessageBuilder.withPayload(value)
|
||||
.copyHeaders(originalMessage.getHeaders())
|
||||
.removeHeader(MessageHeaders.CONTENT_TYPE).build();
|
||||
return returnMessage;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private <T> T resolveArgument(Message<I> message) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Resolving input argument from message: " + message);
|
||||
}
|
||||
|
||||
T argument = (T) (shouldConvertFromMessage(message)
|
||||
? this.messageConverter.fromMessage(message, this.inputClass, this.inputParameterizedType) : message);
|
||||
Assert.notNull(argument, "Failed to resolve argument type '" + this.inputClass
|
||||
+ "' from message: " + message);
|
||||
if (this.batchMode
|
||||
&& this.messagePayloadClass != null
|
||||
&& this.isInputArgumentMessage
|
||||
&& argument instanceof Message
|
||||
&& ((Message<?>) argument).getPayload() instanceof List
|
||||
&& !this.messagePayloadClass.isAssignableFrom(((Message<?>) argument).getPayload().getClass())) {
|
||||
argument = (T) MessageBuilder
|
||||
.withPayload(convertListContents(message.getPayload(), this.messagePayloadClass,
|
||||
this.messagePayloadType))
|
||||
.build();
|
||||
}
|
||||
else if (this.isInputArgumentMessage && !(argument instanceof Message)) {
|
||||
if (shouldBatchConvert(argument)) {
|
||||
argument = convertListContents(argument, this.messagePayloadClass, this.messagePayloadType);
|
||||
}
|
||||
argument = (T) MessageBuilder.withPayload(argument)
|
||||
.copyHeaders(message.getHeaders()).build();
|
||||
}
|
||||
else if (!this.isInputArgumentMessage && argument instanceof Message) {
|
||||
argument = ((Message<T>) argument).getPayload();
|
||||
if (shouldBatchConvert(argument)) {
|
||||
argument = convertListContents(argument, this.listContentClass, this.listContentParameterizedType);
|
||||
}
|
||||
}
|
||||
return argument;
|
||||
}
|
||||
|
||||
private <T> boolean shouldBatchConvert(T argument) {
|
||||
return this.batchMode && argument instanceof List && this.listContentClass != null;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private <T> T convertListContents(T argument, Class<?> targetClass, Type hint) {
|
||||
return (T) ((List<?>) argument).stream()
|
||||
.map(payload -> this.messageConverter.fromMessage(MessageBuilder.withPayload(payload).build(),
|
||||
targetClass, hint))
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
|
||||
private boolean shouldConvertFromMessage(Message<?> message) {
|
||||
return !this.inputClass.isAssignableFrom(Message.class)
|
||||
&& !this.inputClass.isAssignableFrom(message.getPayload().getClass())
|
||||
&& !this.inputClass.isAssignableFrom(Object.class);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,289 +0,0 @@
|
||||
/*
|
||||
* Copyright 2018-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
|
||||
*
|
||||
* https://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.Consumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.core.publisher.MonoSink;
|
||||
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import org.springframework.cloud.function.context.FunctionRegistration;
|
||||
import org.springframework.cloud.function.context.FunctionType;
|
||||
import org.springframework.cloud.function.context.catalog.FunctionInspector;
|
||||
import org.springframework.cloud.function.core.FluxSupplier;
|
||||
import org.springframework.cloud.stream.binder.BindingCreatedEvent;
|
||||
import org.springframework.cloud.stream.config.BindingServiceProperties;
|
||||
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
|
||||
import org.springframework.context.support.GenericApplicationContext;
|
||||
import org.springframework.integration.context.IntegrationObjectSupport;
|
||||
import org.springframework.integration.dsl.IntegrationFlowBuilder;
|
||||
import org.springframework.integration.dsl.IntegrationFlows;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageChannel;
|
||||
import org.springframework.messaging.SubscribableChannel;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.StringUtils;
|
||||
|
||||
/**
|
||||
* @author Oleg Zhurakousky
|
||||
* @author David Turanski
|
||||
* @author Ilayaperumal Gopinathan
|
||||
* @since 2.1
|
||||
*/
|
||||
public class IntegrationFlowFunctionSupport {
|
||||
|
||||
private final FunctionCatalog functionCatalog;
|
||||
|
||||
private final FunctionInspector functionInspector;
|
||||
|
||||
private final CompositeMessageConverterFactory messageConverterFactory;
|
||||
|
||||
private final StreamFunctionProperties functionProperties;
|
||||
|
||||
|
||||
//private final AtomicReference<MonoSink<Object>> triggerRef = new AtomicReference<>();
|
||||
|
||||
private final Publisher<Object> trigger;
|
||||
|
||||
private final GenericApplicationContext context;
|
||||
|
||||
IntegrationFlowFunctionSupport(FunctionCatalog functionCatalog,
|
||||
FunctionInspector functionInspector,
|
||||
CompositeMessageConverterFactory messageConverterFactory,
|
||||
StreamFunctionProperties functionProperties,
|
||||
BindingServiceProperties bindingServiceProperties,
|
||||
GenericApplicationContext context) {
|
||||
|
||||
Assert.notNull(functionCatalog, "'functionCatalog' must not be null");
|
||||
Assert.notNull(functionInspector, "'functionInspector' must not be null");
|
||||
Assert.notNull(messageConverterFactory,
|
||||
"'messageConverterFactory' must not be null");
|
||||
Assert.notNull(functionProperties, "'functionProperties' must not be null");
|
||||
this.functionCatalog = functionCatalog;
|
||||
this.functionInspector = functionInspector;
|
||||
this.messageConverterFactory = messageConverterFactory;
|
||||
this.functionProperties = functionProperties;
|
||||
this.context = context;
|
||||
this.functionProperties.setBindingServiceProperties(bindingServiceProperties);
|
||||
AtomicReference<MonoSink<Object>> triggerRef = null;
|
||||
trigger = Mono.create(emmiter -> {
|
||||
triggerRef.set(emmiter);
|
||||
});
|
||||
context.addApplicationListener(event -> {
|
||||
if (event instanceof BindingCreatedEvent) {
|
||||
if (triggerRef.get() != null) {
|
||||
triggerRef.get().success();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Determines if function specified via 'spring.cloud.stream.function.definition'
|
||||
* property can be located in {@link FunctionCatalog}s.
|
||||
* @param <T> type of function
|
||||
* @param typeOfFunction must be Supplier, Function or Consumer
|
||||
* @return {@code true} if function is already stored
|
||||
*/
|
||||
public <T> boolean containsFunction(Class<T> typeOfFunction) {
|
||||
return StringUtils.hasText(this.functionProperties.getDefinition())
|
||||
&& this.catalogContains(typeOfFunction,
|
||||
this.functionProperties.getDefinition());
|
||||
}
|
||||
|
||||
/**
|
||||
* Determines if function specified via 'spring.cloud.stream.function.definition'
|
||||
* property can be located in {@link FunctionCatalog}.
|
||||
* @param <T> type of function
|
||||
* @param typeOfFunction must be Supplier, Function or Consumer
|
||||
* @param functionName the function name to check
|
||||
* @return {@code true} if function is already stored
|
||||
*/
|
||||
public <T> boolean containsFunction(Class<T> typeOfFunction, String functionName) {
|
||||
return StringUtils.hasText(functionName)
|
||||
&& this.catalogContains(typeOfFunction, functionName);
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the function type basing on the function definition.
|
||||
*/
|
||||
public FunctionType getCurrentFunctionType() {
|
||||
FunctionType functionType = this.functionInspector.getRegistration(
|
||||
this.functionCatalog.lookup(this.functionProperties.getDefinition()))
|
||||
.getType();
|
||||
return functionType;
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an instance of the {@link IntegrationFlowBuilder} from a {@link Supplier}
|
||||
* bean available in the context. The name of the bean must be provided via
|
||||
* `spring.cloud.stream.function.definition` property.
|
||||
* @return instance of {@link IntegrationFlowBuilder}
|
||||
* @throws IllegalStateException if the named Supplier can not be located.
|
||||
*/
|
||||
public IntegrationFlowBuilder integrationFlowFromNamedSupplier() {
|
||||
if (StringUtils.hasText(this.functionProperties.getDefinition())) {
|
||||
Supplier<?> supplier = this.functionCatalog.lookup(Supplier.class,
|
||||
this.functionProperties.getDefinition());
|
||||
if (supplier instanceof FluxSupplier) {
|
||||
supplier = ((FluxSupplier<?>) supplier).getTarget();
|
||||
}
|
||||
return integrationFlowFromProvidedSupplier(supplier).split();
|
||||
}
|
||||
|
||||
throw new IllegalStateException(
|
||||
"A Supplier is not specified in the 'spring.cloud.stream.function.definition' property.");
|
||||
}
|
||||
|
||||
/**
|
||||
* Create an instance of the {@link IntegrationFlowBuilder} from a provided
|
||||
* {@link Supplier}.
|
||||
* @param supplier supplier from which the flow builder will be built
|
||||
* @return instance of {@link IntegrationFlowBuilder}
|
||||
*/
|
||||
@SuppressWarnings({ "rawtypes", "unchecked" })
|
||||
public IntegrationFlowBuilder integrationFlowFromProvidedSupplier(
|
||||
Supplier<?> supplier) {
|
||||
String supplierName = this.functionProperties.getDefinition().split("\\|")[0];
|
||||
FunctionRegistration fr = this.functionInspector.getRegistration(this.functionCatalog.lookup(supplierName));
|
||||
if (fr != null && fr.getType().isWrapper()) {
|
||||
Publisher publisher = (Publisher) supplier.get();
|
||||
publisher = publisher instanceof Flux
|
||||
? ((Flux) publisher).delaySubscription(trigger)
|
||||
: ((Mono) publisher).delaySubscription(trigger);
|
||||
|
||||
return IntegrationFlows.from(publisher);
|
||||
}
|
||||
return IntegrationFlows.from(supplier);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param inputChannel channel for which flow we be built
|
||||
* @return instance of {@link IntegrationFlowBuilder}
|
||||
*/
|
||||
public IntegrationFlowBuilder integrationFlowFromChannel(
|
||||
SubscribableChannel inputChannel) {
|
||||
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel);
|
||||
return flowBuilder;
|
||||
}
|
||||
|
||||
/**
|
||||
* @param inputChannel channel for which flow we be built
|
||||
* @param outputChannel channel for which flow we be built
|
||||
* @return instance of {@link IntegrationFlowBuilder}
|
||||
*/
|
||||
public IntegrationFlowBuilder integrationFlowForFunction(
|
||||
SubscribableChannel inputChannel, MessageChannel outputChannel) {
|
||||
|
||||
if (inputChannel instanceof IntegrationObjectSupport) {
|
||||
String inputBindingName = ((IntegrationObjectSupport) inputChannel)
|
||||
.getComponentName();
|
||||
if (StringUtils.hasText(inputBindingName)) {
|
||||
this.functionProperties.setInputDestinationName(inputBindingName);
|
||||
}
|
||||
}
|
||||
|
||||
if (outputChannel instanceof IntegrationObjectSupport) {
|
||||
String outputBindingName = ((IntegrationObjectSupport) outputChannel)
|
||||
.getComponentName();
|
||||
if (StringUtils.hasText(outputBindingName)) {
|
||||
this.functionProperties.setOutputDestinationName(outputBindingName);
|
||||
}
|
||||
}
|
||||
|
||||
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(inputChannel);
|
||||
|
||||
if (!this.andThenFunction(flowBuilder, outputChannel, this.functionProperties)) {
|
||||
flowBuilder = flowBuilder.channel(outputChannel);
|
||||
}
|
||||
return flowBuilder;
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a {@link Function} bean to the end of an integration flow. The name of the bean
|
||||
* must be provided via `spring.cloud.stream.function.definition` property.
|
||||
* <p>
|
||||
* NOTE: If this method returns true, the integration flow is now represented as a
|
||||
* Reactive Streams {@link Publisher} bean.
|
||||
* </p>
|
||||
* @param flowBuilder instance of the {@link IntegrationFlowBuilder} representing the
|
||||
* current state of the integration flow
|
||||
* @param outputChannel channel where the output of a function will be sent
|
||||
* @param functionProperties the function properties
|
||||
* @return true if {@link Function} was located and added and false if it wasn't.
|
||||
*/
|
||||
public boolean andThenFunction(IntegrationFlowBuilder flowBuilder,
|
||||
MessageChannel outputChannel, StreamFunctionProperties functionProperties) {
|
||||
return andThenFunction(flowBuilder.toReactivePublisher(), outputChannel,
|
||||
functionProperties);
|
||||
}
|
||||
|
||||
/**
|
||||
* @param publisher publisher to subscribe to
|
||||
* @param outputChannel output channel to which a message will be sent
|
||||
* @param functionProperties function properties
|
||||
* @param <I> input of the function
|
||||
* @param <O> output of the function
|
||||
* @return whether the function was properly invoked
|
||||
*/
|
||||
public <I, O> boolean andThenFunction(Publisher<?> publisher,
|
||||
MessageChannel outputChannel, StreamFunctionProperties functionProperties) {
|
||||
if (!StringUtils.hasText(functionProperties.getDefinition())) {
|
||||
return false;
|
||||
}
|
||||
FunctionInvoker<I, O> functionInvoker = new FunctionInvoker<>(functionProperties,
|
||||
this.functionCatalog, this.functionInspector,
|
||||
this.messageConverterFactory, this.context.getBeanFactory());
|
||||
|
||||
if (outputChannel != null) {
|
||||
subscribeToInput(functionInvoker, publisher, outputChannel::send);
|
||||
}
|
||||
else {
|
||||
subscribeToInput(functionInvoker, publisher, null);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private <T> boolean catalogContains(Class<T> functionType, String name) {
|
||||
return this.functionCatalog.lookup(functionType, name) != null;
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private <I, O> void subscribeToInput(FunctionInvoker<I, O> functionInvoker,
|
||||
Publisher<?> publisher, Consumer<Message<O>> outputProcessor) {
|
||||
|
||||
Flux<?> inputPublisher = Flux.from(publisher);
|
||||
subscribeToOutput(outputProcessor,
|
||||
functionInvoker.apply((Flux<Message<I>>) inputPublisher)).subscribe();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,836 +0,0 @@
|
||||
/*
|
||||
* Copyright 2018-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
|
||||
*
|
||||
* https://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.lang.reflect.Field;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
|
||||
import org.junit.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
|
||||
import org.springframework.boot.builder.SpringApplicationBuilder;
|
||||
import org.springframework.cloud.function.context.FunctionCatalog;
|
||||
import org.springframework.cloud.function.context.catalog.FunctionInspector;
|
||||
import org.springframework.cloud.stream.annotation.EnableBinding;
|
||||
import org.springframework.cloud.stream.annotation.StreamMessageConverter;
|
||||
import org.springframework.cloud.stream.binder.test.InputDestination;
|
||||
import org.springframework.cloud.stream.binder.test.OutputDestination;
|
||||
import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration;
|
||||
import org.springframework.cloud.stream.config.BindingServiceProperties;
|
||||
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
|
||||
import org.springframework.cloud.stream.function.pojo.Baz;
|
||||
import org.springframework.cloud.stream.function.pojo.ErrorBaz;
|
||||
import org.springframework.cloud.stream.messaging.Processor;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.integration.support.MessageBuilder;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.converter.MessageConverter;
|
||||
import org.springframework.messaging.support.GenericMessage;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Tolga Kavukcu
|
||||
* @author Gary Russell
|
||||
*
|
||||
*/
|
||||
public class FunctionInvokerTests {
|
||||
|
||||
private static String testWithFluxedConsumerValue;
|
||||
|
||||
@Test
|
||||
public void testSimpleEchoConfiguration() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(
|
||||
SimpleEchoConfiguration.class)).web(WebApplicationType.NONE).run(
|
||||
"--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
Message<byte[]> inputMessage = MessageBuilder
|
||||
.withPayload("{\"name\":\"bob\"}".getBytes()).build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getPayload())
|
||||
.isEqualTo("{\"name\":\"bob\"}".getBytes());
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFluxPojoFunction() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(SimpleFluxFunctionConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
Message<byte[]> inputMessage = MessageBuilder
|
||||
.withPayload("{\"name\":\"bob\"}".getBytes()).build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getPayload()).isEqualTo("Person: bob".getBytes());
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFluxMessagePojoFunction() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(
|
||||
SimpleFluxMessageFunctionConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
Message<byte[]> inputMessage = MessageBuilder
|
||||
.withPayload("{\"name\":\"bob\"}".getBytes()).build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getPayload()).isEqualTo("Person: bob".getBytes());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFunctionHonorsOutboundBindingContentType() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(
|
||||
ConverterDoesNotProduceCTConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func",
|
||||
"--spring.cloud.stream.bindings.output.contentType=text/plain")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
Message<byte[]> inputMessage = MessageBuilder
|
||||
.withPayload("{\"name\":\"bob\"}".getBytes())
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)
|
||||
.toString()).isEqualTo("text/plain");
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testFunctionHonorsConverterSetContentType() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(
|
||||
ConverterInjectingCTConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func",
|
||||
"--spring.cloud.stream.bindings.output.contentType=text/plain")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
Message<byte[]> inputMessage = MessageBuilder
|
||||
.withPayload("{\"name\":\"bob\"}".getBytes())
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE)
|
||||
.toString()).isEqualTo("ping/pong");
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSameMessageTypesAreNotConverted() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(MyFunctionsConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.jmx.enabled=false")) {
|
||||
|
||||
Message<Foo> inputMessage = new GenericMessage<>(new Foo());
|
||||
|
||||
StreamFunctionProperties functionProperties = createStreamFunctionProperties();
|
||||
|
||||
functionProperties.setDefinition("messageToMessageSameType");
|
||||
FunctionInvoker<Foo, Foo> messageToMessageSameType = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
Message<Foo> outputMessage = messageToMessageSameType
|
||||
.apply(Flux.just(inputMessage)).blockFirst();
|
||||
assertThat(inputMessage).isSameAs(outputMessage);
|
||||
|
||||
functionProperties.setDefinition("pojoToPojoSameType");
|
||||
FunctionInvoker<Foo, Foo> pojoToPojoSameType = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage))
|
||||
.blockFirst();
|
||||
assertThat(inputMessage.getPayload()).isEqualTo(outputMessage.getPayload());
|
||||
|
||||
functionProperties.setDefinition("messageToMessageNoType");
|
||||
FunctionInvoker<Foo, Foo> messageToMessageNoType = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
outputMessage = messageToMessageNoType.apply(Flux.just(inputMessage))
|
||||
.blockFirst();
|
||||
assertThat(outputMessage).isInstanceOf(Message.class);
|
||||
|
||||
functionProperties.setDefinition("withException");
|
||||
FunctionInvoker<Foo, Foo> withException = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
|
||||
Flux<Message<Foo>> fluxOfMessages = Flux
|
||||
.just(new GenericMessage<>(new ErrorFoo()), inputMessage);
|
||||
Message<Foo> resultMessage = withException.apply(fluxOfMessages).blockFirst();
|
||||
assertThat(resultMessage.getPayload()).isNotInstanceOf(ErrorFoo.class);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testNativeEncodingEnabled() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(MyFunctionsConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.jmx.enabled=false")) {
|
||||
|
||||
Message<Baz> inputMessage = new GenericMessage<>(new Baz());
|
||||
|
||||
StreamFunctionProperties functionProperties = createStreamFunctionPropertiesWithNativeEncoding();
|
||||
|
||||
functionProperties.setDefinition("pojoToPojoNonEmptyPojo");
|
||||
FunctionInvoker<Baz, Baz> pojoToPojoSameType = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
Message<Baz> outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage))
|
||||
.blockFirst();
|
||||
assertThat(inputMessage.getPayload()).isEqualTo(outputMessage.getPayload());
|
||||
|
||||
Message<Baz> inputMessageWithBaz = new GenericMessage<>(new Baz());
|
||||
|
||||
functionProperties.setDefinition("messageToMessageNoType");
|
||||
FunctionInvoker<Baz, Baz> messageToMessageNoType = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
outputMessage = messageToMessageNoType.apply(Flux.just(inputMessageWithBaz))
|
||||
.blockFirst();
|
||||
assertThat(outputMessage).isInstanceOf(Message.class);
|
||||
|
||||
functionProperties.setDefinition("withExceptionNativeEncodingEnabled");
|
||||
FunctionInvoker<Baz, Baz> withException = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
|
||||
Flux<Message<Baz>> fluxOfMessages = Flux
|
||||
.just(new GenericMessage<>(new ErrorBaz()), inputMessage);
|
||||
Message<Baz> resultMessage = withException.apply(fluxOfMessages).blockFirst();
|
||||
assertThat(resultMessage.getPayload()).isNotInstanceOf(ErrorFoo.class);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithOutNativeEncodingEnabled() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(MyFunctionsConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.jmx.enabled=false")) {
|
||||
|
||||
Message<Baz> inputMessage = new GenericMessage<>(new Baz());
|
||||
|
||||
StreamFunctionProperties functionProperties = createStreamFunctionProperties();
|
||||
|
||||
functionProperties.setDefinition("pojoToPojoNonEmptyPojo");
|
||||
FunctionInvoker<Baz, Baz> pojoToPojoSameType = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
Message<Baz> outputMessage = pojoToPojoSameType.apply(Flux.just(inputMessage))
|
||||
.blockFirst();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(inputMessage.getPayload())
|
||||
.isNotEqualTo(outputMessage.getPayload());
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testWithFluxedConsumer() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration
|
||||
.getCompleteConfiguration(MyFunctionsConfiguration.class))
|
||||
.web(WebApplicationType.NONE)
|
||||
.run("--spring.jmx.enabled=false")) {
|
||||
|
||||
String value = "Hello";
|
||||
Message<String> inputMessage = new GenericMessage<>(value);
|
||||
|
||||
StreamFunctionProperties functionProperties = createStreamFunctionProperties();
|
||||
|
||||
functionProperties.setDefinition("fluxConsumer");
|
||||
FunctionInvoker<String, Void> fluxedConsumer = new FunctionInvoker<>(
|
||||
functionProperties,
|
||||
context.getBean(FunctionCatalog.class),
|
||||
context.getBean(FunctionInspector.class),
|
||||
context.getBean(CompositeMessageConverterFactory.class));
|
||||
|
||||
fluxedConsumer.apply(Flux.just(inputMessage)).blockFirst();
|
||||
|
||||
assertThat(testWithFluxedConsumerValue).isEqualTo(value);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testListPayloadConfiguration() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(
|
||||
ListPayloadNotBatchConfiguration.class)).web(WebApplicationType.NONE).run(
|
||||
"--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
Message<byte[]> inputMessage = MessageBuilder
|
||||
.withPayload("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes())
|
||||
.build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getPayload())
|
||||
.isEqualTo("{\"name\":\"bob\"}".getBytes());
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSimpleBatchConfiguration() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(
|
||||
SimpleBatchConfiguration.class)).web(WebApplicationType.NONE).run(
|
||||
"--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func",
|
||||
"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
List<byte[]> list = new ArrayList<>();
|
||||
list.add("{\"name\":\"bob\"}".getBytes());
|
||||
list.add("{\"name\":\"jill\"}".getBytes());
|
||||
Message<List<byte[]>> inputMessage = MessageBuilder
|
||||
.withPayload(list)
|
||||
.build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getPayload())
|
||||
.isEqualTo("{\"name\":\"bob\"}".getBytes());
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testNestedBatchConfiguration() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(
|
||||
NestedBatchConfiguration.class)).web(WebApplicationType.NONE).run(
|
||||
"--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func",
|
||||
"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
List<byte[]> list = new ArrayList<>();
|
||||
list.add("[{\"name\":\"bob\"},{\"name\":\"jill\"}]".getBytes());
|
||||
Message<List<byte[]>> inputMessage = MessageBuilder
|
||||
.withPayload(list)
|
||||
.build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getPayload())
|
||||
.isEqualTo("{\"name\":\"bob\"}".getBytes());
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageBatchConfiguration() {
|
||||
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
|
||||
TestChannelBinderConfiguration.getCompleteConfiguration(
|
||||
MessageBatchConfiguration.class)).web(WebApplicationType.NONE).run(
|
||||
"--spring.jmx.enabled=false",
|
||||
"--spring.cloud.stream.function.definition=func",
|
||||
"--spring.cloud.stream.bindings.input.consumer.batch-mode=true")) {
|
||||
|
||||
InputDestination inputDestination = context.getBean(InputDestination.class);
|
||||
OutputDestination outputDestination = context
|
||||
.getBean(OutputDestination.class);
|
||||
|
||||
List<byte[]> list = new ArrayList<>();
|
||||
list.add("{\"name\":\"bob\"}".getBytes());
|
||||
list.add("{\"name\":\"jill\"}".getBytes());
|
||||
Message<List<byte[]>> inputMessage = MessageBuilder
|
||||
.withPayload(list)
|
||||
.build();
|
||||
inputDestination.send(inputMessage);
|
||||
|
||||
Message<byte[]> outputMessage = outputDestination.receive();
|
||||
assertThat(outputMessage).isNotNull();
|
||||
assertThat(outputMessage.getPayload())
|
||||
.isEqualTo("{\"name\":\"bob\"}".getBytes());
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
private StreamFunctionProperties createStreamFunctionProperties() {
|
||||
StreamFunctionProperties functionProperties = new StreamFunctionProperties();
|
||||
functionProperties.setInputDestinationName("input");
|
||||
functionProperties.setOutputDestinationName("output");
|
||||
BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
|
||||
bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3);
|
||||
try {
|
||||
Field f = ReflectionUtils.findField(StreamFunctionProperties.class,
|
||||
"bindingServiceProperties");
|
||||
f.setAccessible(true);
|
||||
f.set(functionProperties, bindingServiceProperties);
|
||||
return functionProperties;
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private StreamFunctionProperties createStreamFunctionPropertiesWithNativeEncoding() {
|
||||
StreamFunctionProperties functionProperties = new StreamFunctionProperties();
|
||||
functionProperties.setInputDestinationName("input");
|
||||
functionProperties.setOutputDestinationName("output");
|
||||
BindingServiceProperties bindingServiceProperties = new BindingServiceProperties();
|
||||
bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3);
|
||||
bindingServiceProperties.getProducerProperties("output")
|
||||
.setUseNativeEncoding(true);
|
||||
try {
|
||||
Field bspField = ReflectionUtils.findField(StreamFunctionProperties.class,
|
||||
"bindingServiceProperties");
|
||||
bspField.setAccessible(true);
|
||||
bspField.set(functionProperties, bindingServiceProperties);
|
||||
return functionProperties;
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class SimpleEchoConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<Person, Person> func() {
|
||||
return x -> x;
|
||||
}
|
||||
|
||||
public static class Person {
|
||||
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class SimpleFluxFunctionConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<Flux<Person>, Flux<String>> func() {
|
||||
return x -> x.map(person -> person.toString());
|
||||
}
|
||||
|
||||
public static class Person {
|
||||
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "Person: " + name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class SimpleFluxMessageFunctionConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<Flux<Message<Person>>, Flux<Message<String>>> func() {
|
||||
return x -> x.map(personMessage -> {
|
||||
Person person = personMessage.getPayload();
|
||||
Message<String> message = MessageBuilder.withPayload(person.toString())
|
||||
.copyHeaders(personMessage.getHeaders()).build();
|
||||
return message;
|
||||
});
|
||||
}
|
||||
|
||||
public static class Person {
|
||||
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "Person: " + name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class ConverterDoesNotProduceCTConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<String, String> func() {
|
||||
return x -> x;
|
||||
}
|
||||
|
||||
@StreamMessageConverter
|
||||
public MessageConverter customConverter() {
|
||||
return new MessageConverter() {
|
||||
|
||||
@Override
|
||||
public Message<?> toMessage(Object payload, MessageHeaders headers) {
|
||||
return new GenericMessage<byte[]>(((String) payload).getBytes());
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object fromMessage(Message<?> message, Class<?> targetClass) {
|
||||
String contentType = message.getHeaders()
|
||||
.get(MessageHeaders.CONTENT_TYPE).toString();
|
||||
if (contentType.equals("foo/bar")) {
|
||||
return new String((byte[]) message.getPayload());
|
||||
}
|
||||
return null;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class ConverterInjectingCTConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<String, String> func() {
|
||||
return x -> x;
|
||||
}
|
||||
|
||||
@StreamMessageConverter
|
||||
public MessageConverter customConverter() {
|
||||
return new MessageConverter() {
|
||||
|
||||
@Override
|
||||
public Message<?> toMessage(Object payload, MessageHeaders headers) {
|
||||
return MessageBuilder.withPayload(((String) payload).getBytes())
|
||||
.setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object fromMessage(Message<?> message, Class<?> targetClass) {
|
||||
String contentType = message.getHeaders()
|
||||
.get(MessageHeaders.CONTENT_TYPE).toString();
|
||||
if (contentType.equals("foo/bar")) {
|
||||
return new String((byte[]) message.getPayload());
|
||||
}
|
||||
return null;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class ListPayloadNotBatchConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<List<Person>, Person> func() {
|
||||
return x -> x.get(0);
|
||||
}
|
||||
|
||||
public static class Person {
|
||||
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class SimpleBatchConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<List<Person>, Person> func() {
|
||||
return x -> x.get(0);
|
||||
}
|
||||
|
||||
public static class Person {
|
||||
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class NestedBatchConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<List<List<Person>>, Person> func() {
|
||||
return x -> x.get(0).get(0);
|
||||
}
|
||||
|
||||
public static class Person {
|
||||
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
@EnableBinding(Processor.class)
|
||||
public static class MessageBatchConfiguration {
|
||||
|
||||
@Bean
|
||||
public Function<Message<List<Person>>, Person> func() {
|
||||
return x -> x.getPayload().get(0);
|
||||
}
|
||||
|
||||
public static class Person {
|
||||
|
||||
private String name;
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@EnableAutoConfiguration
|
||||
public static class MyFunctionsConfiguration {
|
||||
|
||||
@Bean
|
||||
public Consumer<Flux<String>> fluxConsumer() {
|
||||
return f -> f.subscribe(v -> {
|
||||
System.out.println("Consuming flux: " + v);
|
||||
testWithFluxedConsumerValue = v;
|
||||
});
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Message<Foo>, Message<Bar>> messageToMessageDifferentType() {
|
||||
return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders())
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Message<?>, Message<?>> messageToMessageAnyType() {
|
||||
return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders())
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Message<?>, Message<?>> messageToMessageNoType() {
|
||||
return x -> MessageBuilder.withPayload(new Bar()).copyHeaders(x.getHeaders())
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Message<Foo>, Message<Foo>> messageToMessageSameType() {
|
||||
return x -> x;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Foo, Foo> pojoToPojoSameType() {
|
||||
return x -> x;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Baz, Baz> pojoToPojoNonEmptyPojo() {
|
||||
return x -> x;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Foo, Foo> withException() {
|
||||
return x -> {
|
||||
if (x instanceof ErrorFoo) {
|
||||
System.out.println("Throwing exception ");
|
||||
throw new RuntimeException("Boom!");
|
||||
}
|
||||
else {
|
||||
System.out.println("All is good ");
|
||||
return x;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Function<Baz, Baz> withExceptionNativeEncodingEnabled() {
|
||||
return x -> {
|
||||
if (x instanceof ErrorBaz) {
|
||||
System.out.println("Throwing exception ");
|
||||
throw new RuntimeException("Boom!");
|
||||
}
|
||||
else {
|
||||
System.out.println("All is good ");
|
||||
return x;
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
private static class Foo {
|
||||
|
||||
}
|
||||
|
||||
private static class ErrorFoo extends Foo {
|
||||
|
||||
}
|
||||
|
||||
private static class Bar {
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user