From a05004ae3812ad7c216fe1041609540d05fdbe69 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 4 Oct 2018 11:12:32 -0400 Subject: [PATCH] INT-4536: Support Kafka Tombstones JIRA: https://jira.spring.io/browse/INT-4536 Represent `@KafkaNull` as `null` when `@Payload(required = false)`. --- ...ltConfiguringBeanFactoryPostProcessor.java | 8 +++- .../support/MessagingMethodInvokerHelper.java | 25 ++++++---- .../NullAwarePayloadArgumentResolver.java | 47 +++++++++++++++++++ ...ServiceActivatorMethodResolutionTests.java | 43 ++++++++++++++++- 4 files changed, 112 insertions(+), 11 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/NullAwarePayloadArgumentResolver.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java index 3ad0dea1a5..a51a4a3fa5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/DefaultConfiguringBeanFactoryPostProcessor.java @@ -58,6 +58,7 @@ import org.springframework.integration.handler.support.MapArgumentResolver; import org.springframework.integration.handler.support.PayloadExpressionArgumentResolver; import org.springframework.integration.handler.support.PayloadsArgumentResolver; import org.springframework.integration.support.DefaultMessageBuilderFactory; +import org.springframework.integration.support.NullAwarePayloadArgumentResolver; import org.springframework.integration.support.SmartLifecycleRoleController; import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter; import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter; @@ -99,6 +100,7 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce this.classLoader = classLoader; } + @Override public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException { if (beanFactory instanceof BeanDefinitionRegistry) { this.beanFactory = beanFactory; @@ -477,9 +479,13 @@ class DefaultConfiguringBeanFactoryPostProcessor implements BeanFactoryPostProce } } - private static BeanDefinition internalArgumentResolversBuilder(boolean listCapable) { + private BeanDefinition internalArgumentResolversBuilder(boolean listCapable) { ManagedList resolvers = new ManagedList<>(); resolvers.add(new RootBeanDefinition(PayloadExpressionArgumentResolver.class)); + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(NullAwarePayloadArgumentResolver.class); + builder.addConstructorArgReference(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME); + // TODO Validator ? + resolvers.add(builder.getBeanDefinition()); resolvers.add(new RootBeanDefinition(PayloadsArgumentResolver.class)); resolvers.add(new RootBeanDefinition(MapArgumentResolver.class)); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java index d507b90a39..35f9b656b8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/support/MessagingMethodInvokerHelper.java @@ -74,6 +74,8 @@ import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.annotation.UseSpelInvoker; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.support.MutableMessage; +import org.springframework.integration.support.NullAwarePayloadArgumentResolver; +import org.springframework.integration.support.converter.ConfigurableCompositeMessageConverter; import org.springframework.integration.support.json.JsonObjectMapper; import org.springframework.integration.support.json.JsonObjectMapperProvider; import org.springframework.integration.util.AbstractExpressionEvaluator; @@ -538,6 +540,19 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator * that don't run in an application context. */ private void configureLocalMessageHandlerFactory() { + MessageConverter messageConverter = null; + if (getBeanFactory() != null && + getBeanFactory() + .containsBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)) { + messageConverter = getBeanFactory() + .getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME, + MessageConverter.class); + this.messageHandlerMethodFactory.setMessageConverter(messageConverter); + } + else { + messageConverter = new ConfigurableCompositeMessageConverter(); + } + NullAwarePayloadArgumentResolver nullResolver = new NullAwarePayloadArgumentResolver(messageConverter); PayloadExpressionArgumentResolver payloadExpressionArgumentResolver = new PayloadExpressionArgumentResolver(); payloadExpressionArgumentResolver.setBeanFactory(getBeanFactory()); @@ -549,6 +564,7 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator List customArgumentResolvers = new LinkedList<>(); customArgumentResolvers.add(payloadExpressionArgumentResolver); + customArgumentResolvers.add(nullResolver); customArgumentResolvers.add(payloadsArgumentResolver); if (this.canProcessMessageList) { @@ -560,15 +576,6 @@ public class MessagingMethodInvokerHelper extends AbstractExpressionEvaluator customArgumentResolvers.add(mapArgumentResolver); this.messageHandlerMethodFactory.setCustomArgumentResolvers(customArgumentResolvers); - - if (getBeanFactory() != null && - getBeanFactory() - .containsBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME)) { - this.messageHandlerMethodFactory - .setMessageConverter(getBeanFactory() - .getBean(IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME, - MessageConverter.class)); - } } @SuppressWarnings("unchecked") diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/NullAwarePayloadArgumentResolver.java b/spring-integration-core/src/main/java/org/springframework/integration/support/NullAwarePayloadArgumentResolver.java new file mode 100644 index 0000000000..29f32fc7c6 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/NullAwarePayloadArgumentResolver.java @@ -0,0 +1,47 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * http://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.integration.support; + +import org.springframework.messaging.converter.MessageConverter; +import org.springframework.messaging.handler.annotation.Payload; +import org.springframework.messaging.handler.annotation.support.PayloadArgumentResolver; +import org.springframework.validation.Validator; + +/** + * A {@link PayloadArgumentResolver} that treats KafkaNull payloads as null. + * {@link Payload @Paylaod} annotation must have required = false. + * + * @author Gary Russell + * @since 5.1 + * + */ +public class NullAwarePayloadArgumentResolver extends PayloadArgumentResolver { + + public NullAwarePayloadArgumentResolver(MessageConverter messageConverter) { + super(messageConverter, null, false); + } + + public NullAwarePayloadArgumentResolver(MessageConverter messageConverter, Validator validator) { + super(messageConverter, validator, false); + } + + @Override + protected boolean isEmptyPayload(Object payload) { + return super.isEmptyPayload(payload) || "KafkaNull".equals(payload.getClass().getSimpleName()); + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorMethodResolutionTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorMethodResolutionTests.java index 61792e42a7..6b5cb86892 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorMethodResolutionTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/ServiceActivatorMethodResolutionTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-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. @@ -30,10 +30,12 @@ import org.springframework.integration.gateway.RequestReplyExchanger; import org.springframework.integration.handler.ServiceActivatingHandler; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.support.GenericMessage; /** * @author Mark Fisher + * @author Gary Russell */ public class ServiceActivatorMethodResolutionTests { @@ -231,6 +233,17 @@ public class ServiceActivatorMethodResolutionTests { assertNotEquals("FOO", outputChannel.receive(10).getPayload()); } + @Test + public void nullOk() { + NullOkTestBean testBean = new NullOkTestBean(); + ServiceActivatingHandler serviceActivator = new ServiceActivatingHandler(testBean); + QueueChannel outputChannel = new QueueChannel(); + serviceActivator.setOutputChannel(outputChannel); + serviceActivator.handleMessage(new GenericMessage<>(new KafkaNull())); + Message result = outputChannel.receive(0); + assertEquals("gotNull", result.getPayload()); + } + @SuppressWarnings("unused") private static class SingleAnnotationTestBean { @@ -305,4 +318,32 @@ public class ServiceActivatorMethodResolutionTests { } + + @SuppressWarnings("unused") + private static class NullOkTestBean { + + NullOkTestBean() { + super(); + } + + @ServiceActivator + public String nullOK(@Payload(required = false) String s) { + if (s == null) { + return "gotNull"; + } + else { + return s; + } + } + + } + + private static class KafkaNull { + + KafkaNull() { + super(); + } + + } + }