diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java index 1f2364ba92..0b72fc2c28 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/support/DefaultAmqpHeaderMapper.java @@ -109,54 +109,40 @@ public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper headers = new HashMap(); try { JavaUtils.INSTANCE - .acceptIfNotNull(AmqpHeaders.APP_ID, amqpMessageProperties.getAppId(), - (key, value) -> headers.put(key, value)) - .acceptIfNotNull(AmqpHeaders.CLUSTER_ID, amqpMessageProperties.getClusterId(), - (key, value) -> headers.put(key, value)) + .acceptIfNotNull(AmqpHeaders.APP_ID, amqpMessageProperties.getAppId(), headers::put) + .acceptIfNotNull(AmqpHeaders.CLUSTER_ID, amqpMessageProperties.getClusterId(), headers::put) .acceptIfNotNull(AmqpHeaders.CONTENT_ENCODING, amqpMessageProperties.getContentEncoding(), - (key, value) -> headers.put(key, value)); + headers::put); long contentLength = amqpMessageProperties.getContentLength(); JavaUtils.INSTANCE - .acceptIfCondition(contentLength > 0, AmqpHeaders.CONTENT_LENGTH, contentLength, - (key, value) -> headers.put(key, value)) - .acceptIfHasText(AmqpHeaders.CONTENT_TYPE, amqpMessageProperties.getContentType(), - (key, value) -> headers.put(key, value)) - .acceptIfHasText(AmqpHeaders.CORRELATION_ID, amqpMessageProperties.getCorrelationId(), - (key, value) -> headers.put(key, value)) + .acceptIfCondition(contentLength > 0, AmqpHeaders.CONTENT_LENGTH, contentLength, headers::put) + .acceptIfHasText(AmqpHeaders.CONTENT_TYPE, amqpMessageProperties.getContentType(), headers::put) + .acceptIfHasText(AmqpHeaders.CORRELATION_ID, amqpMessageProperties.getCorrelationId(), headers::put) .acceptIfNotNull(AmqpHeaders.RECEIVED_DELIVERY_MODE, amqpMessageProperties.getReceivedDeliveryMode(), - (key, value) -> headers.put(key, value)); + headers::put); long deliveryTag = amqpMessageProperties.getDeliveryTag(); JavaUtils.INSTANCE - .acceptIfCondition(deliveryTag > 0, AmqpHeaders.DELIVERY_TAG, deliveryTag, - (key, value) -> headers.put(key, value)) - .acceptIfHasText(AmqpHeaders.EXPIRATION, amqpMessageProperties.getExpiration(), - (key, value) -> headers.put(key, value)); + .acceptIfCondition(deliveryTag > 0, AmqpHeaders.DELIVERY_TAG, deliveryTag, headers::put) + .acceptIfHasText(AmqpHeaders.EXPIRATION, amqpMessageProperties.getExpiration(), headers::put); Integer messageCount = amqpMessageProperties.getMessageCount(); JavaUtils.INSTANCE .acceptIfCondition(messageCount != null && messageCount > 0, AmqpHeaders.MESSAGE_COUNT, messageCount, - (key, value) -> headers.put(key, value)) - .acceptIfHasText(AmqpHeaders.MESSAGE_ID, amqpMessageProperties.getMessageId(), - (key, value) -> headers.put(key, value)); + headers::put) + .acceptIfHasText(AmqpHeaders.MESSAGE_ID, amqpMessageProperties.getMessageId(), headers::put); Integer priority = amqpMessageProperties.getPriority(); JavaUtils.INSTANCE .acceptIfCondition(priority != null && priority > 0, IntegrationMessageHeaderAccessor.PRIORITY, priority, (key, value) -> headers.put(key, value)) - .acceptIfNotNull(AmqpHeaders.RECEIVED_DELAY, amqpMessageProperties.getReceivedDelay(), - (key, value) -> headers.put(key, value)) + .acceptIfNotNull(AmqpHeaders.RECEIVED_DELAY, amqpMessageProperties.getReceivedDelay(), headers::put) .acceptIfNotNull(AmqpHeaders.RECEIVED_EXCHANGE, amqpMessageProperties.getReceivedExchange(), - (key, value) -> headers.put(key, value)) + headers::put) .acceptIfHasText(AmqpHeaders.RECEIVED_ROUTING_KEY, amqpMessageProperties.getReceivedRoutingKey(), - (key, value) -> headers.put(key, value)) - .acceptIfNotNull(AmqpHeaders.REDELIVERED, amqpMessageProperties.isRedelivered(), - (key, value) -> headers.put(key, value)) - .acceptIfNotNull(AmqpHeaders.REPLY_TO, amqpMessageProperties.getReplyTo(), - (key, value) -> headers.put(key, value)) - .acceptIfNotNull(AmqpHeaders.TIMESTAMP, amqpMessageProperties.getTimestamp(), - (key, value) -> headers.put(key, value)) - .acceptIfHasText(AmqpHeaders.TYPE, amqpMessageProperties.getType(), - (key, value) -> headers.put(key, value)) - .acceptIfHasText(AmqpHeaders.RECEIVED_USER_ID, amqpMessageProperties.getReceivedUserId(), - (key, value) -> headers.put(key, value)); + headers::put) + .acceptIfNotNull(AmqpHeaders.REDELIVERED, amqpMessageProperties.isRedelivered(), headers::put) + .acceptIfNotNull(AmqpHeaders.REPLY_TO, amqpMessageProperties.getReplyTo(), headers::put) + .acceptIfNotNull(AmqpHeaders.TIMESTAMP, amqpMessageProperties.getTimestamp(), headers::put) + .acceptIfHasText(AmqpHeaders.TYPE, amqpMessageProperties.getType(), headers::put) + .acceptIfHasText(AmqpHeaders.RECEIVED_USER_ID, amqpMessageProperties.getReceivedUserId(), headers::put); for (String jsonHeader : JsonHeaders.HEADERS) { Object value = amqpMessageProperties.getHeaders().get(jsonHeader.replaceFirst(JsonHeaders.PREFIX, "")); 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 50c57561de..1366ae4765 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 @@ -41,6 +41,7 @@ import org.springframework.integration.core.MessageProducer; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.support.context.NamedComponent; +import org.springframework.integration.util.JavaUtils; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.messaging.core.DestinationResolver; @@ -195,31 +196,32 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean ((ApplicationContextAware) this.handler).setApplicationContext(this.applicationContext)) + .acceptIfCondition(this.handler instanceof BeanFactoryAware && getBeanFactory() != null, + getBeanFactory(), + factory -> ((BeanFactoryAware) this.handler).setBeanFactory(factory)) + .acceptIfCondition(this.handler instanceof BeanNameAware && this.beanName != null, this.beanName, + namne -> ((BeanNameAware) this.handler).setBeanName(this.beanName)) + .acceptIfCondition(this.handler instanceof ApplicationEventPublisherAware + && this.applicationEventPublisher != null, + this.applicationEventPublisher, + publisher -> ((ApplicationEventPublisherAware) this.handler) + .setApplicationEventPublisher(publisher)); configureOutputChannelIfAny(); Object actualHandler = extractTarget(this.handler); if (actualHandler == null) { actualHandler = this.handler; } + final Object handlerToConfigure = actualHandler; // must final for lambdas if (actualHandler instanceof IntegrationObjectSupport) { - if (this.componentName != null) { - ((IntegrationObjectSupport) actualHandler).setComponentName(this.componentName); - } - if (this.channelResolver != null) { - ((IntegrationObjectSupport) actualHandler).setChannelResolver(this.channelResolver); - } + JavaUtils.INSTANCE + .acceptIfNotNull(this.componentName, + name -> ((IntegrationObjectSupport) handlerToConfigure).setComponentName(name)) + .acceptIfNotNull(this.channelResolver, + resolver -> ((IntegrationObjectSupport) handlerToConfigure).setChannelResolver(resolver)); } if (!CollectionUtils.isEmpty(this.adviceChain)) { if (actualHandler instanceof AbstractReplyProducingMessageHandler) { @@ -234,15 +236,12 @@ public abstract class AbstractSimpleMessageHandlerFactoryBean ((AbstractMessageProducingHandler) handlerToConfigure).setAsync(asyncValue)) + .acceptIfCondition(this.handler instanceof Orderable && this.order != null, + this.order, theOrder -> ((Orderable) this.handler).setOrder(theOrder)); this.initialized = true; } if (this.handler instanceof InitializingBean) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java index a0df3e9591..43fb1fcdbb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java @@ -29,6 +29,7 @@ import org.springframework.integration.aggregator.ReleaseStrategy; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.support.management.AbstractMessageHandlerMetrics; +import org.springframework.integration.util.JavaUtils; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.scheduling.TaskScheduler; @@ -107,6 +108,7 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe this.sendTimeout = sendTimeout; } + @Override public void setOutputChannelName(String outputChannelName) { this.outputChannelName = outputChannelName; } @@ -194,86 +196,27 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe } } AggregatingMessageHandler aggregator = new AggregatingMessageHandler(outputProcessor); - - if (this.expireGroupsUponCompletion != null) { - aggregator.setExpireGroupsUponCompletion(this.expireGroupsUponCompletion); - } - - if (this.sendTimeout != null) { - aggregator.setSendTimeout(this.sendTimeout); - } - - if (this.outputChannelName != null) { - aggregator.setOutputChannelName(this.outputChannelName); - } - - if (this.metrics != null) { - aggregator.configureMetrics(this.metrics); - } - - if (this.statsEnabled != null) { - aggregator.setStatsEnabled(this.statsEnabled); - } - - if (this.countsEnabled != null) { - aggregator.setCountsEnabled(this.countsEnabled); - } - - if (this.lockRegistry != null) { - aggregator.setLockRegistry(this.lockRegistry); - } - - if (this.messageStore != null) { - aggregator.setMessageStore(this.messageStore); - } - - if (this.correlationStrategy != null) { - aggregator.setCorrelationStrategy(this.correlationStrategy); - } - - if (this.releaseStrategy != null) { - aggregator.setReleaseStrategy(this.releaseStrategy); - } - - if (this.groupTimeoutExpression != null) { - aggregator.setGroupTimeoutExpression(this.groupTimeoutExpression); - } - - if (this.forceReleaseAdviceChain != null) { - aggregator.setForceReleaseAdviceChain(this.forceReleaseAdviceChain); - } - - if (this.taskScheduler != null) { - aggregator.setTaskScheduler(this.taskScheduler); - } - - if (this.discardChannel != null) { - aggregator.setDiscardChannel(this.discardChannel); - } - - if (this.discardChannelName != null) { - aggregator.setDiscardChannelName(this.discardChannelName); - } - - if (this.sendPartialResultOnExpiry != null) { - aggregator.setSendPartialResultOnExpiry(this.sendPartialResultOnExpiry); - } - - if (this.minimumTimeoutForEmptyGroups != null) { - aggregator.setMinimumTimeoutForEmptyGroups(this.minimumTimeoutForEmptyGroups); - } - - if (this.expireGroupsUponTimeout != null) { - aggregator.setExpireGroupsUponTimeout(this.expireGroupsUponTimeout); - } - - if (this.popSequence != null) { - aggregator.setPopSequence(this.popSequence); - } - - if (this.releaseLockBeforeSend != null) { - aggregator.setReleaseLockBeforeSend(this.releaseLockBeforeSend); - } + JavaUtils.INSTANCE + .acceptIfNotNull(this.expireGroupsUponCompletion, aggregator::setExpireGroupsUponCompletion) + .acceptIfNotNull(this.sendTimeout, aggregator::setSendTimeout) + .acceptIfNotNull(this.outputChannelName, aggregator::setOutputChannelName) + .acceptIfNotNull(this.metrics, aggregator::configureMetrics) + .acceptIfNotNull(this.statsEnabled, aggregator::setStatsEnabled) + .acceptIfNotNull(this.countsEnabled, aggregator::setCountsEnabled) + .acceptIfNotNull(this.lockRegistry, aggregator::setLockRegistry) + .acceptIfNotNull(this.messageStore, aggregator::setMessageStore) + .acceptIfNotNull(this.correlationStrategy, aggregator::setCorrelationStrategy) + .acceptIfNotNull(this.releaseStrategy, aggregator::setReleaseStrategy) + .acceptIfNotNull(this.groupTimeoutExpression, aggregator::setGroupTimeoutExpression) + .acceptIfNotNull(this.forceReleaseAdviceChain, aggregator::setForceReleaseAdviceChain) + .acceptIfNotNull(this.taskScheduler, aggregator::setTaskScheduler) + .acceptIfNotNull(this.discardChannel, aggregator::setDiscardChannel) + .acceptIfNotNull(this.discardChannelName, aggregator::setDiscardChannelName) + .acceptIfNotNull(this.sendPartialResultOnExpiry, aggregator::setSendPartialResultOnExpiry) + .acceptIfNotNull(this.minimumTimeoutForEmptyGroups, aggregator::setMinimumTimeoutForEmptyGroups) + .acceptIfNotNull(this.expireGroupsUponTimeout, aggregator::setExpireGroupsUponTimeout) + .acceptIfNotNull(this.popSequence, aggregator::setPopSequence) + .acceptIfNotNull(this.releaseLockBeforeSend, aggregator::setReleaseLockBeforeSend); return aggregator; }