Additional deprecation removals
This commit is contained in:
Oleg Zhurakousky
@@ -1,383 +0,0 @@
|
||||
/*
|
||||
* Copyright 2019-2021 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.function.context;
|
||||
|
||||
import java.io.Closeable;
|
||||
import java.lang.reflect.Type;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.function.Consumer;
|
||||
import java.util.function.Function;
|
||||
import java.util.function.Supplier;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.reactivestreams.Publisher;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.WebApplicationType;
|
||||
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
|
||||
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry;
|
||||
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry.FunctionInvocationWrapper;
|
||||
import org.springframework.cloud.function.context.config.FunctionContextUtils;
|
||||
import org.springframework.cloud.function.context.config.JsonMessageConverter;
|
||||
import org.springframework.cloud.function.context.config.RoutingFunction;
|
||||
import org.springframework.cloud.function.context.config.SmartCompositeMessageConverter;
|
||||
import org.springframework.cloud.function.json.JacksonMapper;
|
||||
import org.springframework.cloud.function.json.JsonMapper;
|
||||
import org.springframework.cloud.function.utils.FunctionClassUtils;
|
||||
import org.springframework.context.ConfigurableApplicationContext;
|
||||
import org.springframework.core.convert.support.GenericConversionService;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
|
||||
/**
|
||||
* Base implementation for adapter initializers and request handlers.
|
||||
*
|
||||
* @param <C> the type of the target specific (native) context object.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
* @since 2.1
|
||||
*
|
||||
* @deprecated since 3.1 in favor of individual implementations of invokers
|
||||
*/
|
||||
@SuppressWarnings("rawtypes")
|
||||
@Deprecated
|
||||
public abstract class AbstractSpringFunctionAdapterInitializer<C> implements Closeable {
|
||||
|
||||
private static Log logger = LogFactory.getLog(AbstractSpringFunctionAdapterInitializer.class);
|
||||
|
||||
/**
|
||||
* Name of the bean for registering the target execution context passed to `initialize(context)` operation.
|
||||
*/
|
||||
public static final String TARGET_EXECUTION_CTX_NAME = "executionContext";
|
||||
|
||||
private final Class<?> configurationClass;
|
||||
|
||||
private Function function;
|
||||
|
||||
private Consumer consumer;
|
||||
|
||||
private Supplier supplier;
|
||||
|
||||
private FunctionRegistration<?> functionRegistration;
|
||||
|
||||
private AtomicBoolean initialized = new AtomicBoolean();
|
||||
|
||||
@Autowired(required = false)
|
||||
protected FunctionCatalog catalog;
|
||||
|
||||
@Autowired(required = false)
|
||||
protected JsonMapper jsonMapper;
|
||||
|
||||
private ConfigurableApplicationContext context;
|
||||
|
||||
public ConfigurableApplicationContext getContext() {
|
||||
return context;
|
||||
}
|
||||
|
||||
public AbstractSpringFunctionAdapterInitializer(Class<?> configurationClass) {
|
||||
Assert.notNull(configurationClass, "'configurationClass' must not be null");
|
||||
this.configurationClass = configurationClass;
|
||||
}
|
||||
|
||||
public AbstractSpringFunctionAdapterInitializer() {
|
||||
this(FunctionClassUtils.getStartClass());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void close() {
|
||||
if (this.context != null) {
|
||||
this.context.close();
|
||||
}
|
||||
}
|
||||
|
||||
protected void initialize(C targetContext) {
|
||||
if (!this.initialized.compareAndSet(false, true)) {
|
||||
return;
|
||||
}
|
||||
logger.info("Initializing: " + this.configurationClass);
|
||||
SpringApplication builder = springApplication();
|
||||
ConfigurableApplicationContext context = builder.run();
|
||||
context.getAutowireCapableBeanFactory().autowireBean(this);
|
||||
this.context = context;
|
||||
if (this.catalog == null) {
|
||||
SmartCompositeMessageConverter messageConverter =
|
||||
new SmartCompositeMessageConverter(Collections.singletonList(new JsonMessageConverter(new JacksonMapper(new ObjectMapper()))));
|
||||
this.catalog = new SimpleFunctionRegistry(new GenericConversionService(),
|
||||
messageConverter, new JacksonMapper(new ObjectMapper()));
|
||||
initFunctionConsumerOrSupplierFromContext(targetContext);
|
||||
}
|
||||
else {
|
||||
initFunctionConsumerOrSupplierFromCatalog(targetContext);
|
||||
}
|
||||
}
|
||||
|
||||
protected Class<?> getInputType() {
|
||||
|
||||
Object func = function();
|
||||
if (func != null && func instanceof FunctionInvocationWrapper) {
|
||||
return FunctionTypeUtils.getRawType(FunctionTypeUtils.getGenericType(((FunctionInvocationWrapper) func).getInputType()));
|
||||
}
|
||||
if (functionRegistration != null) {
|
||||
return FunctionTypeUtils.getRawType(FunctionTypeUtils.getInputType(functionRegistration.getType()));
|
||||
}
|
||||
return Object.class;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
protected Function<Publisher<?>, Publisher<?>> getFunction() {
|
||||
return function;
|
||||
}
|
||||
|
||||
protected Object function() {
|
||||
if (this.function != null) {
|
||||
return this.function;
|
||||
}
|
||||
else if (this.consumer != null) {
|
||||
return this.consumer;
|
||||
}
|
||||
else if (this.supplier != null) {
|
||||
return this.supplier;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
protected Publisher<?> apply(Publisher<?> input) {
|
||||
if (this.function != null) {
|
||||
Object result = this.function.apply(input);
|
||||
if (result instanceof Publisher) {
|
||||
return Flux.from((Publisher) result);
|
||||
}
|
||||
else {
|
||||
return Flux.just(result);
|
||||
}
|
||||
}
|
||||
if (this.consumer != null) {
|
||||
this.consumer.accept(input);
|
||||
return Flux.empty();
|
||||
}
|
||||
if (this.supplier != null) {
|
||||
Object result = this.supplier.get();
|
||||
if (!(result instanceof Publisher)) {
|
||||
result = Mono.just(result);
|
||||
}
|
||||
return (Publisher<?>) result;
|
||||
}
|
||||
throw new IllegalStateException("No function defined");
|
||||
}
|
||||
|
||||
/**
|
||||
* Allows you to resolve function name for cases where it
|
||||
* could not be located under default name.
|
||||
*
|
||||
* Default implementation returns empty string.
|
||||
*
|
||||
* @param targetContext the target context instance
|
||||
* @return the name of the function
|
||||
*/
|
||||
protected String doResolveName(Object targetContext) {
|
||||
return "";
|
||||
}
|
||||
|
||||
protected Object convertOutput(Object input, Object output) {
|
||||
return output;
|
||||
}
|
||||
|
||||
protected <O> O result(Object input, Publisher<?> output) {
|
||||
List<Object> result = new ArrayList<>();
|
||||
for (Object value : Flux.from(output).toIterable()) {
|
||||
result.add(this.convertOutput(input, value));
|
||||
}
|
||||
if (isSingleInput(getFunction(), input) && result.size() == 1) {
|
||||
@SuppressWarnings("unchecked")
|
||||
O value = (O) result.get(0);
|
||||
return value;
|
||||
}
|
||||
if (isSingleOutput(getFunction(), input) && result.size() == 1) {
|
||||
@SuppressWarnings("unchecked")
|
||||
O value = (O) result.get(0);
|
||||
return value;
|
||||
}
|
||||
@SuppressWarnings("unchecked")
|
||||
O value = (O) result;
|
||||
return CollectionUtils.isEmpty(result) ? null : value;
|
||||
}
|
||||
|
||||
protected void clear(String name) {
|
||||
FunctionInvocationWrapper f = this.catalog.lookup(name);
|
||||
if (f.isFunction()) {
|
||||
this.function = f;
|
||||
}
|
||||
else if (f.isConsumer()) {
|
||||
this.consumer = f;
|
||||
}
|
||||
else {
|
||||
this.supplier = f;
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isSingleInput(Function<?, ?> function, Object input) {
|
||||
if (!(input instanceof Collection)) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (function != null) {
|
||||
return Collection.class
|
||||
.isAssignableFrom(((FunctionInvocationWrapper) function).getRawInputType());
|
||||
}
|
||||
return ((Collection<?>) input).size() <= 1;
|
||||
}
|
||||
|
||||
private boolean isSingleOutput(Function<?, ?> function, Object output) {
|
||||
if (!(output instanceof Collection)) {
|
||||
return true;
|
||||
}
|
||||
if (function != null) {
|
||||
Class<?> outputType = FunctionTypeUtils.getRawType(FunctionTypeUtils.getGenericType(((FunctionInvocationWrapper) function).getOutputType()));
|
||||
return Collection.class.isAssignableFrom(outputType);
|
||||
}
|
||||
return ((Collection<?>) output).size() <= 1;
|
||||
}
|
||||
|
||||
private String resolveName(Class<?> type, Object targetContext) {
|
||||
String functionName = context.getEnvironment().getProperty("function.name");
|
||||
if (functionName != null) {
|
||||
return functionName;
|
||||
}
|
||||
else if (type.isAssignableFrom(Function.class)) {
|
||||
return "function";
|
||||
}
|
||||
else if (type.isAssignableFrom(Consumer.class)) {
|
||||
return "consumer";
|
||||
}
|
||||
else if (type.isAssignableFrom(Supplier.class)) {
|
||||
return "supplier";
|
||||
}
|
||||
throw new IllegalStateException("Unknown type " + type);
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "unchecked" })
|
||||
private <T> T getAndInstrumentFromContext(String name) {
|
||||
this.functionRegistration =
|
||||
new FunctionRegistration(context.getBean(name), name);
|
||||
|
||||
Type type = FunctionContextUtils.
|
||||
findType(name, this.context.getBeanFactory());
|
||||
|
||||
this.functionRegistration = functionRegistration.type(type);
|
||||
|
||||
((FunctionRegistry) this.catalog).register(functionRegistration);
|
||||
return this.catalog.lookup(name);
|
||||
}
|
||||
|
||||
private void initFunctionConsumerOrSupplierFromContext(Object targetContext) {
|
||||
String name = resolveName(Function.class, targetContext);
|
||||
if (context.containsBean(name) && context.getBean(name) instanceof Function) {
|
||||
this.function = getAndInstrumentFromContext(name);
|
||||
return;
|
||||
}
|
||||
|
||||
name = resolveName(Consumer.class, targetContext);
|
||||
if (context.containsBean(name) && context.getBean(name) instanceof Consumer) {
|
||||
this.function = getAndInstrumentFromContext(name); // FluxConsumer or any other consumer wrapper is a Function
|
||||
return;
|
||||
}
|
||||
|
||||
name = resolveName(Supplier.class, targetContext);
|
||||
if (context.containsBean(name) && context.getBean(name) instanceof Supplier) {
|
||||
this.supplier = getAndInstrumentFromContext(name);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
private void initFunctionConsumerOrSupplierFromCatalog(Object targetContext) {
|
||||
String name = resolveName(Function.class, targetContext);
|
||||
this.function = this.catalog.lookup(Function.class, name);
|
||||
if (this.function != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
name = resolveName(Consumer.class, targetContext);
|
||||
this.consumer = this.catalog.lookup(Consumer.class, name);
|
||||
if (this.consumer != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
name = resolveName(Supplier.class, targetContext);
|
||||
this.supplier = this.catalog.lookup(Supplier.class, name);
|
||||
if (this.supplier != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
if (this.catalog.size() >= 1 && this.catalog.size() <= 2) { // we may have RoutingFunction function
|
||||
String functionName = this.catalog.getNames(Function.class).stream()
|
||||
.filter(n -> !n.equals(RoutingFunction.FUNCTION_NAME))
|
||||
.findFirst().orElseGet(() -> null);
|
||||
if (functionName != null) {
|
||||
this.function = this.catalog.lookup(Function.class, functionName);
|
||||
return;
|
||||
}
|
||||
functionName = this.catalog.getNames(Supplier.class).stream()
|
||||
.findFirst().orElseGet(() -> null);
|
||||
if (functionName != null) {
|
||||
this.supplier = this.catalog.lookup(Supplier.class, functionName);
|
||||
return;
|
||||
}
|
||||
functionName = this.catalog.getNames(Consumer.class).stream()
|
||||
.findFirst().orElseGet(() -> null);
|
||||
if (functionName != null) {
|
||||
this.consumer = this.catalog.lookup(Consumer.class, functionName);
|
||||
return;
|
||||
}
|
||||
}
|
||||
else {
|
||||
name = this.doResolveName(targetContext);
|
||||
this.function = this.catalog.lookup(Function.class, name);
|
||||
if (this.function != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.consumer = this.catalog.lookup(Consumer.class, name);
|
||||
if (this.consumer != null) {
|
||||
return;
|
||||
}
|
||||
this.supplier = this.catalog.lookup(Supplier.class, name);
|
||||
if (this.supplier != null) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private SpringApplication springApplication() {
|
||||
Class<?> sourceClass = this.configurationClass;
|
||||
SpringApplication application = new org.springframework.cloud.function.context.FunctionalSpringApplication(
|
||||
sourceClass);
|
||||
application.setWebApplicationType(WebApplicationType.NONE);
|
||||
return application;
|
||||
}
|
||||
}
|
||||
@@ -1,29 +0,0 @@
|
||||
/*
|
||||
* Copyright 2012-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.function.context;
|
||||
|
||||
import java.lang.reflect.Type;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*
|
||||
*/
|
||||
public interface WrapperDetector {
|
||||
|
||||
boolean isWrapper(Type type);
|
||||
|
||||
}
|
||||
@@ -1,320 +0,0 @@
|
||||
/*
|
||||
* Copyright 2016-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.function.context.catalog;
|
||||
|
||||
import java.lang.reflect.ParameterizedType;
|
||||
import java.lang.reflect.Type;
|
||||
import java.lang.reflect.WildcardType;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Optional;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
import org.reactivestreams.Publisher;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.core.publisher.Mono;
|
||||
import reactor.util.function.Tuple2;
|
||||
import reactor.util.function.Tuple3;
|
||||
import reactor.util.function.Tuple4;
|
||||
import reactor.util.function.Tuple5;
|
||||
import reactor.util.function.Tuple6;
|
||||
import reactor.util.function.Tuple7;
|
||||
import reactor.util.function.Tuple8;
|
||||
import reactor.util.function.Tuples;
|
||||
|
||||
import org.springframework.cloud.function.context.FunctionRegistration;
|
||||
import org.springframework.core.convert.ConversionService;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.converter.MessageConverter;
|
||||
import org.springframework.messaging.support.MessageBuilder;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
import org.springframework.util.MimeType;
|
||||
import org.springframework.util.ObjectUtils;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
*
|
||||
*/
|
||||
class FunctionTypeConversionHelper {
|
||||
|
||||
private static Log logger = LogFactory.getLog(FunctionTypeConversionHelper.class);
|
||||
|
||||
private final FunctionRegistration<?> functionRegistration;
|
||||
|
||||
private final Type[] functionArgumentTypes;
|
||||
|
||||
private final ConversionService conversionService;
|
||||
|
||||
private final MessageConverter messageConverter;
|
||||
|
||||
FunctionTypeConversionHelper(FunctionRegistration<?> functionRegistration, ConversionService conversionService,
|
||||
MessageConverter messageConverter) {
|
||||
this.conversionService = conversionService;
|
||||
this.messageConverter = messageConverter;
|
||||
this.functionRegistration = functionRegistration;
|
||||
if ((this.functionRegistration.getType()) instanceof ParameterizedType) {
|
||||
this.functionArgumentTypes = ((ParameterizedType) this.functionRegistration.getType())
|
||||
.getActualTypeArguments();
|
||||
}
|
||||
else {
|
||||
this.functionArgumentTypes = new Type[] { FunctionTypeUtils.getInputType(this.functionRegistration.getType()) };
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
Object convertInputIfNecessary(Object input) {
|
||||
List<Object> convertedResults = new ArrayList<Object>();
|
||||
if (input instanceof Tuple2) {
|
||||
convertedResults.add(this.doConvert(((Tuple2) input).getT1(), getInputArgumentType(0)));
|
||||
convertedResults.add(this.doConvert(((Tuple2) input).getT2(), getInputArgumentType(1)));
|
||||
}
|
||||
if (input instanceof Tuple3) {
|
||||
convertedResults.add(this.doConvert(((Tuple3) input).getT3(), getInputArgumentType(2)));
|
||||
}
|
||||
if (input instanceof Tuple4) {
|
||||
convertedResults.add(this.doConvert(((Tuple4) input).getT4(), getInputArgumentType(3)));
|
||||
}
|
||||
if (input instanceof Tuple5) {
|
||||
convertedResults.add(this.doConvert(((Tuple5) input).getT5(), getInputArgumentType(4)));
|
||||
}
|
||||
if (input instanceof Tuple6) {
|
||||
convertedResults.add(this.doConvert(((Tuple6) input).getT6(), getInputArgumentType(5)));
|
||||
}
|
||||
if (input instanceof Tuple7) {
|
||||
convertedResults.add(this.doConvert(((Tuple7) input).getT7(), getInputArgumentType(6)));
|
||||
}
|
||||
if (input instanceof Tuple8) {
|
||||
convertedResults.add(this.doConvert(((Tuple8) input).getT8(), getInputArgumentType(7)));
|
||||
}
|
||||
|
||||
input = CollectionUtils.isEmpty(convertedResults) ? this.doConvert(input, getInputArgumentType(0))
|
||||
: Tuples.fromArray(convertedResults.toArray());
|
||||
return input;
|
||||
}
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
Object convertOutputIfNecessary(Object output, MimeType... acceptedOutputTypes) {
|
||||
if (ObjectUtils.isEmpty(acceptedOutputTypes)) {
|
||||
return output;
|
||||
}
|
||||
List<Object> convertedResults = new ArrayList<Object>();
|
||||
if (output instanceof Tuple2) {
|
||||
convertedResults.add(this.doConvert(((Tuple2) output).getT1(), acceptedOutputTypes[0]));
|
||||
convertedResults.add(this.doConvert(((Tuple2) output).getT2(), acceptedOutputTypes[1]));
|
||||
}
|
||||
if (output instanceof Tuple3) {
|
||||
convertedResults.add(this.doConvert(((Tuple3) output).getT3(), acceptedOutputTypes[2]));
|
||||
}
|
||||
if (output instanceof Tuple4) {
|
||||
convertedResults.add(this.doConvert(((Tuple4) output).getT4(), acceptedOutputTypes[3]));
|
||||
}
|
||||
if (output instanceof Tuple5) {
|
||||
convertedResults.add(this.doConvert(((Tuple5) output).getT5(), acceptedOutputTypes[4]));
|
||||
}
|
||||
if (output instanceof Tuple6) {
|
||||
convertedResults.add(this.doConvert(((Tuple6) output).getT6(), acceptedOutputTypes[5]));
|
||||
}
|
||||
if (output instanceof Tuple7) {
|
||||
convertedResults.add(this.doConvert(((Tuple7) output).getT7(), acceptedOutputTypes[6]));
|
||||
}
|
||||
if (output instanceof Tuple8) {
|
||||
convertedResults.add(this.doConvert(((Tuple8) output).getT8(), acceptedOutputTypes[7]));
|
||||
}
|
||||
|
||||
output = Tuples.fromArray(convertedResults.toArray());
|
||||
|
||||
return output;
|
||||
}
|
||||
|
||||
int getInputArgumentCount() {
|
||||
Type[] types = ((ParameterizedType) this.functionArgumentTypes[0]).getActualTypeArguments();
|
||||
return types.length;
|
||||
}
|
||||
|
||||
Object getInputArgument(int index) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
Class<?> getInputArgumentRawType(int index) {
|
||||
Type[] types = ((ParameterizedType) this.functionArgumentTypes[0]).getActualTypeArguments();
|
||||
return (Class<?>) ((ParameterizedType) types[index]).getRawType();
|
||||
}
|
||||
|
||||
Type getInputArgumentType(int index) {
|
||||
if (this.functionArgumentTypes[0] instanceof ParameterizedType
|
||||
&& (Publisher.class.isAssignableFrom((Class<?>) ((ParameterizedType) this.functionArgumentTypes[0]).getRawType())
|
||||
|| ((ParameterizedType) this.functionArgumentTypes[0]).getTypeName().startsWith("reactor.util.function.Tuple"))) {
|
||||
Type[] types = ((ParameterizedType) this.functionArgumentTypes[0]).getActualTypeArguments();
|
||||
|
||||
return (types[index]);
|
||||
}
|
||||
else {
|
||||
return this.functionArgumentTypes[0];
|
||||
}
|
||||
}
|
||||
|
||||
Type getOutputArgumentType(int index) {
|
||||
if (this.functionArgumentTypes[1] instanceof ParameterizedType) {
|
||||
Type[] types = ((ParameterizedType) this.functionArgumentTypes[1]).getActualTypeArguments();
|
||||
|
||||
return (types[index]);
|
||||
}
|
||||
else {
|
||||
return this.functionArgumentTypes[1];
|
||||
}
|
||||
}
|
||||
|
||||
private Class<?> getRawType(Type targetType) {
|
||||
Class<?> rawType;
|
||||
if (targetType instanceof ParameterizedType) {
|
||||
if (Publisher.class.isAssignableFrom((Class<?>) ((ParameterizedType) targetType).getRawType())
|
||||
|| Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) targetType).getRawType())) {
|
||||
|
||||
if (((ParameterizedType) targetType).getActualTypeArguments()[0] instanceof ParameterizedType) {
|
||||
return this.getRawType(((ParameterizedType) targetType).getActualTypeArguments()[0]);
|
||||
}
|
||||
else {
|
||||
rawType = (Class<?>) ((ParameterizedType) targetType).getActualTypeArguments()[0];
|
||||
}
|
||||
}
|
||||
else {
|
||||
rawType = (Class<?>) ((ParameterizedType) targetType).getRawType();
|
||||
}
|
||||
}
|
||||
else {
|
||||
if (targetType instanceof WildcardType) {
|
||||
rawType = Object.class;
|
||||
}
|
||||
else {
|
||||
rawType = (Class<?>) targetType;
|
||||
}
|
||||
}
|
||||
return rawType;
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
private Object doConvert(Object incoming, Type targetType) {
|
||||
Class<?> actualType = this.getRawType(targetType);
|
||||
if (incoming instanceof Publisher) {
|
||||
if (!actualType.isAssignableFrom(Void.class)) {
|
||||
incoming = incoming instanceof Mono
|
||||
? Mono.from((Publisher) incoming)
|
||||
.map(value -> this.doConvertArgument(value, targetType, actualType))
|
||||
.doOnError(System.out::println)
|
||||
: Flux.from((Publisher) incoming)
|
||||
.map(value -> this.doConvertArgument(value, targetType, actualType))
|
||||
.doOnError(System.out::println);
|
||||
|
||||
}
|
||||
}
|
||||
else {
|
||||
Assert.isTrue(!FunctionTypeUtils.isPublisher(FunctionTypeUtils.getInputType(this.functionRegistration.getType())),
|
||||
"Invoking reactive function as imperative is not allowed. Function name(s): "
|
||||
+ this.functionRegistration.getNames());
|
||||
incoming = this.doConvertArgument(incoming, targetType, actualType);
|
||||
}
|
||||
return incoming;
|
||||
}
|
||||
|
||||
@SuppressWarnings({ "unchecked", "rawtypes" })
|
||||
private Object doConvert(Object incoming, MimeType mimeType) {
|
||||
MessageHeaders headers = new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, mimeType));
|
||||
if (incoming instanceof Publisher) {
|
||||
incoming = incoming instanceof Mono
|
||||
? Mono.from((Publisher) incoming).map(value -> this.messageConverter.toMessage(value, headers))
|
||||
: Flux.from((Publisher) incoming).map(value -> this.messageConverter.toMessage(value, headers));
|
||||
}
|
||||
else {
|
||||
Assert.isTrue(!FunctionTypeUtils.isPublisher(FunctionTypeUtils.getInputType(this.functionRegistration.getType())),
|
||||
"Invoking reactive function as imperative is not allowed. Function name(s): "
|
||||
+ this.functionRegistration.getNames());
|
||||
incoming = this.messageConverter.toMessage(incoming, headers);
|
||||
}
|
||||
return incoming;
|
||||
}
|
||||
|
||||
private Object doConvertArgument(Object incomingValue, Type targetType, Class<?> actualInputType) {
|
||||
if (!Void.class.isAssignableFrom(actualInputType)) {
|
||||
if (incomingValue instanceof Message<?>) {
|
||||
incomingValue = this.isMessage(targetType)
|
||||
? this.fromMessageToMessage((Message<?>) incomingValue, actualInputType)
|
||||
: this.fromMessageToValue((Message<?>) incomingValue, actualInputType);
|
||||
}
|
||||
else {
|
||||
if (!incomingValue.getClass().isAssignableFrom(actualInputType)) {
|
||||
Assert.isTrue(this.conversionService.canConvert(incomingValue.getClass(), actualInputType),
|
||||
"Failed to convert value of type " + incomingValue.getClass() + " to " + targetType);
|
||||
incomingValue = this.conversionService.convert(incomingValue, actualInputType);
|
||||
}
|
||||
}
|
||||
}
|
||||
else {
|
||||
incomingValue = null;
|
||||
}
|
||||
return incomingValue;
|
||||
}
|
||||
|
||||
private boolean isMessage(Type targetType) {
|
||||
if (targetType instanceof ParameterizedType) {
|
||||
return Message.class.isAssignableFrom((Class<?>) ((ParameterizedType) targetType).getRawType());
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/*
|
||||
* Will conditionally convert Message's payload to a targetType unless such
|
||||
* payload is already of that type.
|
||||
*/
|
||||
private Object fromMessageToValue(Message<?> incomingMessage, Class<?> targetType) {
|
||||
Object incomingValue = ((Message<?>) incomingMessage).getPayload();
|
||||
if (!incomingValue.getClass().isAssignableFrom(targetType)) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Converting message '" + incomingMessage + "' with payload of type '"
|
||||
+ incomingMessage.getPayload().getClass().getName() + "' to value of type '"
|
||||
+ targetType.getName() + "' for invocation of " + functionRegistration.getNames());
|
||||
}
|
||||
if (incomingMessage.getPayload() instanceof Optional && !((Optional<?>) incomingMessage.getPayload()).isPresent()) {
|
||||
incomingValue = incomingMessage;
|
||||
}
|
||||
else {
|
||||
incomingValue = this.messageConverter.fromMessage(incomingMessage, targetType);
|
||||
}
|
||||
}
|
||||
return incomingValue;
|
||||
}
|
||||
|
||||
/*
|
||||
* Will conditionally convert Message's payload to a targetType unless such
|
||||
* payload is already of that type wrapping the result of conversion into a
|
||||
* Message with converted type as a payload.
|
||||
*/
|
||||
private Message<?> fromMessageToMessage(Message<?> incomingMessage, Class<?> targetType) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("Converting message '" + incomingMessage + "' with payload of type '"
|
||||
+ incomingMessage.getPayload().getClass().getName() + "' to message with payload of type '"
|
||||
+ targetType.getName() + "' for invocation of " + functionRegistration.getNames());
|
||||
}
|
||||
return MessageBuilder.withPayload(this.fromMessageToValue(incomingMessage, targetType))
|
||||
.copyHeaders(incomingMessage.getHeaders()).build();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,43 +0,0 @@
|
||||
/*
|
||||
* Copyright 2019-2019 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* 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.function.context.catalog;
|
||||
|
||||
import java.util.function.Consumer;
|
||||
|
||||
import org.reactivestreams.Publisher;
|
||||
import reactor.core.publisher.Flux;
|
||||
|
||||
import org.springframework.messaging.Message;
|
||||
|
||||
/**
|
||||
* @author Dave Syer
|
||||
*/
|
||||
public class MessageConsumer implements Consumer<Publisher<Message<?>>> {
|
||||
|
||||
private Consumer<Object> delegate;
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
public MessageConsumer(Consumer<?> input) {
|
||||
this.delegate = (Consumer<Object>) input;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void accept(Publisher<Message<?>> input) {
|
||||
Flux.from(input).map(Message::getPayload).subscribe(this.delegate::accept);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,190 +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.function.context.config;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.lang.reflect.Field;
|
||||
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;
|
||||
import org.springframework.messaging.Message;
|
||||
import org.springframework.messaging.MessageHeaders;
|
||||
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
|
||||
import org.springframework.messaging.converter.MessageConversionException;
|
||||
import org.springframework.util.MimeType;
|
||||
import org.springframework.util.ReflectionUtils;
|
||||
|
||||
/**
|
||||
* Variation of {@link MappingJackson2MessageConverter} to support marshalling and
|
||||
* unmarshalling of Messages's payload from 'String' or 'byte[]' to an instance of a
|
||||
* 'targetClass' and and back to 'byte[]'.
|
||||
*
|
||||
* @author Oleg Zhurakousky
|
||||
* @author Gary Russell
|
||||
* @since 2.0
|
||||
*/
|
||||
class ApplicationJsonMessageMarshallingConverter extends MappingJackson2MessageConverter {
|
||||
|
||||
private final Field headersField;
|
||||
|
||||
private final Map<Type, JavaType> typeCache = new ConcurrentHashMap<>();
|
||||
|
||||
ApplicationJsonMessageMarshallingConverter(@Nullable ObjectMapper objectMapper) {
|
||||
if (objectMapper != null) {
|
||||
this.setObjectMapper(objectMapper);
|
||||
}
|
||||
this.headersField = ReflectionUtils.findField(MessageHeaders.class, "headers");
|
||||
this.headersField.setAccessible(true);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Object convertToInternal(Object payload, @Nullable MessageHeaders headers,
|
||||
@Nullable Object conversionHint) {
|
||||
if (payload instanceof byte[]) {
|
||||
return payload;
|
||||
}
|
||||
else if (payload instanceof String) {
|
||||
return ((String) payload).getBytes(StandardCharsets.UTF_8);
|
||||
}
|
||||
else {
|
||||
return super.convertToInternal(payload, headers, conversionHint);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Object convertFromInternal(Message<?> message, Class<?> targetClass, @Nullable Object hint) {
|
||||
Object conversionHint = hint;
|
||||
Object result = null;
|
||||
if (conversionHint instanceof MethodParameter) {
|
||||
Class<?> conversionHintType = ((MethodParameter) conversionHint)
|
||||
.getParameterType();
|
||||
if (Message.class.isAssignableFrom(conversionHintType)) {
|
||||
/*
|
||||
* Ensures that super won't attempt to create Message as a result of
|
||||
* conversion and stays at payload conversion only. The Message will
|
||||
* eventually be created in
|
||||
* MessageMethodArgumentResolver.resolveArgument(..)
|
||||
*/
|
||||
conversionHint = null;
|
||||
}
|
||||
else if (((MethodParameter) conversionHint)
|
||||
.getGenericParameterType() instanceof ParameterizedType) {
|
||||
ParameterizedTypeReference<Object> forType = ParameterizedTypeReference
|
||||
.forType(((MethodParameter) conversionHint)
|
||||
.getGenericParameterType());
|
||||
result = convertParameterizedType(message, forType.getType());
|
||||
}
|
||||
}
|
||||
else if (conversionHint instanceof ParameterizedTypeReference) {
|
||||
result = convertParameterizedType(message, ((ParameterizedTypeReference<?>) conversionHint).getType());
|
||||
}
|
||||
else if (conversionHint instanceof ParameterizedType) {
|
||||
result = convertParameterizedType(message, (Type) conversionHint);
|
||||
}
|
||||
if (result == null) {
|
||||
if (message.getPayload() instanceof byte[]
|
||||
&& targetClass.isAssignableFrom(String.class)) {
|
||||
result = new String((byte[]) message.getPayload(),
|
||||
StandardCharsets.UTF_8);
|
||||
}
|
||||
else {
|
||||
result = super.convertFromInternal(message, targetClass, conversionHint);
|
||||
}
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
private Object convertParameterizedType(Message<?> message, Type conversionHint) {
|
||||
ObjectMapper objectMapper = this.getObjectMapper();
|
||||
Object payload = message.getPayload();
|
||||
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);
|
||||
}
|
||||
if (payload instanceof byte[]) {
|
||||
return objectMapper.readValue((byte[]) payload, type);
|
||||
}
|
||||
else if (payload instanceof String) {
|
||||
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());
|
||||
}
|
||||
else {
|
||||
// fall back to simple type-conversion
|
||||
// see https://github.com/spring-cloud/spring-cloud-stream/issues/1898
|
||||
return objectMapper.convertValue(value, typeToUse.getContentType());
|
||||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
logger.error("Failed to convert payload " + value, e);
|
||||
}
|
||||
return null;
|
||||
}).collect(Collectors.toList());
|
||||
|
||||
return collection;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
}
|
||||
catch (IOException e) {
|
||||
throw new MessageConversionException("Cannot parse payload ", e);
|
||||
}
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
@Override
|
||||
@Nullable
|
||||
protected MimeType getMimeType(@Nullable MessageHeaders headers) {
|
||||
Object contentType = headers.get(MessageHeaders.CONTENT_TYPE);
|
||||
if (contentType instanceof byte[]) {
|
||||
contentType = new String((byte[]) contentType, StandardCharsets.UTF_8);
|
||||
contentType = ((String) contentType).replace("\"", "");
|
||||
Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils.getField(this.headersField, headers);
|
||||
headersMap.put(MessageHeaders.CONTENT_TYPE, contentType);
|
||||
}
|
||||
return super.getMimeType(headers);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user