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 2fec2f3fa..b89d47853 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,7 +28,6 @@ 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; @@ -97,7 +96,6 @@ 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; @@ -446,16 +444,12 @@ 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)) { - 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 -> { + MessageChannel outputChannel = this.applicationContext.getBean(outputBindingIter.next(), MessageChannel.class); + flux = flux.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 deleted file mode 100644 index 0adb3c546..000000000 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ReactiveFunctionWithNativeEncodingTests.java +++ /dev/null @@ -1,94 +0,0 @@ -/* - * 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()); - } - - } -}