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

GH-1896Stream Fixed conversion of collection of messages

Browse Source
This commit is contained in:
Oleg Zhurakousky
2020-01-27 17:25:21 +01:00
parent 20206a5342
commit 35d3c45959
2 changed files with 39 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistry.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2019 the original author or authors.
* Copyright 2019-2020 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.
@@ -19,6 +19,7 @@ package org.springframework.cloud.function.context.catalog;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
@@ -28,6 +29,7 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -573,6 +575,7 @@ public class BeanFactoryAwareFunctionRegistry
return result;
}
@SuppressWarnings({ "rawtypes", "unchecked" })
private Object convertOutputValueIfNecessary(Object value, String... acceptedOutputMimeTypes) {
logger.debug("Applying type conversion on output value");
Object convertedValue = null;
@@ -585,7 +588,7 @@ public class BeanFactoryAwareFunctionRegistry
try {
convertedInputArray[i] = outputArgument instanceof Publisher
? this.convertOutputPublisherIfNecessary((Publisher<?>) outputArgument, acceptedOutputMimeTypes[i])
: this.convertOutputValueIfNecessary(outputArgument, acceptedOutputMimeTypes);
: this.convertOutputValueIfNecessary(outputArgument, acceptedOutputMimeTypes[i]);
}
catch (ArrayIndexOutOfBoundsException e) {
throw new IllegalStateException("The number of 'acceptedOutputMimeTypes' for function '" + this.functionDefinition
@@ -599,32 +602,37 @@ public class BeanFactoryAwareFunctionRegistry
else if (value != null) {
List<MimeType> acceptedContentTypes = MimeTypeUtils.parseMimeTypes(acceptedOutputMimeTypes[0].toString());
convertedValue = acceptedContentTypes.stream()
.map(acceptedContentType -> {
Message<?> resultMessage = null;
if (value instanceof Message) {
Message<?> message = (Message<?>) value;
if (message.getPayload() instanceof byte[]) {
resultMessage = message;
}
else {
resultMessage = messageConverter.toMessage(message.getPayload(), message.getHeaders());
}
}
else {
if (value instanceof byte[]) {
resultMessage = MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, acceptedContentType).build();
}
else {
resultMessage = messageConverter
.toMessage(value, new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, acceptedContentType)));
}
}
return resultMessage;
})
.filter(v -> v != null)
.findFirst().orElse(null);
for (int i = 0; i < acceptedContentTypes.size() && convertedValue == null; i++) {
MimeType acceptedContentType = acceptedContentTypes.get(i);
if (value instanceof Message) {
Message<?> message = (Message<?>) value;
if (message.getPayload() instanceof byte[]) {
convertedValue = message;
}
else {
convertedValue = messageConverter.toMessage(message.getPayload(), message.getHeaders());
}
}
else if (value instanceof Iterable || ObjectUtils.isArray(value)) {
boolean isArray = ObjectUtils.isArray(value);
if (isArray) {
value = Arrays.asList((Object[]) value);
}
AtomicReference<List<Message>> messages = new AtomicReference<List<Message>>(new ArrayList<>());
((Iterable) value).forEach(element ->
messages.get().add((Message) convertOutputValueIfNecessary(element, acceptedContentType.toString())));
convertedValue = messages.get();
}
else {
if (value instanceof byte[]) {
convertedValue = MessageBuilder.withPayload(value).setHeader(MessageHeaders.CONTENT_TYPE, acceptedContentType).build();
}
else {
convertedValue = messageConverter
.toMessage(value, new MessageHeaders(Collections.singletonMap(MessageHeaders.CONTENT_TYPE, acceptedContentType)));
}
}
}
}
return convertedValue != null ? convertedValue : value;

3
spring-cloud-function-context/src/test/java/org/springframework/cloud/function/context/catalog/BeanFactoryAwareFunctionRegistryMultiInOutTests.java
View File

@@ -365,12 +365,14 @@ public class BeanFactoryAwareFunctionRegistryMultiInOutTests {
return Integer.class.isAssignableFrom(clazz);
}
@Override
protected Object convertFromInternal(
Message<?> message, Class<?> targetClass, @Nullable Object conversionHint) {
ByteBuffer wrappedPayload = ByteBuffer.wrap((byte[]) message.getPayload());
return wrappedPayload.getInt();
}
@Override
protected Object convertToInternal(
Object payload, @Nullable MessageHeaders headers, @Nullable Object conversionHint) {
@@ -473,6 +475,7 @@ public class BeanFactoryAwareFunctionRegistryMultiInOutTests {
public void setId(int id) {
this.id = id;
}
@Override
public String toString() {
return "Person: " + name + "/" + id;
}