From 99949979176ac8d89d510ac6c922223ff5bc9ee0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 27 Sep 2019 17:14:51 -0400 Subject: [PATCH] GH-3067: Use default NullChannel instance if poss. Resolves https://github.com/spring-projects/spring-integration/issues/3067 Also add missing receive counter. * Late binding of null discard channel; checkstyle * Fix test; fall back to new NullChannel(); always evaluate tx expressions --- .../AbstractCorrelatingMessageHandler.java | 23 ++++++--- .../integration/channel/NullChannel.java | 28 +++++++++++ ...ngTransactionSynchronizationProcessor.java | 16 +++---- .../config/ResequencerParserTests.java | 2 - .../annotation/AggregatorAnnotationTests.java | 3 +- ...PseudoTransactionalMessageSourceTests.java | 6 +-- ...eTailInboundChannelAdapterFactoryBean.java | 48 ++++++++++++------- .../integration/file/dsl/TailAdapterSpec.java | 5 +- 8 files changed, 88 insertions(+), 43 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index c9102d8974..52559610df 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -39,6 +39,7 @@ import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.DiscardingMessageHandler; @@ -55,6 +56,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; import org.springframework.messaging.MessageHandlingException; +import org.springframework.messaging.core.DestinationResolutionException; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -315,7 +317,7 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP super.onInit(); Assert.state(!(this.discardChannelName != null && this.discardChannel != null), "'discardChannelName' and 'discardChannel' are mutually exclusive."); - BeanFactory beanFactory = this.getBeanFactory(); + BeanFactory beanFactory = getBeanFactory(); if (beanFactory != null) { if (this.outputProcessor instanceof BeanFactoryAware) { ((BeanFactoryAware) this.outputProcessor).setBeanFactory(beanFactory); @@ -328,10 +330,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } } - if (this.discardChannel == null) { - this.discardChannel = new NullChannel(); - } - if (this.releasePartialSequences) { Assert.isInstanceOf(SequenceSizeReleaseStrategy.class, this.releaseStrategy, () -> "Release strategy of type [" + this.releaseStrategy.getClass().getSimpleName() + @@ -392,8 +390,21 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP @Override public MessageChannel getDiscardChannel() { String channelName = this.discardChannelName; + if (channelName == null && this.discardChannel == null) { + channelName = IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME; + } if (channelName != null) { - this.discardChannel = getChannelResolver().resolveDestination(channelName); + try { + this.discardChannel = getChannelResolver().resolveDestination(channelName); + } + catch (DestinationResolutionException ex) { + if (channelName.equals(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME)) { + this.discardChannel = new NullChannel(); + } + else { + throw ex; + } + } this.discardChannelName = null; } return this.discardChannel; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java index 21c45bb150..5cdd677c58 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java @@ -27,6 +27,7 @@ import org.springframework.integration.support.management.AbstractMessageChannel import org.springframework.integration.support.management.ConfigurableMetricsAware; import org.springframework.integration.support.management.DefaultMessageChannelMetrics; import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.support.management.metrics.CounterFacade; import org.springframework.integration.support.management.metrics.MetricsCaptor; import org.springframework.integration.support.management.metrics.TimerFacade; import org.springframework.lang.Nullable; @@ -68,6 +69,8 @@ public class NullChannel implements PollableChannel, private TimerFacade successTimer; + private CounterFacade receiveCounter; + @Override public void setBeanName(String beanName) { this.beanName = beanName; @@ -268,6 +271,7 @@ public class NullChannel implements PollableChannel, if (this.loggingEnabled) { this.logger.debug("receive called on null channel"); } + incrementReceiveCounter(); return null; } @@ -276,6 +280,27 @@ public class NullChannel implements PollableChannel, return receive(); } + private void incrementReceiveCounter() { + if (this.metricsCaptor != null) { + if (this.receiveCounter == null) { + this.receiveCounter = buildReceiveCounter(); + } + this.receiveCounter.increment(); + } + } + + private CounterFacade buildReceiveCounter() { + CounterFacade counterFacade = this.metricsCaptor + .counterBuilder(RECEIVE_COUNTER_NAME) + .tag("name", getComponentName() == null ? "unknown" : getComponentName()) + .tag("type", "channel") + .tag("result", "success") + .tag("exception", "none") + .description("Messages received") + .build(); + return counterFacade; + } + @Override public String toString() { return (this.beanName != null) ? this.beanName : super.toString(); @@ -286,6 +311,9 @@ public class NullChannel implements PollableChannel, if (this.successTimer != null) { this.successTimer.remove(); } + if (this.receiveCounter != null) { + this.receiveCounter.remove(); + } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java index 417b046a54..de65fa6c80 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java @@ -21,9 +21,9 @@ import java.util.Map.Entry; import org.springframework.expression.EvaluationContext; import org.springframework.expression.Expression; import org.springframework.expression.spel.support.StandardEvaluationContext; -import org.springframework.integration.channel.NullChannel; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; @@ -66,11 +66,11 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int private volatile Expression afterRollbackExpression; - private volatile MessageChannel beforeCommitChannel = new NullChannel(); + private volatile MessageChannel beforeCommitChannel; - private volatile MessageChannel afterCommitChannel = new NullChannel(); + private volatile MessageChannel afterCommitChannel; - private volatile MessageChannel afterRollbackChannel = new NullChannel(); + private volatile MessageChannel afterRollbackChannel; public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) { this.evaluationContext = evaluationContext; @@ -129,8 +129,8 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int doProcess(holder, this.afterRollbackExpression, this.afterRollbackChannel, "afterRollback"); } - private void doProcess(IntegrationResourceHolder holder, Expression expression, MessageChannel messageChannel, - String expressionType) { + private void doProcess(IntegrationResourceHolder holder, Expression expression, + @Nullable MessageChannel messageChannel, String expressionType) { Message message = holder.getMessage(); if (message != null) { @@ -141,7 +141,7 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int } EvaluationContext evaluationContextToUse = prepareEvaluationContextToUse(holder); Object value = expression.getValue(evaluationContextToUse, message); - if (value != null) { + if (value != null && messageChannel != null) { if (logger.isDebugEnabled()) { logger.debug("Sending expression result message to " + messageChannel + " " + "as part of '" + expressionType + "' transaction synchronization"); @@ -171,7 +171,7 @@ public class ExpressionEvaluatingTransactionSynchronizationProcessor extends Int } } } - else { + else if (messageChannel != null) { if (logger.isDebugEnabled()) { logger.debug("Sending received message to " + messageChannel + " as part of '" + expressionType + "' transaction synchronization"); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java index 750d4130ad..13565920f1 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java @@ -30,7 +30,6 @@ import org.springframework.integration.aggregator.MethodInvokingCorrelationStrat import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy; import org.springframework.integration.aggregator.ReleaseStrategy; import org.springframework.integration.aggregator.ResequencingMessageHandler; -import org.springframework.integration.channel.NullChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageGroup; @@ -60,7 +59,6 @@ public class ResequencerParserTests { ResequencingMessageHandler resequencer = TestUtils.getPropertyValue(endpoint, "handler", ResequencingMessageHandler.class); assertThat(getPropertyValue(resequencer, "outputChannel")).isNull(); - assertThat(getPropertyValue(resequencer, "discardChannel") instanceof NullChannel).isTrue(); assertThat(getPropertyValue( resequencer, "messagingTemplate.sendTimeout")) .as("The ResequencerEndpoint is not set with the appropriate timeout value").isEqualTo(-1L); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java index 5604e4f2e1..522f012289 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java @@ -30,7 +30,6 @@ import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy; import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy; import org.springframework.integration.aggregator.SimpleSequenceSizeReleaseStrategy; -import org.springframework.integration.channel.NullChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.MessageHandler; @@ -39,6 +38,7 @@ import org.springframework.messaging.MessageHandler; * @author Marius Bogoevici * @author Mark Fisher * @author Artem Bilan + * @author Gary Russell */ public class AggregatorAnnotationTests { @@ -51,7 +51,6 @@ public class AggregatorAnnotationTests { assertThat(getPropertyValue(aggregator, "releaseStrategy") instanceof SimpleSequenceSizeReleaseStrategy) .isTrue(); assertThat(getPropertyValue(aggregator, "outputChannel")).isNull(); - assertThat(getPropertyValue(aggregator, "discardChannel") instanceof NullChannel).isTrue(); assertThat(getPropertyValue(aggregator, "messagingTemplate.sendTimeout")).isEqualTo(-1L); assertThat(getPropertyValue(aggregator, "sendPartialResultOnExpiry")).isEqualTo(false); context.close(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java index 7d25649eee..5c15c3c313 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java @@ -96,10 +96,8 @@ public class PseudoTransactionalMessageSourceTests { } }); - MessageChannel afterCommitChannel = TestUtils.getPropertyValue(syncProcessor, "afterCommitChannel", - MessageChannel.class); - assertThat(afterCommitChannel).isInstanceOf(NullChannel.class); - + MessageChannel afterCommitChannel = new NullChannel(); + syncProcessor.setAfterCommitChannel(afterCommitChannel); Log logger = TestUtils.getPropertyValue(afterCommitChannel, "logger", Log.class); logger = Mockito.spy(logger); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java index a2d55d7202..8aa24fef24 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java @@ -45,39 +45,43 @@ import org.springframework.util.StringUtils; public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBean implements BeanNameAware, SmartLifecycle, ApplicationEventPublisherAware { - private volatile String nativeOptions; + private String nativeOptions; - private volatile boolean enableStatusReader = true; + private boolean enableStatusReader = true; - private volatile Long idleEventInterval; + private Long idleEventInterval; - private volatile File file; + private File file; - private volatile TaskExecutor taskExecutor; + private TaskExecutor taskExecutor; - private volatile TaskScheduler taskScheduler; + private TaskScheduler taskScheduler; - private volatile Long delay; + private Long delay; - private volatile Long fileDelay; + private Long fileDelay; - private volatile Boolean end; + private Boolean end; - private volatile Boolean reopen; + private Boolean reopen; - private volatile FileTailingMessageProducerSupport tailAdapter; + private FileTailingMessageProducerSupport tailAdapter; - private volatile String beanName; + private String beanName; - private volatile MessageChannel outputChannel; + private MessageChannel outputChannel; - private volatile MessageChannel errorChannel; + private MessageChannel errorChannel; - private volatile Boolean autoStartup; + private String outputChannelName; - private volatile Integer phase; + private String errorChannelName; - private volatile ApplicationEventPublisher applicationEventPublisher; + private Boolean autoStartup; + + private Integer phase; + + private ApplicationEventPublisher applicationEventPublisher; public void setNativeOptions(String nativeOptions) { if (StringUtils.hasText(nativeOptions)) { @@ -141,10 +145,18 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea this.outputChannel = outputChannel; } + public void setOutputChannelName(String outputChannelName) { + this.outputChannelName = outputChannelName; + } + public void setErrorChannel(MessageChannel errorChannel) { this.errorChannel = errorChannel; } + public void setErrorChannelName(String errorChannelName) { + this.errorChannelName = errorChannelName; + } + public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; } @@ -238,6 +250,8 @@ public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBea .acceptIfNotNull(this.autoStartup, adapter::setAutoStartup) .acceptIfNotNull(this.phase, adapter::setPhase) .acceptIfNotNull(this.applicationEventPublisher, adapter::setApplicationEventPublisher) + .acceptIfNotNull(this.outputChannelName, adapter::setOutputChannelName) + .acceptIfNotNull(this.errorChannelName, adapter::setErrorChannelName) .acceptIfNotNull(beanFactory, adapter::setBeanFactory); adapter.afterPropertiesSet(); this.tailAdapter = adapter; diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java index adfc7ffe0c..cd2edb739a 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java @@ -20,7 +20,6 @@ import java.io.File; import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.core.task.TaskExecutor; -import org.springframework.integration.channel.NullChannel; import org.springframework.integration.dsl.MessageProducerSpec; import org.springframework.integration.file.config.FileTailInboundChannelAdapterFactoryBean; import org.springframework.integration.file.tail.FileTailingMessageProducerSupport; @@ -32,6 +31,7 @@ import org.springframework.util.Assert; * A {@link MessageProducerSpec} for file tailing adapters. * * @author Artem Bilan + * @author Gary Russell * * @since 5.0 */ @@ -186,9 +186,6 @@ public class TailAdapterSpec extends MessageProducerSpec