diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java index e040009220..8fe40f970f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/FixedSubscriberChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2016 the original author or authors. + * Copyright 2014-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. @@ -34,16 +34,18 @@ import org.springframework.messaging.SubscribableChannel; * Note: Stopping ({@link #unsubscribe(MessageHandler)}) the subscribed ({@link MessageHandler}) has no effect. * * @author Gary Russell + * @author Artem Bilan + * * @since 4.0 * */ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNameAware, NamedComponent { - private final Log logger = LogFactory.getLog(FixedSubscriberChannel.class); + private static final Log LOGGER = LogFactory.getLog(FixedSubscriberChannel.class); private final MessageHandler handler; - private volatile String beanName; + private String beanName; public FixedSubscriberChannel() { throw new IllegalArgumentException("Cannot instantiate a " + this.getClass().getSimpleName() @@ -59,9 +61,14 @@ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNa this.beanName = name; } + @Override + public String getBeanName() { + return this.beanName; + } + @Override public boolean send(Message message) { - return this.send(message, 0); + return send(message, 0); } @Override @@ -83,23 +90,23 @@ public final class FixedSubscriberChannel implements SubscribableChannel, BeanNa @Override public boolean subscribe(MessageHandler handler) { - if (handler != this.handler && this.logger.isDebugEnabled()) { - this.logger.debug(this.getComponentName() + ": cannot be subscribed to (it has a fixed single subscriber)."); + if (handler != this.handler && LOGGER.isDebugEnabled()) { + LOGGER.debug(getComponentName() + ": cannot be subscribed to (it has a fixed single subscriber)."); } return false; } @Override public boolean unsubscribe(MessageHandler handler) { - if (this.logger.isDebugEnabled()) { - this.logger.debug(this.getComponentName() + ": cannot be unsubscribed from (it has a fixed single subscriber)."); + if (LOGGER.isDebugEnabled()) { + LOGGER.debug(getComponentName() + ": cannot be unsubscribed from (it has a fixed single subscriber)."); } return false; } @Override public String getComponentType() { - return "Fixed Subscriber Channel"; + return "fixed-subscriber-channel"; } @Override 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 7cc753c166..3702e609a7 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 @@ -85,6 +85,12 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, this.managementOverrides.loggingConfigured = true; } + @Override + @Nullable + public String getBeanName() { + return this.beanName; + } + @Override @Nullable public String getComponentName() { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java index 6662d90ac7..0643242c1e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AbstractSimpleMessageHandlerFactoryBean.java @@ -219,7 +219,7 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean extends AbstractExpressionEvaluat private String managedName; - private volatile boolean countsEnabled; + private boolean countsEnabled; - private volatile boolean loggingEnabled = true; + private boolean loggingEnabled = true; private MetricsCaptor metricsCaptor; @@ -83,6 +83,11 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat this.beanName = name; } + @Override + public String getBeanName() { + return this.beanName; + } + @Override public void setManagedType(String managedType) { this.managedType = managedType; @@ -216,14 +221,9 @@ public abstract class AbstractMessageSource extends AbstractExpressionEvaluat } private Map evaluateHeaders() { - Map results = new HashMap<>(); - for (Map.Entry entry : this.headerExpressions.entrySet()) { - Object headerValue = this.evaluateExpression(entry.getValue()); - if (headerValue != null) { - results.put(entry.getKey(), headerValue); - } - } - return results; + return ExpressionEvalMap.from(this.headerExpressions) + .usingEvaluationContext(getEvaluationContext()) + .build(); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java index b163c581bf..8374a68da4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/filter/MessageFilter.java @@ -16,6 +16,7 @@ package org.springframework.integration.filter; +import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; import org.springframework.core.convert.ConversionService; @@ -47,11 +48,11 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa private final MessageSelector selector; - private volatile boolean throwExceptionOnRejection; + private boolean throwExceptionOnRejection; - private volatile MessageChannel discardChannel; + private MessageChannel discardChannel; - private volatile String discardChannelName; + private String discardChannelName; /** * Create a MessageFilter that will delegate to the given {@link MessageSelector}. @@ -109,13 +110,10 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa @Override public MessageChannel getDiscardChannel() { - if (this.discardChannelName != null) { - synchronized (this) { - if (this.discardChannelName != null) { - this.discardChannel = getChannelResolver().resolveDestination(this.discardChannelName); - this.discardChannelName = null; - } - } + String channelName = this.discardChannelName; + if (channelName != null) { + this.discardChannel = getChannelResolver().resolveDestination(channelName); + this.discardChannelName = null; } return this.discardChannel; } @@ -129,15 +127,16 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa @Override protected void doInit() { Assert.state(!(this.discardChannelName != null && this.discardChannel != null), - "'discardChannelName' and 'discardChannel' are mutually exclusive."); + "'discardChannelName' and 'discardChannel' are mutually exclusive."); if (this.selector instanceof AbstractMessageProcessingSelector) { ConversionService conversionService = getConversionService(); if (conversionService != null) { ((AbstractMessageProcessingSelector) this.selector).setConversionService(conversionService); } } - if (this.selector instanceof BeanFactoryAware && this.getBeanFactory() != null) { - ((BeanFactoryAware) this.selector).setBeanFactory(this.getBeanFactory()); + BeanFactory beanFactory = getBeanFactory(); + if (this.selector instanceof BeanFactoryAware && beanFactory != null) { + ((BeanFactoryAware) this.selector).setBeanFactory(beanFactory); } } @@ -178,8 +177,7 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa this.messagingTemplate.send(channel, message); } if (this.throwExceptionOnRejection) { - throw new MessageRejectedException(message, "MessageFilter '" + this.getComponentName() - + "' rejected Message"); + throw new MessageRejectedException(message, "MessageFilter '" + getBeanName() + "' rejected Message"); } } return result; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/context/NamedComponent.java b/spring-integration-core/src/main/java/org/springframework/integration/support/context/NamedComponent.java index 6880c2d8ea..b2a75fe57f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/context/NamedComponent.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/context/NamedComponent.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-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. @@ -18,6 +18,8 @@ package org.springframework.integration.support.context; /** * @author Mark Fisher + * @author Artem Bilan + * * @since 2.0 */ public interface NamedComponent { @@ -26,4 +28,8 @@ public interface NamedComponent { String getComponentType(); + default String getBeanName() { + return getComponentName(); + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageHandlerMetrics.java index 89359f8026..ec0b8a97a5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageHandlerMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageHandlerMetrics.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-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. @@ -23,6 +23,8 @@ import org.springframework.util.Assert; * Adds {@link TrackableComponent}. * * @author Gary Russell + * @author Artem Bilan + * * @since 4.2 */ @IntegrationManagedResource @@ -37,6 +39,11 @@ public class LifecycleTrackableMessageHandlerMetrics extends LifecycleMessageHan this.trackable = (TrackableComponent) delegate; } + @Override + public String getBeanName() { + return this.trackable.getBeanName(); + } + @Override public String getComponentName() { return this.trackable.getComponentName(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceMetrics.java index bc04b9788e..e5f21f8c74 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceMetrics.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-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. @@ -23,6 +23,8 @@ import org.springframework.util.Assert; * Adds {@link TrackableComponent}. * * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ @IntegrationManagedResource @@ -37,6 +39,11 @@ public class LifecycleTrackableMessageSourceMetrics extends LifecycleMessageSour this.trackable = (TrackableComponent) lifecycle; } + @Override + public String getBeanName() { + return this.trackable.getBeanName(); + } + @Override public String getComponentName() { return this.trackable.getComponentName(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/TrackableRouterMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/TrackableRouterMetrics.java index 1b0dcb65ad..0b45b4480d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/TrackableRouterMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/TrackableRouterMetrics.java @@ -1,5 +1,5 @@ /* - * Copyright 2015 the original author or authors. + * Copyright 2015-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. @@ -23,6 +23,8 @@ import org.springframework.util.Assert; * Adds {@link TrackableComponent}. * * @author Gary Russell + * @author Artem Bilan + * * @since 2.0 */ public class TrackableRouterMetrics extends RouterMetrics implements TrackableComponent { @@ -35,6 +37,11 @@ public class TrackableRouterMetrics extends RouterMetrics implements TrackableCo this.trackable = (TrackableComponent) delegate; } + @Override + public String getBeanName() { + return this.trackable.getBeanName(); + } + @Override public String getComponentName() { return this.trackable.getComponentName(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java index fb925d6f96..eac17505a3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java @@ -72,6 +72,7 @@ import org.springframework.integration.dsl.Transformers; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.GenericHandler; +import org.springframework.integration.handler.LoggingHandler; import org.springframework.integration.handler.advice.ErrorMessageSendingRecoverer; import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice; import org.springframework.integration.handler.advice.RequestHandlerRetryAdvice; @@ -697,11 +698,16 @@ public class IntegrationFlowTests { return tpte; } + @Bean + public MessageHandler loggingMessageHandler() { + return new LoggingHandler(LoggingHandler.Level.DEBUG); + } + @Bean public IntegrationFlow wireTapFlow1() { return IntegrationFlows.from("tappedChannel1") .wireTap("tapChannel", wt -> wt.selector(m -> m.getPayload().equals("foo"))) - .channel("nullChannel") + .handle(loggingMessageHandler()) .get(); } @@ -709,7 +715,7 @@ public class IntegrationFlowTests { public IntegrationFlow wireTapFlow2() { return f -> f .wireTap("tapChannel", wt -> wt.selector(m -> m.getPayload().equals("foo"))) - .channel("nullChannel"); + .handle(loggingMessageHandler()); } @Bean @@ -717,7 +723,7 @@ public class IntegrationFlowTests { return f -> f .transform("payload") .wireTap("tapChannel", wt -> wt.selector("payload == 'foo'")) - .channel("nullChannel"); + .handle(loggingMessageHandler()); } @Bean