diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java index 3b819b8915..1aea460ce5 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/outbound/AmqpOutboundEndpoint.java @@ -30,6 +30,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback; import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.context.Lifecycle; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.amqp.support.MappingUtils; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; @@ -96,6 +97,10 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint return this.expectReply ? "amqp:outbound-gateway" : "amqp:outbound-channel-adapter"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return this.expectReply ? super.getIntegrationPatternType() : IntegrationPatternType.outbound_channel_adapter; + } @Override public RabbitTemplate getRabbitTemplate() { @@ -168,6 +173,7 @@ public class AmqpOutboundEndpoint extends AbstractAmqpOutboundEndpoint private void send(String exchangeName, String routingKey, final Message requestMessage, CorrelationData correlationData) { + if (this.rabbitTemplate != null) { MessageConverter converter = this.rabbitTemplate.getMessageConverter(); org.springframework.amqp.core.Message amqpMessage = MappingUtils.mapMessage(requestMessage, converter, diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationPattern.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationPattern.java new file mode 100644 index 0000000000..f10a4d152c --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationPattern.java @@ -0,0 +1,37 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://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; + +/** + * Indicates that a component implements some Enterprise Integration Pattern. + * + * @author Artem Bilan + * + * @since 5.3 + * + * @see IntegrationPatternType + * @see EIP official site + */ +public interface IntegrationPattern { + + /** + * Return a pattern type this component implements. + * @return the {@link IntegrationPatternType} this component implements. + */ + IntegrationPatternType getIntegrationPatternType(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/IntegrationPatternType.java b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationPatternType.java new file mode 100644 index 0000000000..dc06a466b6 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/IntegrationPatternType.java @@ -0,0 +1,161 @@ +/* + * Copyright 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. + * You may obtain a copy of the License at + * + * https://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; + +import java.util.Arrays; +import java.util.Set; +import java.util.stream.Collectors; + +/** + * The Enterprise Integration Pattern types. + * Used to indicate which pattern a target component implements. + * + * @author Artem Bilan + * + * @since 5.3 + */ +public enum IntegrationPatternType { + + message_channel(IntegrationPatternCategory.messaging_channel), + + publish_subscribe_channel(IntegrationPatternCategory.messaging_channel), + + executor_channel(IntegrationPatternCategory.messaging_channel), + + pollable_channel(IntegrationPatternCategory.messaging_channel), + + reactive_channel(IntegrationPatternCategory.messaging_channel), + + null_channel(IntegrationPatternCategory.messaging_channel), + + bridge(IntegrationPatternCategory.messaging_endpoint), + + service_activator(IntegrationPatternCategory.messaging_endpoint), + + outbound_channel_adapter(IntegrationPatternCategory.messaging_endpoint), + + inbound_channel_adapter(IntegrationPatternCategory.messaging_endpoint), + + outbound_gateway(IntegrationPatternCategory.messaging_endpoint), + + inbound_gateway(IntegrationPatternCategory.messaging_endpoint), + + gateway(IntegrationPatternCategory.messaging_endpoint), + + splitter(IntegrationPatternCategory.message_routing), + + transformer(IntegrationPatternCategory.message_transformation), + + header_enricher(IntegrationPatternCategory.message_transformation), + + filter(IntegrationPatternCategory.message_routing), + + content_enricher(IntegrationPatternCategory.message_transformation), + + header_filter(IntegrationPatternCategory.message_transformation), + + claim_check_in(IntegrationPatternCategory.message_transformation), + + claim_check_out(IntegrationPatternCategory.message_transformation), + + aggregator(IntegrationPatternCategory.message_routing), + + resequencer(IntegrationPatternCategory.message_routing), + + barrier(IntegrationPatternCategory.message_routing), + + chain(IntegrationPatternCategory.message_routing), + + scatter_gather(IntegrationPatternCategory.message_routing), + + delayer(IntegrationPatternCategory.message_routing), + + control_bus(IntegrationPatternCategory.system_management), + + router(IntegrationPatternCategory.message_routing), + + recipient_list_router(IntegrationPatternCategory.message_routing); + + + private final IntegrationPatternCategory patternCategory; + + IntegrationPatternType(IntegrationPatternCategory patternCategory) { + this.patternCategory = patternCategory; + } + + public IntegrationPatternCategory getPatternCategory() { + return this.patternCategory; + } + + /** + * The Enterprise Integration Pattern categories. + * Used to indicate which pattern category a target component belongs. + */ + public enum IntegrationPatternCategory { + + messaging_channel( + message_channel, + publish_subscribe_channel, + executor_channel, + pollable_channel, + reactive_channel, + null_channel), + + messaging_endpoint( + service_activator, + outbound_channel_adapter, + inbound_channel_adapter, + outbound_gateway, + inbound_gateway, + gateway, + bridge), + + message_routing( + splitter, + filter, + aggregator, + resequencer, + barrier, + chain, + scatter_gather, + delayer, + router, + recipient_list_router), + + message_transformation( + transformer, + header_enricher, + content_enricher, + header_filter, + claim_check_in, + claim_check_out), + + system_management(control_bus); + + private final IntegrationPatternType[] patternTypes; + + IntegrationPatternCategory(IntegrationPatternType... patternTypes) { + this.patternTypes = patternTypes; + } + + public Set getPatternTypes() { + return Arrays.stream(this.patternTypes).collect(Collectors.toSet()); + } + + } + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java index 3e05de408c..d165eb598c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AggregatingMessageHandler.java @@ -18,6 +18,7 @@ package org.springframework.integration.aggregator; import java.util.Collection; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; @@ -40,6 +41,7 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler public AggregatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { + super(processor, store, correlationStrategy, releaseStrategy); } @@ -53,15 +55,18 @@ public class AggregatingMessageHandler extends AbstractCorrelatingMessageHandler /** * Will set the 'expireGroupsUponCompletion' flag. - * * @param expireGroupsUponCompletion true when groups should be expired on completion. - * * @see #afterRelease */ public void setExpireGroupsUponCompletion(boolean expireGroupsUponCompletion) { this.expireGroupsUponCompletion = expireGroupsUponCompletion; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.aggregator; + } + @Override protected boolean isExpireGroupsUponCompletion() { return this.expireGroupsUponCompletion; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java index da566ef981..26093e03b8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/BarrierMessageHandler.java @@ -22,6 +22,7 @@ import java.util.concurrent.SynchronousQueue; import java.util.concurrent.TimeUnit; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.DiscardingMessageHandler; import org.springframework.integration.handler.MessageTriggerAction; @@ -149,6 +150,11 @@ public class BarrierMessageHandler extends AbstractReplyProducingMessageHandler return "barrier"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.barrier; + } + @Override protected Object handleRequestMessage(Message requestMessage) { Object key = this.correlationStrategy.getCorrelationKey(requestMessage); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandler.java index 560fd54331..4f90edd02a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/FluxAggregatorMessageHandler.java @@ -23,6 +23,7 @@ import java.util.function.Predicate; import org.springframework.context.Lifecycle; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; import org.springframework.integration.handler.AbstractMessageProducingHandler; import org.springframework.messaging.Message; @@ -216,6 +217,16 @@ public class FluxAggregatorMessageHandler extends AbstractMessageProducingHandle this.windowConfigurer = windowConfigurer; } + @Override + public String getComponentType() { + return "flux-aggregator"; + } + + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.aggregator; + } + @Override public void start() { if (this.subscribed.compareAndSet(false, true)) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java index 2c7215a066..f32c3fee9f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/ResequencingMessageHandler.java @@ -18,6 +18,7 @@ package org.springframework.integration.aggregator; import java.util.Collection; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.SimpleMessageStore; @@ -35,16 +36,15 @@ import org.springframework.messaging.Message; */ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandler { - public ResequencingMessageHandler(MessageGroupProcessor processor, - MessageGroupStore store, CorrelationStrategy correlationStrategy, - ReleaseStrategy releaseStrategy) { + public ResequencingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, + CorrelationStrategy correlationStrategy, ReleaseStrategy releaseStrategy) { + super(processor, store, correlationStrategy, releaseStrategy); this.setExpireGroupsUponTimeout(false); } - public ResequencingMessageHandler(MessageGroupProcessor processor, - MessageGroupStore store) { + public ResequencingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store) { super(processor, store); this.setExpireGroupsUponTimeout(false); } @@ -56,10 +56,8 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle } /** - * {@inheritDoc} - * - * (overridden to false for a resequencer so late messages are immediately discarded rather - * than waiting for the next timeout) + * Overridden to false for a resequencer so late messages are immediately discarded rather + * than waiting for the next timeout */ @Override public final void setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout) { @@ -71,6 +69,11 @@ public class ResequencingMessageHandler extends AbstractCorrelatingMessageHandle return "resequencer"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.resequencer; + } + @Override protected boolean shouldCopyRequestHeaders() { return false; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java index 0fc355b8fd..3517effe53 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractExecutorChannel.java @@ -22,6 +22,7 @@ import java.util.Iterator; import java.util.List; import java.util.concurrent.Executor; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.dispatcher.AbstractDispatcher; import org.springframework.integration.support.MessagingExceptionWrapper; import org.springframework.lang.Nullable; @@ -128,6 +129,11 @@ public abstract class AbstractExecutorChannel extends AbstractSubscribableChanne return this.executorInterceptorsSize > 0; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.executor_channel; + } + protected class MessageHandlingTask implements Runnable { private final MessageHandlingRunnable delegate; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 40a93834be..e8fc1b1bca 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -30,6 +30,8 @@ import org.apache.commons.logging.Log; import org.springframework.beans.factory.BeanFactory; import org.springframework.core.OrderComparator; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.history.MessageHistory; @@ -68,7 +70,8 @@ import org.springframework.util.StringUtils; public abstract class AbstractMessageChannel extends IntegrationObjectSupport implements MessageChannel, TrackableComponent, ChannelInterceptorAware, org.springframework.integration.support.management.MessageChannelMetrics, - ConfigurableMetricsAware { + ConfigurableMetricsAware, + IntegrationPattern { protected final ChannelInterceptorList interceptors; // NOSONAR @@ -109,6 +112,11 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport return "channel"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.message_channel; + } + @Override public void setShouldTrack(boolean shouldTrack) { this.shouldTrack = shouldTrack; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java index 051165592e..dec6701a8b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java @@ -20,6 +20,7 @@ import java.util.ArrayDeque; import java.util.Deque; import java.util.List; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.support.management.PollableChannelManagement; import org.springframework.integration.support.management.metrics.CounterFacade; import org.springframework.integration.support.management.metrics.MetricsCaptor; @@ -64,6 +65,11 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel return getMetrics().getReceiveErrorCountLong(); } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.pollable_channel; + } + /** * Receive the first available message from this channel. If the channel * contains no messages, this method will block. 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 63594de511..adc3164e4c 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 @@ -22,6 +22,8 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.BeanNameAware; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.support.context.NamedComponent; import org.springframework.integration.support.management.AbstractMessageChannelMetrics; import org.springframework.integration.support.management.ConfigurableMetricsAware; @@ -49,7 +51,8 @@ import org.springframework.util.Assert; @SuppressWarnings("deprecation") public class NullChannel implements PollableChannel, org.springframework.integration.support.management.MessageChannelMetrics, - ConfigurableMetricsAware, BeanNameAware, NamedComponent { + ConfigurableMetricsAware, BeanNameAware, NamedComponent, + IntegrationPattern { private final Log logger = LogFactory.getLog(getClass()); @@ -105,6 +108,11 @@ public class NullChannel implements PollableChannel, return "null-channel"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.null_channel; + } + @Override public void registerMetricsCaptor(MetricsCaptor registry) { this.metricsCaptor = registry; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index a1ee3abc51..a60c8ec5a8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -19,6 +19,7 @@ package org.springframework.integration.channel; import java.util.concurrent.Executor; import org.springframework.beans.factory.BeanFactory; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.context.IntegrationProperties; import org.springframework.integration.dispatcher.BroadcastingDispatcher; import org.springframework.integration.util.ErrorHandlingTaskExecutor; @@ -69,6 +70,11 @@ public class PublishSubscribeChannel extends AbstractExecutorChannel { return "publish-subscribe-channel"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.publish_subscribe_channel; + } + /** * Provide an {@link ErrorHandler} strategy for handling Exceptions that * occur downstream from this channel. This will only be applied if diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveStreamsSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveStreamsSubscribableChannel.java index 7a05bf1db0..c05b96cc96 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveStreamsSubscribableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ReactiveStreamsSubscribableChannel.java @@ -18,6 +18,8 @@ package org.springframework.integration.channel; import org.reactivestreams.Publisher; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.messaging.Message; /** @@ -26,8 +28,13 @@ import org.springframework.messaging.Message; * * @since 5.0 */ -public interface ReactiveStreamsSubscribableChannel { +public interface ReactiveStreamsSubscribableChannel extends IntegrationPattern { void subscribeTo(Publisher> publisher); + @Override + default IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.reactive_channel; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java b/spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java index 28a989405d..57d8a4b6de 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/MessageSource.java @@ -16,6 +16,8 @@ package org.springframework.integration.core; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; @@ -23,16 +25,22 @@ import org.springframework.messaging.Message; * Base interface for any source of {@link Message Messages} that can be polled. * * @author Mark Fisher + * @author Artem Bilan */ @FunctionalInterface -public interface MessageSource { +public interface MessageSource extends IntegrationPattern { /** * Retrieve the next available message from this source. - * Returns null if no message is available. + * Returns {@code null} if no message is available. * @return The message or null. */ @Nullable Message receive(); + @Override + default IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.inbound_channel_adapter; + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java index c18733d860..a2efe5d562 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageProducerSupport.java @@ -19,6 +19,8 @@ package org.springframework.integration.endpoint; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.SmartInitializingSingleton; import org.springframework.core.AttributeAccessor; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.history.MessageHistory; @@ -43,7 +45,7 @@ import org.springframework.util.StringUtils; * @author Gary Russell */ public abstract class MessageProducerSupport extends AbstractEndpoint implements MessageProducer, TrackableComponent, - SmartInitializingSingleton { + SmartInitializingSingleton, IntegrationPattern { private final MessagingTemplate messagingTemplate = new MessagingTemplate(); @@ -152,6 +154,11 @@ public abstract class MessageProducerSupport extends AbstractEndpoint implements return this.messagingTemplate; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.inbound_channel_adapter; + } + @Override public void afterSingletonsInstantiated() { Assert.state(this.outputChannel != null || StringUtils.hasText(this.outputChannelName), 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 730bbdae14..2c87872402 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 @@ -20,6 +20,7 @@ import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; import org.springframework.core.convert.ConversionService; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.MessageRejectedException; import org.springframework.integration.core.MessageSelector; import org.springframework.integration.handler.AbstractReplyProducingPostProcessingMessageHandler; @@ -124,6 +125,11 @@ public class MessageFilter extends AbstractReplyProducingPostProcessingMessageHa return "filter"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.filter; + } + @Override protected void doInit() { Assert.state(!(this.discardChannelName != null && this.discardChannel != null), diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java index 120d26c2a0..619d75d81f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/GatewayProxyFactoryBean.java @@ -55,6 +55,7 @@ import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.annotation.Gateway; import org.springframework.integration.annotation.GatewayHeader; import org.springframework.integration.endpoint.AbstractEndpoint; @@ -942,6 +943,11 @@ public class GatewayProxyFactoryBean extends AbstractEndpoint setRequestMapper(messageMapper); } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.gateway; + } + @Nullable Expression getReceiveTimeoutExpression() { return this.receiveTimeoutExpression; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java index 43d70a4570..38362936d7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/gateway/MessagingGatewaySupport.java @@ -25,6 +25,8 @@ import org.reactivestreams.Subscriber; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.core.AttributeAccessor; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.MessageTimeoutException; import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; import org.springframework.integration.core.MessagingTemplate; @@ -75,7 +77,7 @@ import reactor.core.publisher.MonoProcessor; @IntegrationManagedResource public abstract class MessagingGatewaySupport extends AbstractEndpoint implements org.springframework.integration.support.management.TrackableComponent, - org.springframework.integration.support.management.MessageSourceMetrics { + org.springframework.integration.support.management.MessageSourceMetrics, IntegrationPattern { private static final long DEFAULT_TIMEOUT = 1000L; @@ -345,6 +347,11 @@ public abstract class MessagingGatewaySupport extends AbstractEndpoint return this.managementOverrides; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.inbound_gateway; + } + @Override protected void onInit() { Assert.state(!(this.requestChannelName != null && this.requestChannel != null), diff --git a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java index 1c08da38a5..77f5250be7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/graph/IntegrationNode.java @@ -21,9 +21,13 @@ import java.util.HashMap; import java.util.Map; import org.springframework.expression.Expression; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.context.ExpressionCapable; import org.springframework.integration.support.context.NamedComponent; import org.springframework.lang.Nullable; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; /** @@ -45,6 +49,12 @@ public abstract class IntegrationNode { private final String componentType; + @Nullable + private final IntegrationPatternType integrationPatternType; + + @Nullable + private final IntegrationPatternType.IntegrationPatternCategory integrationPatternCategory; + private final Map properties = new HashMap<>(); private final Map unmodifiableProperties = Collections.unmodifiableMap(this.properties); @@ -63,6 +73,26 @@ public abstract class IntegrationNode { this.properties.put("expression", expression.getExpressionString()); } } + + IntegrationPatternType patternType = null; + + if (nodeObject instanceof IntegrationPattern) { + patternType = ((IntegrationPattern) nodeObject).getIntegrationPatternType(); + } + else if (nodeObject instanceof MessageHandler) { + patternType = IntegrationPatternType.service_activator; + } + else if (nodeObject instanceof MessageChannel) { + patternType = IntegrationPatternType.message_channel; + } + + this.integrationPatternType = patternType; + if (this.integrationPatternType != null) { + this.integrationPatternCategory = this.integrationPatternType.getPatternCategory(); + } + else { + this.integrationPatternCategory = null; + } } public int getNodeId() { @@ -73,10 +103,20 @@ public abstract class IntegrationNode { return this.nodeName; } - public final String getComponentType() { + public final String getComponentType() { return this.componentType; } + @Nullable + public IntegrationPatternType getIntegrationPatternType() { + return this.integrationPatternType; + } + + @Nullable + public IntegrationPatternType.IntegrationPatternCategory getIntegrationPatternCategory() { + return this.integrationPatternCategory; + } + public Stats getStats() { return this.stats.isAvailable() ? this.stats : null; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index 719f60e2e6..7338020e02 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -22,6 +22,8 @@ import java.util.concurrent.ConcurrentHashMap; import org.reactivestreams.Subscription; import org.springframework.core.Ordered; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.Orderable; import org.springframework.integration.history.MessageHistory; @@ -60,7 +62,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler, org.springframework.integration.support.management.MessageHandlerMetrics, ConfigurableMetricsAware, - TrackableComponent, Orderable, CoreSubscriber> { + TrackableComponent, Orderable, CoreSubscriber>, + IntegrationPattern { private final ManagementOverrides managementOverrides = new ManagementOverrides(); @@ -139,6 +142,11 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport return this.managementOverrides; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.outbound_channel_adapter; + } + @Override protected void onInit() { if (this.statsEnabled) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java index 7be16dc673..88140738a0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java @@ -23,6 +23,7 @@ import org.aopalliance.aop.Advice; import org.springframework.aop.framework.ProxyFactory; import org.springframework.beans.factory.BeanClassLoaderAware; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.handler.advice.HandleMessageAdvice; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; @@ -91,6 +92,15 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa return this.beanClassLoader; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + // Most out-of-the-box Spring Integration implementations provide an outbound gateway + // for particular external protocol. If an implementation doesn't belong to this category, + // it overrides this method to provide its own specific integration pattern type: + // service-activator, splitter, aggregator, router etc. + return IntegrationPatternType.outbound_gateway; + } + @Override protected final void onInit() { super.onInit(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/BridgeHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/BridgeHandler.java index 3525b4c17d..66fac1a705 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/BridgeHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/BridgeHandler.java @@ -16,6 +16,7 @@ package org.springframework.integration.handler; +import org.springframework.integration.IntegrationPatternType; import org.springframework.messaging.Message; /** @@ -40,6 +41,11 @@ public class BridgeHandler extends AbstractReplyProducingMessageHandler { return "bridge"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.bridge; + } + @Override protected Object handleRequestMessage(Message requestMessage) { return requestMessage; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java index edf1f67878..bc94fa441b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java @@ -36,6 +36,7 @@ import org.springframework.expression.EvaluationContext; import org.springframework.expression.EvaluationException; import org.springframework.expression.Expression; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; @@ -284,6 +285,11 @@ public class DelayHandler extends AbstractReplyProducingMessageHandler implement return "delayer"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.delayer; + } + @Override protected void doInit() { if (this.messageStore == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionCommandMessageProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionCommandMessageProcessor.java index 449b833bff..87e111ee71 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionCommandMessageProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ExpressionCommandMessageProcessor.java @@ -31,6 +31,8 @@ import org.springframework.expression.MethodExecutor; import org.springframework.expression.MethodFilter; import org.springframework.expression.MethodResolver; import org.springframework.expression.spel.support.ReflectiveMethodResolver; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.CollectionUtils; @@ -46,7 +48,8 @@ import org.springframework.util.CollectionUtils; * * @since 2.0 */ -public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor { +public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor + implements IntegrationPattern { @Nullable private final MethodFilter methodFilter; @@ -75,6 +78,11 @@ public class ExpressionCommandMessageProcessor extends AbstractMessageProcessor< } } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.control_bus; + } + /** * Evaluates the Message payload expression as a command. * @throws IllegalArgumentException if the payload is not an Exception or String diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java index b62ac84ca3..3b995a9f47 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/MessageHandlerChain.java @@ -18,10 +18,12 @@ package org.springframework.integration.handler; import java.util.Collections; import java.util.HashSet; +import java.util.LinkedList; import java.util.List; import java.util.concurrent.locks.ReentrantLock; import org.springframework.context.Lifecycle; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.core.MessageProducer; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -66,18 +68,18 @@ import org.springframework.util.Assert; public class MessageHandlerChain extends AbstractMessageProducingHandler implements CompositeMessageHandler, Lifecycle { - private volatile List handlers; - - private volatile boolean initialized; - private final Object initializationMonitor = new Object(); - private volatile boolean running; - private final ReentrantLock lifecycleLock = new ReentrantLock(); + private List handlers; + + private volatile boolean initialized; + + private volatile boolean running; + public void setHandlers(List handlers) { - this.handlers = handlers; + this.handlers = new LinkedList<>(handlers); } @Override @@ -90,28 +92,25 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler return "chain"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.chain; + } + @Override protected void onInit() { super.onInit(); synchronized (this.initializationMonitor) { if (!this.initialized) { Assert.notEmpty(this.handlers, "handler list must not be empty"); - this.configureChain(); + configureChain(); this.initialized = true; } } } - @Override - protected void handleMessageInternal(Message message) { - if (!this.initialized) { - this.onInit(); - } - this.handlers.get(0).handleMessage(message); - } - private void configureChain() { - Assert.isTrue(this.handlers.size() == new HashSet(this.handlers).size(), + Assert.isTrue(this.handlers.size() == new HashSet<>(this.handlers).size(), "duplicate handlers are not allowed in a chain"); for (int i = 0; i < this.handlers.size(); i++) { MessageHandler handler = this.handlers.get(i); @@ -120,10 +119,11 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler "the last one in the chain must implement the MessageProducer interface."); MessageHandler nextHandler = this.handlers.get(i + 1); - MessageChannel nextChannel = (message, timeout) -> { - nextHandler.handleMessage(message); - return true; - }; + MessageChannel nextChannel = + (message, timeout) -> { + nextHandler.handleMessage(message); + return true; + }; ((MessageProducer) handler).setOutputChannel(nextChannel); @@ -146,6 +146,14 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler } } + @Override + protected void handleMessageInternal(Message message) { + if (!this.initialized) { + onInit(); + } + this.handlers.get(0).handleMessage(message); + } + @Override protected boolean shouldCopyRequestHeaders() { return false; @@ -171,7 +179,7 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler this.lifecycleLock.lock(); try { if (!this.running) { - this.doStart(); + doStart(); this.running = true; if (logger.isInfoEnabled()) { logger.info("started " + this); @@ -188,7 +196,7 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler this.lifecycleLock.lock(); try { if (this.running) { - this.doStop(); + doStop(); this.running = false; if (logger.isInfoEnabled()) { logger.info("stopped " + this); @@ -203,7 +211,7 @@ public class MessageHandlerChain extends AbstractMessageProducingHandler public final void stop(Runnable callback) { this.lifecycleLock.lock(); try { - this.stop(); + stop(); callback.run(); } finally { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyProducingMessageHandlerWrapper.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyProducingMessageHandlerWrapper.java index c72e35eaaf..a36487ee8d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyProducingMessageHandlerWrapper.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ReplyProducingMessageHandlerWrapper.java @@ -17,6 +17,8 @@ package org.springframework.integration.handler; import org.springframework.context.Lifecycle; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; @@ -44,6 +46,13 @@ public class ReplyProducingMessageHandlerWrapper extends AbstractReplyProducingM this.target = target; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return (this.target instanceof IntegrationPattern) + ? ((IntegrationPattern) this.target).getIntegrationPatternType() + : IntegrationPatternType.service_activator; + } + @Override protected Object handleRequestMessage(Message requestMessage) { this.target.handleMessage(requestMessage); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java index 5173c3425b..25a9d88368 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/ServiceActivatingHandler.java @@ -21,6 +21,8 @@ import java.lang.reflect.Method; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; import org.springframework.core.convert.ConversionService; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.annotation.ServiceActivator; import org.springframework.lang.Nullable; import org.springframework.messaging.Message; @@ -57,6 +59,13 @@ public class ServiceActivatingHandler extends AbstractReplyProducingMessageHandl return "service-activator"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return (this.processor instanceof IntegrationPattern) + ? ((IntegrationPattern) this.processor).getIntegrationPatternType() + : IntegrationPatternType.service_activator; + } + @Override protected void doInit() { if (this.processor instanceof AbstractMessageProcessor) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java index 7d27c6546f..3612ff62fd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/AbstractMessageRouter.java @@ -19,8 +19,10 @@ package org.springframework.integration.router; import java.util.Collection; import java.util.UUID; +import org.springframework.beans.factory.BeanFactory; import org.springframework.core.convert.ConversionService; import org.springframework.core.convert.support.DefaultConversionService; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.support.management.IntegrationManagedResource; @@ -61,8 +63,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler imple * fails to return any channels. If no default channel is provided and channel * resolution fails to return any channels, the router will throw an * {@link MessageDeliveryException}. - *

- * If messages shall be ignored (dropped) instead, please provide a + *

If messages shall be ignored (dropped) instead, please provide a * {@link org.springframework.integration.channel.NullChannel}. * @param defaultOutputChannel The default output channel. */ @@ -129,6 +130,11 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler imple return "router"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.router; + } + /** * Provides {@link MessagingTemplate} access for subclasses * @return The messaging template. @@ -151,8 +157,9 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler imple super.onInit(); Assert.state(!(this.defaultOutputChannelName != null && this.defaultOutputChannel != null), "'defaultOutputChannelName' and 'defaultOutputChannel' are mutually exclusive."); - if (this.getBeanFactory() != null) { - this.messagingTemplate.setBeanFactory(this.getBeanFactory()); + BeanFactory beanFactory = getBeanFactory(); + if (beanFactory != null) { + this.messagingTemplate.setBeanFactory(beanFactory); } } @@ -167,7 +174,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler imple @Override protected void handleMessageInternal(Message message) { boolean sent = false; - Collection results = this.determineTargetChannels(message); + Collection results = determineTargetChannels(message); if (results != null) { int sequenceSize = results.size(); int sequenceNumber = 1; @@ -179,10 +186,10 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler imple else { UUID id = message.getHeaders().getId(); messageToSend = getMessageBuilderFactory() - .fromMessage(message) - .pushSequenceDetails(id == null ? generateId() : id, - sequenceNumber++, sequenceSize) - .build(); + .fromMessage(message) + .pushSequenceDetails(id == null ? generateId() : id, + sequenceNumber++, sequenceSize) + .build(); } if (channel != null) { sent |= doSend(channel, messageToSend); @@ -195,7 +202,7 @@ public abstract class AbstractMessageRouter extends AbstractMessageHandler imple this.messagingTemplate.send(this.defaultOutputChannel, message); } else { - throw new MessageDeliveryException(message, "No channel resolved by router '" + this.getComponentName() + throw new MessageDeliveryException(message, "No channel resolved by router '" + this + "' and no 'defaultOutputChannel' defined."); } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java index 670b05f90c..bccac4c44e 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/router/RecipientListRouter.java @@ -30,6 +30,7 @@ import java.util.stream.Collectors; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.core.MessageSelector; import org.springframework.integration.filter.ExpressionEvaluatingSelector; import org.springframework.jmx.export.annotation.ManagedAttribute; @@ -250,6 +251,11 @@ public class RecipientListRouter extends AbstractMessageRouter implements Recipi return "recipient-list-router"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.recipient_list_router; + } + @Override protected Collection determineTargetChannels(Message message) { return this.recipients.stream() diff --git a/spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java index 83288d2bf9..8998fffc8f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java @@ -20,6 +20,7 @@ import org.springframework.aop.support.AopUtils; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanInitializationException; import org.springframework.context.Lifecycle; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.channel.FixedSubscriberChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; @@ -107,6 +108,16 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i this.errorChannelName = errorChannelName; } + @Override + public String getComponentType() { + return "scatter-gather"; + } + + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.scatter_gather; + } + @Override protected void doInit() { BeanFactory beanFactory = getBeanFactory(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index eb4f25532d..5e2988cd30 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java @@ -28,6 +28,7 @@ import java.util.stream.Stream; import org.reactivestreams.Publisher; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.DiscardingMessageHandler; @@ -107,6 +108,11 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess return this.discardChannel; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.splitter; + } + @Override protected void doInit() { Assert.state(!(this.discardChannelName != null && this.discardChannel != null), diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckInTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckInTransformer.java index c3bf430495..440d9f8e5f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckInTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckInTransformer.java @@ -18,8 +18,9 @@ package org.springframework.integration.transformer; import java.util.UUID; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.store.MessageStore; -import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -28,9 +29,11 @@ import org.springframework.util.Assert; * is the id of the stored Message. * * @author Mark Fisher + * @author Artem Bilan + * * @since 2.0 */ -public class ClaimCheckInTransformer extends AbstractTransformer { +public class ClaimCheckInTransformer extends AbstractTransformer implements IntegrationPattern { private final MessageStore messageStore; @@ -50,16 +53,18 @@ public class ClaimCheckInTransformer extends AbstractTransformer { return "claim-check-in"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.claim_check_in; + } + @Override protected Object doTransform(Message message) { Assert.notNull(message, "message must not be null"); UUID id = message.getHeaders().getId(); Assert.notNull(id, "ID header must not be null"); this.messageStore.addMessage(message); - AbstractIntegrationMessageBuilder responseBuilder = getMessageBuilderFactory().withPayload(id); - // headers on the 'current' message take precedence - responseBuilder.copyHeaders(message.getHeaders()); - return responseBuilder.build(); + return id; } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckOutTransformer.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckOutTransformer.java index 3f7c81a2ff..54f57f56c5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckOutTransformer.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ClaimCheckOutTransformer.java @@ -18,6 +18,8 @@ package org.springframework.integration.transformer; import java.util.UUID; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.store.MessageStore; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.messaging.Message; @@ -31,9 +33,11 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Oleg Zhurakousky * @author Nick Spacek + * @author Artem Bilan + * * @since 2.0 */ -public class ClaimCheckOutTransformer extends AbstractTransformer { +public class ClaimCheckOutTransformer extends AbstractTransformer implements IntegrationPattern { private final MessageStore messageStore; @@ -59,6 +63,11 @@ public class ClaimCheckOutTransformer extends AbstractTransformer { return "claim-check-out"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.claim_check_in; + } + @Override protected Object doTransform(Message message) { Assert.notNull(message, "message must not be null"); @@ -74,9 +83,9 @@ public class ClaimCheckOutTransformer extends AbstractTransformer { else { retrievedMessage = this.messageStore.getMessage(id); } - Assert.notNull(retrievedMessage, "unable to locate Message for ID: " + id - + " within MessageStore [" + this.messageStore + "]"); - AbstractIntegrationMessageBuilder responseBuilder = this.getMessageBuilderFactory().fromMessage(retrievedMessage); + Assert.notNull(retrievedMessage, + () -> "unable to locate Message for ID: " + id + " within MessageStore [" + this.messageStore + "]"); + AbstractIntegrationMessageBuilder responseBuilder = getMessageBuilderFactory().fromMessage(retrievedMessage); // headers on the 'current' message take precedence responseBuilder.copyHeaders(message.getHeaders()); return responseBuilder.build(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java index ab6e6e8330..d87db1b092 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/ContentEnricher.java @@ -28,6 +28,7 @@ import org.springframework.expression.Expression; import org.springframework.expression.spel.SpelParserConfiguration; import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; @@ -254,6 +255,11 @@ public class ContentEnricher extends AbstractReplyProducingMessageHandler implem return "enricher"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.content_enricher; + } + /** * Initializes the Content Enricher. Will instantiate an internal Gateway if the * requestChannel is set. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.java index ea49030944..f90bffde1a 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderEnricher.java @@ -22,6 +22,8 @@ import java.util.Map.Entry; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.handler.MessageProcessor; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; @@ -41,7 +43,7 @@ import org.springframework.messaging.MessageHeaders; * @author Artem Bilan * @author Gary Russell */ -public class HeaderEnricher extends IntegrationObjectSupport implements Transformer { +public class HeaderEnricher extends IntegrationObjectSupport implements Transformer, IntegrationPattern { private final Map> headersToAdd; @@ -92,6 +94,47 @@ public class HeaderEnricher extends IntegrationObjectSupport implements Transfor return "header-enricher"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.header_enricher; + } + + @Override + public void onInit() { + boolean shouldOverwrite = this.defaultOverwrite; + boolean checkReadOnlyHeaders = getMessageBuilderFactory() instanceof DefaultMessageBuilderFactory; + + for (Entry> entry : this.headersToAdd.entrySet()) { + if (checkReadOnlyHeaders && + (MessageHeaders.ID.equals(entry.getKey()) || MessageHeaders.TIMESTAMP.equals(entry.getKey()))) { + throw new BeanInitializationException( + "HeaderEnricher cannot override 'id' and 'timestamp' read-only headers.\n" + + "Wrong 'headersToAdd' [" + this.headersToAdd + + "] configuration for " + getComponentName()); + } + + HeaderValueMessageProcessor processor = entry.getValue(); + if (processor instanceof BeanFactoryAware && getBeanFactory() != null) { + ((BeanFactoryAware) processor).setBeanFactory(getBeanFactory()); + } + Boolean processorOverwrite = processor.isOverwrite(); + if (processorOverwrite != null) { + shouldOverwrite |= processorOverwrite; + } + } + + if (this.messageProcessor != null + && this.messageProcessor instanceof BeanFactoryAware + && getBeanFactory() != null) { + ((BeanFactoryAware) this.messageProcessor).setBeanFactory(getBeanFactory()); + } + + if (!shouldOverwrite && !this.shouldSkipNulls && logger.isWarnEnabled()) { + logger.warn(getComponentName() + + " is configured to not overwrite existing headers. 'shouldSkipNulls = false' will have no effect"); + } + } + @Override public Message transform(Message message) { MessageHeaders messageHeaders = message.getHeaders(); @@ -152,40 +195,4 @@ public class HeaderEnricher extends IntegrationObjectSupport implements Transfor } } - @Override - public void onInit() { - boolean shouldOverwrite = this.defaultOverwrite; - boolean checkReadOnlyHeaders = getMessageBuilderFactory() instanceof DefaultMessageBuilderFactory; - - for (Entry> entry : this.headersToAdd.entrySet()) { - if (checkReadOnlyHeaders && - (MessageHeaders.ID.equals(entry.getKey()) || MessageHeaders.TIMESTAMP.equals(entry.getKey()))) { - throw new BeanInitializationException( - "HeaderEnricher cannot override 'id' and 'timestamp' read-only headers.\n" + - "Wrong 'headersToAdd' [" + this.headersToAdd - + "] configuration for " + getComponentName()); - } - - HeaderValueMessageProcessor processor = entry.getValue(); - if (processor instanceof BeanFactoryAware && getBeanFactory() != null) { - ((BeanFactoryAware) processor).setBeanFactory(getBeanFactory()); - } - Boolean processorOverwrite = processor.isOverwrite(); - if (processorOverwrite != null) { - shouldOverwrite |= processorOverwrite; - } - } - - if (this.messageProcessor != null - && this.messageProcessor instanceof BeanFactoryAware - && getBeanFactory() != null) { - ((BeanFactoryAware) this.messageProcessor).setBeanFactory(getBeanFactory()); - } - - if (!shouldOverwrite && !this.shouldSkipNulls && logger.isWarnEnabled()) { - logger.warn(getComponentName() + - " is configured to not overwrite existing headers. 'shouldSkipNulls = false' will have no effect"); - } - } - } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderFilter.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderFilter.java index 1b917a47c8..4f211e6e01 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderFilter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/HeaderFilter.java @@ -19,6 +19,8 @@ package org.springframework.integration.transformer; import java.util.Arrays; import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.DefaultMessageBuilderFactory; @@ -36,7 +38,7 @@ import org.springframework.util.Assert; * * @since 2.0 */ -public class HeaderFilter extends IntegrationObjectSupport implements Transformer { +public class HeaderFilter extends IntegrationObjectSupport implements Transformer, IntegrationPattern { private final String[] headersToRemove; @@ -57,6 +59,11 @@ public class HeaderFilter extends IntegrationObjectSupport implements Transforme return "header-filter"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.header_filter; + } + @Override protected void onInit() { super.onInit(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java index 6b0803a0de..118204d0b1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/transformer/MessageTransformingHandler.java @@ -20,6 +20,8 @@ import java.util.Collection; import org.springframework.beans.factory.BeanFactoryAware; import org.springframework.context.Lifecycle; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.support.context.NamedComponent; import org.springframework.messaging.Message; @@ -60,6 +62,13 @@ public class MessageTransformingHandler extends AbstractReplyProducingMessageHan ((NamedComponent) this.transformer).getComponentType() : "transformer"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return (this.transformer instanceof IntegrationPattern) + ? ((IntegrationPattern) this.transformer).getIntegrationPatternType() + : IntegrationPatternType.transformer; + } + @Override public void addNotPropagatedHeaders(String... headers) { super.addNotPropagatedHeaders(headers); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java index 039605ce95..2f4bfa180f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/graph/IntegrationGraphServerTests.java @@ -215,6 +215,24 @@ public class IntegrationGraphServerTests { JsonPathUtils.evaluate(baos.toByteArray(), "$..links[?(@.from == " + routerNodeId + "&& @.to == " + fizChannelNodeId + ")]"); assertThat(jsonArray).hasSize(1); + + jsonArray = JsonPathUtils.evaluate(baos.toByteArray(), + "$..nodes[?(@.name == 'services.foo.serviceActivator.handler')]"); + + assertThat(jsonArray).hasSize(1); + + Map serviceActivator = (Map) jsonArray.get(0); + assertThat(serviceActivator).containsEntry("integrationPatternType", "service_activator"); + + jsonArray = JsonPathUtils.evaluate(baos.toByteArray(), + "$..nodes[?(@.name == 'polling')]"); + + assertThat(jsonArray).hasSize(1); + + serviceActivator = (Map) jsonArray.get(0); + assertThat(serviceActivator) + .containsEntry("integrationPatternType", "service_activator") + .containsEntry("integrationPatternCategory", "messaging_endpoint"); } @Test diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java index a996dfec8a..f60c340f60 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileWritingMessageHandler.java @@ -49,6 +49,7 @@ import org.springframework.context.Lifecycle; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.expression.spel.support.StandardEvaluationContext; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.file.support.FileExistsMode; import org.springframework.integration.file.support.FileUtils; @@ -302,7 +303,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand */ public void setCharset(String charset) { Assert.notNull(charset, "charset must not be null"); - Assert.isTrue(Charset.isSupported(charset), "Charset '" + charset + "' is not supported."); + Assert.isTrue(Charset.isSupported(charset), () -> "Charset '" + charset + "' is not supported."); this.charset = Charset.forName(charset); } @@ -412,6 +413,16 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand this.newFileCallback = newFileCallback; } + @Override + public String getComponentType() { + return this.expectReply ? "file:outbound-gateway" : "file:outbound-channel-adapter"; + } + + @Override + public IntegrationPatternType getIntegrationPatternType() { + return this.expectReply ? super.getIntegrationPatternType() : IntegrationPatternType.outbound_channel_adapter; + } + @Override protected void doInit() { this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); @@ -548,7 +559,7 @@ public class FileWritingMessageHandler extends AbstractReplyProducingMessageHand private File writeMessageToFile(Message requestMessage, File originalFileFromHeader, File tempFile, File resultFile, Object timestamp) throws IOException { - File fileToReturn = null; + File fileToReturn; Object payload = requestMessage.getPayload(); if (payload instanceof File) { fileToReturn = handleFileMessage((File) payload, tempFile, resultFile, requestMessage); diff --git a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyCommandMessageProcessor.java b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyCommandMessageProcessor.java index ccaf46308a..36afb682f0 100644 --- a/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyCommandMessageProcessor.java +++ b/spring-integration-groovy/src/main/java/org/springframework/integration/groovy/GroovyCommandMessageProcessor.java @@ -21,8 +21,11 @@ import java.io.UncheckedIOException; import java.util.Map; import java.util.UUID; +import org.springframework.integration.IntegrationPattern; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.scripting.AbstractScriptExecutingMessageProcessor; import org.springframework.integration.scripting.ScriptVariableGenerator; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.scripting.ScriptSource; import org.springframework.scripting.groovy.GroovyObjectCustomizer; @@ -45,10 +48,12 @@ import groovy.lang.GString; * * @since 2.0 */ -public class GroovyCommandMessageProcessor extends AbstractScriptExecutingMessageProcessor { +public class GroovyCommandMessageProcessor extends AbstractScriptExecutingMessageProcessor + implements IntegrationPattern { private GroovyObjectCustomizer customizer; + @Nullable private Binding binding; @@ -103,6 +108,11 @@ public class GroovyCommandMessageProcessor extends AbstractScriptExecutingMessag this.customizer = customizer; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return IntegrationPatternType.control_bus; + } + @Override protected ScriptSource getScriptSource(Message message) { Object payload = message.getPayload(); diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/BaseHttpInboundEndpoint.java b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/BaseHttpInboundEndpoint.java index 815d1f6aef..a685a82890 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/BaseHttpInboundEndpoint.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/inbound/BaseHttpInboundEndpoint.java @@ -29,6 +29,7 @@ import org.springframework.http.HttpEntity; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; import org.springframework.http.HttpStatus; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.gateway.MessagingGatewaySupport; @@ -341,7 +342,12 @@ public class BaseHttpInboundEndpoint extends MessagingGatewaySupport implements @Override public String getComponentType() { - return (this.expectReply) ? "http:inbound-gateway" : "http:inbound-channel-adapter"; + return this.expectReply ? "http:inbound-gateway" : "http:inbound-channel-adapter"; + } + + @Override + public IntegrationPatternType getIntegrationPatternType() { + return this.expectReply ? super.getIntegrationPatternType() : IntegrationPatternType.inbound_channel_adapter; } /** diff --git a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java index 70e2694768..2bc81894fd 100644 --- a/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java +++ b/spring-integration-http/src/main/java/org/springframework/integration/http/outbound/AbstractHttpRequestExecutingMessageHandler.java @@ -45,6 +45,7 @@ import org.springframework.http.HttpHeaders; import org.springframework.http.HttpMethod; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.expression.ExpressionEvalMap; import org.springframework.integration.expression.ExpressionUtils; import org.springframework.integration.expression.ValueExpression; @@ -266,6 +267,11 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac this.trustedSpel = trustedSpel; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return this.expectReply ? super.getIntegrationPatternType() : IntegrationPatternType.outbound_channel_adapter; + } + @Override protected void doInit() { BeanFactory beanFactory = getBeanFactory(); @@ -319,7 +325,7 @@ public abstract class AbstractHttpRequestExecutingMessageHandler extends Abstrac doConvertSetCookie(headers); } - AbstractIntegrationMessageBuilder replyBuilder = null; + AbstractIntegrationMessageBuilder replyBuilder; MessageBuilderFactory messageBuilderFactory = getMessageBuilderFactory(); if (httpResponse.hasBody()) { Object responseBody = httpResponse.getBody(); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java index d3cbbf4ad9..70bb10ab36 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java @@ -32,6 +32,7 @@ import javax.management.MBeanServerConnection; import javax.management.MalformedObjectNameException; import javax.management.ObjectName; +import org.springframework.integration.IntegrationPatternType; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.util.ClassUtils; import org.springframework.jmx.support.ObjectNameManager; @@ -124,6 +125,11 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa return this.expectReply ? "jmx:operation-invoking-outbound-gateway" : "jmx:operation-invoking-channel-adapter"; } + @Override + public IntegrationPatternType getIntegrationPatternType() { + return this.expectReply ? super.getIntegrationPatternType() : IntegrationPatternType.outbound_channel_adapter; + } + @Override protected Object handleRequestMessage(Message requestMessage) { ObjectName objectName = resolveObjectName(requestMessage); diff --git a/src/reference/asciidoc/graph.adoc b/src/reference/asciidoc/graph.adoc index 15baf67e43..52b5f88d6a 100644 --- a/src/reference/asciidoc/graph.adoc +++ b/src/reference/asciidoc/graph.adoc @@ -14,13 +14,15 @@ A Spring Integration application with only the default components would expose a { "contentDescriptor" : { "providerVersion" : "{project-version}", - "providerFormatVersion" : 1.1, + "providerFormatVersion" : 1.2, "provider" : "spring-integration", "name" : "myAppName:1.0" }, "nodes" : [ { "nodeId" : 1, "componentType" : "null-channel", + "integrationPatternType" : "null_channel", + "integrationPatternCategory" : "messaging_channel", "properties" : { }, "sendTimers" : { "successes" : { @@ -42,6 +44,8 @@ A Spring Integration application with only the default components would expose a }, { "nodeId" : 2, "componentType" : "publish-subscribe-channel", + "integrationPatternType" : "publish_subscribe_channel", + "integrationPatternCategory" : "messaging_channel", "properties" : { }, "sendTimers" : { "successes" : { @@ -59,6 +63,8 @@ A Spring Integration application with only the default components would expose a }, { "nodeId" : 3, "componentType" : "logging-channel-adapter", + "integrationPatternType" : "outbound_channel_adapter", + "integrationPatternCategory" : "messaging_endpoint", "properties" : { }, "output" : null, "input" : "errorChannel", @@ -219,13 +225,14 @@ The preceding gateway produces nodes similar to the following: ==== [source,json] - ---- { "nodeId" : 10, "name" : "gate.bar(class java.lang.String)", "stats" : null, "componentType" : "gateway", + "integrationPatternType" : "gateway", + "integrationPatternCategory" : "messaging_endpoint", "output" : "four", "errors" : null }, @@ -234,6 +241,8 @@ The preceding gateway produces nodes similar to the following: "name" : "gate.foo(class java.lang.String)", "stats" : null, "componentType" : "gateway", + "integrationPatternType" : "gateway", + "integrationPatternCategory" : "messaging_endpoint", "output" : "four", "errors" : null }, @@ -242,6 +251,8 @@ The preceding gateway produces nodes similar to the following: "name" : "gate.foo(class java.lang.Integer)", "stats" : null, "componentType" : "gateway", + "integrationPatternType" : "gateway", + "integrationPatternCategory" : "messaging_endpoint", "output" : "four", "errors" : null } @@ -251,6 +262,9 @@ The preceding gateway produces nodes similar to the following: You can use this `IntegrationNode` hierarchy for parsing the graph model on the client side as well as to understand the general Spring Integration runtime behavior. See also <<./overview.adoc#programming-tips,Programming Tips and Tricks>> for more information. +Version 5.3 introduced an `IntegrationPattern` abstraction and all out-of-the-box components, which represent an Enterprise Integration Pattern (EIP), implement this abstraction and provide an `IntegrationPatternType` enum value. +This information can be useful for some categorizing logic in the target application or, being exposed into the graph node, it can be used by a UI to determine how to draw the component. + === Integration Graph Controller If your application is web-based (or built on top of Spring Boot with an embedded web container) and the Spring Integration HTTP or WebFlux module (see <<./http.adoc#http,HTTP Support>> and <<./webflux.adoc#webflux,WebFlux Support>>, respectively) is present on the classpath, you can use a `IntegrationGraphController` to expose the `IntegrationGraphServer` functionality as a REST service. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index b2dcca40b3..4f169d9765 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -15,6 +15,11 @@ If you are interested in more details, see the Issue Tracker tickets that were r [[x5.3-new-components]] === New Components +[[x5.3-integration-pattern]] +==== Integration Pattern + +The `IntegrationPattern` abstraction has been introduced to indicate which enterprise integration pattern (an `IntegrationPatternType`) and category a Spring Integration component belongs to. +See its JavaDocs and <<./graph.adoc#integration-graph,Integration Graph>> for more information about this abstraction and its use-cases. + [[x5.3-general]] === General Changes -