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

Refactor and simplify Observability related code

Browse Source
This commit is contained in:
Oleg Zhurakousky
2022-10-24 17:14:41 +02:00
parent 5cfa934da3
commit 09b1317620
4 changed files with 19 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionAroundWrapper.java
View File

@@ -43,15 +43,8 @@ public abstract class FunctionAroundWrapper {
String functionalTracingEnabledStr = System.getProperty("spring.cloud.function.observability.enabled");
boolean functionalTracingEnabled = StringUtils.hasText(functionalTracingEnabledStr)
? Boolean.parseBoolean(functionalTracingEnabledStr) : true;
if (functionalTracingEnabled && !(input instanceof Publisher) && input instanceof Message) {
boolean isSkipOutputConversion = targetFunction.isSkipOutputConversion();
targetFunction.setSkipOutputConversion(true);
try {
return this.doApply(input, targetFunction);
}
finally {
targetFunction.setSkipOutputConversion(isSkipOutputConversion);
}
if (functionalTracingEnabled && !(input instanceof Publisher) && input instanceof Message && !FunctionTypeUtils.isCollectionOfMessage(targetFunction.getOutputType())) {
return this.doApply(input, targetFunction);
}
else {
return targetFunction.apply(input);

1
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java
View File

@@ -99,6 +99,7 @@ public final class FunctionTypeUtils {
return true;
}
type = getGenericType(type);
type = type == null ? Object.class : type;
Class<?> rawType = type instanceof ParameterizedType ? getRawType(type) : (Class<?>) type;
return Collection.class.isAssignableFrom(rawType) || JsonNode.class.isAssignableFrom(rawType);
}

58
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java
View File

@@ -224,10 +224,6 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
logger.debug("Function '" + functionDefinition + "' is not found in cache");
}
if (function != null) {
function = this.wrapInAroundAdviceIfNecessary(function);
}
return (T) function;
}
@@ -258,28 +254,6 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
return functionDefinition;
}
/**
* This is primarily to support spring-cloud-sleauth.
* There is no current use cases in functions where it is used.
* The approach may change in the future.
*/
private FunctionInvocationWrapper wrapInAroundAdviceIfNecessary(FunctionInvocationWrapper function) {
FunctionInvocationWrapper wrappedFunction = function;
if (function != null && this.functionAroundWrapper != null) {
wrappedFunction = new FunctionInvocationWrapper(function) {
@Override
Object doApply(Object input) {
if (logger.isDebugEnabled()) {
logger.debug("Executing around advise(s): " + functionAroundWrapper);
}
return functionAroundWrapper.apply(input, function);
}
};
}
return wrappedFunction;
}
/*
*
*/
@@ -429,6 +403,8 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
private boolean propagateInputHeaders;
private boolean wrapped;
/*
* This is primarily to support Stream's ability to access
* un-converted payload (e.g., to evaluate expression on some attribute of a payload)
@@ -437,20 +413,6 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
*/
private Function<Object, Object> enhancer;
FunctionInvocationWrapper(FunctionInvocationWrapper function) {
this.expectedOutputContentType = function.expectedOutputContentType;
this.skipOutputConversion = function.skipOutputConversion;
this.skipInputConversion = function.skipInputConversion;
this.target = function.target;
this.propagateInputHeaders = function.propagateInputHeaders;
this.composed = function.composed;
this.inputType = function.inputType;
this.composed = function.composed;
this.outputType = function.outputType;
this.functionDefinition = function.functionDefinition;
this.message = this.inputType != null && FunctionTypeUtils.isMessage(this.inputType);
}
FunctionInvocationWrapper(String functionDefinition, Object target, Type inputType, Type outputType) {
this.target = target;
this.inputType = this.normalizeType(inputType);
@@ -559,12 +521,14 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
if (logger.isDebugEnabled() && !(input instanceof Publisher)) {
logger.debug("Invoking function " + this);
}
//Object result = (this.getTarget() instanceof PassThruFunction) ? input : this.doApply(input);
Object result = this.doApply(input);
if (result != null && this.outputType != null) {
result = this.convertOutputIfNecessary(result, this.outputType, this.expectedOutputContentType);
Object result;
if (functionAroundWrapper != null && !this.wrapped) {
this.wrapped = true;
result = functionAroundWrapper.apply(input, this);
}
else {
result = this.doApply(input);
}
return result;
@@ -720,6 +684,10 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
else { // Function
result = this.invokeFunction(convertedInput);
}
if (result != null && this.outputType != null) {
result = this.convertOutputIfNecessary(result, this.outputType, this.expectedOutputContentType);
}
return result;
}

10
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/observability/ObservationFunctionAroundWrapper.java
View File

@@ -20,7 +20,6 @@ import io.micrometer.observation.Observation;
import io.micrometer.observation.ObservationRegistry;
import org.springframework.cloud.function.context.catalog.FunctionAroundWrapper;
import org.springframework.cloud.function.context.catalog.FunctionTypeUtils;
import org.springframework.cloud.function.context.catalog.SimpleFunctionRegistry;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
@@ -43,18 +42,15 @@ public class ObservationFunctionAroundWrapper extends FunctionAroundWrapper {
@Override
protected Object doApply(Object message, SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction) {
if (FunctionTypeUtils.isCollectionOfMessage(targetFunction.getOutputType())) {
return targetFunction.apply(message); // no instrumentation
}
return nonReactorStream((Message<?>) message, targetFunction);
}
private Object nonReactorStream(Message<?> message,
SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction) {
if (targetFunction.isConsumer() || targetFunction.isFunction()) {
//if (targetFunction.isConsumer() || targetFunction.isFunction()) {
return functionProcessingObservation(targetFunction, message).observe(() -> targetFunction.apply(message));
}
return functionProcessingObservation(targetFunction, message).observe(targetFunction::get);
// }
// return functionProcessingObservation(targetFunction, message).observe(targetFunction::get);
}
private Observation functionProcessingObservation(SimpleFunctionRegistry.FunctionInvocationWrapper targetFunction, Message<?> message) {