diff --git a/spring-cloud-function-context/pom.xml b/spring-cloud-function-context/pom.xml index e06edc5f7..a482e4fe1 100644 --- a/spring-cloud-function-context/pom.xml +++ b/spring-cloud-function-context/pom.xml @@ -42,7 +42,10 @@ org.springframework spring-web - true + + + + tools.jackson.module diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java index ac8f54878..33d7548d0 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java @@ -79,6 +79,7 @@ import org.springframework.util.MimeTypeUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; +import org.springframework.web.server.ResponseStatusException; /** * Implementation of {@link FunctionCatalog} and {@link FunctionRegistry} which @@ -939,7 +940,19 @@ private Object fluxifyInputIfNecessary(Object input) { logger.debug("Actual input represents a collection while input type of the function does not represent a collection. " + "Therefore framework will attempt invoke function for each element in the collection."); MessageHeaders headers = input instanceof Message ? ((Message) input).getHeaders() : new MessageHeaders(Collections.emptyMap()); - Collection collectionPayload = jsonMapper.fromJson(payload, Collection.class); + // KOMUNE Modification + // Collection Type is needed by kotlin serializer to deserialize object + // Original version: + // Collection collectionPayload = jsonMapper.fromJson(payload, Collection.class); + Type genType = FunctionTypeUtils.getGenericType(this.inputType); + ResolvableType resolvableType = ResolvableType.forType(genType); + if (resolvableType.toClass() == Message.class) { + resolvableType = resolvableType.getGeneric(0); + } + ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, resolvableType); + Collection collectionPayload = jsonMapper.fromJson(payload, listType.getType()); + // KOMUNE End Of Modification + Class inputClass = FunctionTypeUtils.getRawType(this.inputType); if (this.isInputTypeMessage()) { inputClass = FunctionTypeUtils.getRawType(FunctionTypeUtils.getImmediateGenericType(this.inputType, 0)); @@ -1598,6 +1611,12 @@ else if (FunctionTypeUtils.isFlux(type) && publisher instanceof Mono) { try { return this.convertInputIfNecessary(v, actualType == null ? type : actualType); } + // KOMUNE Modification + // force message conversion error propagation + catch (ResponseStatusException e) { + throw e; + } + // KOMUNE End Of Modification catch (Exception e) { throw new IllegalStateException("Failed to convert input", e); } @@ -1606,6 +1625,12 @@ else if (FunctionTypeUtils.isFlux(type) && publisher instanceof Mono) { try { return this.convertInputIfNecessary(v, actualType == null ? type : actualType); } + // KOMUNE Modification + // force message conversion error propagation + catch (ResponseStatusException e) { + throw e; + } + // KOMUNE End Of Modification catch (Exception e) { throw new IllegalStateException("Failed to convert input", e); } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java index 08e09dd55..7a669ac4a 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java @@ -217,6 +217,14 @@ public static class JsonMapperConfiguration { public JsonMapper jsonMapper(ApplicationContext context) { String preferredMapper = context.getEnvironment().getProperty(JSON_MAPPER_PROPERTY); if (StringUtils.hasText(preferredMapper)) { + // KOMUNE Modification + // Kept commented out here: KSerializationMapper lives in f2 (f2.spring.KSerializationMapper) + // and is not on this repository's classpath. The f2 vendored copy of this file enables it. + // if ("kSerialization".equals(preferredMapper) && ClassUtils.isPresent("kotlinx.serialization.json.Json", null)) { + // return kSerialization(context); + // } + // else + // KOMUNE End Of Modification if ("gson".equals(preferredMapper)) { return gson(context); } diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java index 36484a6c9..7775993c0 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/JsonMessageConverter.java @@ -26,12 +26,16 @@ import java.util.ArrayList; import java.util.Enumeration; import java.util.List; +import java.util.Locale; + +import tools.jackson.databind.DatabindException; import org.springframework.cloud.function.cloudevent.CloudEventMessageUtils; import org.springframework.cloud.function.json.JsonMapper; import org.springframework.core.GenericTypeResolver; import org.springframework.core.MethodParameter; import org.springframework.core.ParameterizedTypeReference; +import org.springframework.http.HttpStatus; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHeaders; @@ -41,6 +45,7 @@ import org.springframework.util.CollectionUtils; import org.springframework.util.MimeType; import org.springframework.util.StringUtils; +import org.springframework.web.server.ResponseStatusException; /** * Implementation of {@link MessageConverter} which uses Jackson or Gson libraries to do the @@ -145,6 +150,12 @@ protected Object convertFromInternal(Message message, Class targetClass, @ if (message.getPayload() instanceof byte[] && String.class.isAssignableFrom(targetClass)) { return new String((byte[]) message.getPayload(), StandardCharsets.UTF_8); } + // KOMUNE Modification + // force message conversion error propagation + else if (isJsonContentType(message.getHeaders()) && e.getCause() instanceof DatabindException) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "Error parsing json", e.getCause()); + } + // KOMUNE End Of Modification else if (logger.isDebugEnabled()) { Object payload = message.getPayload(); if (payload instanceof byte[]) { @@ -167,6 +178,20 @@ protected Object convertToInternal(Object payload, @Nullable MessageHeaders head return jsonMapper.toJson(payload); } + // KOMUNE Modification + private static boolean isJsonContentType(MessageHeaders headers) { + Object contentType = headers.get(MessageHeaders.CONTENT_TYPE); + if (contentType == null) { + contentType = headers.get("Content-Type"); + } + if (contentType == null) { + contentType = headers.get("content-type"); + } + return contentType != null + && String.valueOf(contentType).toLowerCase(Locale.ROOT).contains("application/json"); + } + // KOMUNE End Of Modification + private Type getResolvedType(Class targetClass, @Nullable Object conversionHint) { if (conversionHint instanceof MethodParameter param) { param = param.nestedIfOptional(); diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java index 8c9419300..b8febb4d2 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/KotlinLambdaToFunctionAutoConfiguration.java @@ -18,6 +18,8 @@ import java.lang.reflect.ParameterizedType; import java.lang.reflect.Type; +import java.util.ArrayList; +import java.util.List; import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; @@ -213,28 +215,47 @@ private boolean isValidKotlinFunction(Type functionType, Type[] type) { !isTypeRepresentedByClass(type[1], Unit.class); } + // KOMUNE Modification + // Relaxed arity checks so Kotlin suspend lambdas declared without an explicit + // Continuation/Flow parameter are still detected, and match supertypes as well. private boolean isValidKotlinSuspendSupplier(Type functionType, Type[] type) { return isTypeRepresentedByClass(functionType, Function1.class) && - type.length == 2 && - CoroutinesUtils.isContinuationFlowType(type[0]); + (type.length == 1 || (type.length == 2 && CoroutinesUtils.isContinuationFlowType(type[0]))); } private boolean isValidKotlinSuspendConsumer(Type functionType, Type[] type) { return isTypeRepresentedByClass(functionType, Function2.class) && - type.length == 3 && - CoroutinesUtils.isFlowType(type[0]) && - CoroutinesUtils.isContinuationUnitType(type[1]); + ( + type.length == 1 || + (type.length == 3 && CoroutinesUtils.isContinuationUnitType(type[1]) && CoroutinesUtils.isFlowType(type[0])) + ); } private boolean isValidKotlinSuspendFunction(Type functionType, Type[] type) { return isTypeRepresentedByClass(functionType, Function2.class) && - type.length == 3 && - CoroutinesUtils.isContinuationFlowType(type[1]); + ((type.length == 3 && CoroutinesUtils.isContinuationFlowType(type[1])) + || type.length == 2); } private boolean isTypeRepresentedByClass(Type type, Class clazz) { - return type.getTypeName().contains(clazz.getName()); + return concatWithSuperTypes(type) + .stream() + .map(Type::getTypeName) + .map(s -> s.contains(clazz.getName())) + .reduce((a, b) -> a || b) + .orElse(false); + } + + private List concatWithSuperTypes(Type type) { + List all = new ArrayList<>(); + all.add(type); + if (type instanceof ParameterizedType) { + Class[] interfaces = ((Class) ((ParameterizedType) type).getRawType()).getInterfaces(); + all.addAll(List.of(interfaces)); + } + return all; } + // KOMUNE End Of Modification public Class getObjectType() { return FunctionRegistration.class; diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java index bbf9fb351..6be0cc36b 100644 --- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java +++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java @@ -40,8 +40,11 @@ import org.springframework.messaging.support.MessageHeaderAccessor; import org.springframework.util.MimeType; import org.springframework.util.StringUtils; +import org.springframework.web.server.ResponseStatusException; /** + * FIX KOMUNE - https://github.com/spring-cloud/spring-cloud-function/issues/901. + * Return an error instead of logging a warning when an object can't be parsed. * * @author Oleg Zhurakousky * @author Salvatore Bernardo @@ -77,6 +80,12 @@ public Object fromMessage(Message message, Class targetClass) { return result; } } + // KOMUNE Modification + // force message conversion error propagation + catch (ResponseStatusException e) { + throw e; + } + // KOMUNE End Of Modification catch (Exception e) { if (logger.isWarnEnabled()) { logger.warn("Failure during type conversion by " + converter + ". Will try the next converter.", e); diff --git a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebRequestProcessingHelper.java b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebRequestProcessingHelper.java index cb4f95003..8a6960695 100644 --- a/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebRequestProcessingHelper.java +++ b/spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebRequestProcessingHelper.java @@ -59,6 +59,13 @@ private FunctionWebRequestProcessingHelper() { public static FunctionInvocationWrapper findFunction(FunctionProperties functionProperties, HttpMethod method, FunctionCatalog functionCatalog, Map attributes, String path) { + // KOMUNE Modification + // For web browser we need to answer empty OPTIONS REQUEST + // TODO Try to replace it with OptionFunctionController + if (method.equals(HttpMethod.OPTIONS)) { + return null; + } + // KOMUNE End Of Modification if (method.equals(HttpMethod.GET) || method.equals(HttpMethod.POST) || method.equals(HttpMethod.PUT) || method.equals(HttpMethod.DELETE)) { return doFindFunction(functionProperties.getDefinition(), method, functionCatalog, attributes, path); } @@ -149,9 +156,10 @@ public static Publisher processRequest(FunctionWrapper wrapper, Object argume } if (pResult instanceof Flux) { - pResult = ((Flux) pResult).onErrorContinue((e, v) -> { - logger.error("Failed to process value: " + v, (Throwable) e); - }).collectList(); + // KOMUNE Modification + // force message conversion error propagation (no onErrorContinue) + pResult = ((Flux) pResult).collectList(); + // KOMUNE End Of Modification } pResult = Mono.from(pResult); } diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/PojoFunctionIntegrationTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/PojoFunctionIntegrationTests.java index 1faba31dc..ace2d2f20 100644 --- a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/PojoFunctionIntegrationTests.java +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/PojoFunctionIntegrationTests.java @@ -19,6 +19,7 @@ import java.util.Locale; import java.util.function.Function; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -102,6 +103,10 @@ void testCorrectInput() { * by Jackson and produces a {@code MismatchedInputException} in the log * instead of a silent {@code ClassCastException}. */ + // KOMUNE Modification + // JSON deserialization errors are surfaced as 400 (ResponseStatusException) instead of a logged 500. + @Disabled("KOMUNE: JSON deserialization errors are returned as 400, not logged as 500") + // KOMUNE End Of Modification @Test void testMalformedJsonRejectsWithJacksonError(CapturedOutput output) { String json = """ @@ -159,6 +164,10 @@ void testBrokenJsonRejectsWithJacksonError(CapturedOutput output) { * Without the fix the scalar passes through unchecked and the function * lambda fails with a misleading {@code ClassCastException}. */ + // KOMUNE Modification + // JSON deserialization errors are surfaced as 400 (ResponseStatusException) instead of a logged 500. + @Disabled("KOMUNE: JSON deserialization errors are returned as 400, not logged as 500") + // KOMUNE End Of Modification @Test void testScalarStringRejectsWithJacksonError(CapturedOutput output) { String json = "\"hello\""; diff --git a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HttpGetIntegrationTests.java b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HttpGetIntegrationTests.java index ef327c37f..2b12db56b 100644 --- a/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HttpGetIntegrationTests.java +++ b/spring-cloud-function-web/src/test/java/org/springframework/cloud/function/web/flux/HttpGetIntegrationTests.java @@ -29,6 +29,7 @@ import java.util.stream.Collectors; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import reactor.core.publisher.Flux; @@ -101,6 +102,10 @@ public void wordsJson() throws Exception { .getBody()).isEqualTo("[\"foo\",\"bar\"]"); } + // KOMUNE Modification + // Errors are now propagated instead of being swallowed by onErrorContinue, + // so /bang no longer yields a partial ["foo"] body. + @Disabled("KOMUNE: conversion errors are propagated, not skipped") @Test public void errorJson() throws Exception { assertThat(this.rest