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

GH-3067: Use default NullChannel instance if poss.

Browse Source 
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
This commit is contained in:
Gary Russell
2019-09-27 17:14:51 -04:00
committed by Artem Bilan
parent 0c7cae15da
commit 9994997917
8 changed files with 88 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java
View File

@@ -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;

28
spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java
View File

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

16
spring-integration-core/src/main/java/org/springframework/integration/transaction/ExpressionEvaluatingTransactionSynchronizationProcessor.java
View File

@@ -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");

2
spring-integration-core/src/test/java/org/springframework/integration/config/ResequencerParserTests.java
View File

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

3
spring-integration-core/src/test/java/org/springframework/integration/config/annotation/AggregatorAnnotationTests.java
View File

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

6
spring-integration-core/src/test/java/org/springframework/integration/endpoint/PseudoTransactionalMessageSourceTests.java
View File

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

48
spring-integration-file/src/main/java/org/springframework/integration/file/config/FileTailInboundChannelAdapterFactoryBean.java
View File

@@ -45,39 +45,43 @@ import org.springframework.util.StringUtils;
public class FileTailInboundChannelAdapterFactoryBean extends AbstractFactoryBean<FileTailingMessageProducerSupport>
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;

5
spring-integration-file/src/main/java/org/springframework/integration/file/dsl/TailAdapterSpec.java
View File

@@ -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<TailAdapterSpec, FileTa
@Override
protected FileTailingMessageProducerSupport doGet() {
if (this.outputChannel == null) {
this.factoryBean.setOutputChannel(new NullChannel());
}
FileTailingMessageProducerSupport tailingMessageProducerSupport = null;
try {
this.factoryBean.afterPropertiesSet();