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

Add Nullability support into Java DSL

Browse Source
This commit is contained in:
abilan
2023-04-14 14:16:36 -04:00
parent 053cc00484
commit d5181bf0d7
105 changed files with 514 additions and 366 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-core/src/main/java/org/springframework/integration/config/CorrelationStrategyFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2019 the original author or authors.
* Copyright 2002-2023 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.
@@ -25,6 +25,7 @@ import org.springframework.integration.aggregator.CorrelationStrategy;
import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy;
import org.springframework.integration.aggregator.MethodInvokingCorrelationStrategy;
import org.springframework.integration.util.MessagingAnnotationUtils;
import org.springframework.lang.NonNull;
import org.springframework.util.StringUtils;
/**
@@ -75,6 +76,7 @@ public class CorrelationStrategyFactoryBean implements FactoryBean<CorrelationSt
}
}
@NonNull
@Override
public CorrelationStrategy getObject() {
return this.strategy;

4
spring-integration-core/src/main/java/org/springframework/integration/config/ReleaseStrategyFactoryBean.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2020 the original author or authors.
* Copyright 2002-2023 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.
@@ -27,6 +27,7 @@ import org.springframework.integration.aggregator.MethodInvokingReleaseStrategy;
import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.aggregator.SimpleSequenceSizeReleaseStrategy;
import org.springframework.integration.util.MessagingAnnotationUtils;
import org.springframework.lang.NonNull;
import org.springframework.util.StringUtils;
/**
@@ -87,6 +88,7 @@ public class ReleaseStrategyFactoryBean implements FactoryBean<ReleaseStrategy>,
}
}
@NonNull
@Override
public ReleaseStrategy getObject() {
return this.strategy;

2
spring-integration-core/src/main/java/org/springframework/integration/context/IntegrationObjectSupport.java
View File

@@ -98,7 +98,7 @@ public abstract class IntegrationObjectSupport implements BeanNameAware, NamedCo
private boolean initialized;
@Override
public final void setBeanName(String beanName) {
public final void setBeanName(@Nullable String beanName) {
this.beanName = beanName;
}

5
spring-integration-core/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2023 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.
@@ -27,6 +27,7 @@ import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGro
import org.springframework.integration.aggregator.MessageGroupProcessor;
import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor;
import org.springframework.integration.store.MessageGroup;
import org.springframework.lang.Nullable;
/**
* A {@link CorrelationHandlerSpec} for an {@link AggregatingMessageHandler}.
@@ -65,7 +66,7 @@ public class AggregatorSpec extends CorrelationHandlerSpec<AggregatorSpec, Aggre
* target object must have an {@link org.springframework.integration.annotation.Aggregator} annotation).
* @return the handler spec.
*/
public AggregatorSpec processor(Object target, String methodName) {
public AggregatorSpec processor(Object target, @Nullable String methodName) {
return super.processor(target)
.outputProcessor(methodName != null
? new MethodInvokingMessageGroupProcessor(target, methodName)

191
spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java
View File

@@ -150,7 +150,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
}
protected B addComponents(Map<Object, String> components) {
if (components != null) {
if (!CollectionUtils.isEmpty(components)) {
this.integrationComponents.putAll(components);
}
return _this();
@@ -223,7 +223,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param messageChannelName the bean name to use.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B fixedSubscriberChannel(String messageChannelName) {
public B fixedSubscriberChannel(@Nullable String messageChannelName) {
return channel(new FixedSubscriberChannelPrototype(messageChannelName));
}
@@ -305,7 +305,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* {@link PublishSubscribeSpec} options including 'subflow' definition.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B publishSubscribeChannel(Executor executor,
public B publishSubscribeChannel(@Nullable Executor executor,
Consumer<PublishSubscribeSpec> publishSubscribeChannelConfigurer) {
Assert.notNull(publishSubscribeChannelConfigurer, "'publishSubscribeChannelConfigurer' must not be null");
@@ -411,7 +411,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param wireTapConfigurer the {@link Consumer} to accept options for the {@link WireTap}.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B wireTap(IntegrationFlow flow, Consumer<WireTapSpec> wireTapConfigurer) {
public B wireTap(IntegrationFlow flow, @Nullable Consumer<WireTapSpec> wireTapConfigurer) {
MessageChannel wireTapChannel = obtainInputChannelFromFlow(flow);
return wireTap(wireTapChannel, wireTapConfigurer);
@@ -473,7 +473,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param wireTapConfigurer the {@link Consumer} to accept options for the {@link WireTap}.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B wireTap(MessageChannel wireTapChannel, Consumer<WireTapSpec> wireTapConfigurer) {
public B wireTap(MessageChannel wireTapChannel, @Nullable Consumer<WireTapSpec> wireTapConfigurer) {
WireTapSpec wireTapSpec = new WireTapSpec(wireTapChannel);
if (wireTapConfigurer != null) {
wireTapConfigurer.accept(wireTapSpec);
@@ -528,7 +528,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see ExpressionCommandMessageProcessor
* @see GenericEndpointSpec
*/
public B controlBus(Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
public B controlBus(@Nullable Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
return handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(
new ControlBusMethodFilter())), endpointConfigurer);
}
@@ -553,7 +553,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see ExpressionEvaluatingTransformer
*/
public B transform(String expression,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
Assert.hasText(expression, "'expression' must not be empty");
return transform(null,
@@ -580,7 +580,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see MethodInvokingTransformer
*/
public B transform(Object service, String methodName) {
public B transform(Object service, @Nullable String methodName) {
return transform(service, methodName, null);
}
@@ -593,8 +593,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see ExpressionEvaluatingTransformer
*/
public B transform(Object service, String methodName,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
public B transform(Object service, @Nullable String methodName,
@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
MethodInvokingTransformer transformer;
if (StringUtils.hasText(methodName)) {
@@ -676,7 +676,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see MethodInvokingTransformer
* @see LambdaMessageProcessor
*/
public <P, T> B transform(Class<P> expectedType, GenericTransformer<P, T> genericTransformer) {
public <P, T> B transform(@Nullable Class<P> expectedType, GenericTransformer<P, T> genericTransformer) {
return transform(expectedType, genericTransformer, null);
}
@@ -716,8 +716,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see LambdaMessageProcessor
* @see GenericEndpointSpec
*/
public <P, T> B transform(Class<P> expectedType, GenericTransformer<P, T> genericTransformer,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
public <P, T> B transform(@Nullable Class<P> expectedType, GenericTransformer<P, T> genericTransformer,
@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
Assert.notNull(genericTransformer, "'genericTransformer' must not be null");
Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer :
@@ -750,7 +750,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see FilterEndpointSpec
*/
public B filter(String expression, Consumer<FilterEndpointSpec> endpointConfigurer) {
public B filter(String expression, @Nullable Consumer<FilterEndpointSpec> endpointConfigurer) {
Assert.hasText(expression, "'expression' must not be empty");
return filter(null, new ExpressionEvaluatingSelector(expression), endpointConfigurer);
}
@@ -774,7 +774,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see MethodInvokingSelector
*/
public B filter(Object service, String methodName) {
public B filter(Object service, @Nullable String methodName) {
return filter(service, methodName, null);
}
@@ -787,7 +787,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see MethodInvokingSelector
*/
public B filter(Object service, String methodName, Consumer<FilterEndpointSpec> endpointConfigurer) {
public B filter(Object service, @Nullable String methodName,
@Nullable Consumer<FilterEndpointSpec> endpointConfigurer) {
MethodInvokingSelector selector =
StringUtils.hasText(methodName)
? new MethodInvokingSelector(service, methodName)
@@ -826,7 +828,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B filter(MessageProcessorSpec<?> messageProcessorSpec, Consumer<FilterEndpointSpec> endpointConfigurer) {
public B filter(MessageProcessorSpec<?> messageProcessorSpec,
@Nullable Consumer<FilterEndpointSpec> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.getObject();
return addComponent(processor)
@@ -850,7 +854,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see LambdaMessageProcessor
*/
public <P> B filter(Class<P> expectedType, GenericSelector<P> genericSelector) {
public <P> B filter(@Nullable Class<P> expectedType, GenericSelector<P> genericSelector) {
return filter(expectedType, genericSelector, null);
}
@@ -874,15 +878,15 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see LambdaMessageProcessor
* @see FilterEndpointSpec
*/
public <P> B filter(Class<P> expectedType, GenericSelector<P> genericSelector,
Consumer<FilterEndpointSpec> endpointConfigurer) {
public <P> B filter(@Nullable Class<P> expectedType, GenericSelector<P> genericSelector,
@Nullable Consumer<FilterEndpointSpec> endpointConfigurer) {
Assert.notNull(genericSelector, "'genericSelector' must not be null");
MessageSelector selector = genericSelector instanceof MessageSelector ? (MessageSelector) genericSelector :
(ClassUtils.isLambda(genericSelector.getClass())
? new MethodInvokingSelector(new LambdaMessageProcessor(genericSelector, expectedType))
: new MethodInvokingSelector(genericSelector, ClassUtils.SELECTOR_ACCEPT_METHOD));
return this.register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer);
return register(new FilterEndpointSpec(new MessageFilter(selector)), endpointConfigurer);
}
/**
@@ -926,7 +930,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param methodName the method to invoke.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B handle(String beanName, String methodName) {
public B handle(String beanName, @Nullable String methodName) {
return handle(beanName, methodName, null);
}
@@ -940,8 +944,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B handle(String beanName, String methodName,
Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
public B handle(String beanName, @Nullable String methodName,
@Nullable Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
return handle(new ServiceActivatingHandler(new BeanNameMessageProcessor<>(beanName, methodName)),
endpointConfigurer);
}
@@ -966,7 +971,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param methodName the method to invoke.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B handle(Object service, String methodName) {
public B handle(Object service, @Nullable String methodName) {
return handle(service, methodName, null);
}
@@ -980,8 +985,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B handle(Object service, String methodName,
Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
public B handle(Object service, @Nullable String methodName,
@Nullable Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
ServiceActivatingHandler handler;
if (StringUtils.hasText(methodName)) {
@@ -1011,7 +1016,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see LambdaMessageProcessor
*/
public <P> B handle(Class<P> expectedType, GenericHandler<P> handler) {
public <P> B handle(@Nullable Class<P> expectedType, GenericHandler<P> handler) {
return handle(expectedType, handler, null);
}
@@ -1035,8 +1040,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see LambdaMessageProcessor
*/
public <P> B handle(Class<P> expectedType, GenericHandler<P> handler,
Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
public <P> B handle(@Nullable Class<P> expectedType, GenericHandler<P> handler,
@Nullable Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
ServiceActivatingHandler serviceActivatingHandler;
if (ClassUtils.isLambda(handler.getClass())) {
@@ -1108,7 +1113,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public <H extends MessageHandler> B handle(MessageHandlerSpec<?, H> messageHandlerSpec,
Consumer<GenericEndpointSpec<H>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<H>> endpointConfigurer) {
Assert.notNull(messageHandlerSpec, "'messageHandlerSpec' must not be null");
if (messageHandlerSpec instanceof ComponentsRegistration) {
@@ -1132,7 +1137,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param <H> the {@link MessageHandler} type.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public <H extends MessageHandler> B handle(H messageHandler, Consumer<GenericEndpointSpec<H>> endpointConfigurer) {
public <H extends MessageHandler> B handle(H messageHandler,
@Nullable Consumer<GenericEndpointSpec<H>> endpointConfigurer) {
Assert.notNull(messageHandler, "'messageHandler' must not be null");
return register(new GenericEndpointSpec<>(messageHandler), endpointConfigurer);
}
@@ -1160,7 +1167,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see GenericEndpointSpec
*/
public B bridge(Consumer<GenericEndpointSpec<BridgeHandler>> endpointConfigurer) {
public B bridge(@Nullable Consumer<GenericEndpointSpec<BridgeHandler>> endpointConfigurer) {
return handle(new BridgeHandler(), endpointConfigurer);
}
@@ -1172,7 +1179,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B delay(String groupId) {
return this.delay(groupId, null);
return delay(groupId, null);
}
/**
@@ -1183,7 +1190,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see DelayerEndpointSpec
*/
public B delay(String groupId, Consumer<DelayerEndpointSpec> endpointConfigurer) {
public B delay(String groupId, @Nullable Consumer<DelayerEndpointSpec> endpointConfigurer) {
return register(new DelayerEndpointSpec(new DelayHandler(groupId)), endpointConfigurer);
}
@@ -1251,7 +1258,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see GenericEndpointSpec
*/
public B enrichHeaders(MapBuilder<?, String, Object> headers,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
return enrichHeaders(headers.get(), endpointConfigurer);
}
@@ -1279,7 +1286,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see GenericEndpointSpec
*/
public B enrichHeaders(Map<String, Object> headers,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec();
headerEnricherSpec.headers(headers);
@@ -1331,7 +1338,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see SplitterEndpointSpec
*/
public B split(Consumer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
public B split(@Nullable Consumer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
return split(new DefaultMessageSplitter(), endpointConfigurer);
}
@@ -1354,7 +1361,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see SplitterEndpointSpec
*/
public B split(String expression, Consumer<SplitterEndpointSpec<ExpressionEvaluatingSplitter>> endpointConfigurer) {
public B split(String expression,
@Nullable Consumer<SplitterEndpointSpec<ExpressionEvaluatingSplitter>> endpointConfigurer) {
Assert.hasText(expression, "'expression' must not be empty");
return split(new ExpressionEvaluatingSplitter(PARSER.parseExpression(expression)), endpointConfigurer);
}
@@ -1378,7 +1387,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see MethodInvokingSplitter
*/
public B split(Object service, String methodName) {
public B split(Object service, @Nullable String methodName) {
return split(service, methodName, null);
}
@@ -1394,8 +1403,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see SplitterEndpointSpec
* @see MethodInvokingSplitter
*/
public B split(Object service, String methodName,
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
public B split(Object service, @Nullable String methodName,
@Nullable Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
MethodInvokingSplitter splitter;
if (StringUtils.hasText(methodName)) {
@@ -1414,7 +1423,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param methodName the method to invoke at runtime.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B split(String beanName, String methodName) {
public B split(String beanName, @Nullable String methodName) {
return split(beanName, methodName, null);
}
@@ -1429,8 +1438,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see SplitterEndpointSpec
*/
public B split(String beanName, String methodName,
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
public B split(String beanName, @Nullable String methodName,
@Nullable Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
return split(new MethodInvokingSplitter(new BeanNameMessageProcessor<>(beanName, methodName)),
endpointConfigurer);
@@ -1471,7 +1480,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see SplitterEndpointSpec
*/
public B split(MessageProcessorSpec<?> messageProcessorSpec,
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
@Nullable Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.getObject();
@@ -1531,8 +1540,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see LambdaMessageProcessor
* @see SplitterEndpointSpec
*/
public <P> B split(Class<P> expectedType, Function<P, ?> splitter,
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
public <P> B split(@Nullable Class<P> expectedType, Function<P, ?> splitter,
@Nullable Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
MethodInvokingSplitter split =
ClassUtils.isLambda(splitter.getClass())
@@ -1563,7 +1572,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see SplitterEndpointSpec
*/
public <S extends AbstractMessageSplitter> B split(MessageHandlerSpec<?, S> splitterMessageHandlerSpec,
Consumer<SplitterEndpointSpec<S>> endpointConfigurer) {
@Nullable Consumer<SplitterEndpointSpec<S>> endpointConfigurer) {
Assert.notNull(splitterMessageHandlerSpec, "'splitterMessageHandlerSpec' must not be null");
return split(splitterMessageHandlerSpec.getObject(), endpointConfigurer);
}
@@ -1589,7 +1599,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see SplitterEndpointSpec
*/
public <S extends AbstractMessageSplitter> B split(S splitter,
Consumer<SplitterEndpointSpec<S>> endpointConfigurer) {
@Nullable Consumer<SplitterEndpointSpec<S>> endpointConfigurer) {
Assert.notNull(splitter, "'splitter' must not be null");
return register(new SplitterEndpointSpec<>(splitter), endpointConfigurer);
@@ -1627,7 +1637,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see GenericEndpointSpec
*/
public B headerFilter(HeaderFilter headerFilter,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
return transform(null, headerFilter, endpointConfigurer);
}
@@ -1639,7 +1649,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B claimCheckIn(MessageStore messageStore) {
return this.claimCheckIn(messageStore, null);
return claimCheckIn(messageStore, null);
}
/**
@@ -1652,7 +1662,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see GenericEndpointSpec
*/
public B claimCheckIn(MessageStore messageStore,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
return transform(null, new ClaimCheckInTransformer(messageStore), endpointConfigurer);
}
@@ -1692,7 +1702,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see ClaimCheckOutTransformer#setRemoveMessage(boolean)
*/
public B claimCheckOut(MessageStore messageStore, boolean removeMessage,
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore);
claimCheckOutTransformer.setRemoveMessage(removeMessage);
@@ -1727,7 +1737,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see ResequencerSpec
*/
public B resequence(Consumer<ResequencerSpec> resequencer) {
public B resequence(@Nullable Consumer<ResequencerSpec> resequencer) {
return register(new ResequencerSpec(), resequencer);
}
@@ -1765,7 +1775,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see AggregatorSpec
*/
public B aggregate(Consumer<AggregatorSpec> aggregator) {
public B aggregate(@Nullable Consumer<AggregatorSpec> aggregator) {
return register(new AggregatorSpec(), aggregator);
}
@@ -1776,7 +1786,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param method the method to invoke at runtime.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B route(String beanName, String method) {
public B route(String beanName, @Nullable String method) {
return route(beanName, method, null);
}
@@ -1788,8 +1798,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param routerConfigurer the {@link Consumer} to provide {@link MethodInvokingRouter} options.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B route(String beanName, String method, Consumer<RouterSpec<Object,
MethodInvokingRouter>> routerConfigurer) {
public B route(String beanName, @Nullable String method,
@Nullable Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
MethodInvokingRouter methodInvokingRouter =
new MethodInvokingRouter(new BeanNameMessageProcessor<>(beanName, method));
@@ -1815,7 +1825,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see MethodInvokingRouter
*/
public B route(Object service, String methodName) {
public B route(Object service, @Nullable String methodName) {
return route(service, methodName, null);
}
@@ -1828,8 +1838,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see MethodInvokingRouter
*/
public B route(Object service, String methodName,
Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
public B route(Object service, @Nullable String methodName,
@Nullable Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
MethodInvokingRouter router;
if (StringUtils.hasText(methodName)) {
@@ -1859,7 +1869,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param <T> the target result type.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public <T> B route(String expression, Consumer<RouterSpec<T, ExpressionEvaluatingRouter>> routerConfigurer) {
public <T> B route(String expression,
@Nullable Consumer<RouterSpec<T, ExpressionEvaluatingRouter>> routerConfigurer) {
return route(new RouterSpec<>(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression))),
routerConfigurer);
}
@@ -1882,7 +1894,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see LambdaMessageProcessor
*/
public <S, T> B route(Class<S> expectedType, Function<S, T> router) {
public <S, T> B route(@Nullable Class<S> expectedType, Function<S, T> router) {
return route(expectedType, router, null);
}
@@ -1910,8 +1922,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see LambdaMessageProcessor
*/
public <P, T> B route(Class<P> expectedType, Function<P, T> router,
Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
public <P, T> B route(@Nullable Class<P> expectedType, Function<P, T> router,
@Nullable Consumer<RouterSpec<T, MethodInvokingRouter>> routerConfigurer) {
MethodInvokingRouter methodInvokingRouter =
ClassUtils.isLambda(router.getClass())
@@ -1953,7 +1965,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B route(MessageProcessorSpec<?> messageProcessorSpec,
Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
@Nullable Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.getObject();
@@ -1963,7 +1975,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
}
protected <R extends AbstractMessageRouter, S extends AbstractRouterSpec<? super S, R>> B route(S routerSpec,
Consumer<S> routerConfigurer) {
@Nullable Consumer<S> routerConfigurer) {
if (routerConfigurer != null) {
routerConfigurer.accept(routerSpec);
@@ -1974,7 +1986,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
Map<Object, String> componentsToRegister = null;
Map<Object, String> routerComponents = routerSpec.getComponentsToRegister();
if (routerComponents != null) {
if (!CollectionUtils.isEmpty(routerComponents)) {
componentsToRegister = new LinkedHashMap<>(routerComponents);
routerComponents.clear();
}
@@ -2069,7 +2081,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param <R> the {@link AbstractMessageRouter} type.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public <R extends AbstractMessageRouter> B route(R router, Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
public <R extends AbstractMessageRouter> B route(R router,
@Nullable Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
return handle(router, endpointConfigurer);
}
@@ -2098,7 +2112,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* options.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B gateway(String requestChannel, Consumer<GatewayEndpointSpec> endpointConfigurer) {
public B gateway(String requestChannel, @Nullable Consumer<GatewayEndpointSpec> endpointConfigurer) {
return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer);
}
@@ -2127,7 +2141,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* options.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B gateway(MessageChannel requestChannel, Consumer<GatewayEndpointSpec> endpointConfigurer) {
public B gateway(MessageChannel requestChannel, @Nullable Consumer<GatewayEndpointSpec> endpointConfigurer) {
return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer);
}
@@ -2162,7 +2176,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @param endpointConfigurer the {@link Consumer} to provide integration endpoint options.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B gateway(IntegrationFlow flow, Consumer<GatewayEndpointSpec> endpointConfigurer) {
public B gateway(IntegrationFlow flow, @Nullable Consumer<GatewayEndpointSpec> endpointConfigurer) {
MessageChannel requestChannel = obtainInputChannelFromFlow(flow);
return gateway(requestChannel, endpointConfigurer);
}
@@ -2217,7 +2231,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(LoggingHandler.Level level, String category) {
public B log(LoggingHandler.Level level, @Nullable String category) {
return log(level, category, (Expression) null);
}
@@ -2365,7 +2379,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
* @see #wireTap(WireTapSpec)
*/
public B log(LoggingHandler.Level level, String category, Expression logExpression) {
public B log(LoggingHandler.Level level, @Nullable String category, @Nullable Expression logExpression) {
LoggingHandler loggingHandler = new LoggingHandler(level);
if (StringUtils.hasText(category)) {
loggingHandler.setLoggerName(category);
@@ -2457,7 +2471,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see #bridge()
*/
@Deprecated
public IntegrationFlow logAndReply(LoggingHandler.Level level, String category) {
public IntegrationFlow logAndReply(LoggingHandler.Level level, @Nullable String category) {
return logAndReply(level, category, (Expression) null);
}
@@ -2661,7 +2675,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @see #bridge()
*/
@Deprecated
public IntegrationFlow logAndReply(LoggingHandler.Level level, String category, Expression logExpression) {
public IntegrationFlow logAndReply(LoggingHandler.Level level, @Nullable String category,
@Nullable Expression logExpression) {
return log(level, category, logExpression)
.bridge()
.get();
@@ -2687,7 +2703,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* Can be {@code null}.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B scatterGather(MessageChannel scatterChannel, Consumer<AggregatorSpec> gatherer) {
public B scatterGather(MessageChannel scatterChannel, @Nullable Consumer<AggregatorSpec> gatherer) {
return scatterGather(scatterChannel, gatherer, null);
}
@@ -2702,8 +2718,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* {@link ScatterGatherHandler} and its endpoint. Can be {@code null}.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B scatterGather(MessageChannel scatterChannel, Consumer<AggregatorSpec> gatherer,
Consumer<ScatterGatherSpec> scatterGather) {
public B scatterGather(MessageChannel scatterChannel, @Nullable Consumer<AggregatorSpec> gatherer,
@Nullable Consumer<ScatterGatherSpec> scatterGather) {
AggregatorSpec aggregatorSpec = new AggregatorSpec();
if (gatherer != null) {
@@ -2790,7 +2806,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* {@link org.springframework.integration.aggregator.BarrierMessageHandler} options.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B barrier(long timeout, Consumer<BarrierSpec> barrierConfigurer) {
public B barrier(long timeout, @Nullable Consumer<BarrierSpec> barrierConfigurer) {
return register(new BarrierSpec(timeout), barrierConfigurer);
}
@@ -2811,7 +2827,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B trigger(String triggerActionId,
Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
MessageProcessor<Void> trigger = new BeanNameMessageProcessor<>(triggerActionId, "trigger");
return handle(new ServiceActivatingHandler(trigger), endpointConfigurer);
@@ -2834,7 +2850,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B trigger(MessageTriggerAction triggerAction,
Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
@Nullable Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
Consumer<Message<?>> trigger = triggerAction::trigger;
return handle(new ServiceActivatingHandler(new LambdaMessageProcessor(trigger, Message.class)),
@@ -2959,7 +2975,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
}
protected <S extends ConsumerEndpointSpec<? super S, ? extends MessageHandler>> B register(S endpointSpec,
Consumer<S> endpointConfigurer) {
@Nullable Consumer<S> endpointConfigurer) {
if (endpointConfigurer != null) {
endpointConfigurer.accept(endpointSpec);
@@ -2977,7 +2993,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
addComponents(endpointSpec.getComponentsToRegister());
if (inputChannel instanceof MessageChannelReference messageChannelReference) {
factoryBeanTuple2.getT1().setInputChannelName(messageChannelReference.getName());
factoryBeanTuple2.getT1().setInputChannelName(messageChannelReference.name());
}
else {
if (inputChannel instanceof FixedSubscriberChannelPrototype fixedSubscriberChannel) {
@@ -3001,7 +3017,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
if (currComponent != null) {
String channelName = null;
if (outputChannel instanceof MessageChannelReference channelReference) {
channelName = channelReference.getName();
channelName = channelReference.name();
}
if (currComponent instanceof MessageProducer messageProducer) {
@@ -3099,7 +3115,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
REFERENCED_REPLY_PRODUCERS.add(replyHandler);
}
protected static Object extractProxyTarget(Object target) {
@Nullable
protected static Object extractProxyTarget(@Nullable Object target) {
if (!(target instanceof Advised advised)) {
return target;
}

5
spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -38,6 +38,7 @@ import org.springframework.integration.handler.advice.ReactiveRequestHandlerAdvi
import org.springframework.integration.router.AbstractMessageRouter;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.transaction.TransactionInterceptorBuilder;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.scheduling.TaskScheduler;
@@ -61,7 +62,7 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
protected final List<Advice> adviceChain = new LinkedList<>(); // NOSONAR final
protected ConsumerEndpointSpec(H messageHandler) {
protected ConsumerEndpointSpec(@Nullable H messageHandler) {
super(messageHandler, new ConsumerEndpointFactoryBean());
}

11
spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java
View File

@@ -28,6 +28,7 @@ import org.springframework.beans.factory.FactoryBean;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
@@ -54,7 +55,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
protected H handler; // NOSONAR
protected EndpointSpec(H handler, F endpointFactoryBean) {
protected EndpointSpec(@Nullable H handler, F endpointFactoryBean) {
this.endpointFactoryBean = endpointFactoryBean;
this.handler = handler;
}
@@ -83,9 +84,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
*/
public S poller(PollerSpec pollerMetadataSpec) {
Map<Object, String> components = pollerMetadataSpec.getComponentsToRegister();
if (components != null) {
this.componentsToRegister.putAll(components);
}
this.componentsToRegister.putAll(components);
return poller(pollerMetadataSpec.getObject());
}
@@ -122,9 +121,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
@Override
public Map<Object, String> getComponentsToRegister() {
return this.componentsToRegister.isEmpty()
? null
: this.componentsToRegister;
return this.componentsToRegister;
}
@Override

13
spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -31,6 +31,7 @@ import org.springframework.integration.transformer.support.AbstractHeaderValueMe
import org.springframework.integration.transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.HeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.StaticHeaderValueMessageProcessor;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
@@ -235,7 +236,7 @@ public class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnri
* @return the enricher spec.
* @see ContentEnricher#setHeaderExpressions(Map)
*/
public <V> EnricherSpec header(String name, V value, Boolean overwrite) {
public <V> EnricherSpec header(String name, V value, @Nullable Boolean overwrite) {
AbstractHeaderValueMessageProcessor<V> headerValueMessageProcessor =
new StaticHeaderValueMessageProcessor<V>(value);
headerValueMessageProcessor.setOverwrite(overwrite);
@@ -260,7 +261,7 @@ public class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnri
* @return the enricher spec.
* @see ContentEnricher#setHeaderExpressions(Map)
*/
public EnricherSpec headerExpression(String name, String expression, Boolean overwrite) {
public EnricherSpec headerExpression(String name, String expression, @Nullable Boolean overwrite) {
Assert.hasText(expression, "'expression' must not be empty");
return headerExpression(name, PARSER.parseExpression(expression), overwrite);
}
@@ -287,11 +288,13 @@ public class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnri
* @see ContentEnricher#setHeaderExpressions(Map)
* @see FunctionExpression
*/
public <P> EnricherSpec headerFunction(String name, Function<Message<P>, Object> function, Boolean overwrite) {
public <P> EnricherSpec headerFunction(String name, Function<Message<P>, Object> function,
@Nullable Boolean overwrite) {
return headerExpression(name, new FunctionExpression<>(function), overwrite);
}
private EnricherSpec headerExpression(String name, Expression expression, Boolean overwrite) {
private EnricherSpec headerExpression(String name, Expression expression, @Nullable Boolean overwrite) {
AbstractHeaderValueMessageProcessor<?> headerValueMessageProcessor =
new ExpressionEvaluatingHeaderValueMessageProcessor<>(expression, null);
headerValueMessageProcessor.setOverwrite(overwrite);

25
spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -41,6 +41,7 @@ import org.springframework.integration.transformer.support.ExpressionEvaluatingH
import org.springframework.integration.transformer.support.HeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.RoutingSlipHeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.StaticHeaderValueMessageProcessor;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.util.Assert;
@@ -133,7 +134,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
/**
* Add header specifications from the {@link MapBuilder}; if a map value is an
* {@link Expression}, it will be evaluated at run time when the message headers are
* enriched. Otherwise the value is simply added to the headers. Headers derived from
* enriched. Otherwise, the value is simply added to the headers. Headers derived from
* the map will <b>not</b> overwrite existing headers, unless
* {@link #defaultOverwrite(boolean)} is true.
* @param headers the header map builder.
@@ -146,12 +147,12 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
/**
* Add header specifications from the {@link MapBuilder}; if a map value is an
* {@link Expression}, it will be evaluated at run time when the message headers are
* enriched. Otherwise the value is simply added to the headers.
* enriched. Otherwise, the value is simply added to the headers.
* @param headers the header map builder.
* @param overwrite true to overwrite existing headers.
* @return the header enricher spec.
*/
public HeaderEnricherSpec headers(MapBuilder<?, String, Object> headers, Boolean overwrite) {
public HeaderEnricherSpec headers(MapBuilder<?, String, Object> headers, @Nullable Boolean overwrite) {
Assert.notNull(headers, HEADERS_MUST_NOT_BE_NULL);
return headers(headers.get(), overwrite);
}
@@ -159,7 +160,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
/**
* Add header specifications from the {@link Map}; if a map value is an
* {@link Expression}, it will be evaluated at run time when the message headers are
* enriched. Otherwise the value is simply added to the headers. Headers derived from
* enriched. Otherwise, the value is simply added to the headers. Headers derived from
* the map will <em>not</em> overwrite existing headers, unless
* {@link #defaultOverwrite(boolean)} is true.
* @param headers The header builder.
@@ -172,12 +173,12 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
/**
* Add header specifications from the {@link Map}; if a map value is an
* {@link Expression}, it will be evaluated at run time when the message headers are
* enriched. Otherwise the value is simply added to the headers.
* enriched. Otherwise, the value is simply added to the headers.
* @param headers The header builder.
* @param overwrite true to overwrite existing headers.
* @return the header enricher spec.
*/
public HeaderEnricherSpec headers(Map<String, Object> headers, Boolean overwrite) {
public HeaderEnricherSpec headers(Map<String, Object> headers, @Nullable Boolean overwrite) {
Assert.notNull(headers, HEADERS_MUST_NOT_BE_NULL);
for (Entry<String, Object> entry : headers.entrySet()) {
String name = entry.getKey();
@@ -753,7 +754,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
* @param <V> the value type.
* @return the header enricher spec.
*/
public <V> HeaderEnricherSpec header(String name, V value, Boolean overwrite) {
public <V> HeaderEnricherSpec header(String name, V value, @Nullable Boolean overwrite) {
AbstractHeaderValueMessageProcessor<V> headerValueMessageProcessor =
new StaticHeaderValueMessageProcessor<>(value);
headerValueMessageProcessor.setOverwrite(overwrite);
@@ -780,7 +781,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
* @param overwrite true to overwrite an existing header.
* @return the header enricher spec.
*/
public HeaderEnricherSpec headerExpression(String name, String expression, Boolean overwrite) {
public HeaderEnricherSpec headerExpression(String name, String expression, @Nullable Boolean overwrite) {
Assert.hasText(expression, "'expression' must not be empty");
return headerExpression(name, PARSER.parseExpression(expression), overwrite);
}
@@ -810,12 +811,12 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
* @see FunctionExpression
*/
public <P> HeaderEnricherSpec headerFunction(String name, Function<Message<P>, ?> function,
Boolean overwrite) {
@Nullable Boolean overwrite) {
return headerExpression(name, new FunctionExpression<>(function), overwrite);
}
private HeaderEnricherSpec headerExpression(String name, Expression expression, Boolean overwrite) {
private HeaderEnricherSpec headerExpression(String name, Expression expression, @Nullable Boolean overwrite) {
AbstractHeaderValueMessageProcessor<?> headerValueMessageProcessor =
new ExpressionEvaluatingHeaderValueMessageProcessor<>(expression, null);
headerValueMessageProcessor.setOverwrite(overwrite);
@@ -859,7 +860,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
* @return the header enricher spec.
* @see org.springframework.integration.support.channel.HeaderChannelRegistry
*/
public HeaderEnricherSpec headerChannelsToString(String timeToLiveExpression) {
public HeaderEnricherSpec headerChannelsToString(@Nullable String timeToLiveExpression) {
return headerExpression("replyChannel",
"@" + IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME
+ ".channelToChannelName(headers.replyChannel" +

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationComponentSpec.java
View File

@@ -26,6 +26,8 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.context.SmartLifecycle;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.lang.NonNull;
import org.springframework.lang.Nullable;
/**
* The common Builder abstraction.
@@ -57,11 +59,12 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
* @param idToSet the id.
* @return the spec.
*/
protected S id(String idToSet) {
protected S id(@Nullable String idToSet) {
this.id = idToSet;
return _this();
}
@Nullable
public final String getId() {
return this.id;
}
@@ -85,6 +88,7 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
* !!! This method must not be called from the target configuration !!!
* @return the object backed by this factory bean.
*/
@NonNull
@Override
public T getObject() {
if (this.target == null) {

14
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlow.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.dsl;
import java.util.Collections;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -114,6 +115,7 @@ public interface IntegrationFlow {
* @return the channel.
* @since 5.0.4
*/
@Nullable
default MessageChannel getInputChannel() {
return null;
}
@@ -124,7 +126,7 @@ public interface IntegrationFlow {
* @since 5.5.4
*/
default Map<Object, String> getIntegrationComponents() {
return null;
return Collections.emptyMap();
}
/**
@@ -214,7 +216,7 @@ public interface IntegrationFlow {
* @see SourcePollingChannelAdapterSpec
*/
static IntegrationFlowBuilder from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
@Nullable Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Assert.notNull(messageSourceSpec, "'messageSourceSpec' must not be null");
return from(messageSourceSpec.getObject(), endpointConfigurer, registerComponents(messageSourceSpec));
@@ -246,7 +248,7 @@ public interface IntegrationFlow {
* @see Supplier
*/
static <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
@Nullable Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Assert.notNull(messageSource, "'messageSource' must not be null");
return from(new AbstractMessageSource<>() {
@@ -440,7 +442,7 @@ public interface IntegrationFlow {
@SuppressWarnings("overloads")
static IntegrationFlowBuilder from(IntegrationFlow other) {
Map<Object, String> integrationComponents = other.getIntegrationComponents();
Assert.notNull(integrationComponents, () ->
Assert.notEmpty(integrationComponents, () ->
"The provided integration flow to compose from '" + other +
"' must be declared as a bean in the application context");
Object lastIntegrationComponentFromOther =
@@ -488,6 +490,7 @@ public interface IntegrationFlow {
return integrationFlowBuilder.addComponent(inboundGateway);
}
@Nullable
private static IntegrationFlowBuilder registerComponents(Object spec) {
if (spec instanceof ComponentsRegistration componentsRegistration) {
return new IntegrationFlowBuilder()
@@ -496,8 +499,9 @@ public interface IntegrationFlow {
return null;
}
@Nullable
@SuppressWarnings("unchecked")
private static <T> T extractProxyTarget(T target) {
private static <T> T extractProxyTarget(@Nullable T target) {
if (!(target instanceof Advised advised)) {
return target;
}

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowAdapter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -37,7 +37,7 @@ import org.springframework.util.Assert;
* Requires the implementation for the {@link #buildFlow()} method to produce
* {@link IntegrationFlowDefinition} using one of {@link #from} support methods.
* <p>
* Typically is used for target service implementation:
* Typically, is used for target service implementation:
* <pre class="code">
* &#64;Component
* public class MyFlowAdapter extends IntegrationFlowAdapter {
@@ -68,12 +68,12 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, Managea
@Override
public final void configure(IntegrationFlowDefinition<?> flow) {
IntegrationFlowDefinition<?> targetFlow = buildFlow();
Assert.state(targetFlow != null, "the 'buildFlow()' must not return null");
flow.integrationComponents.clear();
flow.integrationComponents.putAll(targetFlow.integrationComponents);
this.targetIntegrationFlow = flow.get();
}
@Nullable
@Override
public MessageChannel getInputChannel() {
assertTargetIntegrationFlow();

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -64,7 +64,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
/**
* Populate the {@link MessageTransformingHandler} instance for the provided
* {@link GenericTransformer}. In addition accept options for the integration endpoint
* {@link GenericTransformer}. In addition, accept options for the integration endpoint
* using {@link GenericEndpointSpec}. Use
* {@link #transform(Class, GenericTransformer, Consumer)} if you need to access the
* entire message.
@@ -226,7 +226,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
/**
* Populate the {@link MethodInvokingRouter} for provided {@link Function}
* with provided options from {@link RouterSpec}.
* In addition accept options for the integration endpoint using {@link GenericEndpointSpec}.
* In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}.
* Typically used with a Java 8 Lambda expression:
* <pre class="code">
* {@code

7
spring-integration-core/src/main/java/org/springframework/integration/dsl/MessageProducerSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -18,6 +18,7 @@ package org.springframework.integration.dsl;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
/**
@@ -34,7 +35,7 @@ import org.springframework.messaging.MessageChannel;
public abstract class MessageProducerSpec<S extends MessageProducerSpec<S, P>, P extends MessageProducerSupport>
extends IntegrationComponentSpec<S, P> {
public MessageProducerSpec(P producer) {
public MessageProducerSpec(@Nullable P producer) {
this.target = producer;
}
@@ -43,7 +44,7 @@ public abstract class MessageProducerSpec<S extends MessageProducerSpec<S, P>, P
* Configure the message producer's bean name.
*/
@Override
public S id(String id) {
public S id(@Nullable String id) {
this.target.setBeanName(id);
return super.id(id);
}

7
spring-integration-core/src/main/java/org/springframework/integration/dsl/MessagingGatewaySpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2023 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.
@@ -19,6 +19,7 @@ package org.springframework.integration.dsl;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.mapping.InboundMessageMapper;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
/**
@@ -34,12 +35,12 @@ import org.springframework.messaging.MessageChannel;
public abstract class MessagingGatewaySpec<S extends MessagingGatewaySpec<S, G>, G extends MessagingGatewaySupport>
extends IntegrationComponentSpec<S, G> {
public MessagingGatewaySpec(G gateway) {
public MessagingGatewaySpec(@Nullable G gateway) {
this.target = gateway;
}
@Override
public S id(String id) {
public S id(@Nullable String id) {
this.target.setBeanName(id);
return super.id(id);
}

23
spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import org.springframework.integration.filter.ExpressionEvaluatingSelector;
import org.springframework.integration.filter.MethodInvokingSelector;
import org.springframework.integration.router.RecipientListRouter;
import org.springframework.integration.util.ClassUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.StringUtils;
@@ -89,7 +90,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
}
/**
* Adds a recipient channel that will be selected if the the selector's accept method returns 'true'.
* Adds a recipient channel that will be selected if the selector's accept method returns 'true'.
* @param channelName the channel name.
* @param selector the selector.
* @param <P> the selector source type.
@@ -113,7 +114,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
}
/**
* Adds a recipient channel that always will be selected.
* Adds a recipient channel that always is selected.
* @param channel the recipient channel.
* @return the router spec.
*/
@@ -122,22 +123,22 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
}
/**
* Adds a recipient channel that will be selected if the the expression evaluates to 'true'.
* Adds a recipient channel that will be selected if the expression evaluates to 'true'.
* @param channel the recipient channel.
* @param expression the expression.
* @return the router spec.
*/
public RecipientListRouterSpec recipient(MessageChannel channel, String expression) {
public RecipientListRouterSpec recipient(MessageChannel channel, @Nullable String expression) {
return recipient(channel, StringUtils.hasText(expression) ? PARSER.parseExpression(expression) : null);
}
/**
* Adds a recipient channel that will be selected if the the expression evaluates to 'true'.
* Adds a recipient channel that will be selected if the expression evaluates to 'true'.
* @param channel the recipient channel.
* @param expression the expression.
* @return the router spec.
*/
public RecipientListRouterSpec recipient(MessageChannel channel, Expression expression) {
public RecipientListRouterSpec recipient(MessageChannel channel, @Nullable Expression expression) {
if (expression != null) {
ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector(expression);
this.handler.addRecipient(channel, selector);
@@ -150,7 +151,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
}
/**
* Adds a recipient channel that will be selected if the the selector's accept method returns 'true'.
* Adds a recipient channel that will be selected if the selector's accept method returns 'true'.
* @param channel the recipient channel.
* @param selector the selector.
* @return the router spec.
@@ -160,7 +161,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
}
/**
* Adds a recipient channel that will be selected if the the selector's accept method returns 'true'.
* Adds a recipient channel that will be selected if the selector's accept method returns 'true'.
* @param channel the recipient channel.
* @param selector the selector.
* @param <P> the selector source type.
@@ -210,7 +211,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
* @param subFlow the subflow.
* @return the router spec.
*/
public RecipientListRouterSpec recipientFlow(String expression, IntegrationFlow subFlow) {
public RecipientListRouterSpec recipientFlow(@Nullable String expression, IntegrationFlow subFlow) {
return recipientFlow(StringUtils.hasText(expression) ? PARSER.parseExpression(expression) : null, subFlow);
}
@@ -220,7 +221,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
* @param subFlow the subflow.
* @return the router spec.
*/
public RecipientListRouterSpec recipientFlow(Expression expression, IntegrationFlow subFlow) {
public RecipientListRouterSpec recipientFlow(@Nullable Expression expression, IntegrationFlow subFlow) {
MessageChannel channel = obtainInputChannelFromFlow(subFlow);
return recipient(channel, expression);
}

5
spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -42,8 +42,7 @@ import org.springframework.util.StringUtils;
*
* @since 5.0
*/
public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
extends AbstractRouterSpec<RouterSpec<K, R>, R> {
public class RouterSpec<K, R extends AbstractMappingMessageRouter> extends AbstractRouterSpec<RouterSpec<K, R>, R> {
private final RouterMappingProvider mappingProvider;

5
spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2023 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.
@@ -19,6 +19,7 @@ package org.springframework.integration.dsl;
import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.lang.Nullable;
/**
* @author Artem Bilan
@@ -43,7 +44,7 @@ public class SourcePollingChannelAdapterSpec extends
return _this();
}
public SourcePollingChannelAdapterSpec poller(PollerMetadata pollerMetadata) {
public SourcePollingChannelAdapterSpec poller(@Nullable PollerMetadata pollerMetadata) {
if (pollerMetadata != null) {
if (PollerMetadata.MAX_MESSAGES_UNBOUNDED == pollerMetadata.getMaxMessagesPerPoll()) {
pollerMetadata.setMaxMessagesPerPoll(1);

10
spring-integration-core/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -27,6 +27,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
/**
@@ -102,6 +103,7 @@ public class StandardIntegrationFlow
throw new UnsupportedOperationException();
}
@Nullable
@Override
public MessageChannel getInputChannel() {
if (this.inputChannel == null) {
@@ -146,8 +148,7 @@ public class StandardIntegrationFlow
public void stop(Runnable callback) {
AggregatingCallback aggregatingCallback = new AggregatingCallback(this.integrationComponents.size(), callback);
for (Object component : this.integrationComponents.keySet()) {
if (component instanceof SmartLifecycle) {
SmartLifecycle lifecycle = (SmartLifecycle) component;
if (component instanceof SmartLifecycle lifecycle) {
if (lifecycle.isRunning()) {
lifecycle.stop(aggregatingCallback);
continue;
@@ -161,8 +162,7 @@ public class StandardIntegrationFlow
@Override
public void stop() {
for (Object component : this.integrationComponents.keySet()) {
if (component instanceof SmartLifecycle) {
SmartLifecycle lifecycle = (SmartLifecycle) component;
if (component instanceof SmartLifecycle lifecycle) {
if (lifecycle.isRunning()) {
lifecycle.stop();
}

4
spring-integration-core/src/main/java/org/springframework/integration/dsl/WireTapSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2023 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.
@@ -104,7 +104,7 @@ public class WireTapSpec extends IntegrationComponentSpec<WireTapSpec, WireTap>
return Collections.singletonMap(this.selector, null);
}
else {
return null;
return Collections.emptyMap();
}
}

12
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java
View File

@@ -72,6 +72,7 @@ import org.springframework.integration.dsl.StandardIntegrationFlow;
import org.springframework.integration.dsl.support.MessageChannelReference;
import org.springframework.integration.gateway.AnnotationGatewayProxyFactoryBean;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -178,7 +179,7 @@ public class IntegrationFlowBeanPostProcessor
targetIntegrationComponents.put(endpoint, id);
}
else if (component instanceof MessageChannelReference messageChannelReference) {
String channelBeanName = messageChannelReference.getName();
String channelBeanName = messageChannelReference.name();
if (!this.beanFactory.containsBean(channelBeanName)) {
DirectChannel directChannel = new DirectChannel();
registerComponent(directChannel, channelBeanName, flowBeanName);
@@ -422,8 +423,7 @@ public class IntegrationFlowBeanPostProcessor
}
return !this.beanFactory.getBeansOfType(instance.getClass(), false, false)
.values()
.contains(instance);
.containsValue(instance);
}
private void registerComponent(Object component, String beanName) {
@@ -431,7 +431,7 @@ public class IntegrationFlowBeanPostProcessor
}
@SuppressWarnings("unchecked")
private void registerComponent(Object component, String beanName, String parentName,
private void registerComponent(Object component, String beanName, @Nullable String parentName,
BeanDefinitionCustomizer... customizers) {
AbstractBeanDefinition beanDefinition =
@@ -461,7 +461,9 @@ public class IntegrationFlowBeanPostProcessor
return generateBeanName(instance, prefix, null, false);
}
private String generateBeanName(Object instance, String prefix, String fallbackId, boolean useFlowIdAsPrefix) {
private String generateBeanName(Object instance, String prefix, @Nullable String fallbackId,
boolean useFlowIdAsPrefix) {
if (instance instanceof NamedComponent namedComponent && namedComponent.getBeanName() != null) {
String beanName = namedComponent.getBeanName();
return useFlowIdAsPrefix

5
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-2023 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.
@@ -21,6 +21,7 @@ import java.util.Map;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.lang.Nullable;
import org.springframework.messaging.MessageChannel;
/**
@@ -200,7 +201,7 @@ public interface IntegrationFlowContext {
* @param bean an additional arbitrary bean to register into the application context.
* @return the current builder instance
*/
IntegrationFlowRegistrationBuilder addBean(String name, Object bean);
IntegrationFlowRegistrationBuilder addBean(@Nullable String name, Object bean);
/**
* Set the configuration source {@code Object} for this manual Integration flow definition.

11
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowLifecycleAdvice.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2021 the original author or authors.
* Copyright 2018-2023 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.
@@ -16,11 +16,15 @@
package org.springframework.integration.dsl.context;
import java.util.Map;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.dsl.StandardIntegrationFlow;
import org.springframework.lang.Nullable;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
/**
@@ -77,7 +81,7 @@ class IntegrationFlowLifecycleAdvice implements MethodInterceptor {
}
else if ("getIntegrationComponents".equals(method)) {
result = invocation.proceed();
if (result == null) {
if (CollectionUtils.isEmpty((Map<?, ?>) result)) {
result = this.delegate.getIntegrationComponents();
}
}
@@ -91,7 +95,8 @@ class IntegrationFlowLifecycleAdvice implements MethodInterceptor {
return result;
}
private Object applyToDelegate(MethodInvocation invocation, String method, Object resultArg) {
@Nullable
private Object applyToDelegate(MethodInvocation invocation, String method, @Nullable Object resultArg) {
Object result = resultArg;
switch (method) {

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2023 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.
@@ -242,7 +242,7 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
return Collections.unmodifiableMap(this.registry);
}
private String generateBeanName(Object instance, String parentName) {
private String generateBeanName(Object instance, @Nullable String parentName) {
if (instance instanceof NamedComponent) {
String beanName = ((NamedComponent) instance).getBeanName();
if (beanName != null) {
@@ -331,7 +331,7 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont
* @return the current builder instance
*/
@Override
public StandardIntegrationFlowRegistrationBuilder addBean(String name, Object bean) {
public StandardIntegrationFlowRegistrationBuilder addBean(@Nullable String name, Object bean) {
this.additionalBeans.put(bean, name);
return this;
}

1
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/package-info.java
View File

@@ -1,4 +1,5 @@
/**
* The context support classes for Spring Integration Java DSL.
*/
@org.springframework.lang.NonNullApi
package org.springframework.integration.dsl.context;

1
spring-integration-core/src/main/java/org/springframework/integration/dsl/package-info.java
View File

@@ -1,4 +1,5 @@
/**
* Root package of the Spring Integration Java DSL.
*/
@org.springframework.lang.NonNullApi
package org.springframework.integration.dsl;

5
spring-integration-core/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2023 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.
@@ -16,6 +16,7 @@
package org.springframework.integration.dsl.support;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -38,7 +39,7 @@ public class FixedSubscriberChannelPrototype implements MessageChannel {
this(null);
}
public FixedSubscriberChannelPrototype(String name) {
public FixedSubscriberChannelPrototype(@Nullable String name) {
this.name = name;
}

14
spring-integration-core/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2023 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.
@@ -25,24 +25,20 @@ import org.springframework.util.Assert;
* {@link MessageChannel} bean on the bean registration phase.
* For internal use only.
*
* @param name the name of the target {@link MessageChannel} bean.
*
* @author Artem Bilan
*
* @since 5.0
*
* @see org.springframework.integration.dsl.context.IntegrationFlowBeanPostProcessor
*/
public class MessageChannelReference implements MessageChannel {
public record MessageChannelReference(String name) implements MessageChannel {
private final String name;
public MessageChannelReference(String name) {
public MessageChannelReference {
Assert.notNull(name, "'name' must not be null");
this.name = name;
}
public String getName() {
return this.name;
}
@Override
public boolean send(Message<?> message) {

1
spring-integration-core/src/main/java/org/springframework/integration/dsl/support/package-info.java
View File

@@ -1,4 +1,5 @@
/**
* Provides various support classes used across Spring Integration Java DSL Components.
*/
@org.springframework.lang.NonNullApi
package org.springframework.integration.dsl.support;

11
spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/KotlinIntegrationFlowDefinition.kt
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2022 the original author or authors.
* Copyright 2020-2023 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.
@@ -901,6 +901,15 @@ class KotlinIntegrationFlowDefinition(@PublishedApi internal val delegate: Integ
this.delegate.gateway(requestChannel, Consumer(endpointConfigurer))
}
/**
* Populate the "artificial"
* [org.springframework.integration.gateway.GatewayMessageHandler] for the
* provided `subflow` with options from [GatewayEndpointSpec].
*/
fun gateway(flow: IntegrationFlow) {
this.delegate.gateway(flow)
}
/**
* Populate the "artificial"
* [org.springframework.integration.gateway.GatewayMessageHandler] for the

5
spring-integration-core/src/test/java/org/springframework/integration/dsl/flowservices/FlowServiceTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import org.jetbrains.annotations.NotNull;
import org.junit.jupiter.api.Test;
import org.springframework.aop.framework.Advised;
@@ -176,7 +177,7 @@ public class FlowServiceTests {
private final AtomicReference<Object> resultOverLoggingHandler = new AtomicReference<>();
@Override
public void configure(IntegrationFlowDefinition<?> f) {
public void configure(@NotNull IntegrationFlowDefinition<?> f) {
f.<String, String>transform(String::toUpperCase)
.log(LoggingHandler.Level.ERROR, m -> {
resultOverLoggingHandler.set(m.getPayload());

6
spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/routers/RouterDslTests.kt
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-2023 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.
@@ -114,8 +114,8 @@ class RouterDslTests {
integrationFlow {
split()
route<Int, Boolean>({ it % 2 == 0 }) {
subFlowMapping(true) { gateway(oddFlow().inputChannel) }
subFlowMapping(false) { gateway(evenFlow().inputChannel) }
subFlowMapping(true) { gateway(oddFlow()) }
subFlowMapping(false) { gateway(evenFlow()) }
}
aggregate()
}