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

GH-3623: Deprecarte an IntegrationFlows

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3623

* `IntegrationFlow` refactoring
* Apply several code style improvements and good practices
* Code style: no empty lines for methods javadocs
* make deprecated implementation reuse actual one instead of the copy-paste approach
* add whats-new comments
* Fix whats-new page according to standards
This commit is contained in:
Artem Vozhdayenko
2022-07-05 22:47:30 +03:00
committed by GitHub
parent 63759d76b9
commit 53dd050c5b
90 changed files with 796 additions and 570 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -426,7 +426,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
MessageChannel messageChannel = flow.getInputChannel();
if (messageChannel == null) {
messageChannel = new DirectChannel();
IntegrationFlowDefinition<?> flowBuilder = IntegrationFlows.from(messageChannel);
IntegrationFlowDefinition<?> flowBuilder = IntegrationFlow.from(messageChannel);
flow.configure(flowBuilder);
addComponent(flowBuilder.get());
}
@@ -457,7 +457,7 @@ public abstract class BaseIntegrationFlowDefinition<B extends BaseIntegrationFlo
*/
public B wireTap(String wireTapChannel, Consumer<WireTapSpec> wireTapConfigurer) {
DirectChannel internalWireTapChannel = new DirectChannel();
addComponent(IntegrationFlows.from(internalWireTapChannel).channel(wireTapChannel).get());
addComponent(IntegrationFlow.from(internalWireTapChannel).channel(wireTapChannel).get());
return wireTap(internalWireTapChannel, wireTapConfigurer);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2020 the original author or authors.
* Copyright 2020-2022 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.
@@ -58,7 +58,7 @@ public class BroadcastPublishSubscribeSpec
Assert.notNull(subFlow, "'subFlow' must not be null");
IntegrationFlowBuilder flowBuilder =
IntegrationFlows.from(this.target)
IntegrationFlow.from(this.target)
.bridge(consumer -> consumer.order(this.order++));
MessageChannel subFlowInput = subFlow.getInputChannel();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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.
@@ -163,7 +163,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
MessageChannel messageChannel = subFlow.getInputChannel();
if (messageChannel == null) {
messageChannel = new DirectChannel();
IntegrationFlowDefinition<?> flowBuilder = IntegrationFlows.from(messageChannel);
IntegrationFlowDefinition<?> flowBuilder = IntegrationFlow.from(messageChannel);
subFlow.configure(flowBuilder);
this.componentsToRegister.put(evaluateInternalBuilder ? flowBuilder.get() : flowBuilder, null);
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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.
@@ -17,8 +17,29 @@
package org.springframework.integration.dsl;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.reactivestreams.Publisher;
import org.springframework.aop.framework.Advised;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
import org.springframework.integration.dsl.support.MessageChannelReference;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.handler.AbstractMessageProducingHandler;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
/**
* The main Integration DSL abstraction.
@@ -29,7 +50,7 @@ import org.springframework.messaging.MessageChannel;
* <pre class="code">
* &#64;Bean
* public IntegrationFlow fileReadingFlow() {
* return IntegrationFlows
* return IntegrationFlow
* .from(Files.inboundAdapter(tmpDir.getRoot()), e -&gt; e.poller(Pollers.fixedDelay(100)))
* .transform(Files.fileToString())
* .channel(MessageChannels.queue("fileReadingResultChannel"))
@@ -67,6 +88,9 @@ import org.springframework.messaging.MessageChannel;
* </pre>
*
* @author Artem Bilan
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Artem Vozhdayenko
*
* @since 5.0
*
@@ -103,4 +127,386 @@ public interface IntegrationFlow {
return null;
}
/**
* Populate the {@link MessageChannel} name to the new {@link IntegrationFlowBuilder} chain.
* The {@link IntegrationFlow} {@code inputChannel}.
* @param messageChannelName the name of existing {@link MessageChannel} bean.
* The new {@link DirectChannel} bean will be created on context startup
* if there is no bean with this name.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
static IntegrationFlowBuilder from(String messageChannelName) {
return from(new MessageChannelReference(messageChannelName));
}
/**
* Populate the {@link MessageChannel} object to the
* {@link IntegrationFlowBuilder} chain using the fluent API from {@link MessageChannelSpec}.
* The {@link IntegrationFlow} {@code inputChannel}.
* @param messageChannelSpec the MessageChannelSpec to populate {@link MessageChannel} instance.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see MessageChannels
*/
static IntegrationFlowBuilder from(MessageChannelSpec<?, ?> messageChannelSpec) {
Assert.notNull(messageChannelSpec, "'messageChannelSpec' must not be null");
return from(messageChannelSpec.get());
}
/**
* Populate the {@link MessageChannel} name to the new {@link IntegrationFlowBuilder} chain.
* Typically for the {@link org.springframework.integration.channel.FixedSubscriberChannel} together
* with {@code fixedSubscriber = true}.
* The {@link IntegrationFlow} {@code inputChannel}.
* @param messageChannelName the name for {@link DirectChannel} or
* {@link org.springframework.integration.channel.FixedSubscriberChannel}
* to be created on context startup, not reference.
* The {@link MessageChannel} depends on the {@code fixedSubscriber} boolean argument.
* @param fixedSubscriber the boolean flag to determine if result {@link MessageChannel} should
* be {@link DirectChannel}, if {@code false} or
* {@link org.springframework.integration.channel.FixedSubscriberChannel}, if {@code true}.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see DirectChannel
* @see org.springframework.integration.channel.FixedSubscriberChannel
*/
static IntegrationFlowBuilder from(String messageChannelName, boolean fixedSubscriber) {
return fixedSubscriber
? from(new FixedSubscriberChannelPrototype(messageChannelName))
: from(messageChannelName);
}
/**
* Populate the provided {@link MessageChannel} object to the {@link IntegrationFlowBuilder} chain.
* The {@link IntegrationFlow} {@code inputChannel}.
* @param messageChannel the {@link MessageChannel} to populate.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
static IntegrationFlowBuilder from(MessageChannel messageChannel) {
return new IntegrationFlowBuilder().channel(messageChannel);
}
/**
* Populate the {@link MessageSource} object to the {@link IntegrationFlowBuilder} chain
* using the fluent API from the provided {@link MessageSourceSpec}.
* The {@link IntegrationFlow} {@code startMessageSource}.
* @param messageSourceSpec the {@link MessageSourceSpec} to use.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see MessageSourceSpec and its implementations.
*/
static IntegrationFlowBuilder from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec) {
return from(messageSourceSpec, null);
}
/**
* Populate the {@link MessageSource} object to the {@link IntegrationFlowBuilder} chain
* using the fluent API from the provided {@link MessageSourceSpec}.
* The {@link IntegrationFlow} {@code startMessageSource}.
* @param messageSourceSpec the {@link MessageSourceSpec} to use.
* @param endpointConfigurer the {@link Consumer} to provide more options for the
* {@link org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean}.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see MessageSourceSpec
* @see SourcePollingChannelAdapterSpec
*/
static IntegrationFlowBuilder from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Assert.notNull(messageSourceSpec, "'messageSourceSpec' must not be null");
return from(messageSourceSpec.get(), endpointConfigurer, registerComponents(messageSourceSpec));
}
/**
* Provides {@link Supplier} as source of messages to the integration flow which will
* be triggered by the application context's default poller (which must be declared).
* @param messageSource the {@link Supplier} to populate.
* @param <T> the supplier type.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see Supplier
*/
static <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource) {
return fromSupplier(messageSource, null);
}
/**
* Provides {@link Supplier} as source of messages to the integration flow.
* which will be triggered by a <b>provided</b>
* {@link org.springframework.integration.endpoint.SourcePollingChannelAdapter}.
* @param messageSource the {@link Supplier} to populate.
* @param endpointConfigurer the {@link Consumer} to provide more options for the
* {@link org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean}.
* @param <T> the supplier type.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see Supplier
*/
static <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Assert.notNull(messageSource, "'messageSource' must not be null");
return from(new AbstractMessageSource<Object>() {
@Override
protected Object doReceive() {
return messageSource.get();
}
@Override
public String getComponentType() {
return "inbound-channel-adapter";
}
}, endpointConfigurer);
}
/**
* Populate the provided {@link MessageSource} object to the {@link IntegrationFlowBuilder} chain.
* The {@link IntegrationFlow} {@code startMessageSource}.
* @param messageSource the {@link MessageSource} to populate.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see MessageSource
*/
static IntegrationFlowBuilder from(MessageSource<?> messageSource) {
return from(messageSource, null);
}
/**
* Populate the provided {@link MessageSource} object to the {@link IntegrationFlowBuilder} chain.
* The {@link IntegrationFlow} {@code startMessageSource}.
* In addition use {@link SourcePollingChannelAdapterSpec} to provide options for the underlying
* {@link org.springframework.integration.endpoint.SourcePollingChannelAdapter} endpoint.
* @param messageSource the {@link MessageSource} to populate.
* @param endpointConfigurer the {@link Consumer} to provide more options for the
* {@link org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean}.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see MessageSource
* @see SourcePollingChannelAdapterSpec
*/
static IntegrationFlowBuilder from(MessageSource<?> messageSource,
@Nullable Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return from(messageSource, endpointConfigurer, null);
}
private static IntegrationFlowBuilder from(MessageSource<?> messageSource,
@Nullable Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer,
@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
SourcePollingChannelAdapterSpec spec = new SourcePollingChannelAdapterSpec(messageSource);
if (endpointConfigurer != null) {
endpointConfigurer.accept(spec);
}
if (integrationFlowBuilder == null) {
integrationFlowBuilder = new IntegrationFlowBuilder();
}
return integrationFlowBuilder.addComponent(spec)
.currentComponent(spec);
}
/**
* Populate the {@link MessageProducerSupport} object to the {@link IntegrationFlowBuilder} chain
* using the fluent API from the {@link MessageProducerSpec}.
* The {@link IntegrationFlow} {@code startMessageProducer}.
* @param messageProducerSpec the {@link MessageProducerSpec} to use.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
* @see MessageProducerSpec
*/
static IntegrationFlowBuilder from(MessageProducerSpec<?, ?> messageProducerSpec) {
return from(messageProducerSpec.get(), registerComponents(messageProducerSpec));
}
/**
* Populate the provided {@link MessageProducerSupport} object to the {@link IntegrationFlowBuilder} chain.
* The {@link IntegrationFlow} {@code startMessageProducer}.
* @param messageProducer the {@link MessageProducerSupport} to populate.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
static IntegrationFlowBuilder from(MessageProducerSupport messageProducer) {
return from(messageProducer, null);
}
private static IntegrationFlowBuilder from(MessageProducerSupport messageProducer,
@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
MessageChannel outputChannel = messageProducer.getOutputChannel();
if (outputChannel == null) {
outputChannel = new DirectChannel();
messageProducer.setOutputChannel(outputChannel);
}
if (integrationFlowBuilder == null) {
integrationFlowBuilder = from(outputChannel);
}
else {
integrationFlowBuilder.channel(outputChannel);
}
return integrationFlowBuilder.addComponent(messageProducer);
}
/**
* Populate the {@link MessagingGatewaySupport} object to the {@link IntegrationFlowBuilder} chain
* using the fluent API from the {@link MessagingGatewaySpec}.
* The {@link IntegrationFlow} {@code startMessagingGateway}.
* @param inboundGatewaySpec the {@link MessagingGatewaySpec} to use.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
static IntegrationFlowBuilder from(MessagingGatewaySpec<?, ?> inboundGatewaySpec) {
return from(inboundGatewaySpec.get(), registerComponents(inboundGatewaySpec));
}
/**
* Populate the provided {@link MessagingGatewaySupport} object to the {@link IntegrationFlowBuilder} chain.
* The {@link IntegrationFlow} {@code startMessageProducer}.
* @param inboundGateway the {@link MessagingGatewaySupport} to populate.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway) {
return from(inboundGateway, null);
}
/**
* Populate the {@link MessageChannel} to the new {@link IntegrationFlowBuilder}
* chain, which becomes as a {@code requestChannel} for the Messaging Gateway(s) built
* on the provided service interface.
* <p>A gateway proxy bean for provided service interface is registered under a name
* from the
* {@link org.springframework.integration.annotation.MessagingGateway#name()} if present
* or from the {@link IntegrationFlow} bean name plus {@code .gateway} suffix.
* @param serviceInterface the service interface class with an optional
* {@link org.springframework.integration.annotation.MessagingGateway} annotation.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
static IntegrationFlowBuilder from(Class<?> serviceInterface) {
return from(serviceInterface, null);
}
/**
* Populate the {@link MessageChannel} to the new {@link IntegrationFlowBuilder}
* chain, which becomes as a {@code requestChannel} for the Messaging Gateway(s) built
* on the provided service interface.
* <p>A gateway proxy bean for provided service interface is based on the options
* configured via provided {@link Consumer}.
* @param serviceInterface the service interface class with an optional
* {@link org.springframework.integration.annotation.MessagingGateway} annotation.
* @param endpointConfigurer the {@link Consumer} to configure proxy bean for gateway.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
static IntegrationFlowBuilder from(Class<?> serviceInterface,
@Nullable Consumer<GatewayProxySpec> endpointConfigurer) {
GatewayProxySpec gatewayProxySpec = new GatewayProxySpec(serviceInterface);
if (endpointConfigurer != null) {
endpointConfigurer.accept(gatewayProxySpec);
}
return from(gatewayProxySpec.getGatewayRequestChannel())
.addComponent(gatewayProxySpec.getGatewayProxyFactoryBean());
}
/**
* Populate a {@link FluxMessageChannel} to the {@link IntegrationFlowBuilder} chain
* and subscribe it to the provided {@link Publisher}.
* @param publisher the {@link Publisher} to subscribe to.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
@SuppressWarnings("overloads")
static IntegrationFlowBuilder from(Publisher<? extends Message<?>> publisher) {
FluxMessageChannel reactiveChannel = new FluxMessageChannel();
reactiveChannel.subscribeTo(publisher);
return from((MessageChannel) reactiveChannel);
}
/**
* Start the flow with a composition from the {@link IntegrationFlow}.
* @param other the {@link IntegrationFlow} from which to compose.
* @return new {@link IntegrationFlowBuilder}.
* @since 6.0
*/
@SuppressWarnings("overloads")
static IntegrationFlowBuilder from(IntegrationFlow other) {
Map<Object, String> integrationComponents = other.getIntegrationComponents();
Assert.notNull(integrationComponents, () ->
"The provided integration flow to compose from '" + other +
"' 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);
}
else if (lastIntegrationComponentFromOther instanceof ConsumerEndpointFactoryBean) {
MessageHandler handler = ((ConsumerEndpointFactoryBean) lastIntegrationComponentFromOther).getHandler();
handler = extractProxyTarget(handler);
if (handler instanceof AbstractMessageProducingHandler) {
return buildFlowFromOutputChannel((AbstractMessageProducingHandler) handler);
}
lastIntegrationComponentFromOther = handler; // for the exception message below
}
throw new BeanCreationException("The 'IntegrationFlow' to start from must end with " +
"a 'MessageChannel' or reply-producing endpoint to let the result from that flow to be " +
"processed in this instance. The provided flow ends with: " + lastIntegrationComponentFromOther);
}
private static IntegrationFlowBuilder buildFlowFromOutputChannel(AbstractMessageProducingHandler handler) {
MessageChannel outputChannel = handler.getOutputChannel();
if (outputChannel == null) {
outputChannel = new PublishSubscribeChannel();
handler.setOutputChannel(outputChannel);
}
return from(outputChannel);
}
private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway,
@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
MessageChannel outputChannel = inboundGateway.getRequestChannel();
if (outputChannel == null) {
outputChannel = new DirectChannel();
inboundGateway.setRequestChannel(outputChannel);
}
if (integrationFlowBuilder == null) {
integrationFlowBuilder = from(outputChannel);
}
else {
integrationFlowBuilder.channel(outputChannel);
}
return integrationFlowBuilder.addComponent(inboundGateway);
}
private static IntegrationFlowBuilder registerComponents(Object spec) {
if (spec instanceof ComponentsRegistration) {
return new IntegrationFlowBuilder()
.addComponents(((ComponentsRegistration) spec).getComponentsToRegister());
}
return null;
}
@SuppressWarnings("unchecked")
private static <T> T extractProxyTarget(T target) {
if (!(target instanceof Advised)) {
return target;
}
Advised advised = (Advised) target;
try {
return (T) extractProxyTarget(advised.getTargetSource().getTarget());
}
catch (Exception e) {
throw new BeanCreationException("Could not extract target", e);
}
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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.
@@ -133,69 +133,69 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, Managea
}
protected IntegrationFlowDefinition<?> from(String messageChannelName) {
return IntegrationFlows.from(messageChannelName);
return IntegrationFlow.from(messageChannelName);
}
protected IntegrationFlowDefinition<?> from(MessageChannel messageChannel) {
return IntegrationFlows.from(messageChannel);
return IntegrationFlow.from(messageChannel);
}
protected IntegrationFlowDefinition<?> from(String messageChannelName, boolean fixedSubscriber) {
return IntegrationFlows.from(messageChannelName, fixedSubscriber);
return IntegrationFlow.from(messageChannelName, fixedSubscriber);
}
protected IntegrationFlowDefinition<?> from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return IntegrationFlows.from(messageSourceSpec, endpointConfigurer);
return IntegrationFlow.from(messageSourceSpec, endpointConfigurer);
}
protected IntegrationFlowDefinition<?> from(MessageSource<?> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return IntegrationFlows.from(messageSource, endpointConfigurer);
return IntegrationFlow.from(messageSource, endpointConfigurer);
}
protected IntegrationFlowDefinition<?> from(MessageProducerSupport messageProducer) {
return IntegrationFlows.from(messageProducer);
return IntegrationFlow.from(messageProducer);
}
protected IntegrationFlowDefinition<?> from(MessageSource<?> messageSource) {
return IntegrationFlows.from(messageSource);
return IntegrationFlow.from(messageSource);
}
protected IntegrationFlowDefinition<?> from(MessagingGatewaySupport inboundGateway) {
return IntegrationFlows.from(inboundGateway);
return IntegrationFlow.from(inboundGateway);
}
protected IntegrationFlowDefinition<?> from(MessageChannelSpec<?, ?> messageChannelSpec) {
return IntegrationFlows.from(messageChannelSpec);
return IntegrationFlow.from(messageChannelSpec);
}
protected IntegrationFlowDefinition<?> from(MessageProducerSpec<?, ?> messageProducerSpec) {
return IntegrationFlows.from(messageProducerSpec);
return IntegrationFlow.from(messageProducerSpec);
}
protected IntegrationFlowDefinition<?> from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec) {
return IntegrationFlows.from(messageSourceSpec);
return IntegrationFlow.from(messageSourceSpec);
}
protected IntegrationFlowDefinition<?> from(MessagingGatewaySpec<?, ?> inboundGatewaySpec) {
return IntegrationFlows.from(inboundGatewaySpec);
return IntegrationFlow.from(inboundGatewaySpec);
}
protected <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource) {
return IntegrationFlows.fromSupplier(messageSource);
return IntegrationFlow.fromSupplier(messageSource);
}
protected <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return IntegrationFlows.fromSupplier(messageSource, endpointConfigurer);
return IntegrationFlow.fromSupplier(messageSource, endpointConfigurer);
}
protected IntegrationFlowBuilder from(Class<?> serviceInterface) {
return IntegrationFlows.from(serviceInterface);
return IntegrationFlow.from(serviceInterface);
}
/**
@@ -208,11 +208,11 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, Managea
protected IntegrationFlowBuilder from(Class<?> serviceInterface,
@Nullable Consumer<GatewayProxySpec> endpointConfigurer) {
return IntegrationFlows.from(serviceInterface, endpointConfigurer);
return IntegrationFlow.from(serviceInterface, endpointConfigurer);
}
protected IntegrationFlowBuilder from(Publisher<? extends Message<?>> publisher) {
return IntegrationFlows.from(publisher);
return IntegrationFlow.from(publisher);
}
protected abstract IntegrationFlowDefinition<?> buildFlow();

182
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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,30 +16,19 @@
package org.springframework.integration.dsl;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Supplier;
import org.reactivestreams.Publisher;
import org.springframework.aop.framework.Advised;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.FluxMessageChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
import org.springframework.integration.dsl.support.MessageChannelReference;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.handler.AbstractMessageProducingHandler;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
/**
* The central factory for fluent {@link IntegrationFlowBuilder} API.
@@ -51,7 +40,10 @@ import org.springframework.util.Assert;
* @since 5.0
*
* @see org.springframework.integration.dsl.context.IntegrationFlowBeanPostProcessor
*
* @deprecated Since version 6.0, this factory is not recommended to use as it shall be removed in the following versions. Use fluent API methods straight from {@link IntegrationFlow} interface.
*/
@Deprecated(since = "6.0", forRemoval = true)
public final class IntegrationFlows {
/**
@@ -63,7 +55,7 @@ public final class IntegrationFlows {
* @return new {@link IntegrationFlowBuilder}.
*/
public static IntegrationFlowBuilder from(String messageChannelName) {
return from(new MessageChannelReference(messageChannelName));
return IntegrationFlow.from(messageChannelName);
}
/**
@@ -83,9 +75,7 @@ public final class IntegrationFlows {
* @see org.springframework.integration.channel.FixedSubscriberChannel
*/
public static IntegrationFlowBuilder from(String messageChannelName, boolean fixedSubscriber) {
return fixedSubscriber
? from(new FixedSubscriberChannelPrototype(messageChannelName))
: from(messageChannelName);
return IntegrationFlow.from(messageChannelName, fixedSubscriber);
}
/**
@@ -97,8 +87,7 @@ public final class IntegrationFlows {
* @see org.springframework.integration.dsl.MessageChannels
*/
public static IntegrationFlowBuilder from(MessageChannelSpec<?, ?> messageChannelSpec) {
Assert.notNull(messageChannelSpec, "'messageChannelSpec' must not be null");
return from(messageChannelSpec.get());
return IntegrationFlow.from(messageChannelSpec);
}
/**
@@ -108,7 +97,7 @@ public final class IntegrationFlows {
* @return new {@link IntegrationFlowBuilder}.
*/
public static IntegrationFlowBuilder from(MessageChannel messageChannel) {
return new IntegrationFlowBuilder().channel(messageChannel);
return IntegrationFlow.from(messageChannel);
}
/**
@@ -120,7 +109,7 @@ public final class IntegrationFlows {
* @see MessageSourceSpec and its implementations.
*/
public static IntegrationFlowBuilder from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec) {
return from(messageSourceSpec, null);
return IntegrationFlow.from(messageSourceSpec);
}
/**
@@ -136,8 +125,7 @@ public final class IntegrationFlows {
*/
public static IntegrationFlowBuilder from(MessageSourceSpec<?, ? extends MessageSource<?>> messageSourceSpec,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Assert.notNull(messageSourceSpec, "'messageSourceSpec' must not be null");
return from(messageSourceSpec.get(), endpointConfigurer, registerComponents(messageSourceSpec));
return IntegrationFlow.from(messageSourceSpec, endpointConfigurer);
}
/**
@@ -149,7 +137,7 @@ public final class IntegrationFlows {
* @see Supplier
*/
public static <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource) {
return fromSupplier(messageSource, null);
return IntegrationFlow.fromSupplier(messageSource);
}
/**
@@ -165,21 +153,7 @@ public final class IntegrationFlows {
*/
public static <T> IntegrationFlowBuilder fromSupplier(Supplier<T> messageSource,
Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
Assert.notNull(messageSource, "'messageSource' must not be null");
return from(new AbstractMessageSource<Object>() {
@Override
protected Object doReceive() {
return messageSource.get();
}
@Override
public String getComponentType() {
return "inbound-channel-adapter";
}
}, endpointConfigurer);
return IntegrationFlow.fromSupplier(messageSource, endpointConfigurer);
}
/**
@@ -190,7 +164,7 @@ public final class IntegrationFlows {
* @see MessageSource
*/
public static IntegrationFlowBuilder from(MessageSource<?> messageSource) {
return from(messageSource, null);
return IntegrationFlow.from(messageSource);
}
/**
@@ -207,24 +181,7 @@ public final class IntegrationFlows {
*/
public static IntegrationFlowBuilder from(MessageSource<?> messageSource,
@Nullable Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer) {
return from(messageSource, endpointConfigurer, null);
}
private static IntegrationFlowBuilder from(MessageSource<?> messageSource,
@Nullable Consumer<SourcePollingChannelAdapterSpec> endpointConfigurer,
@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
SourcePollingChannelAdapterSpec spec = new SourcePollingChannelAdapterSpec(messageSource);
if (endpointConfigurer != null) {
endpointConfigurer.accept(spec);
}
if (integrationFlowBuilder == null) {
integrationFlowBuilder = new IntegrationFlowBuilder();
}
return integrationFlowBuilder.addComponent(spec)
.currentComponent(spec);
return IntegrationFlow.from(messageSource, endpointConfigurer);
}
/**
@@ -236,7 +193,7 @@ public final class IntegrationFlows {
* @see MessageProducerSpec
*/
public static IntegrationFlowBuilder from(MessageProducerSpec<?, ?> messageProducerSpec) {
return from(messageProducerSpec.get(), registerComponents(messageProducerSpec));
return IntegrationFlow.from(messageProducerSpec);
}
/**
@@ -246,25 +203,7 @@ public final class IntegrationFlows {
* @return new {@link IntegrationFlowBuilder}.
*/
public static IntegrationFlowBuilder from(MessageProducerSupport messageProducer) {
return from(messageProducer, null);
}
private static IntegrationFlowBuilder from(MessageProducerSupport messageProducer,
@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
MessageChannel outputChannel = messageProducer.getOutputChannel();
if (outputChannel == null) {
outputChannel = new DirectChannel();
messageProducer.setOutputChannel(outputChannel);
}
if (integrationFlowBuilder == null) {
integrationFlowBuilder = from(outputChannel);
}
else {
integrationFlowBuilder.channel(outputChannel);
}
return integrationFlowBuilder.addComponent(messageProducer);
return IntegrationFlow.from(messageProducer);
}
/**
@@ -275,7 +214,7 @@ public final class IntegrationFlows {
* @return new {@link IntegrationFlowBuilder}.
*/
public static IntegrationFlowBuilder from(MessagingGatewaySpec<?, ?> inboundGatewaySpec) {
return from(inboundGatewaySpec.get(), registerComponents(inboundGatewaySpec));
return IntegrationFlow.from(inboundGatewaySpec);
}
/**
@@ -285,7 +224,7 @@ public final class IntegrationFlows {
* @return new {@link IntegrationFlowBuilder}.
*/
public static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway) {
return from(inboundGateway, null);
return IntegrationFlow.from(inboundGateway);
}
/**
@@ -301,7 +240,7 @@ public final class IntegrationFlows {
* @return new {@link IntegrationFlowBuilder}.
*/
public static IntegrationFlowBuilder from(Class<?> serviceInterface) {
return from(serviceInterface, null);
return IntegrationFlow.from(serviceInterface);
}
/**
@@ -318,14 +257,7 @@ public final class IntegrationFlows {
*/
public static IntegrationFlowBuilder from(Class<?> serviceInterface,
@Nullable Consumer<GatewayProxySpec> endpointConfigurer) {
GatewayProxySpec gatewayProxySpec = new GatewayProxySpec(serviceInterface);
if (endpointConfigurer != null) {
endpointConfigurer.accept(gatewayProxySpec);
}
return from(gatewayProxySpec.getGatewayRequestChannel())
.addComponent(gatewayProxySpec.getGatewayProxyFactoryBean());
return IntegrationFlow.from(serviceInterface, endpointConfigurer);
}
@@ -337,9 +269,7 @@ public final class IntegrationFlows {
*/
@SuppressWarnings("overloads")
public static IntegrationFlowBuilder from(Publisher<? extends Message<?>> publisher) {
FluxMessageChannel reactiveChannel = new FluxMessageChannel();
reactiveChannel.subscribeTo(publisher);
return from((MessageChannel) reactiveChannel);
return IntegrationFlow.from(publisher);
}
/**
@@ -350,75 +280,7 @@ public final class IntegrationFlows {
*/
@SuppressWarnings("overloads")
public static IntegrationFlowBuilder from(IntegrationFlow other) {
Map<Object, String> integrationComponents = other.getIntegrationComponents();
Assert.notNull(integrationComponents, () ->
"The provided integration flow to compose from '" + other +
"' 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);
}
else if (lastIntegrationComponentFromOther instanceof ConsumerEndpointFactoryBean) {
MessageHandler handler = ((ConsumerEndpointFactoryBean) lastIntegrationComponentFromOther).getHandler();
handler = extractProxyTarget(handler);
if (handler instanceof AbstractMessageProducingHandler) {
return buildFlowFromOutputChannel((AbstractMessageProducingHandler) handler);
}
lastIntegrationComponentFromOther = handler; // for the exception message below
}
throw new BeanCreationException("The 'IntegrationFlow' to start from must end with " +
"a 'MessageChannel' or reply-producing endpoint to let the result from that flow to be " +
"processed in this instance. The provided flow ends with: " + lastIntegrationComponentFromOther);
}
private static IntegrationFlowBuilder buildFlowFromOutputChannel(AbstractMessageProducingHandler handler) {
MessageChannel outputChannel = handler.getOutputChannel();
if (outputChannel == null) {
outputChannel = new PublishSubscribeChannel();
handler.setOutputChannel(outputChannel);
}
return from(outputChannel);
}
private static IntegrationFlowBuilder from(MessagingGatewaySupport inboundGateway,
@Nullable IntegrationFlowBuilder integrationFlowBuilderArg) {
IntegrationFlowBuilder integrationFlowBuilder = integrationFlowBuilderArg;
MessageChannel outputChannel = inboundGateway.getRequestChannel();
if (outputChannel == null) {
outputChannel = new DirectChannel();
inboundGateway.setRequestChannel(outputChannel);
}
if (integrationFlowBuilder == null) {
integrationFlowBuilder = from(outputChannel);
}
else {
integrationFlowBuilder.channel(outputChannel);
}
return integrationFlowBuilder.addComponent(inboundGateway);
}
private static IntegrationFlowBuilder registerComponents(Object spec) {
if (spec instanceof ComponentsRegistration) {
return new IntegrationFlowBuilder()
.addComponents(((ComponentsRegistration) spec).getComponentsToRegister());
}
return null;
}
@SuppressWarnings("unchecked")
private static <T> T extractProxyTarget(T target) {
if (!(target instanceof Advised)) {
return target;
}
Advised advised = (Advised) target;
try {
return (T) extractProxyTarget(advised.getTargetSource().getTarget());
}
catch (Exception e) {
throw new BeanCreationException("Could not extract target", e);
}
return IntegrationFlow.from(other);
}
private IntegrationFlows() {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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.
@@ -62,7 +62,7 @@ import org.springframework.messaging.MessageChannel;
*
* @since 5.0
*
* @see IntegrationFlows
* @see IntegrationFlow
* @see org.springframework.integration.dsl.context.IntegrationFlowBeanPostProcessor
* @see org.springframework.integration.dsl.context.IntegrationFlowContext
* @see SmartLifecycle

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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.
@@ -65,7 +65,6 @@ import org.springframework.integration.dsl.IntegrationComponentSpec;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowAdapter;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec;
import org.springframework.integration.dsl.StandardIntegrationFlow;
import org.springframework.integration.dsl.support.MessageChannelReference;
@@ -309,7 +308,7 @@ public class IntegrationFlowBeanPostProcessor
* see its javadocs for more information.
*/
private Object processIntegrationFlowImpl(IntegrationFlow flow, String beanName) {
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(beanName + ".input");
IntegrationFlowBuilder flowBuilder = IntegrationFlow.from(beanName + ".input");
flow.configure(flowBuilder);

72
spring-integration-core/src/main/kotlin/org/springframework/integration/dsl/IntegrationFlowDsl.kt
View File

@@ -38,8 +38,8 @@ fun integrationFlow(flow: KotlinIntegrationFlowDefinition.() -> Unit) =
}
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(Class<?>, Consumer<GatewayProxySpec>)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(Class<?>, Consumer<GatewayProxySpec>)` factory method.
*
* @author Artem Bilan
*/
@@ -47,121 +47,121 @@ inline fun <reified T> integrationFlow(
crossinline gateway: GatewayProxySpec.() -> Unit = {},
flow: KotlinIntegrationFlowDefinition.() -> Unit): IntegrationFlow {
val flowBuilder = IntegrationFlows.from(T::class.java) { gateway(it) }
val flowBuilder = IntegrationFlow.from(T::class.java) { gateway(it) }
KotlinIntegrationFlowDefinition(flowBuilder).flow()
return flowBuilder.get()
}
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(String, Boolean)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(String, Boolean)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(channelName: String, fixedSubscriber: Boolean = false,
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(channelName, fixedSubscriber), flow)
buildIntegrationFlow(IntegrationFlow.from(channelName, fixedSubscriber), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(MessageChannel)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(MessageChannel)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(channel: MessageChannel, flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(channel), flow)
buildIntegrationFlow(IntegrationFlow.from(channel), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(MessageSource<*>, Consumer<SourcePollingChannelAdapterSpec>)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(MessageSource<*>, Consumer<SourcePollingChannelAdapterSpec>)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(messageSource: MessageSource<*>,
options: SourcePollingChannelAdapterSpec.() -> Unit = {},
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(messageSource) { options(it) }, flow)
buildIntegrationFlow(IntegrationFlow.from(messageSource) { options(it) }, flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(MessageSourceSpec<*>, Consumer<SourcePollingChannelAdapterSpec>)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(MessageSourceSpec<*>, Consumer<SourcePollingChannelAdapterSpec>)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(messageSource: MessageSourceSpec<*, out MessageSource<*>>,
options: SourcePollingChannelAdapterSpec.() -> Unit = {},
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(messageSource, options), flow)
buildIntegrationFlow(IntegrationFlow.from(messageSource, options), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(Supplier<*>, Consumer<SourcePollingChannelAdapterSpec>)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(Supplier<*>, Consumer<SourcePollingChannelAdapterSpec>)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(source: () -> Any,
options: SourcePollingChannelAdapterSpec.() -> Unit = {},
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.fromSupplier(source, options), flow)
buildIntegrationFlow(IntegrationFlow.fromSupplier(source, options), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(Publisher<out Message<*>>)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(Publisher<out Message<*>>)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(publisher: Publisher<out Message<*>>,
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(publisher), flow)
buildIntegrationFlow(IntegrationFlow.from(publisher), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(MessagingGatewaySupport)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(MessagingGatewaySupport)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(gateway: MessagingGatewaySupport,
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(gateway), flow)
buildIntegrationFlow(IntegrationFlow.from(gateway), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(MessagingGatewaySpec<*, *>)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(MessagingGatewaySpec<*, *>)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(gatewaySpec: MessagingGatewaySpec<*, *>,
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(gatewaySpec), flow)
buildIntegrationFlow(IntegrationFlow.from(gatewaySpec), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(MessageProducerSupport)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(MessageProducerSupport)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(producer: MessageProducerSupport,
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(producer), flow)
buildIntegrationFlow(IntegrationFlow.from(producer), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(MessageProducerSpec<*, *>)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(MessageProducerSpec<*, *>)` factory method.
*
* @author Artem Bilan
*/
fun integrationFlow(producerSpec: MessageProducerSpec<*, *>,
flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(producerSpec), flow)
buildIntegrationFlow(IntegrationFlow.from(producerSpec), flow)
/**
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlows.from] -
* `IntegrationFlows.from(IntegrationFlow)` factory method.
* Functional [IntegrationFlow] definition in Kotlin DSL for [IntegrationFlow.from] -
* `IntegrationFlow.from(IntegrationFlow)` factory method.
*
* @author Artem Bilan
*
* @since 5.5.8
*/
fun integrationFlow(sourceFlow: IntegrationFlow, flow: KotlinIntegrationFlowDefinition.() -> Unit) =
buildIntegrationFlow(IntegrationFlows.from(sourceFlow), flow)
buildIntegrationFlow(IntegrationFlow.from(sourceFlow), flow)

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

@@ -1,5 +1,5 @@
/*
* Copyright 2021 the original author or authors.
* Copyright 2021-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -31,7 +31,6 @@ 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.IntegrationFlows;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.context.IntegrationFlowContext;
import org.springframework.integration.scheduling.PollerMetadata;
@@ -124,14 +123,14 @@ public class IntegrationFlowCompositionTests {
IntegrationFlow startFlow = f -> f.handle(m -> { });
assertThatIllegalArgumentException()
.isThrownBy(() -> IntegrationFlows.from(startFlow))
.isThrownBy(() -> IntegrationFlow.from(startFlow))
.withMessageContaining("must be declared as a bean in the application context");
IntegrationFlowContext.IntegrationFlowRegistration startRegistration =
this.integrationFlowContext.registration(startFlow).register();
assertThatExceptionOfType(BeanCreationException.class)
.isThrownBy(() -> IntegrationFlows.from(startRegistration.getIntegrationFlow()))
.isThrownBy(() -> IntegrationFlow.from(startRegistration.getIntegrationFlow()))
.withMessageContaining("The 'IntegrationFlow' to start from must end with " +
"a 'MessageChannel' or reply-producing endpoint");
@@ -175,14 +174,14 @@ public class IntegrationFlowCompositionTests {
@Bean
IntegrationFlow templateSourceFlow() {
return IntegrationFlows.fromSupplier(() -> "test data")
return IntegrationFlow.fromSupplier(() -> "test data")
.channel("sourceChannel")
.get();
}
@Bean
IntegrationFlow compositionMainFlow(IntegrationFlow templateSourceFlow) {
return IntegrationFlows.from(templateSourceFlow)
return IntegrationFlow.from(templateSourceFlow)
.<String, String>transform(String::toUpperCase)
.channel(c -> c.queue("compositionMainFlowResult"))
.get();
@@ -196,7 +195,7 @@ public class IntegrationFlowCompositionTests {
@Bean
IntegrationFlow middleFlow(IntegrationFlow firstFlow, IntegrationFlow lastFlow) {
return IntegrationFlows.from(firstFlow)
return IntegrationFlow.from(firstFlow)
.<String, String>transform(p -> p + ", and middle flow")
.to(lastFlow);
}

7
spring-integration-core/src/test/java/org/springframework/integration/dsl/correlation/CorrelationHandlerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2019 the original author or authors.
* Copyright 2016-2022 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.
@@ -39,7 +39,6 @@ import org.springframework.integration.aggregator.HeaderAttributeCorrelationStra
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannelSpec;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.Transformers;
@@ -249,7 +248,7 @@ public class CorrelationHandlerTests {
@Bean
public IntegrationFlow splitAggregateFlow() {
return IntegrationFlows.from("splitAggregateInput", true)
return IntegrationFlow.from("splitAggregateInput", true)
.transform(Transformers.toJson(ObjectToJsonTransformer.ResultType.NODE))
.split((splitter) -> splitter
.discardFlow((subFlow) -> subFlow.channel((c) -> c.queue("discardChannel"))))
@@ -312,7 +311,7 @@ public class CorrelationHandlerTests {
@Bean
public IntegrationFlow releaseBarrierFlow(MessageTriggerAction barrierTriggerAction) {
return IntegrationFlows.from(releaseChannel())
return IntegrationFlow.from(releaseChannel())
.trigger(barrierTriggerAction,
e -> e.poller(p -> p.fixedDelay(100)))
.get();

39
spring-integration-core/src/test/java/org/springframework/integration/dsl/flows/IntegrationFlowTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2022 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.
@@ -60,7 +60,6 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.Pollers;
import org.springframework.integration.dsl.Transformers;
@@ -570,7 +569,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow supplierFlow() {
return IntegrationFlows.fromSupplier(stringSupplier(), c -> c.id("stringSupplierEndpoint"))
return IntegrationFlow.fromSupplier(stringSupplier(), c -> c.id("stringSupplierEndpoint"))
.transform(toUpperCaseFunction())
.channel("suppliedChannel")
.get();
@@ -602,7 +601,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow supplierFlow2() {
return IntegrationFlows.fromSupplier(() -> "foo",
return IntegrationFlow.fromSupplier(() -> "foo",
c -> c.poller(Pollers.fixedDelay(100).maxMessagesPerPoll(1)))
.<String, String>transform(String::toUpperCase)
.channel("suppliedChannel2")
@@ -622,7 +621,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow controlBusFlow() {
return IntegrationFlows.from(ControlBusGateway.class, (gateway) -> gateway.beanName("controlBusGateway"))
return IntegrationFlow.from(ControlBusGateway.class, (gateway) -> gateway.beanName("controlBusGateway"))
.controlBus((endpoint) -> endpoint.id("controlBus"))
.get();
}
@@ -661,7 +660,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow flow2() {
return IntegrationFlows.from(this.inputChannel)
return IntegrationFlow.from(this.inputChannel)
.filter(p -> p instanceof String, e -> e
.id("filter")
.discardFlow(df -> df
@@ -715,7 +714,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow wireTapFlow1() {
return IntegrationFlows.from("tappedChannel1")
return IntegrationFlow.from("tappedChannel1")
.wireTap("tapChannel", wt -> wt.selector(m -> m.getPayload().equals("foo")))
.handle(new ReactiveMessageHandlerAdapter((message) -> Mono.just(message).log().then()))
.get();
@@ -738,7 +737,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow wireTapFlow4() {
return IntegrationFlows.from("tappedChannel4")
return IntegrationFlow.from("tappedChannel4")
.wireTap(tapChannel())
.channel("nullChannel")
.get();
@@ -792,14 +791,14 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow bridgeFlow() {
return IntegrationFlows.from(MessageChannels.queue("bridgeFlowInput"))
return IntegrationFlow.from(MessageChannels.queue("bridgeFlowInput"))
.channel(MessageChannels.queue("bridgeFlowOutput"))
.get();
}
@Bean
public IntegrationFlow bridgeFlow2() {
return IntegrationFlows.from("bridgeFlow2Input")
return IntegrationFlow.from("bridgeFlow2Input")
.bridge(c -> c.autoStartup(false).id("bridge"))
.fixedSubscriberChannel()
.delay("delayer", d -> d
@@ -817,7 +816,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow claimCheckFlow() {
return IntegrationFlows.from("claimCheckInput")
return IntegrationFlow.from("claimCheckInput")
.claimCheckIn(this.messageStore())
.claimCheckOut(this.messageStore())
.get();
@@ -851,14 +850,14 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow methodInvokingFlow() {
return IntegrationFlows.from("methodInvokingInput")
return IntegrationFlow.from("methodInvokingInput")
.handle(this.greetingService)
.get();
}
@Bean
public IntegrationFlow lambdasFlow() {
return IntegrationFlows.from("lambdasInput")
return IntegrationFlow.from("lambdasInput")
.filter(String.class, "World"::equals)
.transform(String.class, "Hello "::concat)
.get();
@@ -866,7 +865,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow errorRecovererFlow() {
return IntegrationFlows.from(Function.class, (gateway) -> gateway.beanName("errorRecovererFunction"))
return IntegrationFlow.from(Function.class, (gateway) -> gateway.beanName("errorRecovererFunction"))
.<Object>handle((p, h) -> {
throw new RuntimeException("intentional");
},
@@ -888,7 +887,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow recoveryFlow() {
return IntegrationFlows.from(recoveryChannel())
return IntegrationFlow.from(recoveryChannel())
.<MessagingException, Message<?>>transform(MessagingException::getFailedMessage)
.get();
@@ -896,7 +895,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow dedicatedPollingThreadFlow() {
return IntegrationFlows.from(MessageChannels.queue("dedicatedQueueChannel"))
return IntegrationFlow.from(MessageChannels.queue("dedicatedQueueChannel"))
.bridge(e -> e
.poller(Pollers.fixedDelay(0).receiveTimeout(-1))
.taskScheduler(dedicatedTaskScheduler()))
@@ -912,7 +911,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow flowWithNullChannel() {
return IntegrationFlows.from("flowWithNullChannelInput")
return IntegrationFlow.from("flowWithNullChannelInput")
.nullChannel();
}
@@ -948,7 +947,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow globalErrorChannelResolutionFlow(@Qualifier("taskScheduler") TaskExecutor taskExecutor) {
return IntegrationFlows.from(Consumer.class,
return IntegrationFlow.from(Consumer.class,
(gateway) -> gateway.beanName("globalErrorChannelResolutionFunction"))
.channel(c -> c.executor(taskExecutor))
.handle((p, h) -> {
@@ -969,7 +968,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow interceptorFlow(List<String> outputStringList) {
return IntegrationFlows.from("interceptorChannelIn")
return IntegrationFlow.from("interceptorChannelIn")
.intercept(new ChannelInterceptor() {
@Override
@@ -1029,7 +1028,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow wrongLastComponent() {
return IntegrationFlows.from(MessageChannels.direct())
return IntegrationFlow.from(MessageChannels.direct())
.fixedSubscriberChannel()
.get();
}

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

@@ -48,7 +48,6 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowAdapter;
import org.springframework.integration.dsl.IntegrationFlowDefinition;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.handler.LoggingHandler;
import org.springframework.integration.support.MessageBuilder;
@@ -132,7 +131,7 @@ public class FlowServiceTests {
@Bean
public IntegrationFlow subFlow() {
return IntegrationFlows
return IntegrationFlow
.from("processChannel")
.<String, String>transform(String::toUpperCase)
.channel("replyChannel")

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

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2022 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.
@@ -39,7 +39,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.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.gateway.GatewayProxyFactoryBean;
import org.springframework.integration.gateway.MessagingGatewaySupport;
@@ -174,7 +173,7 @@ public class GatewayDslTests {
@Bean
public IntegrationFlow gatewayFlow() {
return IntegrationFlows.from("gatewayInput")
return IntegrationFlow.from("gatewayInput")
.gateway("gatewayRequest", g -> g.errorChannel("gatewayError").replyTimeout(10L))
.gateway((f) -> f.transform("From Gateway SubFlow: "::concat))
.get();
@@ -182,7 +181,7 @@ public class GatewayDslTests {
@Bean
public IntegrationFlow gatewayRequestFlow() {
return IntegrationFlows.from("gatewayRequest")
return IntegrationFlow.from("gatewayRequest")
.filter("foo"::equals, (f) -> f.throwExceptionOnRejection(true))
.<String, String>transform(String::toUpperCase)
.get();
@@ -207,7 +206,7 @@ public class GatewayDslTests {
@Bean
public IntegrationFlow functionGateway() {
return IntegrationFlows.from(MessageFunction.class,
return IntegrationFlow.from(MessageFunction.class,
(gateway) -> gateway
.header("gatewayMethod", MethodArgsHolder::getMethod)
.header("gatewayArgs", MethodArgsHolder::getArgs)
@@ -219,7 +218,7 @@ public class GatewayDslTests {
@Bean
public IntegrationFlow routingGatewayFlow() {
return IntegrationFlows.from(RoutingGateway.class,
return IntegrationFlow.from(RoutingGateway.class,
(gateway) -> gateway.beanName("routingGateway").header("gatewayMethod", MethodArgsHolder::getMethod))
.route(Message.class, (message) ->
message.getHeaders().get("gatewayMethod", Method.class).getName(),

11
spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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.
@@ -62,7 +62,6 @@ import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowAdapter;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlowDefinition;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.MessageProducerSpec;
import org.springframework.integration.dsl.StandardIntegrationFlow;
@@ -131,7 +130,7 @@ public class ManualFlowTests {
};
QueueChannel channel = new QueueChannel();
channel.setBeanName("channel");
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(producer);
IntegrationFlowBuilder flowBuilder = IntegrationFlow.from(producer);
BridgeHandler bridgeHandler = new BridgeHandler();
bridgeHandler.setBeanName("bridge");
@@ -175,7 +174,7 @@ public class ManualFlowTests {
}
MyProducerSpec spec = new MyProducerSpec(new MyProducer());
QueueChannel channel = new QueueChannel();
IntegrationFlow flow = IntegrationFlows.from(spec.id("fooChannel"))
IntegrationFlow flow = IntegrationFlow.from(spec.id("fooChannel"))
.channel(channel)
.get();
IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(flow).register();
@@ -396,7 +395,7 @@ public class ManualFlowTests {
QueueChannel resultChannel = new QueueChannel();
IntegrationFlow integrationFlow = IntegrationFlows
IntegrationFlow integrationFlow = IntegrationFlow
.from(messageFlux)
.<Integer, Boolean>route(p -> p % 2 == 0, m -> m
.subFlowMapping(true, sf -> sf.<Integer, String>transform(em -> "even:" + em))
@@ -426,7 +425,7 @@ public class ManualFlowTests {
String testId = "testId";
StandardIntegrationFlow testFlow =
IntegrationFlows.from(Supplier.class)
IntegrationFlow.from(Supplier.class)
.get();
IntegrationFlowRegistration flowRegistration =

13
spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2022 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,6 @@ import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.context.IntegrationFlowContext;
import org.springframework.integration.endpoint.AbstractEndpoint;
@@ -173,7 +172,7 @@ public class ReactiveStreamsTests {
QueueChannel resultChannel = new QueueChannel();
IntegrationFlow integrationFlow =
IntegrationFlows.from(messageFlux)
IntegrationFlow.from(messageFlux)
.<Integer, Integer>transform(p -> p * 2)
.channel(resultChannel)
.get();
@@ -250,7 +249,7 @@ public class ReactiveStreamsTests {
@Bean
public Publisher<Message<String>> reactiveFlow() {
return IntegrationFlows
return IntegrationFlow
.from(() -> new GenericMessage<>("a,b,c,d,e,f"),
e -> e.poller(p -> p.trigger(ctx -> this.invoked.getAndSet(true) ? null : new Date()))
.id("reactiveStreamsMessageSource"))
@@ -261,7 +260,7 @@ public class ReactiveStreamsTests {
@Bean
public Publisher<Message<Integer>> pollableReactiveFlow() {
return IntegrationFlows
return IntegrationFlow
.from("inputChannel")
.split(s -> s.delimiters(","))
.<String, Integer>transform(Integer::parseInt,
@@ -273,7 +272,7 @@ public class ReactiveStreamsTests {
@Bean
public Publisher<Message<String>> singleChannelFlow() {
return IntegrationFlows
return IntegrationFlow
.from(MessageChannels.direct("singleChannel"))
.log()
.toReactivePublisher();
@@ -281,7 +280,7 @@ public class ReactiveStreamsTests {
@Bean
public Publisher<Message<String>> fixedSubscriberChannelFlow() {
return IntegrationFlows
return IntegrationFlow
.from("fixedSubscriberChannel", true)
.log()
.toReactivePublisher();

13
spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java
View File

@@ -44,7 +44,6 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.EnableMessageHistory;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowDefinition;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.expression.FunctionExpression;
import org.springframework.integration.store.MessageGroup;
@@ -639,7 +638,7 @@ public class RouterTests {
@Bean
public IntegrationFlow routeFlow() {
return IntegrationFlows.from("routerInput")
return IntegrationFlow.from("routerInput")
.<Integer, Boolean>route(p -> p % 2 == 0,
m -> m.channelMapping(true, evenChannel())
.subFlowMapping(false, f ->
@@ -702,7 +701,7 @@ public class RouterTests {
@Bean
public IntegrationFlow recipientListFlow() {
return IntegrationFlows.from("recipientListInput")
return IntegrationFlow.from("recipientListInput")
.<String, String>transform(p -> p.replaceFirst("Payload", ""))
.routeToRecipients(r -> r
.recipient("foo-channel", "'foo' == payload")
@@ -729,14 +728,14 @@ public class RouterTests {
@Bean
public IntegrationFlow routeMethodInvocationFlow() {
return IntegrationFlows.from("routerMethodInput")
return IntegrationFlow.from("routerMethodInput")
.route("routingTestBean", "routeMessage")
.get();
}
@Bean
public IntegrationFlow routeMethodInvocationFlow2() {
return IntegrationFlows.from("routerMethod2Input")
return IntegrationFlow.from("routerMethod2Input")
.route(new RoutingTestBean())
.get();
}
@@ -748,7 +747,7 @@ public class RouterTests {
@Bean
public IntegrationFlow routeMultiMethodInvocationFlow() {
return IntegrationFlows.from("routerMultiInput")
return IntegrationFlow.from("routerMultiInput")
.route(String.class, p -> p.equals("foo") || p.equals("bar")
? new String[]{ "foo", "bar" }
: null,
@@ -918,7 +917,7 @@ public class RouterTests {
@Bean
public IntegrationFlow propagateErrorFromGatherer(TaskExecutor taskExecutor) {
return IntegrationFlows.from(Function.class)
return IntegrationFlow.from(Function.class)
.scatterGather(s -> s
.recipientFlow(subFlow -> subFlow
.channel(c -> c.executor(taskExecutor))

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

@@ -38,7 +38,6 @@ 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.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.integration.dsl.Transformers;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
@@ -288,7 +287,7 @@ public class TransformerTests {
@Bean
public IntegrationFlow enricherFlow() {
return IntegrationFlows.from("enricherInput", true)
return IntegrationFlow.from("enricherInput", true)
.enrich(e -> e.requestChannel("enrichChannel")
.errorChannel(enricherErrorChannel())
.requestPayloadExpression("payload")
@@ -303,7 +302,7 @@ public class TransformerTests {
@Bean
public IntegrationFlow enricherFlow2() {
return IntegrationFlows.from("enricherInput2", true)
return IntegrationFlow.from("enricherInput2", true)
.enrich(e -> e.requestChannel("enrichChannel")
.requestPayloadExpression("payload")
.shouldClonePayload(false)
@@ -315,7 +314,7 @@ public class TransformerTests {
@Bean
public IntegrationFlow enricherFlow3() {
return IntegrationFlows.from("enricherInput3", true)
return IntegrationFlow.from("enricherInput3", true)
.enrich(e -> e.requestChannel("enrichChannel")
.requestPayload(Message::getPayload)
.shouldClonePayload(false)
@@ -325,7 +324,7 @@ public class TransformerTests {
@Bean
public IntegrationFlow enrichFlow() {
return IntegrationFlows.from("enrichChannel")
return IntegrationFlow.from("enrichChannel")
.<TestPojo, Map<?, ?>>transform(p -> {
if ("junk".equals(p.getName())) {
throw new RuntimeException("intentional");

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

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2019 the original author or authors.
* Copyright 2017-2022 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,7 +35,6 @@ import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageChannels;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -108,7 +107,7 @@ public class ContentTypeConversionTests {
@Bean
public IntegrationFlow serviceFlow() {
return IntegrationFlows.from(ServiceGateway.class)
return IntegrationFlow.from(ServiceGateway.class)
.channel("serviceChannel")
.handle(this)
.get();

5
spring-integration-core/src/test/java/org/springframework/integration/support/management/micrometer/MicrometerMetricsTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2018-2020 the original author or authors.
* Copyright 2018-2022 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -42,7 +42,6 @@ import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.EnableIntegrationManagement;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.gateway.GatewayProxyFactoryBean;
import org.springframework.integration.gateway.MessagingGatewaySupport;
@@ -355,7 +354,7 @@ public class MicrometerMetricsTests {
@Bean
IntegrationFlow gatesFlow() {
return IntegrationFlows.from(Gate.class)
return IntegrationFlow.from(Gate.class)
.nullChannel();
}

15
spring-integration-core/src/test/java/org/springframework/integration/transformer/AvroTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2019-2020 the original author or authors.
* Copyright 2019-2022 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.
@@ -33,7 +33,6 @@ 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.IntegrationFlows;
import org.springframework.integration.test.condition.LogLevels;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.transformer.support.AvroHeaders;
@@ -166,7 +165,7 @@ public class AvroTests {
@Bean
public IntegrationFlow flow1() {
return IntegrationFlows.from(in1())
return IntegrationFlow.from(in1())
.transform(new SimpleToAvroTransformer())
.wireTap(tapped())
.transform(fromTransformer())
@@ -177,7 +176,7 @@ public class AvroTests {
@Bean
public IntegrationFlow flow2() {
return IntegrationFlows.from(in2())
return IntegrationFlow.from(in2())
.transform(new SimpleToAvroTransformer())
.wireTap(tapped())
.enrichHeaders(h -> h.header(AvroHeaders.TYPE, AvroTestClass2.class, true))
@@ -189,7 +188,7 @@ public class AvroTests {
@Bean
public IntegrationFlow flow3() {
return IntegrationFlows.from(in3())
return IntegrationFlow.from(in3())
.transform(new SimpleToAvroTransformer())
.wireTap(tapped())
.enrichHeaders(h -> h.header(AvroHeaders.TYPE, AvroTestClass2.class.getName(), true))
@@ -201,7 +200,7 @@ public class AvroTests {
@Bean
public IntegrationFlow flow4() {
return IntegrationFlows.from(in4())
return IntegrationFlow.from(in4())
.transform(new SimpleToAvroTransformer())
.wireTap(tapped())
.enrichHeaders(h -> h.header(AvroHeaders.TYPE, null, true)
@@ -214,7 +213,7 @@ public class AvroTests {
@Bean
public IntegrationFlow flow5() {
return IntegrationFlows.from(in5())
return IntegrationFlow.from(in5())
.transform(new SimpleToAvroTransformer().typeExpression("'avroTest'"))
.wireTap(tapped())
.transform(new SimpleFromAvroTransformer(AvroTestClass1.class)
@@ -227,7 +226,7 @@ public class AvroTests {
@Bean
public IntegrationFlow flow6() {
return IntegrationFlows.from(in6())
return IntegrationFlow.from(in6())
.transform(new SimpleToAvroTransformer().typeExpression("'wontFindThisHeader'"))
.wireTap(tapped())
.transform(new SimpleFromAvroTransformer(AvroTestClass1.class)