Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4523: Add DSL convert(Class<> cls) operator (#2556)

Browse Source 
* INT-4523: Add DSL `convert(Class<> cls)` operator

JIRA: https://jira.spring.io/browse/INT-4523

* Change the `LambdaMessageProcessor` to rely on the `MessageConverter`
populated by the `IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME`
This way all the Lambda-based handlers are going to work the same way
as POJO-based via `MessagingMethodInvokerHelper`
* Add `convert(Class<P> payloadType)` EIP-operator to perform similar
to POJO-based method invocation argument conversion

* * Fix `LambdaMessageProcessorTests` to inject a
`ConfigurableCompositeMessageConverter` from the mocked `BeanFactory`
* Make a `mappingJackson2MessageConverter.setStrictContentTypeMatch(true)`
* Also obtain an `ObjectMapper` from the `Jackson2JsonObjectMapper`
which is configured with the scanned possible Jackson modules.
in the `ConfigurableCompositeMessageConverter` do not try to convert
all the potential content without an appropriate JSON content-type
header
This commit is contained in:
Artem Bilan
2018-09-17 10:40:56 -04:00
committed by Gary Russell
parent 5bfd6971db
commit 54094da76b
5 changed files with 110 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -587,6 +587,21 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
.transform(new MethodInvokingTransformer(processor), endpointConfigurer);
}
/**
* Populate the {@link MessageTransformingHandler} instance
* for the provided {@code payloadType} to convert at runtime.
* @param payloadType the {@link Class} for expected payload type.
* @param <P> the payload type - 'convert to'.
* @return the current {@link IntegrationFlowDefinition}.
* @since 5.1
* @see MethodInvokingTransformer
* @see LambdaMessageProcessor
*/
public <P> B convert(Class<P> payloadType) {
return transform(payloadType, p -> p);
}
/**
* Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer}
* for the specific {@code payloadType} to convert at runtime.
@@ -599,7 +614,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
* @see LambdaMessageProcessor
*/
public <P, T> B transform(Class<P> payloadType, GenericTransformer<P, T> genericTransformer) {
return this.transform(payloadType, genericTransformer, null);
return transform(payloadType, genericTransformer, null);
}
/**
@@ -619,6 +634,25 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
return this.transform(null, genericTransformer, endpointConfigurer);
}
/**
* Populate the {@link MessageTransformingHandler} instance
* for the provided {@code payloadType} to convert at runtime.
* In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
* @param payloadType the {@link Class} for expected payload type.
* @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
* @param <P> the payload type - 'transform to'.
* @return the current {@link IntegrationFlowDefinition}.
* @since 5.1
* @see MethodInvokingTransformer
* @see LambdaMessageProcessor
* @see GenericEndpointSpec
*/
public <P> B convert(Class<P> payloadType,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
return transform(payloadType, p -> p, endpointConfigurer);
}
/**
* Populate the {@link MessageTransformingHandler} instance for the provided {@link GenericTransformer}
* for the specific {@code payloadType} to convert at runtime.

28
spring-integration-core/src/main/java/org/springframework/integration/handler/LambdaMessageProcessor.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 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.
@@ -25,12 +25,10 @@ import java.util.concurrent.atomic.AtomicReference;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.convert.ConversionService;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.util.Assert;
import org.springframework.util.ReflectionUtils;
@@ -48,11 +46,11 @@ public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFac
private final Method method;
private final TypeDescriptor payloadType;
private final Class<?> payloadType;
private final Class<?>[] parameterTypes;
private ConversionService conversionService;
private MessageConverter messageConverter;
public LambdaMessageProcessor(Object target, Class<?> payloadType) {
Assert.notNull(target, "'target' must not be null");
@@ -79,16 +77,14 @@ public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFac
this.method = methodValue.get();
this.method.setAccessible(true);
this.parameterTypes = this.method.getParameterTypes();
this.payloadType = payloadType != null ? TypeDescriptor.valueOf(payloadType) : null;
this.payloadType = payloadType;
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
ConversionService conversionService = IntegrationUtils.getConversionService(beanFactory);
if (conversionService == null) {
conversionService = DefaultConversionService.getSharedInstance();
}
this.conversionService = conversionService;
this.messageConverter =
beanFactory.getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME,
MessageConverter.class);
}
@Override
@@ -109,14 +105,12 @@ public class LambdaMessageProcessor implements MessageProcessor<Object>, BeanFac
}
else {
if (this.payloadType != null) {
if (Message.class.isAssignableFrom(this.payloadType.getType())) {
if (Message.class.isAssignableFrom(this.payloadType)) {
args[i] = message;
}
else {
args[i] = this.conversionService.convert(message.getPayload(),
TypeDescriptor.forObject(message.getPayload()), this.payloadType);
args[i] = this.messageConverter.fromMessage(message, this.payloadType);
}
}
else {
args[i] = message.getPayload();

8
spring-integration-core/src/main/java/org/springframework/integration/support/converter/ConfigurableCompositeMessageConverter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017 the original author or authors.
* Copyright 2017-2018 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.
@@ -22,6 +22,7 @@ import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.springframework.integration.support.json.Jackson2JsonObjectMapper;
import org.springframework.integration.support.json.JacksonPresent;
import org.springframework.messaging.converter.ByteArrayMessageConverter;
import org.springframework.messaging.converter.CompositeMessageConverter;
@@ -78,7 +79,10 @@ public class ConfigurableCompositeMessageConverter extends CompositeMessageConve
List<MessageConverter> converters = new LinkedList<>();
if (JacksonPresent.isJackson2Present()) {
converters.add(new MappingJackson2MessageConverter());
MappingJackson2MessageConverter mappingJackson2MessageConverter = new MappingJackson2MessageConverter();
mappingJackson2MessageConverter.setStrictContentTypeMatch(true);
mappingJackson2MessageConverter.setObjectMapper(new Jackson2JsonObjectMapper().getObjectMapper());
converters.add(mappingJackson2MessageConverter);
}
converters.add(new ByteArrayMessageConverter());
converters.add(new ObjectStringMessageConverter());

19
spring-integration-core/src/test/java/org/springframework/integration/dsl/LambdaMessageProcessorTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2017 the original author or authors.
* Copyright 2016-2018 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.
@@ -20,20 +20,25 @@ import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.fail;
import static org.mockito.BDDMockito.given;
import static org.mockito.Mockito.mock;
import org.junit.Test;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.handler.GenericHandler;
import org.springframework.integration.handler.LambdaMessageProcessor;
import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter;
import org.springframework.integration.transformer.GenericTransformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.GenericMessage;
/**
* @author Gary Russell
* @author Artem Bilan
*
* @since 5.0
*/
@@ -69,7 +74,8 @@ public class LambdaMessageProcessorTests {
private void handle(GenericHandler<?> h) {
LambdaMessageProcessor lmp = new LambdaMessageProcessor(h, String.class);
lmp.setBeanFactory(mock(BeanFactory.class));
lmp.setBeanFactory(getBeanFactory());
lmp.processMessage(new GenericMessage<>("foo"));
}
@@ -77,4 +83,13 @@ public class LambdaMessageProcessorTests {
return message;
}
private BeanFactory getBeanFactory() {
BeanFactory mockBeanFactory = mock(BeanFactory.class);
given(mockBeanFactory.getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME,
MessageConverter.class))
.willReturn(new ConfigurableCompositeMessageConverter());
return mockBeanFactory;
}
}

45
spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java
View File

@@ -196,7 +196,7 @@ public class TransformerTests {
private PollableChannel codecReplyChannel;
@Test
public void testCodec() throws Exception {
public void testCodec() {
this.encodingFlowInput.send(new GenericMessage<>("bar"));
Message<?> receive = this.codecReplyChannel.receive(10000);
assertNotNull(receive);
@@ -246,6 +246,34 @@ public class TransformerTests {
assertNotNull(this.adviceChannel.receive(10000));
}
@Autowired
@Qualifier("convertFlow.input")
private MessageChannel convertFlowInput;
@Test
public void testConvertOperator() {
QueueChannel replyChannel = new QueueChannel();
Date date = new Date();
this.convertFlowInput.send(
MessageBuilder.withPayload("{\"name\": \"Baz\",\"date\": " + date.getTime() + "}")
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json")
.setReplyChannel(replyChannel)
.build());
Message<?> receive = replyChannel.receive(10_000);
assertNotNull(receive);
Object payload = receive.getPayload();
assertThat(payload, instanceOf(TestPojo.class));
TestPojo testPojo = (TestPojo) payload;
assertEquals("Baz", testPojo.getName());
assertEquals(date, testPojo.getDate());
}
@Configuration
@EnableIntegration
public static class ContextConfiguration {
@@ -413,6 +441,12 @@ public class TransformerTests {
return new SomeService();
}
@Bean
public IntegrationFlow convertFlow() {
return f -> f
.convert(TestPojo.class);
}
}
private static final class TestPojo {
@@ -421,6 +455,9 @@ public class TransformerTests {
private Date date;
private TestPojo() {
}
private TestPojo(String name) {
this.name = name;
}
@@ -448,7 +485,7 @@ public class TransformerTests {
public static class MyCodec implements Codec {
@Override
public void encode(Object object, OutputStream outputStream) throws IOException {
public void encode(Object object, OutputStream outputStream) {
}
@Override
@@ -457,13 +494,13 @@ public class TransformerTests {
}
@Override
public <T> T decode(InputStream inputStream, Class<T> type) throws IOException {
public <T> T decode(InputStream inputStream, Class<T> type) {
return null;
}
@SuppressWarnings("unchecked")
@Override
public <T> T decode(byte[] bytes, Class<T> type) throws IOException {
public <T> T decode(byte[] bytes, Class<T> type) {
return (T) (type.equals(String.class) ? new String(bytes) :
type.equals(Integer.class) ? Integer.valueOf(42) : Integer.valueOf(43));
}