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

GH-8586: Deprecate IntegrationComponentSpec.get() (#8594)

Browse Source 
* GH-8586: Deprecate IntegrationComponentSpec.get()

Fixes https://github.com/spring-projects/spring-integration/issues/8586

The `IntegrationComponentSpec` is not a plain wrapper around single component.
Sometimes it comes with several components where all of them must be registered
as beans.
If `IntegrationComponentSpec.get()` is called from end-user code, we may lose
other related components, for example filters in the `FileInboundChannelAdapterSpec`.

* Deprecate `IntegrationComponentSpec.get()` with no-op for end-user,
rather encourage to leave it as is and let the framework take care about its lifecycle
and related components registration
* Fix `IntegrationComponentSpec` logic to deal as a simple `FactoryBean` instead of
extra overhead via `AbstractFactoryBean`
* Use `IntegrationComponentSpec.getObject()` in the framework code where `get()` was called
* Fix tests to expose `IntegrationComponentSpec` as beans instead of previously called `get()`
* Some other clean up and typos fixes in the affected classes
* Document the change

* * Revert `ObjectStringMapBuilder` in the `KafkaInboundGatewaySpec.getComponentsToRegister()`

* Fix language in docs

Co-authored-by: Gary Russell <grussell@vmware.com>

* * Remove trailing whitespace in the `ScriptMessageSourceSpec`

---------

Co-authored-by: Gary Russell <grussell@vmware.com>
This commit is contained in:
Artem Bilan
2023-04-13 09:16:42 -04:00
committed by GitHub
parent bbaffb22bf
commit b99729544d
46 changed files with 401 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundChannelAdapterSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-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.
@@ -43,13 +43,13 @@ public abstract class AmqpInboundChannelAdapterSpec
protected final MessageListenerContainerSpec<?, C> listenerContainerSpec; // NOSONAR final
protected AmqpInboundChannelAdapterSpec(MessageListenerContainerSpec<?, C> listenerContainerSpec) {
super(new AmqpInboundChannelAdapter(listenerContainerSpec.get()));
super(new AmqpInboundChannelAdapter(listenerContainerSpec.getObject()));
this.listenerContainerSpec = listenerContainerSpec;
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.listenerContainerSpec.get(), this.listenerContainerSpec.getId());
return Collections.singletonMap(this.listenerContainerSpec.getObject(), this.listenerContainerSpec.getId());
}
}

12
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/AmqpInboundGatewaySpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-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.
@@ -43,7 +43,7 @@ public abstract class AmqpInboundGatewaySpec
protected final AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec; // NOSONAR final
protected AmqpInboundGatewaySpec(AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec) {
super(new AmqpInboundGateway(listenerContainerSpec.get()));
super(new AmqpInboundGateway(listenerContainerSpec.getObject()));
this.listenerContainerSpec = listenerContainerSpec;
}
@@ -53,16 +53,16 @@ public abstract class AmqpInboundGatewaySpec
* @param listenerContainerSpec the {@link AbstractMessageListenerContainerSpec} to use.
* @param amqpTemplate the {@link AmqpTemplate} to use.
*/
AmqpInboundGatewaySpec(
AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec,
AmqpInboundGatewaySpec(AbstractMessageListenerContainerSpec<?, C> listenerContainerSpec,
AmqpTemplate amqpTemplate) {
super(new AmqpInboundGateway(listenerContainerSpec.get(), amqpTemplate));
super(new AmqpInboundGateway(listenerContainerSpec.getObject(), amqpTemplate));
this.listenerContainerSpec = listenerContainerSpec;
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.listenerContainerSpec.get(), this.listenerContainerSpec.getId());
return Collections.singletonMap(this.listenerContainerSpec.getObject(), this.listenerContainerSpec.getId());
}
}

4
spring-integration-amqp/src/main/java/org/springframework/integration/amqp/dsl/RabbitStreamMessageHandlerSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2022 the original author or authors.
* Copyright 2022-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.
@@ -122,7 +122,7 @@ public class RabbitStreamMessageHandlerSpec
* Set to true to wait for a confirmation.
* @param sync true to wait.
* @return this spec.
* @see #setConfirmTimeout(long)
* @see #confirmTimeout(long)
*/
public RabbitStreamMessageHandlerSpec sync(boolean sync) {
this.target.setSync(sync);

10
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/dsl/AmqpTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2022 the original author or authors.
* Copyright 2014-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.
@@ -50,6 +50,7 @@ import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.amqp.channel.AbstractAmqpChannel;
import org.springframework.integration.amqp.channel.PollableAmqpChannel;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter.BatchMode;
import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
import org.springframework.integration.amqp.support.AmqpHeaderMapper;
@@ -472,15 +473,16 @@ public class AmqpTests {
}
@Bean
public AbstractAmqpChannel unitChannel(ConnectionFactory rabbitConnectionFactory) {
public AmqpPollableMessageChannelSpec<?, PollableAmqpChannel> unitChannel(
ConnectionFactory rabbitConnectionFactory) {
return Amqp.pollableChannel(rabbitConnectionFactory)
.queueName("si.dsl.test")
.channelTransacted(true)
.extractPayload(true)
.inboundHeaderMapper(mapperIn())
.outboundHeaderMapper(mapperOut())
.defaultDeliveryMode(MessageDeliveryMode.NON_PERSISTENT)
.get();
.defaultDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
}
@Bean

6
spring-integration-amqp/src/test/java/org/springframework/integration/amqp/outbound/RabbitStreamMessageHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-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.
@@ -36,6 +36,8 @@ import static org.assertj.core.api.Assertions.assertThat;
/**
* @author Gary Russell
* @author Chris Bono
* @author Artem Bilan
*
* @since 6.0
*/
public class RabbitStreamMessageHandlerTests implements RabbitTestContainer {
@@ -56,7 +58,7 @@ public class RabbitStreamMessageHandlerTests implements RabbitTestContainer {
RabbitStreamMessageHandler handler = RabbitStream.outboundStreamAdapter(streamTemplate)
.sync(true)
.get();
.getObject();
handler.handleMessage(MessageBuilder.withPayload("foo")
.setHeader("bar", "baz")

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-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,7 +19,6 @@ package org.springframework.integration.dsl;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.function.Consumer;
@@ -233,7 +232,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* at the current {@link IntegrationFlow} chain position.
* The provided {@code messageChannelName} is used for the bean registration
* ({@link org.springframework.integration.channel.DirectChannel}), if there is no such a bean
* in the application context. Otherwise the existing {@link MessageChannel} bean is used
* in the application context. Otherwise, the existing {@link MessageChannel} bean is used
* to wire integration endpoints.
* @param messageChannelName the bean name to use.
* @return the current {@link BaseIntegrationFlowDefinition}.
@@ -252,7 +251,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
*/
public B channel(MessageChannelSpec<?, ?> messageChannelSpec) {
Assert.notNull(messageChannelSpec, "'messageChannelSpec' must not be null");
return channel(messageChannelSpec.get());
return channel(messageChannelSpec.getObject());
}
/**
@@ -367,7 +366,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* }
* </pre>
* This method can be used after any {@link #channel} for explicit {@link MessageChannel},
* but with the caution do not impact existing {@link org.springframework.messaging.support.ChannelInterceptor}s.
* but with the caution do not impact existing {@link ChannelInterceptor}s.
* @param wireTapChannel the {@link MessageChannel} bean name to wire-tap.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
@@ -377,8 +376,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
/**
* Populate the {@code Wire Tap} EI Pattern specific
* {@link org.springframework.messaging.support.ChannelInterceptor} implementation
* to the current {@link #currentMessageChannel}.
* {@link ChannelInterceptor} implementation to the current {@link #currentMessageChannel}.
* It is useful when an implicit {@link MessageChannel} is used between endpoints:
* <pre class="code">
* {@code
@@ -388,7 +386,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* }
* </pre>
* This method can be used after any {@link #channel} for explicit {@link MessageChannel},
* but with the caution do not impact existing {@link org.springframework.messaging.support.ChannelInterceptor}s.
* but with the caution do not impact existing {@link ChannelInterceptor}s.
* @param wireTapChannel the {@link MessageChannel} to wire-tap.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
@@ -398,8 +396,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
/**
* Populate the {@code Wire Tap} EI Pattern specific
* {@link org.springframework.messaging.support.ChannelInterceptor} implementation
* to the current {@link #currentMessageChannel}.
* {@link ChannelInterceptor} implementation to the current {@link #currentMessageChannel}.
* It is useful when an implicit {@link MessageChannel} is used between endpoints:
* <pre class="code">
* {@code
@@ -409,7 +406,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* }
* </pre>
* This method can be used after any {@link #channel} for explicit {@link MessageChannel},
* but with the caution do not impact existing {@link org.springframework.messaging.support.ChannelInterceptor}s.
* but with the caution do not impact existing {@link ChannelInterceptor}s.
* @param flow the {@link IntegrationFlow} for wire-tap subflow as an alternative to the {@code wireTapChannel}.
* @param wireTapConfigurer the {@link Consumer} to accept options for the {@link WireTap}.
* @return the current {@link BaseIntegrationFlowDefinition}.
@@ -438,8 +435,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
/**
* Populate the {@code Wire Tap} EI Pattern specific
* {@link org.springframework.messaging.support.ChannelInterceptor} implementation
* to the current {@link #currentMessageChannel}.
* {@link ChannelInterceptor} implementation to the current {@link #currentMessageChannel}.
* It is useful when an implicit {@link MessageChannel} is used between endpoints:
* <pre class="code">
* {@code
@@ -449,7 +445,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* }
* </pre>
* This method can be used after any {@link #channel} for explicit {@link MessageChannel},
* but with the caution do not impact existing {@link org.springframework.messaging.support.ChannelInterceptor}s.
* but with the caution do not impact existing {@link ChannelInterceptor}s.
* @param wireTapChannel the {@link MessageChannel} bean name to wire-tap.
* @param wireTapConfigurer the {@link Consumer} to accept options for the {@link WireTap}.
* @return the current {@link BaseIntegrationFlowDefinition}.
@@ -462,8 +458,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
/**
* Populate the {@code Wire Tap} EI Pattern specific
* {@link org.springframework.messaging.support.ChannelInterceptor} implementation
* to the current {@link #currentMessageChannel}.
* {@link ChannelInterceptor} implementation to the current {@link #currentMessageChannel}.
* It is useful when an implicit {@link MessageChannel} is used between endpoints:
* <pre class="code">
* {@code
@@ -473,7 +468,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* }
* </pre>
* This method can be used after any {@link #channel} for explicit {@link MessageChannel},
* but with the caution do not impact existing {@link org.springframework.messaging.support.ChannelInterceptor}s.
* but with the caution do not impact existing {@link ChannelInterceptor}s.
* @param wireTapChannel the {@link MessageChannel} to wire-tap.
* @param wireTapConfigurer the {@link Consumer} to accept options for the {@link WireTap}.
* @return the current {@link BaseIntegrationFlowDefinition}.
@@ -489,8 +484,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
/**
* Populate the {@code Wire Tap} EI Pattern specific
* {@link org.springframework.messaging.support.ChannelInterceptor} implementation
* to the current {@link #currentMessageChannel}.
* {@link ChannelInterceptor} implementation to the current {@link #currentMessageChannel}.
* <p> It is useful when an implicit {@link MessageChannel} is used between endpoints:
* <pre class="code">
* {@code
@@ -500,14 +494,16 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
* }
* </pre>
* This method can be used after any {@link #channel} for explicit {@link MessageChannel},
* but with the caution do not impact existing {@link org.springframework.messaging.support.ChannelInterceptor}s.
* but with the caution do not impact existing {@link ChannelInterceptor}s.
* @param wireTapSpec the {@link WireTapSpec} to use.
* <p> When this EIP-method is used in the end of flow, it appends {@code nullChannel} to terminate flow properly,
* Otherwise {@code Dispatcher has no subscribers} exception is thrown for implicit {@link DirectChannel}.
* <p> When this EIP-method is used in the end of flow,
* it appends a {@code nullChannel} to terminate flow properly,
* Otherwise a {@code Dispatcher has no subscribers} exception
* is thrown for implicit {@link DirectChannel}.
* @return the current {@link BaseIntegrationFlowDefinition}.
*/
public B wireTap(WireTapSpec wireTapSpec) {
WireTap interceptor = wireTapSpec.get();
WireTap interceptor = wireTapSpec.getObject();
InterceptableChannel currentChannel = currentInterceptableChannel();
addComponent(wireTapSpec);
currentChannel.addInterceptor(interceptor);
@@ -613,7 +609,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
/**
* Populate the {@link MessageTransformingHandler} instance for the
* {@link org.springframework.integration.handler.MessageProcessor} from provided {@link MessageProcessorSpec}.
* {@link MessageProcessor} from provided {@link MessageProcessorSpec}.
* <pre class="code">
* {@code
* .transform(Scripts.script("classpath:myScript.py").variable("foo", bar()))
@@ -629,7 +625,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
/**
* Populate the {@link MessageTransformingHandler} instance for the
* {@link org.springframework.integration.handler.MessageProcessor} from provided {@link MessageProcessorSpec}.
* {@link MessageProcessor} from provided {@link MessageProcessorSpec}.
* In addition, accept options for the integration endpoint using {@link GenericEndpointSpec}.
* <pre class="code">
* {@code
@@ -646,7 +642,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
Consumer<GenericEndpointSpec<MessageTransformingHandler>> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
MessageProcessor<?> processor = messageProcessorSpec.getObject();
return addComponent(processor)
.transform(null, new MethodInvokingTransformer(processor), endpointConfigurer);
}
@@ -832,7 +828,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
*/
public B filter(MessageProcessorSpec<?> messageProcessorSpec, Consumer<FilterEndpointSpec> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
MessageProcessor<?> processor = messageProcessorSpec.getObject();
return addComponent(processor)
.filter(null, new MethodInvokingSelector(processor), endpointConfigurer);
}
@@ -1089,7 +1085,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
MessageProcessor<?> processor = messageProcessorSpec.getObject();
return addComponent(processor)
.handle(new ServiceActivatingHandler(processor), endpointConfigurer);
}
@@ -1118,7 +1114,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
if (messageHandlerSpec instanceof ComponentsRegistration) {
addComponents(((ComponentsRegistration) messageHandlerSpec).getComponentsToRegister());
}
return handle(messageHandlerSpec.get(), endpointConfigurer);
return handle(messageHandlerSpec.getObject(), endpointConfigurer);
}
/**
@@ -1287,7 +1283,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec();
headerEnricherSpec.headers(headers);
Tuple2<ConsumerEndpointFactoryBean, MessageTransformingHandler> tuple2 = headerEnricherSpec.get();
Tuple2<ConsumerEndpointFactoryBean, MessageTransformingHandler> tuple2 = headerEnricherSpec.getObject();
return addComponents(headerEnricherSpec.getComponentsToRegister())
.handle(tuple2.getT2(), endpointConfigurer);
}
@@ -1478,7 +1474,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
MessageProcessor<?> processor = messageProcessorSpec.getObject();
return addComponent(processor)
.split(new MethodInvokingSplitter(processor), endpointConfigurer);
}
@@ -1569,7 +1565,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
public <S extends AbstractMessageSplitter> B split(MessageHandlerSpec<?, S> splitterMessageHandlerSpec,
Consumer<SplitterEndpointSpec<S>> endpointConfigurer) {
Assert.notNull(splitterMessageHandlerSpec, "'splitterMessageHandlerSpec' must not be null");
return split(splitterMessageHandlerSpec.get(), endpointConfigurer);
return split(splitterMessageHandlerSpec.getObject(), endpointConfigurer);
}
/**
@@ -1960,7 +1956,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
Consumer<RouterSpec<Object, MethodInvokingRouter>> routerConfigurer) {
Assert.notNull(messageProcessorSpec, MESSAGE_PROCESSOR_SPEC_MUST_NOT_BE_NULL);
MessageProcessor<?> processor = messageProcessorSpec.get();
MessageProcessor<?> processor = messageProcessorSpec.getObject();
addComponent(processor);
return route(new RouterSpec<>(new MethodInvokingRouter(processor)), routerConfigurer);
@@ -2713,7 +2709,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
if (gatherer != null) {
gatherer.accept(aggregatorSpec);
}
AggregatingMessageHandler aggregatingMessageHandler = aggregatorSpec.get().getT2();
AggregatingMessageHandler aggregatingMessageHandler = aggregatorSpec.getObject().getT2();
addComponent(aggregatingMessageHandler);
ScatterGatherHandler messageHandler = new ScatterGatherHandler(scatterChannel, aggregatingMessageHandler);
return register(new ScatterGatherSpec(messageHandler), scatterGather);
@@ -2766,10 +2762,10 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
if (gatherer != null) {
gatherer.accept(aggregatorSpec);
}
RecipientListRouter recipientListRouter = recipientListRouterSpec.get().getT2();
RecipientListRouter recipientListRouter = recipientListRouterSpec.getObject().getT2();
addComponent(recipientListRouter)
.addComponents(recipientListRouterSpec.getComponentsToRegister());
AggregatingMessageHandler aggregatingMessageHandler = aggregatorSpec.get().getT2();
AggregatingMessageHandler aggregatingMessageHandler = aggregatorSpec.getObject().getT2();
addComponent(aggregatingMessageHandler);
ScatterGatherHandler messageHandler = new ScatterGatherHandler(recipientListRouter, aggregatingMessageHandler);
return register(new ScatterGatherSpec(messageHandler), scatterGather);
@@ -2974,16 +2970,16 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
this.registerOutputChannelIfCan(inputChannel);
}
Tuple2<ConsumerEndpointFactoryBean, ? extends MessageHandler> factoryBeanTuple2 = endpointSpec.get();
Tuple2<ConsumerEndpointFactoryBean, ? extends MessageHandler> factoryBeanTuple2 = endpointSpec.getObject();
addComponents(endpointSpec.getComponentsToRegister());
if (inputChannel instanceof MessageChannelReference) {
factoryBeanTuple2.getT1().setInputChannelName(((MessageChannelReference) inputChannel).getName());
if (inputChannel instanceof MessageChannelReference messageChannelReference) {
factoryBeanTuple2.getT1().setInputChannelName(messageChannelReference.getName());
}
else {
if (inputChannel instanceof FixedSubscriberChannelPrototype) {
String beanName = ((FixedSubscriberChannelPrototype) inputChannel).getName();
if (inputChannel instanceof FixedSubscriberChannelPrototype fixedSubscriberChannel) {
String beanName = fixedSubscriberChannel.getName();
inputChannel = new FixedSubscriberChannel(factoryBeanTuple2.getT2());
if (beanName != null) {
((FixedSubscriberChannel) inputChannel).setBeanName(beanName);
@@ -3002,8 +2998,8 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
Object currComponent = getCurrentComponent();
if (currComponent != null) {
String channelName = null;
if (outputChannel instanceof MessageChannelReference) {
channelName = ((MessageChannelReference) outputChannel).getName();
if (outputChannel instanceof MessageChannelReference channelReference) {
channelName = channelReference.getName();
}
if (currComponent instanceof MessageProducer messageProducer) {
@@ -3015,9 +3011,9 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
messageProducer.setOutputChannel(outputChannel);
}
}
else if (currComponent instanceof SourcePollingChannelAdapterSpec) {
else if (currComponent instanceof SourcePollingChannelAdapterSpec sourcePollingChannelAdapterSpec) {
SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean =
((SourcePollingChannelAdapterSpec) currComponent).get().getT1();
sourcePollingChannelAdapterSpec.getObject().getT1();
if (channelName != null) {
pollingChannelAdapterFactoryBean.setOutputChannelName(channelName);
}
@@ -3081,13 +3077,11 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
}
if (isImplicitChannel()) {
Optional<Object> lastComponent =
components.keySet()
.stream()
.reduce((first, second) -> second);
if (lastComponent.get() instanceof WireTapSpec) {
bridge();
}
components.keySet()
.stream()
.reduce((first, second) -> second)
.filter(WireTapSpec.class::isInstance)
.ifPresent((wireTap) -> bridge());
}
this.integrationFlow = new StandardIntegrationFlow(components);

4
spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.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.
@@ -86,7 +86,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
if (components != null) {
this.componentsToRegister.putAll(components);
}
return poller(pollerMetadataSpec.get());
return poller(pollerMetadataSpec.getObject());
}
/**

89
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationComponentSpec.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.
@@ -16,15 +16,21 @@
package org.springframework.integration.dsl;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.BeanInitializationException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.Lifecycle;
import org.springframework.context.SmartLifecycle;
import org.springframework.expression.spel.standard.SpelExpressionParser;
/**
* The common Builder abstraction. The {@link #get()} method returns the final component.
* The common Builder abstraction.
* If used as a bean definition, must be treated as an {@link FactoryBean},
* therefore its {@link #getObject()} method must not be called in the target configuration.
*
* @param <S> the target {@link IntegrationComponentSpec} implementation type.
* @param <T> the target type.
@@ -35,11 +41,12 @@ import org.springframework.expression.spel.standard.SpelExpressionParser;
*/
@IntegrationDsl
public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpec<S, T>, T>
extends AbstractFactoryBean<T>
implements SmartLifecycle {
implements FactoryBean<T>, InitializingBean, DisposableBean, SmartLifecycle {
protected static final SpelExpressionParser PARSER = new SpelExpressionParser();
protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR - final
protected volatile T target; // NOSONAR
private String id;
@@ -61,8 +68,25 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
/**
* @return the configured component.
* @deprecated since 6.1 with no-op for end-user:
* the {@link #getObject()} is called by the framework at the appropriate phase.
*/
@Deprecated(since = "6.1", forRemoval = true)
public T get() {
return getObject();
}
@Override
public Class<?> getObjectType() {
return getObject().getClass();
}
/**
* !!! This method must not be called from the target configuration !!!
* @return the object backed by this factory bean.
*/
@Override
public T getObject() {
if (this.target == null) {
this.target = doGet();
}
@@ -70,69 +94,57 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
}
@Override
public Class<?> getObjectType() {
return get().getClass();
}
@Override
protected T createInstance() {
T instance = get();
if (instance instanceof InitializingBean) {
try {
((InitializingBean) instance).afterPropertiesSet();
}
catch (Exception e) {
throw new IllegalStateException("Cannot initialize bean: " + instance, e);
public void afterPropertiesSet() {
try {
if (this.target instanceof InitializingBean initializingBean) {
initializingBean.afterPropertiesSet();
}
}
return instance;
catch (Exception ex) {
throw new BeanInitializationException("Cannot initialize bean: " + this.target, ex);
}
}
@Override
protected void destroyInstance(T instance) {
if (instance instanceof DisposableBean) {
public void destroy() {
if (this.target instanceof DisposableBean disposableBean) {
try {
((DisposableBean) instance).destroy();
disposableBean.destroy();
}
catch (Exception e) {
throw new IllegalStateException("Cannot destroy bean: " + instance, e);
throw new IllegalStateException("Cannot destroy bean: " + this.target, e);
}
}
}
@Override
public void start() {
T instance = get();
if (instance instanceof Lifecycle) {
((Lifecycle) instance).start();
if (this.target instanceof Lifecycle lifecycle) {
lifecycle.start();
}
}
@Override
public void stop() {
T instance = get();
if (instance instanceof Lifecycle) {
((Lifecycle) instance).stop();
if (this.target instanceof Lifecycle lifecycle) {
lifecycle.stop();
}
}
@Override
public boolean isRunning() {
T instance = get();
return !(instance instanceof Lifecycle) || ((Lifecycle) instance).isRunning();
return !(this.target instanceof Lifecycle lifecycle) || lifecycle.isRunning();
}
@Override
public boolean isAutoStartup() {
T instance = get();
return instance instanceof SmartLifecycle && ((SmartLifecycle) instance).isAutoStartup();
return this.target instanceof SmartLifecycle lifecycle && lifecycle.isAutoStartup();
}
@Override
public void stop(Runnable callback) {
T instance = get();
if (instance instanceof SmartLifecycle) {
((SmartLifecycle) instance).stop(callback);
if (this.target instanceof SmartLifecycle lifecycle) {
lifecycle.stop(callback);
}
else {
callback.run();
@@ -141,9 +153,8 @@ public abstract class IntegrationComponentSpec<S extends IntegrationComponentSpe
@Override
public int getPhase() {
T instance = get();
if (instance instanceof SmartLifecycle) {
return ((SmartLifecycle) instance).getPhase();
if (this.target instanceof SmartLifecycle lifecycle) {
return lifecycle.getPhase();
}
else {
return 0;

26
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlow.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.
@@ -151,7 +151,7 @@ public interface IntegrationFlow {
*/
static IntegrationFlowBuilder from(MessageChannelSpec<?, ?> messageChannelSpec) {
Assert.notNull(messageChannelSpec, "'messageChannelSpec' must not be null");
return from(messageChannelSpec.get());
return from(messageChannelSpec.getObject());
}
/**
@@ -217,7 +217,7 @@ public interface IntegrationFlow {
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Assert.notNull(messageSourceSpec, "'messageSourceSpec' must not be null");
return from(messageSourceSpec.get(), endpointConfigurer, registerComponents(messageSourceSpec));
return from(messageSourceSpec.getObject(), endpointConfigurer, registerComponents(messageSourceSpec));
}
/**
@@ -321,7 +321,7 @@ public interface IntegrationFlow {
* @see MessageProducerSpec
*/
static IntegrationFlowBuilder from(MessageProducerSpec<?, ?> messageProducerSpec) {
return from(messageProducerSpec.get(), registerComponents(messageProducerSpec));
return from(messageProducerSpec.getObject(), registerComponents(messageProducerSpec));
}
/**
@@ -362,7 +362,7 @@ public interface IntegrationFlow {
* @since 6.0
*/
static IntegrationFlowBuilder from(MessagingGatewaySpec<?, ?> inboundGatewaySpec) {
return from(inboundGatewaySpec.get(), registerComponents(inboundGatewaySpec));
return from(inboundGatewaySpec.getObject(), registerComponents(inboundGatewaySpec));
}
/**
@@ -445,14 +445,14 @@ public interface IntegrationFlow {
"' must be declared as a bean in the application context");
Object lastIntegrationComponentFromOther =
integrationComponents.keySet().stream().reduce((prev, next) -> next).orElse(null);
if (lastIntegrationComponentFromOther instanceof MessageChannel) {
return from((MessageChannel) lastIntegrationComponentFromOther);
if (lastIntegrationComponentFromOther instanceof MessageChannel messageChannel) {
return from(messageChannel);
}
else if (lastIntegrationComponentFromOther instanceof ConsumerEndpointFactoryBean) {
MessageHandler handler = ((ConsumerEndpointFactoryBean) lastIntegrationComponentFromOther).getHandler();
else if (lastIntegrationComponentFromOther instanceof ConsumerEndpointFactoryBean factoryBean) {
MessageHandler handler = factoryBean.getHandler();
handler = extractProxyTarget(handler);
if (handler instanceof AbstractMessageProducingHandler) {
return buildFlowFromOutputChannel((AbstractMessageProducingHandler) handler);
if (handler instanceof AbstractMessageProducingHandler producingHandler) {
return buildFlowFromOutputChannel(producingHandler);
}
lastIntegrationComponentFromOther = handler; // for the exception message below
}
@@ -489,9 +489,9 @@ public interface IntegrationFlow {
}
private static IntegrationFlowBuilder registerComponents(Object spec) {
if (spec instanceof ComponentsRegistration) {
if (spec instanceof ComponentsRegistration componentsRegistration) {
return new IntegrationFlowBuilder()
.addComponents(((ComponentsRegistration) spec).getComponentsToRegister());
.addComponents(componentsRegistration.getComponentsToRegister());
}
return null;
}

4
spring-integration-core/src/main/java/org/springframework/integration/dsl/MessageChannelSpec.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.
@@ -100,7 +100,7 @@ public abstract class MessageChannelSpec<S extends MessageChannelSpec<S, C>, C e
* @see WireTap
*/
public S wireTap(WireTapSpec wireTapSpec) {
WireTap interceptor = wireTapSpec.get();
WireTap interceptor = wireTapSpec.getObject();
this.componentsToRegister.put(interceptor, null);
return interceptor(interceptor);
}

3
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/DslIntegrationConfigurationInitializer.java
View File

@@ -31,11 +31,12 @@ import org.springframework.util.Assert;
* Registers {@link IntegrationFlowBeanPostProcessor} and checks if all
* {@link org.springframework.integration.dsl.IntegrationComponentSpec} are extracted to
* the target object using
* {@link org.springframework.integration.dsl.IntegrationComponentSpec#get()}.
* {@link org.springframework.integration.dsl.IntegrationComponentSpec#getObject()}.
*
* @author Artem Bilan
* @author Gary Russell
* @author Chris Bono
*
* @since 5.0
*
* @see org.springframework.integration.config.IntegrationConfigurationBeanFactoryPostProcessor

35
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.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.
@@ -113,14 +113,14 @@ public class IntegrationFlowBeanPostProcessor
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof StandardIntegrationFlow) {
return processStandardIntegrationFlow((StandardIntegrationFlow) bean, beanName);
if (bean instanceof StandardIntegrationFlow standardIntegrationFlow) {
return processStandardIntegrationFlow(standardIntegrationFlow, beanName);
}
else if (bean instanceof IntegrationFlow) {
return processIntegrationFlowImpl((IntegrationFlow) bean, beanName);
else if (bean instanceof IntegrationFlow integrationFlow) {
return processIntegrationFlowImpl(integrationFlow, beanName);
}
if (bean instanceof IntegrationComponentSpec) {
processIntegrationComponentSpec(beanName, (IntegrationComponentSpec<?, ?>) bean);
if (bean instanceof IntegrationComponentSpec<?, ?> integrationComponentSpec) {
processIntegrationComponentSpec(beanName, integrationComponentSpec);
}
return bean;
}
@@ -156,8 +156,8 @@ public class IntegrationFlowBeanPostProcessor
for (Map.Entry<Object, String> entry : integrationComponents.entrySet()) {
Object component = entry.getKey();
if (component instanceof ConsumerEndpointSpec<?, ?> endpointSpec) {
MessageHandler messageHandler = endpointSpec.get().getT2();
ConsumerEndpointFactoryBean endpoint = endpointSpec.get().getT1();
MessageHandler messageHandler = endpointSpec.getObject().getT2();
ConsumerEndpointFactoryBean endpoint = endpointSpec.getObject().getT1();
String id = endpointSpec.getId();
if (id == null) {
@@ -177,8 +177,8 @@ public class IntegrationFlowBeanPostProcessor
registerComponent(endpoint, id, flowBeanName);
targetIntegrationComponents.put(endpoint, id);
}
else if (component instanceof MessageChannelReference) {
String channelBeanName = ((MessageChannelReference) component).getName();
else if (component instanceof MessageChannelReference messageChannelReference) {
String channelBeanName = messageChannelReference.getName();
if (!this.beanFactory.containsBean(channelBeanName)) {
DirectChannel directChannel = new DirectChannel();
registerComponent(directChannel, channelBeanName, flowBeanName);
@@ -196,7 +196,7 @@ public class IntegrationFlowBeanPostProcessor
generateBeanName(o.getKey(), flowNamePrefix, o.getValue(),
useFlowIdAsPrefix)));
}
SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1();
SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.getObject().getT1();
String id = spec.getId();
if (id == null) {
id = generateBeanName(pollingChannelAdapterFactoryBean, flowNamePrefix, entry.getValue(),
@@ -209,12 +209,13 @@ public class IntegrationFlowBeanPostProcessor
registerComponent(pollingChannelAdapterFactoryBean, id, flowBeanName);
targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id);
MessageSource<?> messageSource = spec.get().getT2();
MessageSource<?> messageSource = spec.getObject().getT2();
if (noBeanPresentForComponent(messageSource, flowBeanName)) {
String messageSourceId = id + ".source";
if (messageSource instanceof NamedComponent
&& ((NamedComponent) messageSource).getComponentName() != null) {
messageSourceId = ((NamedComponent) messageSource).getComponentName();
if (messageSource instanceof NamedComponent namedComponent
&& namedComponent.getComponentName() != null) {
messageSourceId = namedComponent.getComponentName();
}
registerComponent(messageSource, messageSourceId, flowBeanName);
}
@@ -342,7 +343,7 @@ public class IntegrationFlowBeanPostProcessor
}
private void processIntegrationComponentSpec(String beanName, IntegrationComponentSpec<?, ?> bean) {
Object target = bean.get();
Object target = bean.getObject();
invokeBeanInitializationHooks(beanName, target);

4
spring-integration-core/src/test/java/org/springframework/integration/channel/reactive/FluxMessageChannelTests.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.
@@ -119,7 +119,7 @@ public class FluxMessageChannelTests {
@Test
void testFluxMessageChannelCleanUp() throws InterruptedException {
FluxMessageChannel flux = MessageChannels.flux().get();
FluxMessageChannel flux = MessageChannels.flux().getObject();
CountDownLatch finishLatch = new CountDownLatch(1);

10
spring-integration-core/src/test/java/org/springframework/integration/dsl/PollersTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-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.
@@ -35,19 +35,19 @@ public class PollersTests {
@Test
public void testDurations() {
PeriodicTrigger trigger = (PeriodicTrigger) Pollers.fixedDelay(Duration.ofMinutes(1L)).get().getTrigger();
PeriodicTrigger trigger = (PeriodicTrigger) Pollers.fixedDelay(Duration.ofMinutes(1L)).getObject().getTrigger();
assertThat(trigger.getPeriodDuration()).isEqualTo(Duration.ofSeconds(60));
assertThat(trigger.isFixedRate()).isFalse();
trigger = (PeriodicTrigger) Pollers.fixedDelay(Duration.ofMinutes(1L), Duration.ofSeconds(10L))
.get().getTrigger();
.getObject().getTrigger();
assertThat(trigger.getPeriodDuration()).isEqualTo(Duration.ofSeconds(60));
assertThat(trigger.getInitialDelayDuration()).isEqualTo(Duration.ofSeconds(10));
assertThat(trigger.isFixedRate()).isFalse();
trigger = (PeriodicTrigger) Pollers.fixedRate(Duration.ofMinutes(1L)).get().getTrigger();
trigger = (PeriodicTrigger) Pollers.fixedRate(Duration.ofMinutes(1L)).getObject().getTrigger();
assertThat(trigger.getPeriodDuration()).isEqualTo(Duration.ofSeconds(60));
assertThat(trigger.isFixedRate()).isTrue();
trigger = (PeriodicTrigger) Pollers.fixedRate(Duration.ofMinutes(1L), Duration.ofSeconds(10L))
.get().getTrigger();
.getObject().getTrigger();
assertThat(trigger.getPeriodDuration()).isEqualTo(Duration.ofSeconds(60));
assertThat(trigger.getInitialDelayDuration()).isEqualTo(Duration.ofSeconds(10));
assertThat(trigger.isFixedRate()).isTrue();

7
spring-integration-core/src/test/java/org/springframework/integration/dsl/composition/IntegrationFlowCompositionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2021-2022 the original author or authors.
* Copyright 2021-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.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.PollerSpec;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.context.IntegrationFlowContext;
import org.springframework.integration.scheduling.PollerMetadata;
@@ -142,8 +143,8 @@ public class IntegrationFlowCompositionTests {
public static class ContextConfiguration {
@Bean(PollerMetadata.DEFAULT_POLLER)
PollerMetadata defaultPoller() {
return Pollers.fixedDelay(100).get();
PollerSpec defaultPoller() {
return Pollers.fixedDelay(100);
}
@Bean

27
spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.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.
@@ -54,13 +54,16 @@ import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.core.GenericTransformer;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.PollerSpec;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.QueueChannelSpec;
import org.springframework.integration.dsl.Transformers;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
@@ -575,8 +578,8 @@ public class IntegrationFlowTests {
}
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
return Pollers.fixedRate(100).get();
public PollerSpec poller() {
return Pollers.fixedRate(100);
}
@Bean(name = IntegrationContextUtils.TASK_SCHEDULER_BEAN_NAME)
@@ -588,8 +591,8 @@ public class IntegrationFlowTests {
@Bean
public MessageChannel suppliedChannel() {
return MessageChannels.queue(10).get();
public QueueChannelSpec suppliedChannel() {
return MessageChannels.queue(10);
}
}
@@ -608,8 +611,8 @@ public class IntegrationFlowTests {
}
@Bean
public MessageChannel suppliedChannel2() {
return MessageChannels.queue(10).get();
public QueueChannelSpec suppliedChannel2() {
return MessageChannels.queue(10);
}
}
@@ -627,12 +630,12 @@ public class IntegrationFlowTests {
@Bean
public MessageChannel inputChannel() {
return MessageChannels.direct().get();
return new DirectChannel();
}
@Bean
public MessageChannel foo() {
return MessageChannels.publishSubscribe().get();
return new PublishSubscribeChannel();
}
}
@@ -682,7 +685,7 @@ public class IntegrationFlowTests {
@Bean
public MessageChannel publishSubscribeChannel() {
return MessageChannels.publishSubscribe().get();
return new PublishSubscribeChannel();
}
@Bean
@@ -784,8 +787,8 @@ public class IntegrationFlowTests {
private MethodInterceptor delayedAdvice;
@Bean
public QueueChannel successChannel() {
return MessageChannels.queue().get();
public QueueChannelSpec successChannel() {
return MessageChannels.queue();
}
@Bean

5
spring-integration-core/src/test/java/org/springframework/integration/dsl/gateway/GatewayDslTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2022 the original author or authors.
* Copyright 2019-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.
@@ -36,7 +36,6 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.gateway.GatewayProxyFactoryBean;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.gateway.MethodArgsHolder;
@@ -189,7 +188,7 @@ public class GatewayDslTests {
@Bean
public MessageChannel gatewayError() {
return MessageChannels.queue().get();
return new QueueChannel();
}

4
spring-integration-core/src/test/java/org/springframework/integration/dsl/transformers/TransformerTests.java
View File

@@ -30,12 +30,12 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.annotation.Transformer;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.codec.Codec;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.Transformers;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.integration.handler.advice.ExpressionEvaluatingRequestHandlerAdvice;
@@ -423,7 +423,7 @@ public class TransformerTests {
@Bean
public MessageChannel enricherReplyChannel() {
return MessageChannels.direct().get();
return new DirectChannel();
}
@Bean

15
spring-integration-core/src/test/java/org/springframework/integration/json/ContentTypeConversionTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2022 the original author or authors.
* Copyright 2017-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.
@@ -20,8 +20,7 @@ import java.util.Collections;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
@@ -32,6 +31,7 @@ import org.springframework.integration.annotation.GatewayHeader;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.DirectChannelSpec;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.messaging.Message;
@@ -39,7 +39,7 @@ import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
import static org.assertj.core.api.Assertions.assertThat;
@@ -48,7 +48,7 @@ import static org.assertj.core.api.Assertions.assertThat;
*
* @since 5.0
*/
@RunWith(SpringRunner.class)
@SpringJUnitConfig
public class ContentTypeConversionTests {
@Autowired
@@ -91,7 +91,7 @@ public class ContentTypeConversionTests {
}
@Bean
public MessageChannel serviceChannel(final AtomicReference<Object> sendData) {
public DirectChannelSpec serviceChannel(AtomicReference<Object> sendData) {
return MessageChannels.direct()
.interceptor(new ChannelInterceptor() {
@@ -101,8 +101,7 @@ public class ContentTypeConversionTests {
return message;
}
})
.get();
});
}
@Bean

4
spring-integration-core/src/test/kotlin/org/springframework/integration/dsl/KotlinDslTests.kt
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2021 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.
@@ -231,7 +231,7 @@ class KotlinDslTests {
@Bean(PollerMetadata.DEFAULT_POLLER)
fun defaultPoller() =
Pollers.fixedDelay(100).maxMessagesPerPoll(1).get()
Pollers.fixedDelay(100).maxMessagesPerPoll(1)
@Bean
fun convertFlow() =

2
spring-integration-groovy/src/test/groovy/org/springframework/integration/groovy/dsl/test/GroovyDslTests.groovy
View File

@@ -216,7 +216,7 @@ class GroovyDslTests {
@Bean(PollerMetadata.DEFAULT_POLLER)
poller() {
Pollers.fixedDelay(1000).get()
Pollers.fixedDelay(1000)
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundChannelAdapterSpec.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.
@@ -56,7 +56,7 @@ public class TcpInboundChannelAdapterSpec
*/
protected TcpInboundChannelAdapterSpec(AbstractConnectionFactorySpec<?, ?> connectionFactorySpec) {
super(new TcpReceivingChannelAdapter());
this.connectionFactory = connectionFactorySpec.get();
this.connectionFactory = connectionFactorySpec.getObject();
this.target.setConnectionFactory(this.connectionFactory);
}
@@ -94,7 +94,7 @@ public class TcpInboundChannelAdapterSpec
public Map<Object, String> getComponentsToRegister() {
return this.connectionFactory != null
? Collections.singletonMap(this.connectionFactory, this.connectionFactory.getComponentName())
: null;
: Collections.emptyMap();
}
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpInboundGatewaySpec.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.
@@ -55,7 +55,7 @@ public class TcpInboundGatewaySpec extends MessagingGatewaySpec<TcpInboundGatewa
*/
protected TcpInboundGatewaySpec(AbstractConnectionFactorySpec<?, ?> connectionFactorySpec) {
super(new TcpInboundGateway());
this.connectionFactory = connectionFactorySpec.get();
this.connectionFactory = connectionFactorySpec.getObject();
this.target.setConnectionFactory(this.connectionFactory);
}
@@ -93,7 +93,7 @@ public class TcpInboundGatewaySpec extends MessagingGatewaySpec<TcpInboundGatewa
public Map<Object, String> getComponentsToRegister() {
return this.connectionFactory != null
? Collections.singletonMap(this.connectionFactory, this.connectionFactory.getComponentName())
: null;
: Collections.emptyMap();
}
}

6
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundChannelAdapterSpec.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.
@@ -56,7 +56,7 @@ public class TcpOutboundChannelAdapterSpec
*/
protected TcpOutboundChannelAdapterSpec(AbstractConnectionFactorySpec<?, ?> connectionFactorySpec) {
this.target = new TcpSendingMessageHandler();
this.connectionFactory = connectionFactorySpec.get();
this.connectionFactory = connectionFactorySpec.getObject();
this.target.setConnectionFactory(this.connectionFactory);
}
@@ -94,7 +94,7 @@ public class TcpOutboundChannelAdapterSpec
public Map<Object, String> getComponentsToRegister() {
return this.connectionFactory != null
? Collections.singletonMap(this.connectionFactory, this.connectionFactory.getComponentName())
: null;
: Collections.emptyMap();
}
}

30
spring-integration-ip/src/main/java/org/springframework/integration/ip/dsl/TcpOutboundGatewaySpec.java
View File

@@ -58,7 +58,7 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
*/
public TcpOutboundGatewaySpec(TcpClientConnectionFactorySpec<?, ?> connectionFactorySpec) {
this.target = new TcpOutboundGateway();
this.connectionFactory = connectionFactorySpec.get();
this.connectionFactory = connectionFactorySpec.getObject();
this.target.setConnectionFactory(this.connectionFactory);
}
@@ -120,8 +120,21 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
* @param channelName the name.
* @return the spec.
* @since 5.4
* @deprecated in favor of {@link #unsolicitedMessageChannelName(String)}
* due to the typo in method name.
*/
@Deprecated(since = "6.1", forRemoval = true)
public TcpOutboundGatewaySpec unsolictedMessageChannelName(String channelName) {
return unsolicitedMessageChannelName(channelName);
}
/**
* Set the unsolicited message channel name.
* @param channelName the name.
* @return the spec.
* @since 6.1
*/
public TcpOutboundGatewaySpec unsolicitedMessageChannelName(String channelName) {
this.target.setUnsolicitedMessageChannelName(channelName);
return this;
}
@@ -131,8 +144,21 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
* @param channel the channel.
* @return the spec.
* @since 5.4
* @deprecated in favor of {@link #unsolicitedMessageChannel(MessageChannel)}
* due to the typo in method name.
*/
@Deprecated(since = "6.1", forRemoval = true)
public TcpOutboundGatewaySpec unsolictedMessageChannelName(MessageChannel channel) {
return unsolicitedMessageChannel(channel);
}
/**
* Set the unsolicited message channel.
* @param channel the channel.
* @return the spec.
* @since 6.1
*/
public TcpOutboundGatewaySpec unsolicitedMessageChannel(MessageChannel channel) {
this.target.setUnsolicitedMessageChannel(channel);
return this;
}
@@ -141,7 +167,7 @@ public class TcpOutboundGatewaySpec extends MessageHandlerSpec<TcpOutboundGatewa
public Map<Object, String> getComponentsToRegister() {
return this.connectionFactory != null
? Collections.singletonMap(this.connectionFactory, this.connectionFactory.getComponentName())
: null;
: Collections.emptyMap();
}
}

28
spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/ConnectionFacforyTests.java → spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/ConnectionFactoryTests.java
View File

@@ -45,16 +45,18 @@ import static org.mockito.Mockito.mock;
/**
* @author Gary Russell
* @author Tim Ysewyn
* @author Artem Bilan
*
* @since 5.0
*
*/
public class ConnectionFacforyTests {
public class ConnectionFactoryTests {
@Test
public void test() throws Exception {
ApplicationEventPublisher publisher = e -> {
};
AbstractServerConnectionFactory server = Tcp.netServer(0).backlog(2).soTimeout(5000).get();
AbstractServerConnectionFactory server = Tcp.netServer(0).backlog(2).soTimeout(5000).getObject();
final AtomicReference<Message<?>> received = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(1);
server.registerListener(m -> {
@@ -66,7 +68,7 @@ public class ConnectionFacforyTests {
server.afterPropertiesSet();
server.start();
TestingUtilities.waitListening(server, null);
AbstractClientConnectionFactory client = Tcp.netClient("localhost", server.getPort()).get();
AbstractClientConnectionFactory client = Tcp.netClient("localhost", server.getPort()).getObject();
client.setApplicationEventPublisher(publisher);
client.afterPropertiesSet();
client.start();
@@ -78,20 +80,20 @@ public class ConnectionFacforyTests {
}
@Test
public void shouldReturnNioFlavor() throws Exception {
AbstractServerConnectionFactory server = Tcp.nioServer(0).get();
public void shouldReturnNioFlavor() {
AbstractServerConnectionFactory server = Tcp.nioServer(0).getObject();
assertThat(server instanceof TcpNioServerConnectionFactory).isTrue();
AbstractClientConnectionFactory client = Tcp.nioClient("localhost", server.getPort()).get();
AbstractClientConnectionFactory client = Tcp.nioClient("localhost", server.getPort()).getObject();
assertThat(client instanceof TcpNioClientConnectionFactory).isTrue();
}
@Test
public void shouldReturnNetFlavor() throws Exception {
AbstractServerConnectionFactory server = Tcp.netServer(0).get();
public void shouldReturnNetFlavor() {
AbstractServerConnectionFactory server = Tcp.netServer(0).getObject();
assertThat(server instanceof TcpNetServerConnectionFactory).isTrue();
AbstractClientConnectionFactory client = Tcp.netClient("localhost", server.getPort()).get();
AbstractClientConnectionFactory client = Tcp.netClient("localhost", server.getPort()).getObject();
assertThat(client instanceof TcpNetClientConnectionFactory).isTrue();
}
@@ -104,7 +106,7 @@ public class ConnectionFacforyTests {
.socketSupport(sockSupp)
.connectionSupport(conSupp)
.socketFactorySupport(factSupp)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(server, "tcpSocketSupport")).isSameAs(sockSupp);
assertThat(TestUtils.getPropertyValue(server, "tcpNetConnectionSupport")).isSameAs(conSupp);
assertThat(TestUtils.getPropertyValue(server, "tcpSocketFactorySupport")).isSameAs(factSupp);
@@ -118,7 +120,7 @@ public class ConnectionFacforyTests {
.socketSupport(sockSupp)
.directBuffers(true)
.connectionSupport(conSupp)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(server, "tcpSocketSupport")).isSameAs(sockSupp);
assertThat(TestUtils.getPropertyValue(server, "usingDirectBuffers", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(server, "tcpNioConnectionSupport")).isSameAs(conSupp);
@@ -133,7 +135,7 @@ public class ConnectionFacforyTests {
.socketSupport(sockSupp)
.connectionSupport(conSupp)
.socketFactorySupport(factSupp)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(client, "tcpSocketSupport")).isSameAs(sockSupp);
assertThat(TestUtils.getPropertyValue(client, "tcpNetConnectionSupport")).isSameAs(conSupp);
assertThat(TestUtils.getPropertyValue(client, "tcpSocketFactorySupport")).isSameAs(factSupp);
@@ -147,7 +149,7 @@ public class ConnectionFacforyTests {
.socketSupport(sockSupp)
.directBuffers(true)
.connectionSupport(conSupp)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(client, "tcpSocketSupport")).isSameAs(sockSupp);
assertThat(TestUtils.getPropertyValue(client, "usingDirectBuffers", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(client, "tcpNioConnectionSupport")).isSameAs(conSupp);

60
spring-integration-ip/src/test/java/org/springframework/integration/ip/dsl/IpIntegrationTests.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.
@@ -48,6 +48,8 @@ import org.springframework.integration.ip.tcp.TcpSendingMessageHandler;
import org.springframework.integration.ip.tcp.connection.AbstractClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.AbstractServerConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpConnectionServerListeningEvent;
import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionFactory;
import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory;
import org.springframework.integration.ip.tcp.serializer.ByteArrayRawSerializer;
import org.springframework.integration.ip.tcp.serializer.TcpCodecs;
import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
@@ -126,21 +128,21 @@ public class IpIntegrationTests {
void testTcpAdapters() {
ApplicationEventPublisher publisher = e -> {
};
AbstractServerConnectionFactory server = Tcp.netServer(0).backlog(2).soTimeout(5000).id("server").get();
AbstractServerConnectionFactory server = Tcp.netServer(0).backlog(2).soTimeout(5000).id("server").getObject();
assertThat(server.getComponentName()).isEqualTo("server");
server.setApplicationEventPublisher(publisher);
server.afterPropertiesSet();
TcpReceivingChannelAdapter inbound = Tcp.inboundAdapter(server).get();
TcpReceivingChannelAdapter inbound = Tcp.inboundAdapter(server).getObject();
QueueChannel received = new QueueChannel();
inbound.setOutputChannel(received);
inbound.afterPropertiesSet();
inbound.start();
TestingUtilities.waitListening(server, null);
AbstractClientConnectionFactory client = Tcp.netClient("localhost", server.getPort()).id("client").get();
AbstractClientConnectionFactory client = Tcp.netClient("localhost", server.getPort()).id("client").getObject();
assertThat(client.getComponentName()).isEqualTo("client");
client.setApplicationEventPublisher(publisher);
client.afterPropertiesSet();
TcpSendingMessageHandler handler = Tcp.outboundAdapter(client).get();
TcpSendingMessageHandler handler = Tcp.outboundAdapter(client).getObject();
handler.start();
handler.handleMessage(new GenericMessage<>("foo"));
Message<?> receivedMessage = received.receive(10000);
@@ -193,7 +195,8 @@ public class IpIntegrationTests {
UdpMulticastOutboundChannelAdapterSpec udpMulticastOutboundChannelAdapterSpec2 =
udpMulticastOutboundChannelAdapterSpec1.timeToLive(10);
assertThat(udpMulticastOutboundChannelAdapterSpec2.get()).isInstanceOf(MulticastSendingMessageHandler.class);
assertThat(udpMulticastOutboundChannelAdapterSpec2.getObject())
.isInstanceOf(MulticastSendingMessageHandler.class);
}
@Test
@@ -249,17 +252,16 @@ public class IpIntegrationTests {
private volatile String connectionId;
@Bean
public AbstractServerConnectionFactory server1() {
public TcpNetServerConnectionFactorySpec server1() {
return Tcp.netServer(0)
.serializer(TcpCodecs.lengthHeader1())
.deserializer(TcpCodecs.crlf())
.get();
.deserializer(TcpCodecs.crlf());
}
@Bean
public IntegrationFlow inTcpGateway() {
public IntegrationFlow inTcpGateway(TcpNetServerConnectionFactory server1) {
return IntegrationFlow.from(
Tcp.inboundGateway(server1())
Tcp.inboundGateway(server1)
.replyTimeout(1)
.errorOnTimeout(true)
.errorChannel("inTcpGatewayErrorFlow.input"))
@@ -276,8 +278,8 @@ public class IpIntegrationTests {
}
@Bean
public IntegrationFlow unsolicitedServerSide() {
return f -> f.handle(Tcp.outboundAdapter(server1()));
public IntegrationFlow unsolicitedServerSide(TcpNetServerConnectionFactory server1) {
return f -> f.handle(Tcp.outboundAdapter(server1));
}
@Bean
@@ -321,19 +323,17 @@ public class IpIntegrationTests {
}
@Bean
public AbstractClientConnectionFactory client1() {
return Tcp.netClient("localhost", server1().getPort())
public TcpNetClientConnectionFactorySpec client1(TcpNetServerConnectionFactory server1) {
return Tcp.netClient("localhost", server1.getPort())
.serializer(TcpCodecs.crlf())
.deserializer(TcpCodecs.lengthHeader1())
.get();
.deserializer(TcpCodecs.lengthHeader1());
}
@Bean
public TcpOutboundGateway tcpOut() {
return Tcp.outboundGateway(client1())
public TcpOutboundGatewaySpec tcpOut(TcpNetClientConnectionFactory client1) {
return Tcp.outboundGateway(client1)
.remoteTimeout(m -> 5000)
.unsolictedMessageChannelName("unsolicited")
.get();
.unsolicitedMessageChannelName("unsolicited");
}
@Bean
@@ -342,19 +342,17 @@ public class IpIntegrationTests {
}
@Bean
public AbstractClientConnectionFactory client2() {
return Tcp.netClient("localhost", server1().getPort())
public TcpNetClientConnectionFactorySpec client2(TcpNetServerConnectionFactory server1) {
return Tcp.netClient("localhost", server1.getPort())
.serializer(TcpCodecs.crlf())
.deserializer(TcpCodecs.lengthHeader1())
.get();
.deserializer(TcpCodecs.lengthHeader1());
}
@Bean
public TcpOutboundGateway tcpOutAsync() {
return Tcp.outboundGateway(client2())
public TcpOutboundGatewaySpec tcpOutAsync(TcpNetClientConnectionFactory client2) {
return Tcp.outboundGateway(client2)
.async(true)
.remoteTimeout(m -> 5000)
.get();
.remoteTimeout(m -> 5000);
}
@Bean
@@ -371,9 +369,9 @@ public class IpIntegrationTests {
}
@Bean
public IntegrationFlow clientTcpFlow() {
public IntegrationFlow clientTcpFlow(TcpOutboundGateway tcpOut) {
return f -> f
.handle(tcpOut(), e -> e.advice(testAdvice()))
.handle(tcpOut, e -> e.advice(testAdvice()))
.transform(Transformers.objectToString());
}

58
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/Jms.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2021 the original author or authors.
* Copyright 2014-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.
@@ -208,7 +208,7 @@ public final class Jms {
public static JmsMessageDrivenChannelAdapterSpec<?> messageDrivenChannelAdapter(
JmsListenerContainerSpec<?, ? extends AbstractMessageListenerContainer> jmsListenerContainerSpec) {
return new JmsMessageDrivenChannelAdapterSpec<>(jmsListenerContainerSpec.get());
return new JmsMessageDrivenChannelAdapterSpec<>(jmsListenerContainerSpec.getObject());
}
/**
@@ -216,7 +216,9 @@ public final class Jms {
* @param listenerContainer the {@link AbstractMessageListenerContainer} to build on
* @return the {@link JmsMessageDrivenChannelAdapterSpec} instance
*/
public static JmsMessageDrivenChannelAdapterSpec<?> messageDrivenChannelAdapter(AbstractMessageListenerContainer listenerContainer) {
public static JmsMessageDrivenChannelAdapterSpec<?> messageDrivenChannelAdapter(
AbstractMessageListenerContainer listenerContainer) {
return new JmsMessageDrivenChannelAdapterSpec<>(listenerContainer);
}
@@ -227,14 +229,10 @@ public final class Jms {
*/
public static JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<JmsDefaultListenerContainerSpec, DefaultMessageListenerContainer>
messageDrivenChannelAdapter(ConnectionFactory connectionFactory) {
try {
return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<>(
new JmsDefaultListenerContainerSpec()
.connectionFactory(connectionFactory));
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<>(
new JmsDefaultListenerContainerSpec()
.connectionFactory(connectionFactory));
}
/**
@@ -249,15 +247,11 @@ public final class Jms {
public static <C extends AbstractMessageListenerContainer>
JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec<?, C>
messageDrivenChannelAdapter(ConnectionFactory connectionFactory, Class<C> containerClass) {
try {
JmsListenerContainerSpec<?, C> spec =
new JmsListenerContainerSpec<>(containerClass)
.connectionFactory(connectionFactory);
return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec(spec);
}
catch (Exception e) {
throw new IllegalStateException(e);
}
JmsListenerContainerSpec<?, C> spec =
new JmsListenerContainerSpec<>(containerClass)
.connectionFactory(connectionFactory);
return new JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec(spec);
}
/**
@@ -268,14 +262,10 @@ public final class Jms {
*/
public static JmsDefaultListenerContainerSpec container(ConnectionFactory connectionFactory,
Destination destination) {
try {
return new JmsDefaultListenerContainerSpec()
.connectionFactory(connectionFactory)
.destination(destination);
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return new JmsDefaultListenerContainerSpec()
.connectionFactory(connectionFactory)
.destination(destination);
}
/**
@@ -286,14 +276,10 @@ public final class Jms {
*/
public static JmsDefaultListenerContainerSpec container(ConnectionFactory connectionFactory,
String destinationName) {
try {
return new JmsDefaultListenerContainerSpec()
.connectionFactory(connectionFactory)
.destination(destinationName);
}
catch (Exception e) {
throw new IllegalStateException(e);
}
return new JmsDefaultListenerContainerSpec()
.connectionFactory(connectionFactory)
.destination(destinationName);
}
private Jms() {

7
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundChannelAdapterSpec.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.
@@ -49,7 +49,8 @@ public class JmsInboundChannelAdapterSpec<S extends JmsInboundChannelAdapterSpec
}
private JmsInboundChannelAdapterSpec(ConnectionFactory connectionFactory) {
this.target = new JmsDestinationPollingSource(this.jmsTemplateSpec.connectionFactory(connectionFactory).get());
this.target =
new JmsDestinationPollingSource(this.jmsTemplateSpec.connectionFactory(connectionFactory).getObject());
}
/**
@@ -118,7 +119,7 @@ public class JmsInboundChannelAdapterSpec<S extends JmsInboundChannelAdapterSpec
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.jmsTemplateSpec.get(), this.jmsTemplateSpec.getId());
return Collections.singletonMap(this.jmsTemplateSpec.getObject(), this.jmsTemplateSpec.getId());
}
}

4
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsInboundGatewaySpec.java
View File

@@ -246,9 +246,9 @@ public class JmsInboundGatewaySpec<S extends JmsInboundGatewaySpec<S>>
private final S spec;
protected JmsInboundGatewayListenerContainerSpec(S spec) {
super(spec.get());
super(spec.getObject());
this.spec = spec;
this.spec.get().setAutoStartup(false);
this.spec.getObject().setAutoStartup(false);
}
/**

13
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsMessageDrivenChannelAdapterSpec.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.
@@ -80,9 +80,9 @@ public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChanne
}
/**
* Set to false to prevent listener container shutdown when the endpoint is stopped.
* Set to 'false' to prevent listener container shutdown when the endpoint is stopped.
* Then, if so configured, any cached consumer(s) in the container will remain.
* Otherwise the shared connection and will be closed and the listener invokers shut
* Otherwise, the shared connection and will be closed and the listener invokers shut
* down; this behavior is new starting with version 5.1. Default: true.
* @param shutdown false to not shutdown.
* @return the spec.
@@ -106,9 +106,9 @@ public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChanne
private final S spec;
protected JmsMessageDrivenChannelAdapterListenerContainerSpec(S spec) {
super(spec.get());
super(spec.getObject());
this.spec = spec;
this.spec.get().setAutoStartup(false);
this.spec.getObject().setAutoStartup(false);
}
@@ -141,6 +141,7 @@ public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChanne
*/
public JmsMessageDrivenChannelAdapterListenerContainerSpec<S, C> configureListenerContainer(
Consumer<S> configurer) {
Assert.notNull(configurer, "'configurer' must not be null");
configurer.accept(this.spec);
return _this();
@@ -148,7 +149,7 @@ public class JmsMessageDrivenChannelAdapterSpec<S extends JmsMessageDrivenChanne
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.spec.get(), this.spec.getId());
return Collections.singletonMap(this.spec.getObject(), this.spec.getId());
}
}

11
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundChannelAdapterSpec.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.
@@ -51,7 +51,8 @@ public class JmsOutboundChannelAdapterSpec<S extends JmsOutboundChannelAdapterSp
}
private JmsOutboundChannelAdapterSpec(ConnectionFactory connectionFactory) {
this.target = new JmsSendingMessageHandler(this.jmsTemplateSpec.connectionFactory(connectionFactory).get());
this.target =
new JmsSendingMessageHandler(this.jmsTemplateSpec.connectionFactory(connectionFactory).getObject());
}
/**
@@ -101,7 +102,7 @@ public class JmsOutboundChannelAdapterSpec<S extends JmsOutboundChannelAdapterSp
* which a message will be sent.
* @param destination the destination name.
* @return the current {@link JmsOutboundChannelAdapterSpec}.
* @see JmsSendingMessageHandler#setDestinationExpression(Expression)
* @see JmsSendingMessageHandler#setDestinationExpression
*/
public S destinationExpression(String destination) {
this.target.setDestinationExpression(PARSER.parseExpression(destination));
@@ -119,7 +120,7 @@ public class JmsOutboundChannelAdapterSpec<S extends JmsOutboundChannelAdapterSp
* @param destinationFunction the destination function.
* @param <P> the expected payload type.
* @return the current {@link JmsOutboundChannelAdapterSpec}.
* @see JmsSendingMessageHandler#setDestinationExpression(Expression)
* @see JmsSendingMessageHandler#setDestinationExpression
* @see FunctionExpression
*/
public <P> S destination(Function<Message<P>, ?> destinationFunction) {
@@ -194,7 +195,7 @@ public class JmsOutboundChannelAdapterSpec<S extends JmsOutboundChannelAdapterSp
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.jmsTemplateSpec.get(), this.jmsTemplateSpec.getId());
return Collections.singletonMap(this.jmsTemplateSpec.getObject(), this.jmsTemplateSpec.getId());
}
}

4
spring-integration-jms/src/main/java/org/springframework/integration/jms/dsl/JmsOutboundGatewaySpec.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.
@@ -296,7 +296,7 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec<JmsOutboundGatewa
Assert.notNull(configurer, "'configurer' must not be null");
ReplyContainerSpec spec = new ReplyContainerSpec();
configurer.accept(spec);
this.target.setReplyContainerProperties(spec.get());
this.target.setReplyContainerProperties(spec.getObject());
return _this();
}

28
spring-integration-jms/src/test/java/org/springframework/integration/jms/dsl/JmsTests.java
View File

@@ -36,7 +36,6 @@ import org.springframework.integration.annotation.InboundChannelAdapter;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.Poller;
import org.springframework.integration.channel.BroadcastCapableChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.QueueChannel;
@@ -46,11 +45,14 @@ import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowDefinition;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.PollerSpec;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.QueueChannelSpec;
import org.springframework.integration.endpoint.MethodInvokingMessageSource;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.jms.ActiveMQMultiContextTests;
import org.springframework.integration.jms.JmsDestinationPollingSource;
import org.springframework.integration.jms.SubscribableJmsChannel;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
@@ -302,8 +304,8 @@ public class JmsTests extends ActiveMQMultiContextTests {
}
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
return Pollers.fixedDelay(1000).get();
public PollerSpec poller() {
return Pollers.fixedDelay(1000);
}
@Bean
@@ -338,30 +340,29 @@ public class JmsTests extends ActiveMQMultiContextTests {
}
@Bean
public MessageChannel jmsOutboundInboundReplyChannel() {
return MessageChannels.queue().get();
public QueueChannelSpec jmsOutboundInboundReplyChannel() {
return MessageChannels.queue();
}
@Bean
public IntegrationFlow jmsInboundFlow() {
public IntegrationFlow jmsInboundFlow(QueueChannel jmsOutboundInboundReplyChannel) {
return IntegrationFlow
.from(Jms.inboundAdapter(amqFactory).destination("jmsInbound"))
.<String, String>transform(String::toUpperCase)
.channel(this.jmsOutboundInboundReplyChannel())
.channel(jmsOutboundInboundReplyChannel)
.get();
}
@Bean
public BroadcastCapableChannel jmsPublishSubscribeChannel() {
public JmsPublishSubscribeMessageChannelSpec jmsPublishSubscribeChannel() {
return Jms.publishSubscribeChannel(amqFactory)
.destination("pubsub")
.get();
.destination("pubsub");
}
@Bean
public IntegrationFlow pubSubFlow() {
public IntegrationFlow pubSubFlow(SubscribableJmsChannel jmsPublishSubscribeChannel) {
return f -> f
.publishSubscribeChannel(jmsPublishSubscribeChannel(),
.publishSubscribeChannel(jmsPublishSubscribeChannel,
pubsub -> pubsub
.subscribe(subFlow -> subFlow
.channel(c -> c.queue("jmsPubSubBridgeChannel")))
@@ -408,8 +409,7 @@ public class JmsTests extends ActiveMQMultiContextTests {
.from(Jms.messageDrivenChannelAdapter(
Jms.container(amqFactory, "containerSpecDestination")
.pubSubDomain(false)
.taskExecutor(Executors.newCachedThreadPool())
.get()))
.taskExecutor(Executors.newCachedThreadPool())))
.transform(String::trim)
.channel(jmsOutboundInboundReplyChannel())
.get();

9
spring-integration-jms/src/test/kotlin/org/springframework/integration/jms/dsl/JmsDslKotlinTests.kt
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.
@@ -29,6 +29,7 @@ import org.springframework.beans.factory.annotation.Qualifier
import org.springframework.context.annotation.Bean
import org.springframework.context.annotation.Configuration
import org.springframework.integration.IntegrationMessageHeaderAccessor
import org.springframework.integration.channel.QueueChannel
import org.springframework.integration.config.EnableIntegration
import org.springframework.integration.dsl.MessageChannels
import org.springframework.integration.dsl.integrationFlow
@@ -112,10 +113,10 @@ class JmsDslKotlinTests : ActiveMQMultiContextTests() {
}
@Bean
fun jmsOutboundInboundReplyChannel() = MessageChannels.queue().get()
fun jmsOutboundInboundReplyChannel() = MessageChannels.queue()
@Bean
fun jmsMessageDrivenFlowWithContainer() =
fun jmsMessageDrivenFlowWithContainer(jmsOutboundInboundReplyChannel: QueueChannel) =
integrationFlow(
Jms.messageDrivenChannelAdapter(
Jms.container(amqFactory, "containerSpecDestination")
@@ -125,7 +126,7 @@ class JmsDslKotlinTests : ActiveMQMultiContextTests() {
.headerMapper(jmsHeaderMapper())
) {
transform { it: String -> it.trim { it <= ' ' } }
channel(jmsOutboundInboundReplyChannel())
channel(jmsOutboundInboundReplyChannel)
}
}

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.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.
@@ -102,6 +102,7 @@ public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<
*/
public S onPartitionsAssignedSeekCallback(
BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {
this.target.setOnPartitionsAssignedSeekCallback(onPartitionsAssignedCallback);
return _this();
}
@@ -128,7 +129,7 @@ public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<
KafkaInboundGatewayListenerContainerSpec(KafkaMessageListenerContainerSpec<K, V> containerSpec,
KafkaTemplateSpec<K, R> templateSpec) {
super(containerSpec.get(), templateSpec.getTemplate());
super(containerSpec.getObject(), templateSpec.getTemplate());
this.containerSpec = containerSpec;
this.templateSpec = templateSpec;
}
@@ -164,8 +165,8 @@ public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<
@Override
public Map<Object, String> getComponentsToRegister() {
return new ObjectStringMapBuilder()
.put(this.containerSpec.get(), this.containerSpec.getId())
.put(this.templateSpec.get(), this.templateSpec.getId())
.put(this.containerSpec.getObject(), this.containerSpec.getId())
.put(this.templateSpec.getObject(), this.templateSpec.getId())
.get();
}

9
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.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.
@@ -57,6 +57,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
KafkaMessageDrivenChannelAdapterSpec(AbstractMessageListenerContainer<K, V> messageListenerContainer,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
super(new KafkaMessageDrivenChannelAdapter<>(messageListenerContainer, listenerMode));
this.container = messageListenerContainer;
}
@@ -174,6 +175,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
*/
public S onPartitionsAssignedSeekCallback(
BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {
this.target.setOnPartitionsAssignedSeekCallback(onPartitionsAssignedCallback);
return _this();
}
@@ -196,7 +198,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
KafkaMessageDrivenChannelAdapterListenerContainerSpec(KafkaMessageListenerContainerSpec<K, V> spec,
KafkaMessageDrivenChannelAdapter.ListenerMode listenerMode) {
super(spec.get(), listenerMode);
super(spec.getObject(), listenerMode);
this.spec = spec;
}
@@ -208,6 +210,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
*/
public KafkaMessageDrivenChannelAdapterListenerContainerSpec<K, V> configureListenerContainer(
Consumer<KafkaMessageListenerContainerSpec<K, V>> configurer) {
Assert.notNull(configurer, "The 'configurer' cannot be null");
configurer.accept(this.spec);
return _this();
@@ -215,7 +218,7 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.spec.get(), this.spec.getId());
return Collections.singletonMap(this.spec.getObject(), this.spec.getId());
}
}

4
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaOutboundGatewaySpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2022 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.
@@ -110,7 +110,7 @@ public class KafkaOutboundGatewaySpec<K, V, R, S extends KafkaOutboundGatewaySpe
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.kafkaTemplateSpec.get(), this.kafkaTemplateSpec.getId());
return Collections.singletonMap(this.kafkaTemplateSpec.getTemplate(), this.kafkaTemplateSpec.getId());
}
}

5
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaProducerMessageHandlerSpec.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.
@@ -429,10 +429,9 @@ public class KafkaProducerMessageHandlerSpec<K, V, S extends KafkaProducerMessag
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.kafkaTemplateSpec.get(), this.kafkaTemplateSpec.getId());
return Collections.singletonMap(this.kafkaTemplateSpec.getTemplate(), this.kafkaTemplateSpec.getId());
}
}
}

14
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2022 the original author or authors.
* Copyright 2015-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,12 +38,12 @@ import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.channel.BroadcastCapableChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.kafka.channel.PollableKafkaChannel;
import org.springframework.integration.kafka.channel.PublishSubscribeKafkaChannel;
import org.springframework.integration.kafka.inbound.KafkaErrorSendingMessageRecoverer;
import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
@@ -427,10 +427,11 @@ public class KafkaDslTests {
@Bean
public IntegrationFlow channels(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory,
KafkaMessageSource<?, ?> channelSource) {
KafkaMessageSource<?, ?> channelSource,
PublishSubscribeKafkaChannel publishSubscribeKafkaChannel) {
return IntegrationFlow.from(topic6Channel(template, containerFactory))
.publishSubscribeChannel(pubSub(template, containerFactory), channel -> channel
.publishSubscribeChannel(publishSubscribeKafkaChannel, channel -> channel
.subscribe(f -> f.channel(
Kafka.pollableChannel(template, channelSource).id("topic8Channel")))
.subscribe(f -> f.channel(
@@ -439,11 +440,10 @@ public class KafkaDslTests {
}
@Bean
public BroadcastCapableChannel pubSub(KafkaTemplate<Integer, String> template,
public KafkaPublishSubscribeChannelSpec pubSub(KafkaTemplate<Integer, String> template,
ConcurrentKafkaListenerContainerFactory<Integer, String> containerFactory) {
return Kafka.publishSubscribeChannel(template, containerFactory, TEST_TOPIC7)
.get();
return Kafka.publishSubscribeChannel(template, containerFactory, TEST_TOPIC7);
}
@Bean

2
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -539,7 +539,7 @@ class MessageDrivenAdapterTests {
.messageDrivenChannelAdapter(container, ListenerMode.record)
.recordMessageConverter(new StringJsonMessageConverter())
.payloadType(Foo.class)
.get();
.getObject();
QueueChannel out = new QueueChannel();
adapter.setOutputChannel(out);
adapter.afterPropertiesSet();

12
spring-integration-mongodb/src/test/java/org/springframework/integration/mongodb/dsl/MongoDbTests.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.
@@ -40,9 +40,11 @@ import org.springframework.data.mongodb.core.mapping.MongoMappingContext;
import org.springframework.data.mongodb.core.query.BasicQuery;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.QueueChannelSpec;
import org.springframework.integration.handler.ReplyRequiredException;
import org.springframework.integration.mongodb.MongoDbContainerTest;
import org.springframework.integration.mongodb.outbound.MessageCollectionCallback;
@@ -333,16 +335,16 @@ class MongoDbTests implements MongoDbContainerTest {
}
@Bean
public IntegrationFlow gatewayCollectionCallbackFlow() {
public IntegrationFlow gatewayCollectionCallbackFlow(QueueChannel getResultChannel) {
return f -> f
.handle(collectionCallbackOutboundGateway(
(collection, requestMessage) -> collection.countDocuments()))
.channel(getResultChannel());
.channel(getResultChannel);
}
@Bean
public MessageChannel getResultChannel() {
return MessageChannels.queue().get();
public QueueChannelSpec getResultChannel() {
return MessageChannels.queue();
}
@Bean

8
spring-integration-scripting/src/main/java/org/springframework/integration/scripting/dsl/ScriptMessageSourceSpec.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.
@@ -67,7 +67,7 @@ public class ScriptMessageSourceSpec extends MessageSourceSpec<ScriptMessageSour
* The {@link ScriptVariableGenerator} to use.
* @param variableGenerator the {@link ScriptVariableGenerator}
* @return the current spec
* @see ScriptSpec#variableGenerator
* @see ScriptSpec#variableGenerator(ScriptVariableGenerator)
*/
public ScriptMessageSourceSpec variableGenerator(ScriptVariableGenerator variableGenerator) {
this.delegate.variableGenerator(variableGenerator);
@@ -121,12 +121,12 @@ public class ScriptMessageSourceSpec extends MessageSourceSpec<ScriptMessageSour
@Override
protected MessageSource<?> doGet() {
return new MessageProcessorMessageSource(this.delegate.get());
return new MessageProcessorMessageSource(this.delegate.getObject());
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.delegate.get(), this.delegate.getId());
return Collections.singletonMap(this.delegate.getObject(), this.delegate.getId());
}
}

17
spring-integration-ws/src/test/java/org/springframework/integration/ws/dsl/WsDslTests.java
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.
@@ -59,12 +59,12 @@ public class WsDslTests {
Unmarshaller unmarshaller = mock(Unmarshaller.class);
MarshallingWebServiceInboundGateway gateway = Ws.marshallingInboundGateway(marshaller)
.unmarshaller(unmarshaller)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(gateway, "marshaller")).isSameAs(marshaller);
assertThat(TestUtils.getPropertyValue(gateway, "unmarshaller")).isSameAs(unmarshaller);
marshaller = mock(Both.class);
gateway = Ws.marshallingInboundGateway(marshaller).get();
gateway = Ws.marshallingInboundGateway(marshaller).getObject();
assertThat(TestUtils.getPropertyValue(gateway, "marshaller")).isSameAs(marshaller);
assertThat(TestUtils.getPropertyValue(gateway, "unmarshaller")).isSameAs(marshaller);
}
@@ -73,7 +73,7 @@ public class WsDslTests {
void simpleInbound() {
SimpleWebServiceInboundGateway gateway = Ws.simpleInboundGateway()
.extractPayload(false)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(gateway, "extractPayload", Boolean.class)).isFalse();
}
@@ -104,7 +104,7 @@ public class WsDslTests {
.messageSenders(messageSender)
.requestCallback(requestCallback)
.uriVariableExpressions(uriVariableExpressions)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(gateway, "webServiceTemplate.marshaller")).isSameAs(marshaller);
assertThat(TestUtils.getPropertyValue(gateway, "webServiceTemplate.unmarshaller")).isSameAs(unmarshaller);
assertThat(TestUtils.getPropertyValue(gateway, "webServiceTemplate.messageFactory")).isSameAs(messageFactory);
@@ -147,7 +147,7 @@ public class WsDslTests {
.requestCallback(requestCallback)
.uriVariableExpressions(uriVariableExpressions)
.extractPayload(false)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(gateway, "webServiceTemplate.messageFactory")).isSameAs(messageFactory);
assertThat(TestUtils.getPropertyValue(gateway, "webServiceTemplate.faultMessageResolver"))
.isSameAs(faultMessageResolver);
@@ -178,7 +178,7 @@ public class WsDslTests {
.ignoreEmptyResponses(true)
.requestCallback(requestCallback)
.uriVariableExpressions(uriVariableExpressions)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(gateway, "uri")).isSameAs(uri);
assertThat(TestUtils.getPropertyValue(gateway, "headerMapper")).isSameAs(headerMapper);
assertThat(TestUtils.getPropertyValue(gateway, "requestCallback")).isSameAs(requestCallback);
@@ -209,7 +209,7 @@ public class WsDslTests {
.requestCallback(requestCallback)
.uriVariableExpressions(uriVariableExpressions)
.extractPayload(false)
.get();
.getObject();
assertThat(TestUtils.getPropertyValue(gateway, "headerMapper")).isSameAs(headerMapper);
assertThat(TestUtils.getPropertyValue(gateway, "requestCallback")).isSameAs(requestCallback);
assertThat(TestUtils.getPropertyValue(gateway, "uriVariableExpressions")).isEqualTo(uriVariableExpressions);
@@ -225,4 +225,3 @@ public class WsDslTests {
}
}

40
src/reference/asciidoc/dsl.adoc
View File

@@ -39,8 +39,8 @@ public class MyConfiguration {
}
@Bean
public IntegrationFlow myFlow() {
return IntegrationFlow.fromSupplier(integerSource()::getAndIncrement,
public IntegrationFlow myFlow(AtomicInteger integerSource) {
return IntegrationFlow.fromSupplier(integerSource::getAndIncrement,
c -> c.poller(Pollers.fixedRate(100)))
.channel("inputChannel")
.filter((Integer p) -> p > 0)
@@ -63,6 +63,10 @@ You need not replace all of your existing XML configuration to use Java configur
The `org.springframework.integration.dsl` package contains the `IntegrationFlowBuilder` API mentioned earlier and a number of `IntegrationComponentSpec` implementations, which are also builders and provide the fluent API to configure concrete endpoints.
The `IntegrationFlowBuilder` infrastructure provides common https://www.enterpriseintegrationpatterns.com/[enterprise integration patterns] (EIP) for message-based applications, such as channels, endpoints, pollers, and channel interceptors.
IMPORTANT:: The `IntegrationComponentSpec` is a `FactoryBean` implementation, therefore its `getObject()` method must not be called from bean definitions.
The `IntegrationComponentSpec` implementation must be left as is for bean definitions and the framework will manage its lifecycle.
Bean method parameter injection for the target `IntegrationComponentSpec` type (a `FactoryBean` value) must be used for `IntegrationFlow` bean definitions instead of bean method references.
Endpoints are expressed as verbs in the DSL to improve readability.
The following list includes the common DSL method names and the associated EIP endpoint:
@@ -163,10 +167,9 @@ The following example shows how to use it:
[source,java]
----
@Bean
public MessageChannel priorityChannel() {
public PriorityChannelSpec priorityChannel() {
return MessageChannels.priority(this.mongoDbChannelMessageStore, "priorityGroup")
.interceptor(wireTap())
.get();
.interceptor(wireTap());
}
----
====
@@ -181,13 +184,13 @@ The following example shows the possible ways to use the `channel()` EIP method:
[source,java]
----
@Bean
public MessageChannel queueChannel() {
return MessageChannels.queue().get();
public QueueChannelSpec queueChannel() {
return MessageChannels.queue();
}
@Bean
public MessageChannel publishSubscribe() {
return MessageChannels.publishSubscribe().get();
public PublishSubscribeChannelSpec<?> publishSubscribe() {
return MessageChannels.publishSubscribe();
}
@Bean
@@ -261,7 +264,7 @@ public PollerSpec poller() {
See https://docs.spring.io/spring-integration/api/org/springframework/integration/dsl/Pollers.html[`Pollers`] and https://docs.spring.io/spring-integration/api/org/springframework/integration/dsl/PollerSpec.html[`PollerSpec`] in the Javadoc for more information.
IMPORTANT: If you use the DSL to construct a `PollerSpec` as a `@Bean`, do not call the `get()` method in the bean definition.
IMPORTANT: If you use the DSL to construct a `PollerSpec` as a `@Bean`, do not call the `getObject()` method in the bean definition.
The `PollerSpec` is a `FactoryBean` that generates the `PollerMetadata` object from the specification and initializes all of its properties.
[[java-dsl-reactive]]
@@ -833,30 +836,21 @@ For example, we now can configure several subscribers as sub-flows on the `Jms.p
[source,java]
----
@Bean
public BroadcastCapableChannel jmsPublishSubscribeChannel() {
public JmsPublishSubscribeMessageChannelSpec jmsPublishSubscribeChannel() {
return Jms.publishSubscribeChannel(jmsConnectionFactory())
.destination("pubsub")
.get();
.destination("pubsub");
}
@Bean
public IntegrationFlow pubSubFlow() {
public IntegrationFlow pubSubFlow(BroadcastCapableChannel jmsPublishSubscribeChannel) {
return f -> f
.publishSubscribeChannel(jmsPublishSubscribeChannel(),
.publishSubscribeChannel(jmsPublishSubscribeChannel,
pubsub -> pubsub
.subscribe(subFlow -> subFlow
.channel(c -> c.queue("jmsPubSubBridgeChannel1")))
.subscribe(subFlow -> subFlow
.channel(c -> c.queue("jmsPubSubBridgeChannel2"))));
}
@Bean
public BroadcastCapableChannel jmsPublishSubscribeChannel(ConnectionFactory jmsConnectionFactory) {
return (BroadcastCapableChannel) Jms.publishSubscribeChannel(jmsConnectionFactory)
.destination("pubsub")
.get();
}
----
====

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -41,6 +41,11 @@ See <<./filter.adoc#filter, Filter>> for more information.
- The default timeout for send and receive operations in gateways and replying channel adapters has been changed from infinity to `30` seconds.
Only one left as a `1` second is a `receiveTimeout` for `PollingConsumer` to not block a scheduler thread too long and let other queued tasks to be performed with the `TaskScheduler`.
- The `IntegrationComponentSpec.get()` method has been deprecated with removal planned for the next version.
Since `IntegrationComponentSpec` is a `FactoryBean`, its bean definition must stay as is without any target object resolutions.
The Java DSL and the framework by itself will manage the `IntegrationComponentSpec` lifecycle.
See <<./dsl.adoc#java-dsl, Java DSL>> for more information.
[[x6.1-web-sockets]]
=== Web Sockets Changes