From 726a79cdae8d660b32909c7bf8a21207a7833569 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Tue, 20 Oct 2020 16:46:48 -0400 Subject: [PATCH] Reactive functions and native encoding When native encoding is enabled with a reactive function, Spring Cloud Stream is still trying to convert the message causing ClassCastException and other potential issues downstream. Fixing this issue by signaling the framework to not perform any message conversion on the outbound when useNativeEncoding is enabled with a reactive function. Resovles https://github.com/spring-cloud/spring-cloud-stream/issues/2037 --- .../function/FunctionConfiguration.java | 12 ++- ...activeFunctionWithNativeEncodingTests.java | 94 +++++++++++++++++++ 2 files changed, 103 insertions(+), 3 deletions(-) create mode 100644 spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ReactiveFunctionWithNativeEncodingTests.java diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java index b89d47853..2fec2f3fa 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java @@ -28,6 +28,7 @@ import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.function.Supplier; @@ -96,6 +97,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.SubscribableChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; import org.springframework.util.ClassUtils; @@ -444,12 +446,16 @@ public class FunctionConfiguration { resultPublishers = Collections.singletonList(resultPublishers); } Iterator outputBindingIter = outputBindingNames.iterator(); - + AtomicInteger index = new AtomicInteger(); ((Iterable) resultPublishers).forEach(publisher -> { Flux flux = Flux.from((Publisher) publisher); if (!CollectionUtils.isEmpty(outputBindingNames)) { - MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class); - flux = flux.doOnNext(message -> { + final String outputDestinationName = outputBindingIter.next(); + this.adjustFunctionForNativeEncodingIfNecessary(outputDestinationName, function, index.getAndIncrement()); + MessageChannel outputChannel = this.applicationContext.getBean(outputDestinationName, MessageChannel.class); + flux = flux + .map(value -> value instanceof Message ? value : new GenericMessage(value)) + .doOnNext(message -> { if (message instanceof Message && ((Message) message).getHeaders().get("spring.cloud.stream.sendto.destination") != null) { String destinationName = (String) ((Message) message).getHeaders().get("spring.cloud.stream.sendto.destination"); ProducerProperties producerProperties = this.serviceProperties.getBindings().get(outputBindingNames.iterator().next()).getProducer(); diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ReactiveFunctionWithNativeEncodingTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ReactiveFunctionWithNativeEncodingTests.java new file mode 100644 index 000000000..0adb3c546 --- /dev/null +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ReactiveFunctionWithNativeEncodingTests.java @@ -0,0 +1,94 @@ +/* + * Copyright 2020-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. + * 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.function.Function; + +import org.junit.Test; +import org.mockito.Mockito; +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.catalog.SimpleFunctionRegistry; +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.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.Message; +import org.springframework.messaging.converter.CompositeMessageConverter; +import org.springframework.util.ReflectionUtils; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** + * @author Soby Chacko + */ +public class ReactiveFunctionWithNativeEncodingTests { + + @Test + public void testFunctionIsAppliedToExistingMessageSource() throws Exception { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration( + FunctionsConfiguration.class)).web(WebApplicationType.NONE).run( + "--spring.cloud.stream.function.definition=toUpperCaseReactive", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.toUpperCaseReactive-in-0.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.toUpperCaseReactive-out-0.producer.useNativeEncoding=true" + )) { + SimpleFunctionRegistry simpleFunctionRegistry = context.getBean(SimpleFunctionRegistry.class); + + final Field messageConverter = ReflectionUtils + .findField(SimpleFunctionRegistry.class, "messageConverter", CompositeMessageConverter.class); + messageConverter.setAccessible(true); + final CompositeMessageConverter o = (CompositeMessageConverter) messageConverter.get(simpleFunctionRegistry); + + final CompositeMessageConverter spy = Mockito.spy(o); + ReflectionUtils.setField(messageConverter, simpleFunctionRegistry, spy); + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + Message inputMessageOne = MessageBuilder.withPayload("Hello".getBytes()).build(); + inputDestination.send(inputMessageOne); + + Message outputMessage = outputDestination.receive(); + + assertThat(outputMessage.getPayload()).isEqualTo("HELLO".getBytes()); + // Verify that no message conversion done by the framework on the outbound. + // If useNativeEncoding is not enabled (see the properties above), + // then the following verification should fail. + verify(spy, never()).toMessage(any(), any()); + } + } + + @EnableAutoConfiguration + public static class FunctionsConfiguration { + + @Bean + public Function, Flux> toUpperCaseReactive() { + return flux -> flux.map(f -> new String(f).toUpperCase().getBytes()); + } + + } +}