From c6d238085f11a086b62813da5d08cd90310da5ea Mon Sep 17 00:00:00 2001 From: Oleg Zhurakousky Date: Thu, 31 Jan 2019 08:08:06 +0100 Subject: [PATCH] GH-1594 Ensured Function conversion honors outbound CT Ensured that Message conversion that happenes within FunctionInvoker honors the outbound binding content-type in the event MessageConverter itself did not sent one. Restructured some code/API around function invocation Added tests Resolves #1594 --- .../binder/AbstractMessageChannelBinder.java | 30 ---- .../binding/MessageConverterConfigurer.java | 10 +- .../stream/function/FunctionInvoker.java | 41 +++-- .../IntegrationFlowFunctionSupport.java | 10 +- .../function/StreamFunctionProperties.java | 36 +++-- .../stream/function/FunctionInvokerTests.java | 146 ++++++++++++++++-- 6 files changed, 197 insertions(+), 76 deletions(-) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java index ecb33c249..7a479b007 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.binder; -import java.lang.reflect.Method; import java.util.LinkedHashMap; import java.util.Map; import java.util.function.Consumer; @@ -65,7 +64,6 @@ import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.retry.RecoveryCallback; import org.springframework.util.Assert; -import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; /** @@ -854,7 +852,6 @@ public abstract class AbstractMessageChannelBinder doPreSend(Message message, MessageChannel channel) { + + // If handler is a function, FunctionInvoker will already perform message conversion. + // In fact in the future we should consider propagating knowledge of the default content type + //to MessageConverters instead of interceptors + if (message.getPayload() instanceof byte[] && message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) { + return message; + } + + // ===== 1.3 backward compatibility code part-1 === String oct = message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE) ? message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString() : null; String ct = message.getPayload() instanceof String diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java index c7fdc9945..df1444d2a 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -16,7 +16,9 @@ package org.springframework.cloud.stream.function; +import java.lang.reflect.Field; import java.time.Duration; +import java.util.Map; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Consumer; import java.util.function.Function; @@ -30,6 +32,7 @@ import org.springframework.cloud.function.context.FunctionType; import org.springframework.cloud.function.context.catalog.FunctionInspector; import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -38,6 +41,8 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.converter.CompositeMessageConverter; import org.springframework.messaging.support.ErrorMessage; import org.springframework.util.Assert; +import org.springframework.util.MimeType; +import org.springframework.util.ReflectionUtils; /** * @@ -54,6 +59,13 @@ class FunctionInvoker implements Function>, Flux inputClass; private final Class outputClass; @@ -70,6 +82,10 @@ class FunctionInvoker implements Function>, Flux implements Function>, Flux implements Function>, Flux implements Function>, Flux) ReflectionUtils.getField(MESSAGE_HEADERS_FIELD, returnMessage.getHeaders())) + .put(MessageHeaders.CONTENT_TYPE, MimeType.valueOf(bindingServiceProperties.getBindingProperties("output").getContentType())); } Assert.notNull(returnMessage, "Failed to convert result value '" + value + "' to message."); } return returnMessage; } + @SuppressWarnings("unchecked") private Message wrapOutputToMessage(T value, Message originalMessage) { Message returnMessage = (Message) MessageBuilder.withPayload(value).copyHeaders(originalMessage.getHeaders()) diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java index 29705f4f0..d9df46d49 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -58,8 +58,6 @@ public class IntegrationFlowFunctionSupport { private final StreamFunctionProperties functionProperties; - private final BindingServiceProperties bindingServiceProperties; - @Autowired private MessageChannel errorChannel; @@ -81,7 +79,7 @@ public class IntegrationFlowFunctionSupport { this.functionInspector = functionInspector; this.messageConverterFactory = messageConverterFactory; this.functionProperties = functionProperties; - this.bindingServiceProperties = bindingServiceProperties; + this.functionProperties.setBindingServiceProperties(bindingServiceProperties); } /** @@ -158,14 +156,14 @@ public class IntegrationFlowFunctionSupport { if (inputChannel instanceof IntegrationObjectSupport) { String inputBindingName = ((IntegrationObjectSupport)inputChannel).getComponentName(); if (StringUtils.hasText(inputBindingName)) { - this.functionProperties.setConsumerProperties(this.bindingServiceProperties.getConsumerProperties(inputBindingName)); + this.functionProperties.setInputDestinationName(inputBindingName); } } if (outputChannel instanceof IntegrationObjectSupport) { String outputBindingName = ((IntegrationObjectSupport)outputChannel).getComponentName(); if (StringUtils.hasText(outputBindingName)) { - this.functionProperties.setProducerProperties(this.bindingServiceProperties.getProducerProperties(outputBindingName)); + this.functionProperties.setOutputDestinationName(outputBindingName); } } diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java index 96ebfab05..dfb9f4df8 100644 --- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java +++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamFunctionProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -17,8 +17,8 @@ package org.springframework.cloud.stream.function; import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.cloud.stream.messaging.Processor; /** * @@ -36,9 +36,11 @@ public class StreamFunctionProperties { */ private String definition; - private ConsumerProperties consumerProperties; + private BindingServiceProperties bindingServiceProperties; - private ProducerProperties producerProperties; + private String inputDestinationName = Processor.INPUT; + + private String outputDestinationName = Processor.OUTPUT; public String getDefinition() { return this.definition; @@ -48,19 +50,27 @@ public class StreamFunctionProperties { this.definition = definition; } - protected ConsumerProperties getConsumerProperties() { - return consumerProperties; + BindingServiceProperties getBindingServiceProperties() { + return bindingServiceProperties; } - void setConsumerProperties(ConsumerProperties consumerProperties) { - this.consumerProperties = consumerProperties; + void setBindingServiceProperties(BindingServiceProperties bindingServiceProperties) { + this.bindingServiceProperties = bindingServiceProperties; } - protected ProducerProperties getProducerProperties() { - return producerProperties; + String getInputDestinationName() { + return inputDestinationName; } - void setProducerProperties(ProducerProperties producerProperties) { - this.producerProperties = producerProperties; + void setInputDestinationName(String inputDestinationName) { + this.inputDestinationName = inputDestinationName; + } + + String getOutputDestinationName() { + return outputDestinationName; + } + + void setOutputDestinationName(String outputDestinationName) { + this.outputDestinationName = outputDestinationName; } } diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java index 51d16a350..b5f7bd906 100644 --- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java +++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/FunctionInvokerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018 the original author or authors. + * Copyright 2018-2019 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. @@ -30,16 +30,22 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.cloud.function.context.FunctionCatalog; import org.springframework.cloud.function.context.catalog.FunctionInspector; -import org.springframework.cloud.stream.binder.ConsumerProperties; -import org.springframework.cloud.stream.binder.ProducerProperties; +import org.springframework.cloud.stream.annotation.EnableBinding; +import org.springframework.cloud.stream.annotation.StreamMessageConverter; +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.cloud.stream.config.BindingServiceProperties; import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory; import org.springframework.cloud.stream.function.pojo.Baz; import org.springframework.cloud.stream.function.pojo.ErrorBaz; +import org.springframework.cloud.stream.messaging.Processor; 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.MessageHeaders; +import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.ReflectionUtils; @@ -54,6 +60,112 @@ import static org.junit.Assert.assertEquals; */ public class FunctionInvokerTests { + @Test + public void testFunctionHonorsOutboundBindingContentType() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(ConverterDoesNotProduceCTConfiguration.class)).web( + WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.output.contentType=text/plain")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder.withPayload("{\"name\":\"bob\"}".getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertEquals("text/plain", outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()); + + } + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class ConverterDoesNotProduceCTConfiguration { + + @Bean + public Function func() { + return x -> x; + } + + @StreamMessageConverter + @Bean + public MessageConverter customConverter() { + return new MessageConverter() { + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + return new GenericMessage(((String)payload).getBytes()); + } + + @Override + public Object fromMessage(Message message, Class targetClass) { + String contentType = (String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(); + if (contentType.equals("foo/bar")) { + return new String((byte[])message.getPayload()); + } + return null; + } + }; + } + } + + @Test + public void testFunctionHonorsConverterSetContentType() { + try (ConfigurableApplicationContext context = new SpringApplicationBuilder( + TestChannelBinderConfiguration.getCompleteConfiguration(ConverterInjectingCTConfiguration.class)).web( + WebApplicationType.NONE) + .run("--spring.jmx.enabled=false", + "--spring.cloud.stream.function.definition=func", + "--spring.cloud.stream.bindings.output.contentType=text/plain")) { + + InputDestination inputDestination = context.getBean(InputDestination.class); + OutputDestination outputDestination = context.getBean(OutputDestination.class); + + Message inputMessage = MessageBuilder.withPayload("{\"name\":\"bob\"}".getBytes()) + .setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build(); + inputDestination.send(inputMessage); + + Message outputMessage = outputDestination.receive(); + assertEquals("ping/pong", outputMessage.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString()); + + } + } + + @EnableAutoConfiguration + @EnableBinding(Processor.class) + public static class ConverterInjectingCTConfiguration { + + @Bean + public Function func() { + return x -> x; + } + + @StreamMessageConverter + @Bean + public MessageConverter customConverter() { + return new MessageConverter() { + + @Override + public Message toMessage(Object payload, MessageHeaders headers) { + return MessageBuilder.withPayload(((String)payload).getBytes()).setHeader(MessageHeaders.CONTENT_TYPE, "ping/pong").build(); + } + + @Override + public Object fromMessage(Message message, Class targetClass) { + String contentType = (String) message.getHeaders().get(MessageHeaders.CONTENT_TYPE).toString(); + if (contentType.equals("foo/bar")) { + return new String((byte[])message.getPayload()); + } + return null; + } + }; + } + } + @Test public void testSameMessageTypesAreNotConverted() { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( @@ -176,12 +288,14 @@ public class FunctionInvokerTests { private StreamFunctionProperties createStreamFunctionProperties() { StreamFunctionProperties functionProperties = new StreamFunctionProperties(); - ConsumerProperties consumerProperties = new ConsumerProperties(); - consumerProperties.setMaxAttempts(3); + functionProperties.setInputDestinationName("input"); + functionProperties.setOutputDestinationName("output"); + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3); try { - Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties"); + Field f = ReflectionUtils.findField(StreamFunctionProperties.class, "bindingServiceProperties"); f.setAccessible(true); - f.set(functionProperties, consumerProperties); + f.set(functionProperties, bindingServiceProperties); return functionProperties; } catch (Exception e) { throw new IllegalStateException(e); @@ -190,17 +304,15 @@ public class FunctionInvokerTests { private StreamFunctionProperties createStreamFunctionPropertiesWithNativeEncoding() { StreamFunctionProperties functionProperties = new StreamFunctionProperties(); - ConsumerProperties consumerProperties = new ConsumerProperties(); - consumerProperties.setMaxAttempts(3); - ProducerProperties producerProperties = new ProducerProperties(); - producerProperties.setUseNativeEncoding(true); + functionProperties.setInputDestinationName("input"); + functionProperties.setOutputDestinationName("output"); + BindingServiceProperties bindingServiceProperties = new BindingServiceProperties(); + bindingServiceProperties.getConsumerProperties("input").setMaxAttempts(3); + bindingServiceProperties.getProducerProperties("output").setUseNativeEncoding(true); try { - Field c = ReflectionUtils.findField(StreamFunctionProperties.class, "consumerProperties"); - Field p = ReflectionUtils.findField(StreamFunctionProperties.class, "producerProperties"); - c.setAccessible(true); - c.set(functionProperties, consumerProperties); - p.setAccessible(true); - p.set(functionProperties, producerProperties); + Field bspField = ReflectionUtils.findField(StreamFunctionProperties.class, "bindingServiceProperties"); + bspField.setAccessible(true); + bspField.set(functionProperties, bindingServiceProperties); return functionProperties; } catch (Exception e) { throw new IllegalStateException(e);