diff --git a/spring-integration-java-dsl/build.gradle b/spring-integration-java-dsl/build.gradle index 49cc8f6..a10be01 100644 --- a/spring-integration-java-dsl/build.gradle +++ b/spring-integration-java-dsl/build.gradle @@ -33,7 +33,7 @@ ext { mailVersion = '1.4.7' slf4jVersion = '1.7.7' springIntegrationVersion = '4.0.4.RELEASE' - springBootVersion = '1.1.6.RELEASE' + springBootVersion = '1.1.7.RELEASE' linkHomepage = 'https://github.com/spring-projects/spring-integration-extensions' linkCi = 'https://build.spring.io/browse/INTEXT' diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Adapters.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Adapters.java new file mode 100644 index 0000000..102af56 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Adapters.java @@ -0,0 +1,151 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.dsl; + +import java.io.File; + +import javax.jms.ConnectionFactory; + +import org.apache.commons.net.ftp.FTPFile; + +import org.springframework.amqp.core.AmqpTemplate; +import org.springframework.integration.dsl.amqp.Amqp; +import org.springframework.integration.dsl.amqp.AmqpOutboundEndpointSpec; +import org.springframework.integration.dsl.file.FileWritingMessageHandlerSpec; +import org.springframework.integration.dsl.file.Files; +import org.springframework.integration.dsl.ftp.Ftp; +import org.springframework.integration.dsl.ftp.FtpMessageHandlerSpec; +import org.springframework.integration.dsl.ftp.FtpOutboundGatewaySpec; +import org.springframework.integration.dsl.jms.Jms; +import org.springframework.integration.dsl.jms.JmsOutboundChannelAdapterSpec; +import org.springframework.integration.dsl.jms.JmsOutboundGatewaySpec; +import org.springframework.integration.dsl.mail.Mail; +import org.springframework.integration.dsl.mail.MailSendingMessageHandlerSpec; +import org.springframework.integration.dsl.sftp.Sftp; +import org.springframework.integration.dsl.sftp.SftpMessageHandlerSpec; +import org.springframework.integration.dsl.sftp.SftpOutboundGatewaySpec; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.jms.core.JmsTemplate; + +import com.jcraft.jsch.ChannelSftp; + +/** + * @author Artem Bilan + */ +public class Adapters { + + public AmqpOutboundEndpointSpec amqp(AmqpTemplate amqpTemplate) { + return Amqp.outboundAdapter(amqpTemplate); + } + + public AmqpOutboundEndpointSpec amqpGateway(AmqpTemplate amqpTemplate) { + return Amqp.outboundGateway(amqpTemplate); + } + + public FileWritingMessageHandlerSpec file(File destinationDirectory) { + return Files.outboundAdapter(destinationDirectory); + } + + public FileWritingMessageHandlerSpec file(String directoryExpression) { + return Files.outboundAdapter(directoryExpression); + } + + public FileWritingMessageHandlerSpec fileGateway(File destinationDirectory) { + return Files.outboundGateway(destinationDirectory); + } + + public FileWritingMessageHandlerSpec fileGateway(String directoryExpression) { + return Files.outboundGateway(directoryExpression); + } + + public FtpMessageHandlerSpec ftp(SessionFactory sessionFactory) { + return Ftp.outboundAdapter(sessionFactory); + } + + public FtpMessageHandlerSpec ftp(SessionFactory sessionFactory, FileExistsMode fileExistsMode) { + return Ftp.outboundAdapter(sessionFactory, fileExistsMode); + } + + public FtpMessageHandlerSpec ftp(RemoteFileTemplate remoteFileTemplate) { + return Ftp.outboundAdapter(remoteFileTemplate); + } + + public FtpMessageHandlerSpec ftp(RemoteFileTemplate remoteFileTemplate, FileExistsMode fileExistsMode) { + return Ftp.outboundAdapter(remoteFileTemplate, fileExistsMode); + } + + public FtpOutboundGatewaySpec ftpGateway(SessionFactory sessionFactory, + AbstractRemoteFileOutboundGateway.Command command, String expression) { + return Ftp.outboundGateway(sessionFactory, command, expression); + } + + public FtpOutboundGatewaySpec ftpGateway(SessionFactory sessionFactory, String command, + String expression) { + return Ftp.outboundGateway(sessionFactory, command, expression); + } + + public SftpMessageHandlerSpec ftps(SessionFactory sessionFactory) { + return Sftp.outboundAdapter(sessionFactory); + } + + public SftpMessageHandlerSpec sftp(SessionFactory sessionFactory, + FileExistsMode fileExistsMode) { + return Sftp.outboundAdapter(sessionFactory, fileExistsMode); + } + + public SftpMessageHandlerSpec sftp(RemoteFileTemplate remoteFileTemplate) { + return Sftp.outboundAdapter(remoteFileTemplate); + } + + public SftpMessageHandlerSpec sftp(RemoteFileTemplate remoteFileTemplate, + FileExistsMode fileExistsMode) { + return Sftp.outboundAdapter(remoteFileTemplate, fileExistsMode); + } + + public SftpOutboundGatewaySpec sftpGateway(SessionFactory sessionFactory, + AbstractRemoteFileOutboundGateway.Command command, String expression) { + return Sftp.outboundGateway(sessionFactory, command, expression); + } + + public SftpOutboundGatewaySpec sftpGateway(SessionFactory sessionFactory, String command, + String expression) { + return Sftp.outboundGateway(sessionFactory, command, expression); + } + + public JmsOutboundChannelAdapterSpec.JmsOutboundChannelSpecTemplateAware jms(ConnectionFactory connectionFactory) { + return Jms.outboundAdapter(connectionFactory); + } + + public JmsOutboundChannelAdapterSpec> jms(JmsTemplate jmsTemplate) { + return Jms.outboundAdapter(jmsTemplate); + } + + public JmsOutboundGatewaySpec jmsGateway(ConnectionFactory connectionFactory) { + return Jms.outboundGateway(connectionFactory); + } + + public MailSendingMessageHandlerSpec mail(String host) { + return Mail.outboundAdapter(host); + } + + Adapters() { + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Channels.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Channels.java new file mode 100644 index 0000000..0abd121 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Channels.java @@ -0,0 +1,197 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.dsl; + +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.Executor; + +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.integration.dsl.amqp.Amqp; +import org.springframework.integration.dsl.amqp.AmqpMessageChannelSpec; +import org.springframework.integration.dsl.amqp.AmqpPollableMessageChannelSpec; +import org.springframework.integration.dsl.amqp.AmqpPublishSubscribeMessageChannelSpec; +import org.springframework.integration.dsl.channel.DirectChannelSpec; +import org.springframework.integration.dsl.channel.ExecutorChannelSpec; +import org.springframework.integration.dsl.channel.MessageChannels; +import org.springframework.integration.dsl.channel.PriorityChannelSpec; +import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec; +import org.springframework.integration.dsl.channel.QueueChannelSpec; +import org.springframework.integration.dsl.channel.RendezvousChannelSpec; +import org.springframework.integration.dsl.jms.Jms; +import org.springframework.integration.dsl.jms.JmsMessageChannelSpec; +import org.springframework.integration.dsl.jms.JmsPollableMessageChannelSpec; +import org.springframework.integration.dsl.jms.JmsPublishSubscribeMessageChannelSpec; +import org.springframework.integration.store.ChannelMessageStore; +import org.springframework.integration.store.PriorityCapableChannelMessageStore; +import org.springframework.messaging.Message; + +/** + * @author Artem Bilan + */ +public class Channels { + + public DirectChannelSpec direct() { + return MessageChannels.direct(); + } + + public DirectChannelSpec direct(String id) { + return MessageChannels.direct(id); + } + + public QueueChannelSpec queue() { + return MessageChannels.queue(); + } + + public QueueChannelSpec queue(String id) { + return MessageChannels.queue(id); + } + + public QueueChannelSpec queue(Integer capacity) { + return MessageChannels.queue(capacity); + } + + public QueueChannelSpec queue(String id, Integer capacity) { + return MessageChannels.queue(id, capacity); + } + + public QueueChannelSpec queue(BlockingQueue> queue) { + return MessageChannels.queue(queue); + } + + public QueueChannelSpec queue(String id, BlockingQueue> queue) { + return MessageChannels.queue(id, queue); + } + + public QueueChannelSpec.MessageStoreSpec queue(ChannelMessageStore messageGroupStore, Object groupId) { + return MessageChannels.queue(messageGroupStore, groupId); + } + + public QueueChannelSpec.MessageStoreSpec queue(String id, ChannelMessageStore messageGroupStore, Object groupId) { + return MessageChannels.queue(id, messageGroupStore, groupId); + } + + public PriorityChannelSpec priority() { + return MessageChannels.priority(); + } + + public PriorityChannelSpec priority(String id) { + return MessageChannels.priority(id); + } + + public QueueChannelSpec.MessageStoreSpec priority(String id, PriorityCapableChannelMessageStore messageGroupStore, + Object groupId) { + return MessageChannels.priority(id, messageGroupStore, groupId); + } + + public QueueChannelSpec.MessageStoreSpec priority(PriorityCapableChannelMessageStore messageGroupStore, + Object groupId) { + return MessageChannels.priority(messageGroupStore, groupId); + } + + public RendezvousChannelSpec rendezvous() { + return MessageChannels.rendezvous(); + } + + public RendezvousChannelSpec rendezvous(String id) { + return MessageChannels.rendezvous(id); + } + + public PublishSubscribeChannelSpec publishSubscribe() { + return MessageChannels.publishSubscribe(); + } + + public PublishSubscribeChannelSpec publishSubscribe(Executor executor) { + return MessageChannels.publishSubscribe(executor); + } + + public ExecutorChannelSpec executor(Executor executor) { + return MessageChannels.executor(executor); + } + + public ExecutorChannelSpec executor(String id, Executor executor) { + return MessageChannels.executor(id, executor); + } + + public PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) { + return MessageChannels.publishSubscribe(id, executor); + } + + public PublishSubscribeChannelSpec publishSubscribe(String id) { + return MessageChannels.publishSubscribe(id); + } + + public AmqpPollableMessageChannelSpec> amqpPollable( + ConnectionFactory connectionFactory) { + return Amqp.pollableChannel(connectionFactory); + } + + public AmqpPollableMessageChannelSpec> amqpPollable(String id, + ConnectionFactory connectionFactory) { + return Amqp.pollableChannel(id, connectionFactory); + } + + public AmqpMessageChannelSpec> amqp(ConnectionFactory connectionFactory) { + return Amqp.channel(connectionFactory); + } + + public AmqpMessageChannelSpec> amqp(String id, + ConnectionFactory connectionFactory) { + return Amqp.channel(id, connectionFactory); + } + + public static AmqpPublishSubscribeMessageChannelSpec amqpPublishSubscribe(ConnectionFactory connectionFactory) { + return Amqp.publishSubscribeChannel(connectionFactory); + } + + public static AmqpPublishSubscribeMessageChannelSpec amqpPublishSubscribe(String id, + ConnectionFactory connectionFactory) { + return Amqp.publishSubscribeChannel(id, connectionFactory); + } + + public JmsPollableMessageChannelSpec> jmsPollable( + javax.jms.ConnectionFactory connectionFactory) { + return Jms.pollableChannel(connectionFactory); + } + + public JmsPollableMessageChannelSpec> jmsPollable(String id, + javax.jms.ConnectionFactory connectionFactory) { + return Jms.pollableChannel(id, connectionFactory); + } + + public JmsMessageChannelSpec> jms( + javax.jms.ConnectionFactory connectionFactory) { + return Jms.channel(connectionFactory); + } + + public JmsMessageChannelSpec> jms(String id, + javax.jms.ConnectionFactory connectionFactory) { + return Jms.channel(id, connectionFactory); + } + + public JmsPublishSubscribeMessageChannelSpec jmsPublishSubscribe(javax.jms.ConnectionFactory connectionFactory) { + return Jms.publishSubscribeChannel(connectionFactory); + } + + public JmsPublishSubscribeMessageChannelSpec jmsPublishSubscribe(String id, + javax.jms.ConnectionFactory connectionFactory) { + return Jms.publishSubscribeChannel(id, connectionFactory); + } + + Channels() { + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java index 2c24e47..16685e9 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java @@ -34,7 +34,8 @@ import org.springframework.util.StringUtils; /** * @author Artem Bilan */ -public abstract class CorrelationHandlerSpec, H extends AbstractCorrelatingMessageHandler> +public abstract class + CorrelationHandlerSpec, H extends AbstractCorrelatingMessageHandler> extends MessageHandlerSpec { protected MessageGroupStore messageStore; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java index 20d1ef8..e321f91 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java @@ -69,7 +69,8 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec(PARSER.parseExpression(expression))); + return this.messageProcessor(new ExpressionEvaluatingMessageProcessor( + PARSER.parseExpression(expression))); } public HeaderEnricherSpec messageProcessor(String beanName, String methodName) { @@ -99,7 +100,8 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec configurer) { + public HeaderEnricherSpec headerExpressions( + MapBuilderConfigurer configurer) { StringStringMapBuilder builder = new StringStringMapBuilder(); configurer.configure(builder); return headerExpressions(builder.get()); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java index 710b9c0..d23ef49 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java @@ -37,8 +37,8 @@ public final class IntegrationFlowBuilder extends IntegrationFlowDefinition> channels) { + Assert.notNull(channels); + return channel(channels.apply(new Channels())); + } + public B channel(MessageChannelSpec messageChannelSpec) { Assert.notNull(messageChannelSpec); return this.channel(messageChannelSpec.get()); @@ -144,9 +149,9 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { - return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(new ControlBusMethodFilter())), - endpointConfigurer); + public B controlBus(Consumer> endpointConfigurer) { + return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor( + new ControlBusMethodFilter())), endpointConfigurer); } public B transform(String expression) { @@ -163,12 +168,12 @@ public abstract class IntegrationFlowDefinition B transform(GenericTransformer genericTransformer, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { return this.transform(null, genericTransformer, endpointConfigurer); } public B transform(Class

payloadType, GenericTransformer genericTransformer, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { Assert.notNull(genericTransformer); Transformer transformer = genericTransformer instanceof Transformer ? (Transformer) genericTransformer : (isLambda(genericTransformer) @@ -192,12 +197,12 @@ public abstract class IntegrationFlowDefinition B filter(GenericSelector

genericSelector, - EndpointConfigurer endpointConfigurer) { + Consumer endpointConfigurer) { return filter(null, genericSelector, endpointConfigurer); } public

B filter(Class

payloadType, GenericSelector

genericSelector, - EndpointConfigurer endpointConfigurer) { + Consumer endpointConfigurer) { Assert.notNull(genericSelector); MessageSelector selector = genericSelector instanceof MessageSelector ? (MessageSelector) genericSelector : (isLambda(genericSelector) @@ -206,9 +211,19 @@ public abstract class IntegrationFlowDefinition> B handle(S messageHandlerSpec) { - Assert.notNull(messageHandlerSpec); - return handle(messageHandlerSpec.get()); + public B handleWithAdapter( + Function> adapters) { + return handleWithAdapter(adapters, null); + } + + public B handleWithAdapter( + Function> adapters, + Consumer> endpointConfigurer) { + return handle(adapters.apply(new Adapters()), endpointConfigurer); + } + + public B handle(MessageHandlerSpec messageHandlerSpec) { + return handle(messageHandlerSpec, null); } public B handle(MessageHandler messageHandler) { @@ -220,7 +235,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + Consumer> endpointConfigurer) { return this.handle(new ServiceActivatingHandler(new BeanNameMessageProcessor(beanName, methodName)), endpointConfigurer); } @@ -230,7 +245,7 @@ public abstract class IntegrationFlowDefinition B handle(GenericHandler

handler, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { return this.handle(null, handler, endpointConfigurer); } @@ -239,7 +254,7 @@ public abstract class IntegrationFlowDefinition B handle(Class

payloadType, GenericHandler

handler, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { ServiceActivatingHandler serviceActivatingHandler = null; if (isLambda(handler)) { serviceActivatingHandler = new ServiceActivatingHandler(new LambdaMessageProcessor(handler, payloadType)); @@ -250,19 +265,19 @@ public abstract class IntegrationFlowDefinition> - B handle(S messageHandlerSpec, EndpointConfigurer> endpointConfigurer) { + public B handle(MessageHandlerSpec messageHandlerSpec, + Consumer> endpointConfigurer) { Assert.notNull(messageHandlerSpec); return handle(messageHandlerSpec.get(), endpointConfigurer); } public B handle(H messageHandler, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { Assert.notNull(messageHandler); return this.register(new GenericEndpointSpec(messageHandler), endpointConfigurer); } - public B bridge(EndpointConfigurer> endpointConfigurer) { + public B bridge(Consumer> endpointConfigurer) { return this.register(new GenericEndpointSpec(new BridgeHandler()), endpointConfigurer); } @@ -271,7 +286,7 @@ public abstract class IntegrationFlowDefinition endpointConfigurer) { + Consumer endpointConfigurer) { DelayHandler delayHandler = new DelayHandler(groupId); if (StringUtils.hasText(expression)) { delayHandler.setDelayExpression(PARSER.parseExpression(expression)); @@ -279,15 +294,15 @@ public abstract class IntegrationFlowDefinition enricherConfigurer) { + public B enrich(Consumer enricherConfigurer) { return this.enrich(enricherConfigurer, null); } - public B enrich(ComponentConfigurer enricherConfigurer, - EndpointConfigurer> endpointConfigurer) { + public B enrich(Consumer enricherConfigurer, + Consumer> endpointConfigurer) { Assert.notNull(enricherConfigurer); EnricherSpec enricherSpec = new EnricherSpec(); - enricherConfigurer.configure(enricherSpec); + enricherConfigurer.accept(enricherSpec); return this.handle(enricherSpec.get(), endpointConfigurer); } @@ -296,7 +311,7 @@ public abstract class IntegrationFlowDefinition headers, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { return enrichHeaders(headers.get(), endpointConfigurer); } @@ -313,33 +328,35 @@ public abstract class IntegrationFlowDefinition headers, - EndpointConfigurer> endpointConfigurer) { - return enrichHeaders(new ComponentConfigurer() { + Consumer> endpointConfigurer) { + return enrichHeaders(new Consumer() { + @Override - public void configure(HeaderEnricherSpec spec) { + public void accept(HeaderEnricherSpec spec) { spec.headers(headers); } + }, endpointConfigurer); } - public B enrichHeaders(ComponentConfigurer headerEnricherConfigurer) { + public B enrichHeaders(Consumer headerEnricherConfigurer) { return this.enrichHeaders(headerEnricherConfigurer, null); } - public B enrichHeaders(ComponentConfigurer headerEnricherConfigurer, - EndpointConfigurer> endpointConfigurer) { + public B enrichHeaders(Consumer headerEnricherConfigurer, + Consumer> endpointConfigurer) { Assert.notNull(headerEnricherConfigurer); HeaderEnricherSpec headerEnricherSpec = new HeaderEnricherSpec(); - headerEnricherConfigurer.configure(headerEnricherSpec); + headerEnricherConfigurer.accept(headerEnricherSpec); return transform(headerEnricherSpec.get(), endpointConfigurer); } - public B split(EndpointConfigurer> endpointConfigurer) { + public B split(Consumer> endpointConfigurer) { return this.split(new DefaultMessageSplitter(), endpointConfigurer); } public B split(String expression, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { return this.split(new ExpressionEvaluatingSplitter(PARSER.parseExpression(expression)), endpointConfigurer); } @@ -348,7 +365,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + Consumer> endpointConfigurer) { return this.split(new MethodInvokingSplitter(new BeanNameMessageProcessor>(beanName, methodName)), endpointConfigurer); } @@ -358,12 +375,12 @@ public abstract class IntegrationFlowDefinition B split(GenericSplitter splitter, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { return split(null, splitter, endpointConfigurer); } public

B split(Class

payloadType, GenericSplitter

splitter, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { MethodInvokingSplitter split = isLambda(splitter) ? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, payloadType)) : new MethodInvokingSplitter(splitter, "split"); @@ -371,13 +388,13 @@ public abstract class IntegrationFlowDefinition B split(S splitter, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { Assert.notNull(splitter); return this.register(new SplitterEndpointSpec(splitter), endpointConfigurer); } /** - * Provides the {@link HeaderFilter} to the current {@link org.springframework.integration.dsl.IntegrationFlowBuilder.StandardIntegrationFlow}. + * Provides the {@link HeaderFilter} to the current {@link IntegrationFlowBuilder.StandardIntegrationFlow}. * @param headersToRemove the array of headers (or patterns) * to remove from {@link org.springframework.messaging.MessageHeaders}. * @return this {@link IntegrationFlowDefinition}. @@ -387,7 +404,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + Consumer> endpointConfigurer) { return this.transform(headerFilter, endpointConfigurer); } @@ -410,7 +427,7 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + Consumer> endpointConfigurer) { return this.transform(new ClaimCheckInTransformer(messageStore), endpointConfigurer); } @@ -423,42 +440,42 @@ public abstract class IntegrationFlowDefinition> endpointConfigurer) { + Consumer> endpointConfigurer) { ClaimCheckOutTransformer claimCheckOutTransformer = new ClaimCheckOutTransformer(messageStore); claimCheckOutTransformer.setRemoveMessage(removeMessage); return this.transform(claimCheckOutTransformer, endpointConfigurer); } public B resequence() { - return this.resequence((EndpointConfigurer>) null); + return this.resequence((Consumer>) null); } - public B resequence(EndpointConfigurer> endpointConfigurer) { + public B resequence(Consumer> endpointConfigurer) { return this.handle(new ResequencerSpec().get(), endpointConfigurer); } - public B resequence(ComponentConfigurer resequencerConfigurer, - EndpointConfigurer> endpointConfigurer) { + public B resequence(Consumer resequencerConfigurer, + Consumer> endpointConfigurer) { Assert.notNull(resequencerConfigurer); ResequencerSpec spec = new ResequencerSpec(); - resequencerConfigurer.configure(spec); + resequencerConfigurer.accept(spec); return this.handle(spec.get(), endpointConfigurer); } public B aggregate() { - return aggregate((EndpointConfigurer>) null); + return aggregate((Consumer>) null); } public B - aggregate(EndpointConfigurer> endpointConfigurer) { + aggregate(Consumer> endpointConfigurer) { return handle(new AggregatorSpec().get(), endpointConfigurer); } - public B aggregate(ComponentConfigurer aggregatorConfigurer, - EndpointConfigurer> endpointConfigurer) { + public B aggregate(Consumer aggregatorConfigurer, + Consumer> endpointConfigurer) { Assert.notNull(aggregatorConfigurer); AggregatorSpec spec = new AggregatorSpec(); - aggregatorConfigurer.configure(spec); + aggregatorConfigurer.accept(spec); return this.handle(spec.get(), endpointConfigurer); } @@ -467,29 +484,27 @@ public abstract class IntegrationFlowDefinition> routerConfigurer) { + Consumer> routerConfigurer) { return this.route(beanName, method, routerConfigurer, null); } - public B route(String beanName, String method, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + public B route(String beanName, String method, Consumer> routerConfigurer, + Consumer> endpointConfigurer) { return this.route(new MethodInvokingRouter(new BeanNameMessageProcessor(beanName, method)), routerConfigurer, endpointConfigurer); } public B route(String expression) { - return this.route(expression, (ComponentConfigurer>) null); + return this.route(expression, (Consumer>) null); } - public B route(String expression, - ComponentConfigurer> routerConfigurer) { + public B route(String expression, Consumer> routerConfigurer) { return this.route(expression, routerConfigurer, null); } public B route(String expression, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + Consumer> routerConfigurer, + Consumer> endpointConfigurer) { return this.route(new ExpressionEvaluatingRouter(PARSER.parseExpression(expression)), routerConfigurer, endpointConfigurer); } @@ -499,7 +514,7 @@ public abstract class IntegrationFlowDefinition B route(GenericRouter router, - ComponentConfigurer> routerConfigurer) { + Consumer> routerConfigurer) { return this.route(null, router, routerConfigurer); } @@ -508,19 +523,19 @@ public abstract class IntegrationFlowDefinition B route(Class

payloadType, GenericRouter router, - ComponentConfigurer> routerConfigurer) { + Consumer> routerConfigurer) { return this.route(payloadType, router, routerConfigurer, null); } public B route(GenericRouter router, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + Consumer> routerConfigurer, + Consumer> endpointConfigurer) { return route(null, router, routerConfigurer, endpointConfigurer); } public B route(Class

payloadType, GenericRouter router, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + Consumer> routerConfigurer, + Consumer> endpointConfigurer) { MethodInvokingRouter methodInvokingRouter = isLambda(router) ? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType)) : new MethodInvokingRouter(router); @@ -528,24 +543,24 @@ public abstract class IntegrationFlowDefinition B route(R router, - ComponentConfigurer> routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + Consumer> routerConfigurer, + Consumer> endpointConfigurer) { if (routerConfigurer != null) { RouterSpec routerSpec = new RouterSpec(router); - routerConfigurer.configure(routerSpec); + routerConfigurer.accept(routerSpec); } return this.route(router, endpointConfigurer); } - public B routeToRecipients(ComponentConfigurer routerConfigurer) { + public B routeToRecipients(Consumer routerConfigurer) { return this.routeToRecipients(routerConfigurer, null); } - public B routeToRecipients(ComponentConfigurer routerConfigurer, - EndpointConfigurer> endpointConfigurer) { + public B routeToRecipients(Consumer routerConfigurer, + Consumer> endpointConfigurer) { Assert.notNull(routerConfigurer); RecipientListRouterSpec spec = new RecipientListRouterSpec(); - routerConfigurer.configure(spec); + routerConfigurer.accept(spec); DslRecipientListRouter recipientListRouter = (DslRecipientListRouter) spec.get(); Assert.notEmpty(recipientListRouter.get(), "recipient list must not be empty"); return this.route(recipientListRouter, endpointConfigurer); @@ -556,7 +571,7 @@ public abstract class IntegrationFlowDefinition B route(R router, - EndpointConfigurer> endpointConfigurer) { + Consumer> endpointConfigurer) { return this.handle(router, endpointConfigurer); } @@ -565,7 +580,7 @@ public abstract class IntegrationFlowDefinition endpointConfigurer) { + Consumer endpointConfigurer) { return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer); } @@ -574,14 +589,14 @@ public abstract class IntegrationFlowDefinition endpointConfigurer) { + Consumer endpointConfigurer) { return register(new GatewayEndpointSpec(requestChannel), endpointConfigurer); } private > B register(S endpointSpec, - EndpointConfigurer endpointConfigurer) { + Consumer endpointConfigurer) { if (endpointConfigurer != null) { - endpointConfigurer.configure(endpointSpec); + endpointConfigurer.accept(endpointSpec); } MessageChannel inputChannel = this.currentMessageChannel; this.currentMessageChannel = null; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java index 8a888ec..11eec72 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlows.java @@ -24,12 +24,14 @@ import org.springframework.integration.dsl.core.ComponentsRegistration; import org.springframework.integration.dsl.core.MessageProducerSpec; import org.springframework.integration.dsl.core.MessageSourceSpec; import org.springframework.integration.dsl.core.MessagingGatewaySpec; -import org.springframework.integration.dsl.support.EndpointConfigurer; +import org.springframework.integration.dsl.support.Consumer; import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype; +import org.springframework.integration.dsl.support.Function; import org.springframework.integration.dsl.support.MessageChannelReference; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; /** * The central factory for fluent {@link IntegrationFlowBuilder} API. @@ -39,9 +41,9 @@ import org.springframework.messaging.MessageChannel; public final class IntegrationFlows { /** - * @param messageChannelName the name of existing {@link org.springframework.messaging.MessageChannel} bean. - * The new {@link org.springframework.integration.channel.DirectChannel} bean will be - * created on context startup, if there is no bean with this name. + * @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} */ public static IntegrationFlowBuilder from(String messageChannelName) { @@ -49,15 +51,28 @@ public final class IntegrationFlows { } /** - * @param messageChannelName the name for {@link org.springframework.integration.channel.FixedSubscriberChannel} - * to be created on context startup, not reference. + * @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} */ - public static IntegrationFlowBuilder fromFixedMessageChannel(String messageChannelName) { - return from(new FixedSubscriberChannelPrototype(messageChannelName)); + public static IntegrationFlowBuilder from(String messageChannelName, boolean fixedSubscriber) { + return fixedSubscriber + ? from(new FixedSubscriberChannelPrototype(messageChannelName)) + : from(messageChannelName); + } + + public static IntegrationFlowBuilder from(ChannelsFunction channels) { + Assert.notNull(channels); + return from(channels.apply(new Channels())); } public static IntegrationFlowBuilder from(MessageChannelSpec messageChannelSpec) { + Assert.notNull(messageChannelSpec); return from(messageChannelSpec.get()); } @@ -65,13 +80,23 @@ public final class IntegrationFlows { return new IntegrationFlowBuilder().channel(messageChannel); } - public static >> IntegrationFlowBuilder - from(S messageSourceSpec) { + public static IntegrationFlowBuilder from(MessageSourcesFunction sources) { + return from(sources, null); + } + + public static IntegrationFlowBuilder from(MessageSourcesFunction sources, + Consumer endpointConfigurer) { + Assert.notNull(sources); + return from(sources.apply(new MessageSources()), endpointConfigurer); + } + + public static IntegrationFlowBuilder from(MessageSourceSpec> messageSourceSpec) { return from(messageSourceSpec, null); } - public static >> IntegrationFlowBuilder - from(S messageSourceSpec, EndpointConfigurer endpointConfigurer) { + public static IntegrationFlowBuilder from(MessageSourceSpec> messageSourceSpec, + Consumer endpointConfigurer) { + Assert.notNull(messageSourceSpec); return from(messageSourceSpec.get(), endpointConfigurer, registerComponents(messageSourceSpec)); } @@ -80,16 +105,16 @@ public final class IntegrationFlows { } public static IntegrationFlowBuilder from(MessageSource messageSource, - EndpointConfigurer endpointConfigurer) { + Consumer endpointConfigurer) { return from(messageSource, endpointConfigurer, null); } private static IntegrationFlowBuilder from(MessageSource messageSource, - EndpointConfigurer endpointConfigurer, + Consumer endpointConfigurer, IntegrationFlowBuilder integrationFlowBuilder) { SourcePollingChannelAdapterSpec spec = new SourcePollingChannelAdapterSpec(messageSource); if (endpointConfigurer != null) { - endpointConfigurer.configure(spec); + endpointConfigurer.accept(spec); } if (integrationFlowBuilder == null) { integrationFlowBuilder = new IntegrationFlowBuilder(); @@ -98,6 +123,10 @@ public final class IntegrationFlows { .currentComponent(spec); } + public static IntegrationFlowBuilder from(MessageProducersFunction producers) { + return from(producers.apply(new MessageProducers())); + } + public static IntegrationFlowBuilder from(MessageProducerSpec messageProducerSpec) { return from(messageProducerSpec.get(), registerComponents(messageProducerSpec)); } @@ -123,6 +152,10 @@ public final class IntegrationFlows { return integrationFlowBuilder.addComponent(messageProducer); } + public static IntegrationFlowBuilder from(MessagingGatewaysFunction gateways) { + return from(gateways.apply(new MessagingGateways())); + } + public static IntegrationFlowBuilder from(MessagingGatewaySpec inboundGatewaySpec) { return from(inboundGatewaySpec.get(), registerComponents(inboundGatewaySpec)); } @@ -165,4 +198,12 @@ public final class IntegrationFlows { private IntegrationFlows() { } + public interface ChannelsFunction extends Function> {} + + public interface MessageSourcesFunction extends Function> {} + + public interface MessageProducersFunction extends Function> {} + + public interface MessagingGatewaysFunction extends Function> {} + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java index 641e57f..b8e9a97 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/LambdaMessageProcessor.java @@ -55,16 +55,20 @@ class LambdaMessageProcessor implements MessageProcessor, BeanFactoryAwa this.target = target; final AtomicReference methodValue = new AtomicReference(); ReflectionUtils.doWithMethods(target.getClass(), new ReflectionUtils.MethodCallback() { + @Override public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { methodValue.set(method); } + }, new ReflectionUtils.MethodFilter() { + @Override public boolean matches(Method method) { return !method.isBridge() && method.getDeclaringClass() != Object.class && Modifier.isPublic(method.getModifiers()) && !Modifier.isStatic(method.getModifiers()); } + }); Assert.notNull(methodValue.get(), "LambdaMessageProcessor is applicable for inline or lambda " + @@ -125,4 +129,5 @@ class LambdaMessageProcessor implements MessageProcessor, BeanFactoryAwa throw new MessageHandlingException(message, e); } } + } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessageProducers.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessageProducers.java new file mode 100644 index 0000000..3d8d2e6 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessageProducers.java @@ -0,0 +1,75 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.dsl; + +import java.io.File; + +import org.springframework.amqp.core.Queue; +import org.springframework.integration.dsl.amqp.Amqp; +import org.springframework.integration.dsl.amqp.AmqpInboundChannelAdapterSpec; +import org.springframework.integration.dsl.file.Files; +import org.springframework.integration.dsl.file.TailAdapterSpec; +import org.springframework.integration.dsl.jms.Jms; +import org.springframework.integration.dsl.jms.JmsMessageDrivenChannelAdapterSpec; +import org.springframework.integration.dsl.mail.ImapIdleChannelAdapterSpec; +import org.springframework.integration.dsl.mail.Mail; +import org.springframework.jms.listener.AbstractMessageListenerContainer; + +/** + * @author Artem Bilan + */ +public class MessageProducers { + + public AmqpInboundChannelAdapterSpec amqp( + org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory, String... queueNames) { + return Amqp.inboundAdapter(connectionFactory, queueNames); + } + + public AmqpInboundChannelAdapterSpec amqp( + org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory, Queue... queues) { + return Amqp.inboundAdapter(connectionFactory, queues); + } + + public TailAdapterSpec tail(File file) { + return Files.tailAdapter(file); + } + + public ImapIdleChannelAdapterSpec imap(String url) { + return Mail.imapIdleAdapter(url); + } + + public JmsMessageDrivenChannelAdapterSpec> jms( + AbstractMessageListenerContainer listenerContainer) { + return Jms.messageDriverChannelAdapter(listenerContainer); + } + + public JmsMessageDrivenChannelAdapterSpec> jms( + javax.jms.ConnectionFactory connectionFactory) { + return Jms.messageDriverChannelAdapter(connectionFactory); + } + + public + JmsMessageDrivenChannelAdapterSpec> jms( + javax.jms.ConnectionFactory connectionFactory, + Class containerClass) { + return Jms.messageDriverChannelAdapter(connectionFactory, containerClass); + } + + MessageProducers() { + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessageSources.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessageSources.java new file mode 100644 index 0000000..f625e6c --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessageSources.java @@ -0,0 +1,108 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.dsl; + +import java.io.File; +import java.util.Comparator; + +import javax.jms.ConnectionFactory; + +import org.apache.commons.net.ftp.FTPFile; + +import org.springframework.integration.dsl.file.FileInboundChannelAdapterSpec; +import org.springframework.integration.dsl.file.Files; +import org.springframework.integration.dsl.ftp.Ftp; +import org.springframework.integration.dsl.ftp.FtpInboundChannelAdapterSpec; +import org.springframework.integration.dsl.jms.Jms; +import org.springframework.integration.dsl.jms.JmsInboundChannelAdapterSpec; +import org.springframework.integration.dsl.mail.ImapMailInboundChannelAdapterSpec; +import org.springframework.integration.dsl.mail.Mail; +import org.springframework.integration.dsl.mail.Pop3MailInboundChannelAdapterSpec; +import org.springframework.integration.dsl.sftp.Sftp; +import org.springframework.integration.dsl.sftp.SftpInboundChannelAdapterSpec; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.jms.core.JmsTemplate; + +import com.jcraft.jsch.ChannelSftp; + +/** + * @author Artem Bilan + */ +public class MessageSources { + + public FileInboundChannelAdapterSpec file(File directory) { + return file(directory, null); + } + + public FileInboundChannelAdapterSpec file(File directory, Comparator receptionOrderComparator) { + return Files.inboundAdapter(directory, receptionOrderComparator); + } + + public FtpInboundChannelAdapterSpec ftp(SessionFactory sessionFactory) { + return ftp(sessionFactory, null); + } + + public FtpInboundChannelAdapterSpec ftp(SessionFactory sessionFactory, + Comparator receptionOrderComparator) { + return Ftp.inboundAdapter(sessionFactory, receptionOrderComparator); + } + + public SftpInboundChannelAdapterSpec sftp(SessionFactory sessionFactory) { + return sftp(sessionFactory, null); + } + + public SftpInboundChannelAdapterSpec sftp(SessionFactory sessionFactory, + Comparator receptionOrderComparator) { + return Sftp.inboundAdapter(sessionFactory, receptionOrderComparator); + } + + public JmsInboundChannelAdapterSpec> jms(JmsTemplate jmsTemplate) { + return Jms.inboundAdapter(jmsTemplate); + } + + public JmsInboundChannelAdapterSpec.JmsInboundChannelSpecTemplateAware jms(ConnectionFactory connectionFactory) { + return Jms.inboundAdapter(connectionFactory); + } + + public ImapMailInboundChannelAdapterSpec imap() { + return Mail.imapInboundAdapter(); + } + + public ImapMailInboundChannelAdapterSpec imap(String url) { + return Mail.imapInboundAdapter(url); + } + + public Pop3MailInboundChannelAdapterSpec pop3() { + return Mail.pop3InboundAdapter(); + } + + public Pop3MailInboundChannelAdapterSpec pop3(String url) { + return Mail.pop3InboundAdapter(url); + } + + public Pop3MailInboundChannelAdapterSpec pop3(String host, String username, String password) { + return pop3(host, -1, username, password); + } + + public Pop3MailInboundChannelAdapterSpec pop3(String host, int port, String username, String password) { + return Mail.pop3InboundAdapter(host, port, username, password); + } + + MessageSources() { + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessagingGateways.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessagingGateways.java new file mode 100644 index 0000000..4827a70 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/MessagingGateways.java @@ -0,0 +1,68 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.dsl; + +import javax.jms.ConnectionFactory; + +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; +import org.springframework.integration.dsl.amqp.Amqp; +import org.springframework.integration.dsl.amqp.AmqpInboundGatewaySpec; +import org.springframework.integration.dsl.jms.Jms; +import org.springframework.integration.dsl.jms.JmsInboundGatewaySpec; +import org.springframework.jms.listener.AbstractMessageListenerContainer; +import org.springframework.jms.listener.DefaultMessageListenerContainer; + +/** + * @author Artem Bilan + */ +public class MessagingGateways { + + public AmqpInboundGatewaySpec amqp(org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory, + String... queueNames) { + return Amqp.inboundGateway(connectionFactory, queueNames); + } + + public AmqpInboundGatewaySpec amqp(org.springframework.amqp.rabbit.connection.ConnectionFactory connectionFactory, + Queue... queues) { + return Amqp.inboundGateway(connectionFactory, queues); + } + + public AmqpInboundGatewaySpec amqp(SimpleMessageListenerContainer listenerContainer) { + return (AmqpInboundGatewaySpec) Amqp.inboundGateway(listenerContainer); + } + + public JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec jms( + javax.jms.ConnectionFactory connectionFactory) { + return Jms.inboundGateway(connectionFactory); + } + + public + JmsInboundGatewaySpec.JmsInboundGatewayListenerContainerSpec jms(ConnectionFactory connectionFactory, + Class containerClass) { + return Jms.inboundGateway(connectionFactory, containerClass); + } + + public JmsInboundGatewaySpec> jms( + AbstractMessageListenerContainer listenerContainer) { + return Jms.inboundGateway(listenerContainer); + } + + MessagingGateways() { + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java index 4686f56..530ff45 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java @@ -25,8 +25,8 @@ import org.springframework.integration.scheduling.PollerMetadata; * @author Artem Bilan */ -public final class SourcePollingChannelAdapterSpec extends EndpointSpec> { +public final class SourcePollingChannelAdapterSpec extends + EndpointSpec> { SourcePollingChannelAdapterSpec(MessageSource messageSource) { super(messageSource); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java index d0e51f8..0bbd4d4 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/Amqp.java @@ -47,7 +47,8 @@ public abstract class Amqp { return new AmqpInboundGatewaySpec(listenerContainer); } - public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, String... queueNames) { + public static AmqpInboundChannelAdapterSpec inboundAdapter(ConnectionFactory connectionFactory, + String... queueNames) { SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer(connectionFactory); listenerContainer.setQueueNames(queueNames); return (AmqpInboundChannelAdapterSpec) inboundAdapter(listenerContainer); @@ -77,12 +78,13 @@ public abstract class Amqp { return pollableChannel(null, connectionFactory); } - public static > AmqpPollableMessageChannelSpec pollableChannel(String id, - ConnectionFactory connectionFactory) { + public static > AmqpPollableMessageChannelSpec pollableChannel( + String id, ConnectionFactory connectionFactory) { return new AmqpPollableMessageChannelSpec(connectionFactory).id(id); } - public static > AmqpMessageChannelSpec channel(ConnectionFactory connectionFactory) { + public static > AmqpMessageChannelSpec channel( + ConnectionFactory connectionFactory) { return channel(null, connectionFactory); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java index 23ff88e..56bdeb7 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/amqp/AmqpInboundChannelAdapterSpec.java @@ -34,7 +34,8 @@ import org.springframework.util.ErrorHandler; /** * @author Artem Bilan */ -public class AmqpInboundChannelAdapterSpec extends MessageProducerSpec { +public class AmqpInboundChannelAdapterSpec + extends MessageProducerSpec { private final SimpleMessageListenerContainer listenerContainer; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java index 8f7b8c7..220e9b7 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java @@ -107,7 +107,8 @@ public class QueueChannelSpec extends MessageChannelSpec, F extends Be @SuppressWarnings("unchecked") protected EndpointSpec(H handler) { + Assert.notNull(handler); try { Class fClass = ResolvableType.forClass(this.getClass()).as(EndpointSpec.class).resolveGenerics()[1]; F endpointFactoryBean = (F) fClass.newInstance(); @@ -53,6 +55,10 @@ public abstract class EndpointSpec, F extends Be public abstract S poller(PollerMetadata pollerMetadata); + public S poller(Function pollers) { + return poller(pollers.apply(new PollerFactory())); + } + public S poller(PollerSpec pollerMetadataSpec) { return this.poller(pollerMetadataSpec.get()); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/PollerFactory.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/PollerFactory.java new file mode 100644 index 0000000..204812f --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/PollerFactory.java @@ -0,0 +1,76 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.dsl.core; + +import java.util.TimeZone; +import java.util.concurrent.TimeUnit; + +import org.springframework.scheduling.Trigger; + +/** + * @author Artem Bilan + */ +public class PollerFactory { + + public PollerSpec trigger(Trigger trigger) { + return Pollers.trigger(trigger); + } + + public PollerSpec cron(String cronExpression) { + return Pollers.cron(cronExpression); + } + + public PollerSpec cron(String cronExpression, TimeZone timeZone) { + return Pollers.cron(cronExpression, timeZone); + } + + public PollerSpec fixedRate(long period) { + return Pollers.fixedRate(period); + } + + public PollerSpec fixedRate(long period, TimeUnit timeUnit) { + return Pollers.fixedRate(period, timeUnit); + } + + public PollerSpec fixedRate(long period, long initialDelay) { + return Pollers.fixedRate(period, initialDelay); + } + + public PollerSpec fixedDelay(long period, TimeUnit timeUnit, long initialDelay) { + return Pollers.fixedDelay(period, timeUnit, initialDelay); + } + + public PollerSpec fixedRate(long period, TimeUnit timeUnit, long initialDelay) { + return Pollers.fixedRate(period, timeUnit, initialDelay); + } + + public PollerSpec fixedDelay(long period, TimeUnit timeUnit) { + return Pollers.fixedDelay(period, timeUnit); + } + + public PollerSpec fixedDelay(long period, long initialDelay) { + return Pollers.fixedDelay(period, initialDelay); + } + + public PollerSpec fixedDelay(long period) { + return Pollers.fixedDelay(period); + } + + PollerFactory() { + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/PollerSpec.java similarity index 91% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/PollerSpec.java index 6e90d0a..ea7a346 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PollerSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/PollerSpec.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dsl.support; +package org.springframework.integration.dsl.core; import java.util.Arrays; import java.util.LinkedList; @@ -23,7 +23,6 @@ import java.util.concurrent.Executor; import org.aopalliance.aop.Advice; -import org.springframework.integration.dsl.core.IntegrationComponentSpec; import org.springframework.integration.scheduling.PollerMetadata; import org.springframework.integration.transaction.TransactionSynchronizationFactory; import org.springframework.scheduling.Trigger; @@ -46,7 +45,8 @@ public final class PollerSpec extends IntegrationComponentSpec remoteFileTemplate, FileExistsMode fileExistsMode) { + protected FileTransferringMessageHandlerSpec(RemoteFileTemplate remoteFileTemplate, + FileExistsMode fileExistsMode) { Constructor fileExistsModeConstructor = ClassUtils.getConstructorIfAvailable(FileTransferringMessageHandler.class, RemoteFileTemplate.class, FileExistsMode.class); @@ -64,7 +65,8 @@ public abstract class FileTransferringMessageHandlerSpec) fileExistsModeConstructor.newInstance(remoteFileTemplate, + this.target = + (FileTransferringMessageHandler) fileExistsModeConstructor.newInstance(remoteFileTemplate, fileExistsMode); } catch (Exception e) { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileInboundChannelAdapterSpec.java similarity index 94% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteInboundChannelAdapterSpec.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileInboundChannelAdapterSpec.java index be84053..5757073 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteInboundChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileInboundChannelAdapterSpec.java @@ -30,7 +30,7 @@ import org.springframework.util.Assert; /** * @author Artem Bilan */ -public abstract class RemoteInboundChannelAdapterSpec, +public abstract class RemoteFileInboundChannelAdapterSpec, MS extends AbstractInboundFileSynchronizingMessageSource> extends MessageSourceSpec implements ComponentsRegistration { @@ -38,7 +38,7 @@ public abstract class RemoteInboundChannelAdapterSpec filter; - protected RemoteInboundChannelAdapterSpec(AbstractInboundFileSynchronizer synchronizer) { + protected RemoteFileInboundChannelAdapterSpec(AbstractInboundFileSynchronizer synchronizer) { this.synchronizer = synchronizer; } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/Ftp.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/Ftp.java index 9ec3648..1c278df 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/Ftp.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/Ftp.java @@ -16,6 +16,7 @@ package org.springframework.integration.dsl.ftp; +import java.io.File; import java.util.Comparator; import org.apache.commons.net.ftp.FTPFile; @@ -36,7 +37,7 @@ public abstract class Ftp { } public static FtpInboundChannelAdapterSpec inboundAdapter(SessionFactory sessionFactory, - Comparator receptionOrderComparator) { + Comparator receptionOrderComparator) { return new FtpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpInboundChannelAdapterSpec.java index 046ba8d..b4c10dd 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpInboundChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ftp/FtpInboundChannelAdapterSpec.java @@ -21,7 +21,7 @@ import java.util.Comparator; import org.apache.commons.net.ftp.FTPFile; -import org.springframework.integration.dsl.file.RemoteInboundChannelAdapterSpec; +import org.springframework.integration.dsl.file.RemoteFileInboundChannelAdapterSpec; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; @@ -32,8 +32,8 @@ import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMe * @author Artem Bilan */ public class FtpInboundChannelAdapterSpec - extends RemoteInboundChannelAdapterSpec { + extends RemoteFileInboundChannelAdapterSpec { FtpInboundChannelAdapterSpec(SessionFactory sessionFactory, Comparator comparator) { super(new FtpInboundFileSynchronizer(sessionFactory)); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java index e1e9353..3841e74 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/Jms.java @@ -111,13 +111,13 @@ public abstract class Jms { public static JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec - messageDriverAdapter(ConnectionFactory connectionFactory) { - return messageDriverAdapter(connectionFactory, DefaultMessageListenerContainer.class); + messageDriverChannelAdapter(ConnectionFactory connectionFactory) { + return messageDriverChannelAdapter(connectionFactory, DefaultMessageListenerContainer.class); } public static JmsMessageDrivenChannelAdapterSpec.JmsMessageDrivenChannelAdapterListenerContainerSpec - messageDriverAdapter(ConnectionFactory connectionFactory, Class containerClass) { + messageDriverChannelAdapter(ConnectionFactory connectionFactory, Class containerClass) { try { JmsListenerContainerSpec spec = new JmsListenerContainerSpec(containerClass) .connectionFactory(connectionFactory); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java index 5fea5b9..b592c0e 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsDestinationAccessorSpec.java @@ -25,7 +25,8 @@ import org.springframework.jms.support.destination.JmsDestinationAccessor; /** * @author Artem Bilan */ -public abstract class JmsDestinationAccessorSpec, A extends JmsDestinationAccessor> +public abstract class + JmsDestinationAccessorSpec, A extends JmsDestinationAccessor> extends IntegrationComponentSpec { protected JmsDestinationAccessorSpec(A accessor) { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundChannelAdapterSpec.java index f22a62c..c56e10a 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundChannelAdapterSpec.java @@ -20,7 +20,7 @@ import javax.jms.ConnectionFactory; import javax.jms.Destination; import org.springframework.integration.dsl.core.MessageSourceSpec; -import org.springframework.integration.dsl.support.ComponentConfigurer; +import org.springframework.integration.dsl.support.Consumer; import org.springframework.integration.jms.JmsDestinationPollingSource; import org.springframework.integration.jms.JmsHeaderMapper; import org.springframework.jms.core.JmsTemplate; @@ -74,9 +74,9 @@ public class JmsInboundChannelAdapterSpec configurer) { + public JmsInboundChannelSpecTemplateAware configureJmsTemplate(Consumer configurer) { Assert.notNull(configurer); - configurer.configure(this.jmsTemplateSpec); + configurer.accept(this.jmsTemplateSpec); return _this(); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java index 0166f33..2f15757 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsInboundGatewaySpec.java @@ -19,7 +19,7 @@ package org.springframework.integration.dsl.jms; import javax.jms.Destination; import org.springframework.integration.dsl.core.MessagingGatewaySpec; -import org.springframework.integration.dsl.support.ComponentConfigurer; +import org.springframework.integration.dsl.support.Consumer; import org.springframework.integration.jms.ChannelPublishingJmsMessageListener; import org.springframework.integration.jms.JmsHeaderMapper; import org.springframework.jms.listener.AbstractMessageListenerContainer; @@ -124,10 +124,10 @@ public class JmsInboundGatewaySpec> return _this(); } - public JmsInboundGatewayListenerContainerSpec configureListenerContainer - (ComponentConfigurer> configurer) { + public JmsInboundGatewayListenerContainerSpec configureListenerContainer( + Consumer> configurer) { Assert.notNull(configurer); - configurer.configure(this.spec); + configurer.accept(this.spec); return _this(); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapterSpec.java index 6216bbe..b52c039 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsMessageDrivenChannelAdapterSpec.java @@ -19,7 +19,7 @@ package org.springframework.integration.dsl.jms; import javax.jms.Destination; import org.springframework.integration.dsl.core.MessageProducerSpec; -import org.springframework.integration.dsl.support.ComponentConfigurer; +import org.springframework.integration.dsl.support.Consumer; import org.springframework.integration.jms.ChannelPublishingJmsMessageListener; import org.springframework.integration.jms.JmsHeaderMapper; import org.springframework.jms.listener.AbstractMessageListenerContainer; @@ -53,7 +53,8 @@ public class JmsMessageDrivenChannelAdapterSpec extends + public static class + JmsMessageDrivenChannelAdapterListenerContainerSpec extends JmsMessageDrivenChannelAdapterSpec> { private final JmsListenerContainerSpec spec; @@ -74,10 +75,10 @@ public class JmsMessageDrivenChannelAdapterSpec configureListenerContainer - (ComponentConfigurer> configurer) { + public JmsMessageDrivenChannelAdapterListenerContainerSpec configureListenerContainer( + Consumer> configurer) { Assert.notNull(configurer); - configurer.configure(this.spec); + configurer.accept(this.spec); return _this(); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundChannelAdapterSpec.java index d65f150..e0e9200 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundChannelAdapterSpec.java @@ -20,7 +20,7 @@ import javax.jms.ConnectionFactory; import javax.jms.Destination; import org.springframework.integration.dsl.core.MessageHandlerSpec; -import org.springframework.integration.dsl.support.ComponentConfigurer; +import org.springframework.integration.dsl.support.Consumer; import org.springframework.integration.jms.JmsHeaderMapper; import org.springframework.integration.jms.JmsSendingMessageHandler; import org.springframework.jms.core.JmsTemplate; @@ -79,9 +79,9 @@ public class JmsOutboundChannelAdapterSpec configurer) { + public JmsOutboundChannelSpecTemplateAware configureJmsTemplate(Consumer configurer) { Assert.notNull(configurer); - configurer.configure(this.jmsTemplateSpec); + configurer.accept(this.jmsTemplateSpec); return _this(); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java index f0153e6..36931bc 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java @@ -23,7 +23,7 @@ import javax.jms.Destination; import org.springframework.integration.dsl.core.IntegrationComponentSpec; import org.springframework.integration.dsl.core.MessageHandlerSpec; -import org.springframework.integration.dsl.support.ComponentConfigurer; +import org.springframework.integration.dsl.support.Consumer; import org.springframework.integration.jms.JmsHeaderMapper; import org.springframework.integration.jms.JmsOutboundGateway; import org.springframework.jms.support.converter.MessageConverter; @@ -140,10 +140,10 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec configurer) { + public JmsOutboundGatewaySpec replyContainer(Consumer configurer) { Assert.notNull(configurer); ReplyContainerSpec spec = new ReplyContainerSpec(); - configurer.configure(spec); + configurer.accept(spec); this.target.setReplyContainerProperties(spec.get()); return _this(); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java index 75707e9..0cbd37b 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/Mail.java @@ -35,6 +35,10 @@ public class Mail { return new Pop3MailInboundChannelAdapterSpec(url); } + public static Pop3MailInboundChannelAdapterSpec pop3InboundAdapter(String host, String username, String password) { + return pop3InboundAdapter(host, -1, username, password); + } + public static Pop3MailInboundChannelAdapterSpec pop3InboundAdapter(String host, int port, String username, String password) { return new Pop3MailInboundChannelAdapterSpec(host, port, username, password); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/Sftp.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/Sftp.java index 2f862cc..9bab536 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/Sftp.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/Sftp.java @@ -16,16 +16,17 @@ package org.springframework.integration.dsl.sftp; +import java.io.File; import java.util.Comparator; -import com.jcraft.jsch.ChannelSftp; - import org.springframework.integration.file.remote.RemoteFileTemplate; import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.support.FileExistsMode; import org.springframework.integration.sftp.gateway.SftpOutboundGateway; +import com.jcraft.jsch.ChannelSftp; + /** * @author Artem Bilan */ @@ -36,7 +37,7 @@ public abstract class Sftp { } public static SftpInboundChannelAdapterSpec inboundAdapter(SessionFactory sessionFactory, - Comparator receptionOrderComparator) { + Comparator receptionOrderComparator) { return new SftpInboundChannelAdapterSpec(sessionFactory, receptionOrderComparator); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpInboundChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpInboundChannelAdapterSpec.java index 3295bfe..ecac1b0 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpInboundChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpInboundChannelAdapterSpec.java @@ -21,7 +21,7 @@ import java.util.Comparator; import com.jcraft.jsch.ChannelSftp; -import org.springframework.integration.dsl.file.RemoteInboundChannelAdapterSpec; +import org.springframework.integration.dsl.file.RemoteFileInboundChannelAdapterSpec; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter; @@ -32,8 +32,8 @@ import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizing * @author Artem Bilan */ public class SftpInboundChannelAdapterSpec - extends RemoteInboundChannelAdapterSpec { + extends RemoteFileInboundChannelAdapterSpec { SftpInboundChannelAdapterSpec(SessionFactory sessionFactory, Comparator comparator) { super(new SftpInboundFileSynchronizer(sessionFactory)); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpMessageHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpMessageHandlerSpec.java index 5e86d51..69a203c 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpMessageHandlerSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpMessageHandlerSpec.java @@ -16,17 +16,18 @@ package org.springframework.integration.dsl.sftp; -import com.jcraft.jsch.ChannelSftp; - import org.springframework.integration.dsl.file.FileTransferringMessageHandlerSpec; import org.springframework.integration.file.remote.RemoteFileTemplate; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.support.FileExistsMode; +import com.jcraft.jsch.ChannelSftp; + /** * @author Artem Bilan */ -public class SftpMessageHandlerSpec extends FileTransferringMessageHandlerSpec { +public class SftpMessageHandlerSpec + extends FileTransferringMessageHandlerSpec { SftpMessageHandlerSpec(SessionFactory sessionFactory) { super(sessionFactory); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.java index e0ed014..baba636 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/sftp/SftpOutboundGatewaySpec.java @@ -16,17 +16,18 @@ package org.springframework.integration.dsl.sftp; -import com.jcraft.jsch.ChannelSftp; - import org.springframework.integration.dsl.file.RemoteFileOutboundGatewaySpec; import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter; +import com.jcraft.jsch.ChannelSftp; + /** * @author Artem Bilan */ -public class SftpOutboundGatewaySpec extends RemoteFileOutboundGatewaySpec { +public class SftpOutboundGatewaySpec + extends RemoteFileOutboundGatewaySpec { SftpOutboundGatewaySpec(AbstractRemoteFileOutboundGateway outboundGateway) { diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/ComponentConfigurer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Consumer.java similarity index 65% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/ComponentConfigurer.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Consumer.java index 9f65d1a..01089fe 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/ComponentConfigurer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Consumer.java @@ -16,14 +16,20 @@ package org.springframework.integration.dsl.support; -import org.springframework.integration.dsl.core.IntegrationComponentSpec; - /** - * @author Artem Bilan - + * Implementations accept a given value and perform work on the argument. + * + * @param the type of values to accept + * + * @author Jon Brisbin + * @author Stephane Maldini */ -public interface ComponentConfigurer> { +public interface Consumer { - void configure(S spec); + /** + * Execute the logic of the action, accepting the given parameter. + * @param t The parameter to pass to the consumer. + */ + void accept(T t); } diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Function.java similarity index 53% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Function.java index d22caa5..b263598 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/EndpointConfigurer.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Function.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -16,14 +16,24 @@ package org.springframework.integration.dsl.support; -import org.springframework.integration.dsl.core.EndpointSpec; - /** - * @author Artem Bilan - + * Implementations of this class perform work on the given parameter + * and return a result of an optionally different type. + * + * @param The type of the input to the apply operation + * @param The type of the result of the apply operation + * + * @author Jon Brisbin + * @author Stephane Maldini */ -public interface EndpointConfigurer> { +public interface Function { - void configure(S spec); + /** + * Execute the logic of the action, accepting the given parameter. + * @param t The parameter to pass to the action. + * @return result + */ + R apply(T t); } + diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Transformers.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Transformers.java index 993c98d..bf66a6d 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Transformers.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Transformers.java @@ -27,7 +27,7 @@ import org.springframework.core.serializer.Deserializer; import org.springframework.core.serializer.Serializer; import org.springframework.expression.Expression; import org.springframework.expression.spel.standard.SpelExpressionParser; -import org.springframework.integration.dsl.tuple.Tuple2; +import org.springframework.integration.dsl.support.tuple.Tuple2; import org.springframework.integration.file.transformer.FileToByteArrayTransformer; import org.springframework.integration.file.transformer.FileToStringTransformer; import org.springframework.integration.json.JsonToObjectTransformer; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple.java similarity index 96% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple.java index cfcc63e..5576982 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dsl.tuple; +package org.springframework.integration.dsl.support.tuple; import java.io.Serializable; import java.util.Arrays; @@ -24,7 +24,8 @@ import java.util.Iterator; import org.springframework.util.Assert; /** - * A {@literal Tuple} is an immutable {@link java.util.Collection} of objects, each of which can be of an arbitrary type. + * A {@literal Tuple} is an immutable {@link java.util.Collection} of objects, + * each of which can be of an arbitrary type. * * @author Jon Brisbin * @author Stephane Maldini diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple1.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple1.java similarity index 94% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple1.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple1.java index 3d284aa..66166ba 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple1.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple1.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.dsl.tuple; +package org.springframework.integration.dsl.support.tuple; /** * A tuple that holds a single value diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple2.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple2.java similarity index 90% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple2.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple2.java index 43a3cd3..1098936 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple2.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/Tuple2.java @@ -14,13 +14,13 @@ * limitations under the License. */ -package org.springframework.integration.dsl.tuple; +package org.springframework.integration.dsl.support.tuple; /** * A tuple that holds two values * * @param The type of the first value held by this tuple - * @param The type of the second balue held by this tuple + * @param The type of the second value held by this tuple * * @author Jon Brisbin */ diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/package-info.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/package-info.java similarity index 55% rename from spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/package-info.java rename to spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/package-info.java index a55a4a2..65ad5a2 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/package-info.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/tuple/package-info.java @@ -1,4 +1,4 @@ /** * Tuples provide a type-safe way to specify multiple parameters. */ -package org.springframework.integration.dsl.tuple; +package org.springframework.integration.dsl.support.tuple; diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java index d4816d9..81d76c7 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/flows/IntegrationFlowTests.java @@ -101,8 +101,12 @@ import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.config.GlobalChannelInterceptor; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.core.MessageSource; +import org.springframework.integration.dsl.Channels; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.MessageProducers; +import org.springframework.integration.dsl.MessageSources; +import org.springframework.integration.dsl.MessagingGateways; import org.springframework.integration.dsl.amqp.Amqp; import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; @@ -110,7 +114,7 @@ import org.springframework.integration.dsl.file.Files; import org.springframework.integration.dsl.ftp.Ftp; import org.springframework.integration.dsl.jms.Jms; import org.springframework.integration.dsl.sftp.Sftp; -import org.springframework.integration.dsl.support.Pollers; +import org.springframework.integration.dsl.core.Pollers; import org.springframework.integration.dsl.support.Transformers; import org.springframework.integration.dsl.test.TestFtpServer; import org.springframework.integration.dsl.test.TestSftpServer; @@ -119,6 +123,7 @@ import org.springframework.integration.event.core.MessagingEvent; import org.springframework.integration.event.outbound.ApplicationEventPublishingMessageHandler; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.FileWritingMessageHandler; import org.springframework.integration.file.remote.RemoteFileTemplate; import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway; import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer; @@ -163,7 +168,6 @@ import org.springframework.util.StreamUtils; import com.jcraft.jsch.ChannelSftp; import com.mongodb.MongoClient; - import de.flapdoodle.embed.mongo.MongodExecutable; import de.flapdoodle.embed.mongo.MongodStarter; import de.flapdoodle.embed.mongo.config.MongodConfigBuilder; @@ -442,8 +446,7 @@ public class IntegrationFlowTests { assertEquals("test", reply.getPayload()); assertTrue(this.beanFactory.containsBean("bridgeFlow2.channel#0")); - assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0"), instanceOf(FixedSubscriberChannel - .class)); + assertThat(this.beanFactory.getBean("bridgeFlow2.channel#0"), instanceOf(FixedSubscriberChannel.class)); try { this.bridgeFlow2Input.send(message); @@ -1357,7 +1360,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow jmsOutboundFlow() { - return f -> f.handle(Jms.outboundAdapter(this.jmsConnectionFactory) + return f -> f.handleWithAdapter(h -> h.jms(this.jmsConnectionFactory) .destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER)); } @@ -1369,8 +1372,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow jmsInboundFlow() { return IntegrationFlows - .from(Jms.inboundAdapter(this.jmsConnectionFactory) - .destination("jmsInbound")) + .from((MessageSources s) -> s.jms(this.jmsConnectionFactory).destination("jmsInbound")) .transform(String::toUpperCase) .channel(this.jmsOutboundInboundReplyChannel()) .get(); @@ -1379,7 +1381,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow jmsMessageDriverFlow() { return IntegrationFlows - .from(Jms.messageDriverAdapter(this.jmsConnectionFactory) + .from(Jms.messageDriverChannelAdapter(this.jmsConnectionFactory) .destination("jmsMessageDriver")) .transform(String::toLowerCase) .channel(this.jmsOutboundInboundReplyChannel()) @@ -1388,14 +1390,14 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow jmsOutboundGatewayFlow() { - return f -> f.handle(Jms.outboundGateway(this.jmsConnectionFactory) + return f -> f.handleWithAdapter(a -> a.jmsGateway(this.jmsConnectionFactory) .replyContainer() .requestDestination("jmsPipelineTest")); } @Bean public IntegrationFlow jmsInboundGatewayFlow() { - return IntegrationFlows.from(Jms.inboundGateway(this.jmsConnectionFactory) + return IntegrationFlows.from((MessagingGateways g) -> g.jms(this.jmsConnectionFactory) .destination("jmsPipelineTest")) .transform(String::toUpperCase) .get(); @@ -1416,7 +1418,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow ftpInboundFlow() { return IntegrationFlows - .from(Ftp.inboundAdapter(this.ftpSessionFactory) + .from(s -> s.ftp(this.ftpSessionFactory) .preserveTimestamp(true) .remoteDirectory("ftpSource") .regexFilter(".*\\.txt$") @@ -1430,7 +1432,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow sftpInboundFlow() { return IntegrationFlows - .from(Sftp.inboundAdapter(this.sftpSessionFactory) + .from(s -> s.sftp(this.sftpSessionFactory) .preserveTimestamp(true) .remoteDirectory("sftpSource") .regexFilter(".*\\.txt$") @@ -1482,15 +1484,18 @@ public class IntegrationFlowTests { .channel(remoteFileOutputChannel()) .get(); } + @Bean public IntegrationFlow sftpMGetFlow() { return IntegrationFlows.from("sftpMgetInputChannel") - .handle(Sftp.outboundGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, - "payload") - .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) - .regexFileNameFilter("(subSftpSource|.*1.txt)") - .localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory") - .localFilenameGeneratorExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')")) + .handleWithAdapter(h -> + h.sftpGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET, + "payload") + .options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE) + .regexFileNameFilter("(subSftpSource|.*1.txt)") + .localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory") + .localFilenameGeneratorExpression( + "#remoteFileName.replaceFirst('sftpSource', 'localTarget')")) .channel(remoteFileOutputChannel()) .get(); } @@ -1559,7 +1564,8 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) { - return IntegrationFlows.from(MessageChannels.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup")) + return IntegrationFlows.from((Channels c) -> + c.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup")) .bridge(s -> s.poller(Pollers.fixedDelay(100)) .autoStartup(false) .id("priorityChannelBridge")) @@ -1627,7 +1633,7 @@ public class IntegrationFlowTests { .bridge(c -> c.autoStartup(false).id("bridge")) .fixedSubscriberChannel() .delay("delayer", "200", c -> c.advice(this.delayedAdvice).messageStore(this.messageStore())) - .channel(MessageChannels.queue("bridgeFlow2Output")) + .channel(c -> c.queue("bridgeFlow2Output")) .get(); } @@ -1695,8 +1701,8 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow fileFlow1() { return IntegrationFlows.from("fileFlow1Input") - .handle(Files.outboundAdapter(tmpDir).fileNameGenerator(message -> null), - c -> c.id("fileWriting")) + .handleWithAdapter(h -> h.file(tmpDir).fileNameGenerator(message -> null) + , c -> c.id("fileWriting")) .get(); } @@ -1722,7 +1728,7 @@ public class IntegrationFlowTests { @Bean @DependsOn("enrichFlow") public IntegrationFlow enricherFlow() { - return IntegrationFlows.fromFixedMessageChannel("enricherInput") + return IntegrationFlows.from("enricherInput", true) .enrich(e -> e.requestChannel("enrichChannel") .requestPayloadExpression("payload") .shouldClonePayload(false) @@ -1736,7 +1742,7 @@ public class IntegrationFlowTests { @Bean @DependsOn("enrichFlow") public IntegrationFlow enricherFlow2() { - return IntegrationFlows.fromFixedMessageChannel("enricherInput2") + return IntegrationFlows.from("enricherInput2", true) .enrich(e -> e.requestChannel("enrichChannel") .requestPayloadExpression("payload") .shouldClonePayload(false) @@ -1749,7 +1755,7 @@ public class IntegrationFlowTests { @Bean @DependsOn("enrichFlow") public IntegrationFlow enricherFlow3() { - return IntegrationFlows.fromFixedMessageChannel("enricherInput3") + return IntegrationFlows.from("enricherInput3", true) .enrich(e -> e.requestChannel("enrichChannel") .requestPayloadExpression("payload") .shouldClonePayload(false) @@ -1787,11 +1793,11 @@ public class IntegrationFlowTests { public IntegrationFlow splitResequenceFlow() { return f -> f.enrichHeaders(s -> s.header("FOO", "BAR")) .split("testSplitterData", "buildList", c -> c.applySequence(false)) - .channel(MessageChannels.executor(this.taskExecutor())) + .channel(c -> c.executor(this.taskExecutor())) .split(Message.class, target -> (List) target.getPayload(), c -> c.applySequence(false)) .channel(MessageChannels.executor(this.taskExecutor())) .split(s -> s.applySequence(false).get().getT2().setDelimiters(",")) - .channel(MessageChannels.executor(this.taskExecutor())) + .channel(c -> c.executor(this.taskExecutor())) .transform(Integer::parseInt) .enrichHeaders(s -> s.headerExpression(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, "payload")) .resequence(r -> r.releasePartialSequences(true).correlationExpression("'foo'"), null) @@ -1800,7 +1806,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow splitAggregateFlow() { - return IntegrationFlows.fromFixedMessageChannel("splitAggregateInput") + return IntegrationFlows.from("splitAggregateInput", true) .split(null) .channel(MessageChannels.executor(this.taskExecutor())) .resequence() @@ -1812,8 +1818,10 @@ public class IntegrationFlowTests { public IntegrationFlow xpathHeaderEnricherFlow() { return IntegrationFlows.from("xpathHeaderEnricherInput") .enrichHeaders( - s -> s.header("one", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne")) - .header("two", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo")) + s -> s.header("one", + new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne")) + .header("two", + new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo")) .headerChannelsToString(), c -> c.autoStartup(false).id("xpathHeaderEnricher") ) @@ -1874,11 +1882,11 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow tailFlow() { - return IntegrationFlows.from(Files.tailAdapter(new File(tmpDir, "TailTest")) - .delay(500) - .end(false) - .id("tailer") - .autoStartup(false)) + return IntegrationFlows.from((MessageProducers p) -> p.tail(new File(tmpDir, "TailTest")) + .delay(500) + .end(false) + .id("tailer") + .autoStartup(false)) .transform("hello "::concat) .channel(MessageChannels.queue("tailChannel")) .get(); @@ -1922,7 +1930,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow amqpInboundFlow() { - return IntegrationFlows.from(Amqp.inboundAdapter(this.rabbitConnectionFactory, fooQueue())) + return IntegrationFlows.from((MessageProducers p) -> p.amqp(this.rabbitConnectionFactory, fooQueue())) .transform(String.class, String::toUpperCase) .channel(Amqp.pollableChannel(this.rabbitConnectionFactory) .queueName("amqpReplyChannel") @@ -1955,7 +1963,7 @@ public class IntegrationFlowTests { @Bean public IntegrationFlow fileReadingFlow() { return IntegrationFlows - .from(Files.inboundAdapter(tmpDir).patternFilter("*.sitest"), + .from(s -> s.file(tmpDir).patternFilter("*.sitest"), e -> e.poller(Pollers.fixedDelay(100))) .transform(Transformers.fileToString()) .aggregate(a -> a.correlationExpression("1") diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java index 68e2b75..94f87d6 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java @@ -48,9 +48,10 @@ import org.springframework.context.annotation.Configuration; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.MessageProducers; import org.springframework.integration.dsl.channel.MessageChannels; import org.springframework.integration.dsl.mail.Mail; -import org.springframework.integration.dsl.support.Pollers; +import org.springframework.integration.dsl.core.Pollers; import org.springframework.integration.dsl.test.mail.PoorMansMailServer.ImapServer; import org.springframework.integration.dsl.test.mail.PoorMansMailServer.Pop3Server; import org.springframework.integration.dsl.test.mail.PoorMansMailServer.SmtpServer; @@ -197,7 +198,7 @@ public class MailTests { public IntegrationFlow sendMailFlow() { return IntegrationFlows.from("sendMailChannel") .enrichHeaders(Mail.headers().subject("foo").from("foo@bar").to("bar@baz")) - .handle(Mail.outboundAdapter("localhost") + .handleWithAdapter(h -> h.mail("localhost") .port(smtpPort) .credentials("user", "pw") .protocol("smtp") @@ -209,10 +210,9 @@ public class MailTests { @Bean public IntegrationFlow pop3MailFlow() { return IntegrationFlows - .from(Mail.pop3InboundAdapter("localhost", pop3Port, "user", "pw") + .from(s -> s.pop3("localhost", pop3Port, "user", "pw") .javaMailProperties(p -> p.put("mail.debug", "true")), - e -> e.autoStartup(true) - .poller(Pollers.fixedDelay(1000))) + e -> e.autoStartup(true).poller(p -> p.fixedDelay(1000))) .enrichHeaders(s -> s.headerExpressions(c -> c.put(MailHeaders.SUBJECT, "payload.subject") .put(MailHeaders.FROM, "payload.from[0].toString()"))) .channel(MessageChannels.queue("pop3Channel")) @@ -222,11 +222,11 @@ public class MailTests { @Bean public IntegrationFlow imapMailFlow() { return IntegrationFlows - .from(Mail.imapInboundAdapter("imap://user:pw@localhost:" + imapPort + "/INBOX") + .from(s -> s.imap("imap://user:pw@localhost:" + imapPort + "/INBOX") .searchTermStrategy(this::fromAndNotSeenTerm) .javaMailProperties(p -> p.put("mail.debug", "true")), e -> e.autoStartup(true) - .poller(Pollers.fixedDelay(1000))) + .poller(p -> p.fixedDelay(1000))) .channel(MessageChannels.queue("imapChannel")) .get(); } @@ -234,7 +234,7 @@ public class MailTests { @Bean public IntegrationFlow imapIdleFlow() { return IntegrationFlows - .from(Mail.imapIdleAdapter("imap://user:pw@localhost:" + imapIdlePort + "/INBOX") + .from((MessageProducers mp) -> mp.imap("imap://user:pw@localhost:" + imapIdlePort + "/INBOX") .searchTermStrategy(this::fromAndNotSeenTerm) .javaMailProperties(p -> p.put("mail.debug", "true") .put("mail.imap.connectionpoolsize", "5")) diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java index 7dbfd7b..87d6008 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/PoorMansMailServer.java @@ -276,7 +276,8 @@ public class PoorMansMailServer { + "((\"Bar\" NIL \"bar\" \"baz.net\")) NIL NIL " + "\"<4DA0A7E4.3010506@baz.net>\" " + "\"\") " - + "BODYSTRUCTURE (\"TEXT\" \"PLAIN\" (\"CHARSET\" \"ISO-8859-1\") NIL NIL \"7BIT\" 1176 43)))"); + + "BODYSTRUCTURE (\"TEXT\" \"PLAIN\" (\"CHARSET\" \"ISO-8859-1\") NIL NIL " + + "\"7BIT\" 1176 43)))"); write(tag + "OK FETCH completed"); } else if (line.contains("STORE 1 +FLAGS (\\Flagged)")) {