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

Revert "Reactive functions and native encoding"

Browse Source 
This reverts commit 726a79cdae.
This commit is contained in:
Oleg Zhurakousky
2020-10-23 17:29:43 +02:00
parent 726a79cdae
commit 705665f638
2 changed files with 3 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionConfiguration.java
View File

@@ -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<String> 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();

94
spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/ReactiveFunctionWithNativeEncodingTests.java
View File

@@ -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<byte[]> inputMessageOne = MessageBuilder.withPayload("Hello".getBytes()).build();
inputDestination.send(inputMessageOne);
Message<byte[]> 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<byte[]>, Flux<byte[]>> toUpperCaseReactive() {
return flux -> flux.map(f -> new String(f).toUpperCase().getBytes());
}
}
}