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

DSL: Introduce SubFlows

Browse Source 
Add `subFlow` support for `router`s
Move some tests to separate domain classes
Fix some component registrations

DSL: Fix `SubFlow` registration according PR discussion

* Add IO plugin
* Fix SI 4.0 <-> 4.1 compatibility by `springIoCheck` results
* Further divide for tests to separate domain classes

DSL: Add `discardFlow` to the `.filter()`

Make `.router()` `subFlow`s as `always return to the main flow`

Add `.publishSubscribeChannel()` EIP-method with `.subscriber()`

to specify subscribers as `subFlow`s

Polishing
This commit is contained in:
Artem Bilan
2014-10-29 23:50:42 +02:00
committed by Gary Russell
parent 74102a022d
commit be9767b8a9
22 changed files with 1841 additions and 960 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/Channels.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.dsl;
import java.util.concurrent.BlockingQueue;
import java.util.Queue;
import java.util.concurrent.Executor;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -68,11 +68,11 @@ public class Channels {
return MessageChannels.queue(id, capacity);
}
public QueueChannelSpec queue(BlockingQueue<Message<?>> queue) {
public QueueChannelSpec queue(Queue<Message<?>> queue) {
return MessageChannels.queue(queue);
}
public QueueChannelSpec queue(String id, BlockingQueue<Message<?>> queue) {
public QueueChannelSpec queue(String id, Queue<Message<?>> queue) {
return MessageChannels.queue(id, queue);
}
@@ -110,14 +110,23 @@ public class Channels {
return MessageChannels.rendezvous(id);
}
public PublishSubscribeChannelSpec publishSubscribe() {
public PublishSubscribeChannelSpec<? extends PublishSubscribeChannelSpec<?>> publishSubscribe() {
return MessageChannels.publishSubscribe();
}
public PublishSubscribeChannelSpec publishSubscribe(Executor executor) {
public PublishSubscribeChannelSpec<? extends PublishSubscribeChannelSpec<?>> publishSubscribe(Executor executor) {
return MessageChannels.publishSubscribe(executor);
}
public PublishSubscribeChannelSpec<? extends PublishSubscribeChannelSpec<?>> publishSubscribe(String id,
Executor executor) {
return MessageChannels.publishSubscribe(id, executor);
}
public PublishSubscribeChannelSpec<? extends PublishSubscribeChannelSpec<?>> publishSubscribe(String id) {
return MessageChannels.publishSubscribe(id);
}
public ExecutorChannelSpec executor(Executor executor) {
return MessageChannels.executor(executor);
}
@@ -126,14 +135,6 @@ public class Channels {
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<? extends AmqpPollableMessageChannelSpec<?>> amqpPollable(
ConnectionFactory connectionFactory) {
return Amqp.pollableChannel(connectionFactory);

28
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java
View File

@@ -16,14 +16,23 @@
package org.springframework.integration.dsl;
import java.util.Collection;
import java.util.Collections;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.dsl.core.ConsumerEndpointSpec;
import org.springframework.integration.filter.MessageFilter;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
* @author Artem Bilan
*/
public final class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpointSpec, MessageFilter> {
public final class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpointSpec, MessageFilter>
implements ComponentsRegistration {
private IntegrationFlow discardFlow;
FilterEndpointSpec(MessageFilter messageFilter) {
super(messageFilter);
@@ -44,9 +53,26 @@ public final class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpoin
return _this();
}
public FilterEndpointSpec discardFlow(IntegrationFlow discardFlow) {
Assert.notNull(discardFlow);
DirectChannel channel = new DirectChannel();
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
discardFlow.accept(flowBuilder);
this.discardFlow = flowBuilder.get();
return discardChannel(channel);
}
public FilterEndpointSpec discardWithinAdvice(boolean discardWithinAdvice) {
this.target.getT2().setDiscardWithinAdvice(discardWithinAdvice);
return _this();
}
@Override
public Collection<Object> getComponentsToRegister() {
if (this.discardFlow != null) {
return Collections.<Object>singletonList(this.discardFlow);
}
return null;
}
}

46
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
View File

@@ -16,56 +16,14 @@
package org.springframework.integration.dsl;
import java.util.Set;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.integration.dsl.support.FixedSubscriberChannelPrototype;
/**
* @author Artem Bilan
*/
public final class IntegrationFlowBuilder extends IntegrationFlowDefinition<IntegrationFlowBuilder> {
@Override
public StandardIntegrationFlow get() {
if (this.currentMessageChannel instanceof FixedSubscriberChannelPrototype) {
throw new BeanCreationException("The 'currentMessageChannel' (" + this.currentMessageChannel +
") is a prototype for FixedSubscriberChannel which can't be created without MessageHandler " +
"constructor argument. That means that '.fixedSubscriberChannel()' can't be the last EIP-method " +
"in the IntegrationFlow definition.");
}
if (this.integrationComponents.size() == 1) {
if (this.currentComponent != null) {
if (this.currentComponent instanceof SourcePollingChannelAdapterSpec) {
throw new BeanCreationException("The 'SourcePollingChannelAdapter' (" + this.currentComponent
+ ") " + "must be configured with at least one 'MessageChanel' or 'MessageHandler'.");
}
}
else if (this.currentMessageChannel != null) {
throw new BeanCreationException("The 'IntegrationFlow' can't consist of only one 'MessageChannel'. " +
"Add at lest '.bridge()' EIP-method before the end of flow.");
}
}
return new StandardIntegrationFlow(this.integrationComponents);
}
public static final class StandardIntegrationFlow implements IntegrationFlow {
private final Set<Object> integrationComponents;
StandardIntegrationFlow(Set<Object> integrationComponents) {
this.integrationComponents = integrationComponents;
}
public Set<Object> getIntegrationComponents() {
return integrationComponents;
}
@Override
public void accept(IntegrationFlowDefinition<?> flow) {
throw new UnsupportedOperationException();
}
return super.get();
}
}

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

@@ -20,6 +20,7 @@ import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executor;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
@@ -35,6 +36,7 @@ import org.springframework.integration.config.SourcePollingChannelAdapterFactory
import org.springframework.integration.core.GenericSelector;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.dsl.channel.MessageChannelSpec;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.dsl.core.ConsumerEndpointSpec;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.BeanNameMessageProcessor;
@@ -77,6 +79,7 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
/**
@@ -116,24 +119,15 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
public B fixedSubscriberChannel() {
return this.fixedSubscriberChannel(null);
return fixedSubscriberChannel(null);
}
public B fixedSubscriberChannel(String messageChannelName) {
return this.channel(new FixedSubscriberChannelPrototype(messageChannelName));
return channel(new FixedSubscriberChannelPrototype(messageChannelName));
}
public B channel(String messageChannelName) {
return this.channel(new MessageChannelReference(messageChannelName));
}
public B channel(MessageChannel messageChannel) {
Assert.notNull(messageChannel);
if (this.currentMessageChannel != null) {
this.register(new GenericEndpointSpec<BridgeHandler>(new BridgeHandler()), null);
}
this.currentMessageChannel = messageChannel;
return this.registerOutputChannelIfCan(this.currentMessageChannel);
return channel(new MessageChannelReference(messageChannelName));
}
public B channel(Function<Channels, MessageChannelSpec<?, ?>> channels) {
@@ -143,7 +137,28 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
public B channel(MessageChannelSpec<?, ?> messageChannelSpec) {
Assert.notNull(messageChannelSpec);
return this.channel(messageChannelSpec.get());
return channel(messageChannelSpec.get());
}
public B channel(MessageChannel messageChannel) {
Assert.notNull(messageChannel);
if (this.currentMessageChannel != null) {
this.register(new GenericEndpointSpec<BridgeHandler>(new BridgeHandler()), null);
}
this.currentMessageChannel = messageChannel;
return registerOutputChannelIfCan(this.currentMessageChannel);
}
public B publishSubscribeChannel(Consumer<PublishSubscribeSpec> publishSubscribeChannelConfigurer) {
return publishSubscribeChannel(null, publishSubscribeChannelConfigurer);
}
public B publishSubscribeChannel(Executor executor,
Consumer<PublishSubscribeSpec> publishSubscribeChannelConfigurer) {
Assert.notNull(publishSubscribeChannelConfigurer);
PublishSubscribeSpec spec = new PublishSubscribeSpec(executor);
publishSubscribeChannelConfigurer.accept(spec);
return addComponents(spec.getComponentsToRegister()).channel(spec);
}
public B controlBus() {
@@ -269,6 +284,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
public <H extends MessageHandler> B handle(MessageHandlerSpec<?, H> messageHandlerSpec,
Consumer<GenericEndpointSpec<H>> endpointConfigurer) {
Assert.notNull(messageHandlerSpec);
if (messageHandlerSpec instanceof ComponentsRegistration) {
addComponents(((ComponentsRegistration) messageHandlerSpec).getComponentsToRegister());
}
return handle(messageHandlerSpec.get(), endpointConfigurer);
}
@@ -422,7 +440,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
/**
* Provides the {@link HeaderFilter} to the current {@link IntegrationFlowBuilder.StandardIntegrationFlow}.
* Provides the {@link HeaderFilter} to the current {@link StandardIntegrationFlow}.
* @param headersToRemove the array of headers (or patterns)
* to remove from {@link org.springframework.messaging.MessageHeaders}.
* @return this {@link IntegrationFlowDefinition}.
@@ -432,7 +450,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
/**
* Provides the {@link HeaderFilter} to the current {@link IntegrationFlowBuilder.StandardIntegrationFlow}.
* Provides the {@link HeaderFilter} to the current {@link StandardIntegrationFlow}.
* @param headersToRemove the comma separated headers (or patterns) to remove from
* {@link org.springframework.messaging.MessageHeaders}.
* @param patternMatch the {@code boolean} flag to indicate if {@code headersToRemove}
@@ -567,17 +585,48 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
MethodInvokingRouter methodInvokingRouter = isLambda(router)
? new MethodInvokingRouter(new LambdaMessageProcessor(router, payloadType))
: new MethodInvokingRouter(router);
return this.route(methodInvokingRouter, routerConfigurer, endpointConfigurer);
return route(methodInvokingRouter, routerConfigurer, endpointConfigurer);
}
public <R extends AbstractMappingMessageRouter> B route(R router,
Consumer<RouterSpec<R>> routerConfigurer,
Consumer<GenericEndpointSpec<R>> endpointConfigurer) {
Collection<Object> componentsToRegister = null;
if (routerConfigurer != null) {
RouterSpec<R> routerSpec = new RouterSpec<R>(router);
routerConfigurer.accept(routerSpec);
componentsToRegister = routerSpec.getComponentsToRegister();
}
return this.route(router, endpointConfigurer);
route(router, endpointConfigurer);
final MessageChannel afterRouterChannel = new DirectChannel();
boolean hasSubFlows = false;
if (!CollectionUtils.isEmpty(componentsToRegister)) {
for (Object component : componentsToRegister) {
if (component instanceof IntegrationFlowDefinition) {
hasSubFlows = true;
IntegrationFlowDefinition<?> flowBuilder = (IntegrationFlowDefinition<?>) component;
addComponent(flowBuilder.fixedSubscriberChannel()
.bridge(new Consumer<GenericEndpointSpec<BridgeHandler>>() {
@Override
public void accept(GenericEndpointSpec<BridgeHandler> bridge) {
bridge.get().getT2().setOutputChannel(afterRouterChannel);
}
})
.get());
}
else {
addComponent(component);
}
}
}
if (hasSubFlows) {
channel(afterRouterChannel);
}
return _this();
}
public B routeToRecipients(Consumer<RecipientListRouterSpec> routerConfigurer) {
@@ -626,6 +675,9 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
if (endpointConfigurer != null) {
endpointConfigurer.accept(endpointSpec);
}
if (endpointSpec instanceof ComponentsRegistration) {
addComponents(((ComponentsRegistration) endpointSpec).getComponentsToRegister());
}
MessageChannel inputChannel = this.currentMessageChannel;
this.currentMessageChannel = null;
if (inputChannel == null) {
@@ -742,4 +794,27 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
}
protected StandardIntegrationFlow get() {
if (this.currentMessageChannel instanceof FixedSubscriberChannelPrototype) {
throw new BeanCreationException("The 'currentMessageChannel' (" + this.currentMessageChannel +
") is a prototype for FixedSubscriberChannel which can't be created without MessageHandler " +
"constructor argument. That means that '.fixedSubscriberChannel()' can't be the last EIP-method " +
"in the IntegrationFlow definition.");
}
if (this.integrationComponents.size() == 1) {
if (this.currentComponent != null) {
if (this.currentComponent instanceof SourcePollingChannelAdapterSpec) {
throw new BeanCreationException("The 'SourcePollingChannelAdapter' (" + this.currentComponent
+ ") " + "must be configured with at least one 'MessageChanel' or 'MessageHandler'.");
}
}
else if (this.currentMessageChannel != null) {
throw new BeanCreationException("The 'IntegrationFlow' can't consist of only one 'MessageChannel'. " +
"Add at lest '.bridge()' EIP-method before the end of flow.");
}
}
return new StandardIntegrationFlow(this.integrationComponents);
}
}

60
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java Normal file
View File

@@ -0,0 +1,60 @@
/*
* 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.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.Executor;
import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec;
import org.springframework.integration.dsl.core.ComponentsRegistration;
/**
* @author Artem Bilan
*/
public class PublishSubscribeSpec extends PublishSubscribeChannelSpec<PublishSubscribeSpec>
implements ComponentsRegistration {
private final List<Object> subscriberFlows = new ArrayList<Object>();
PublishSubscribeSpec() {
super();
}
PublishSubscribeSpec(Executor executor) {
super(executor);
}
@Override
public PublishSubscribeSpec id(String id) {
return super.id(id);
}
public PublishSubscribeSpec subscribe(IntegrationFlow flow) {
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(this.channel);
flow.accept(flowBuilder);
this.subscriberFlows.add(flowBuilder.get());
return _this();
}
@Override
public Collection<Object> getComponentsToRegister() {
return this.subscriberFlows;
}
}

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

@@ -16,12 +16,35 @@
package org.springframework.integration.dsl;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import javax.annotation.PostConstruct;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.router.AbstractMappingMessageRouter;
import org.springframework.integration.router.MappingMessageRouterManagement;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* @author Artem Bilan
*/
public final class RouterSpec<R extends AbstractMappingMessageRouter> extends AbstractRouterSpec<RouterSpec<R>, R> {
public final class RouterSpec<R extends AbstractMappingMessageRouter> extends AbstractRouterSpec<RouterSpec<R>, R>
implements ComponentsRegistration {
private final List<Object> subFlows = new ArrayList<Object>();
private String prefix;
private String suffix;
private RouterSubFlowMappingProvider mappingProvider;
RouterSpec(R router) {
super(router);
@@ -33,18 +56,73 @@ public final class RouterSpec<R extends AbstractMappingMessageRouter> extends Ab
}
public RouterSpec<R> prefix(String prefix) {
Assert.state(this.subFlows.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive");
this.prefix = prefix;
this.target.setPrefix(prefix);
return _this();
}
public RouterSpec<R> suffix(String suffix) {
Assert.state(this.subFlows.isEmpty(), "The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive");
this.suffix = suffix;
this.target.setSuffix(suffix);
return _this();
}
public RouterSpec<R> channelMapping(String key, String channelName) {
Assert.hasText(key);
Assert.hasText(channelName);
this.target.setChannelMapping(key, channelName);
return _this();
}
public RouterSpec<R> subFlowMapping(String key, IntegrationFlow subFlow) {
Assert.hasText(key);
Assert.notNull(subFlow);
Assert.state(!(StringUtils.hasText(this.prefix) || StringUtils.hasText(this.suffix)),
"The 'prefix'('suffix') and 'subFlowMapping' are mutually exclusive");
DirectChannel channel = new DirectChannel();
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
subFlow.accept(flowBuilder);
this.subFlows.add(flowBuilder);
if (this.mappingProvider == null) {
this.mappingProvider = new RouterSubFlowMappingProvider(this.target);
this.subFlows.add(this.mappingProvider);
}
this.mappingProvider.addMapping(key, channel);
return _this();
}
@Override
public Collection<Object> getComponentsToRegister() {
return this.subFlows;
}
private static class RouterSubFlowMappingProvider {
private final MappingMessageRouterManagement router;
private final Map<String, NamedComponent> mapping = new HashMap<String, NamedComponent>();
public RouterSubFlowMappingProvider(MappingMessageRouterManagement router) {
this.router = router;
}
void addMapping(String key, NamedComponent channel) {
this.mapping.put(key, channel);
}
@PostConstruct
public void init() {
for (Map.Entry<String, NamedComponent> entry : this.mapping.entrySet()) {
this.router.setChannelMapping(entry.getKey(), entry.getValue().getComponentName());
}
}
}
}

41
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* 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.Set;
/**
* @author Artem Bilan
*/
public class StandardIntegrationFlow implements IntegrationFlow {
private final Set<Object> integrationComponents;
StandardIntegrationFlow(Set<Object> integrationComponents) {
this.integrationComponents = integrationComponents;
}
public Set<Object> getIntegrationComponents() {
return integrationComponents;
}
@Override
public void accept(IntegrationFlowDefinition<?> flow) {
throw new UnsupportedOperationException();
}
}

25
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/MessageChannels.java
View File

@@ -16,7 +16,7 @@
package org.springframework.integration.dsl.channel;
import java.util.concurrent.BlockingQueue;
import java.util.Queue;
import java.util.concurrent.Executor;
import org.springframework.integration.store.ChannelMessageStore;
@@ -44,11 +44,11 @@ public final class MessageChannels {
return queue().id(id);
}
public static QueueChannelSpec queue(BlockingQueue<Message<?>> queue) {
public static QueueChannelSpec queue(Queue<Message<?>> queue) {
return new QueueChannelSpec(queue);
}
public static QueueChannelSpec queue(String id, BlockingQueue<Message<?>> queue) {
public static QueueChannelSpec queue(String id, Queue<Message<?>> queue) {
return queue(queue).id(id);
}
@@ -103,20 +103,23 @@ public final class MessageChannels {
return queue(messageGroupStore, groupId).id(id);
}
public static PublishSubscribeChannelSpec publishSubscribe() {
return new PublishSubscribeChannelSpec();
public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannelSpec<S> publishSubscribe() {
return new PublishSubscribeChannelSpec<S>();
}
public static PublishSubscribeChannelSpec publishSubscribe(String id) {
return publishSubscribe().id(id);
public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannelSpec<S> publishSubscribe(
String id) {
return MessageChannels.<S>publishSubscribe().id(id);
}
public static PublishSubscribeChannelSpec publishSubscribe(Executor executor) {
return new PublishSubscribeChannelSpec(executor);
public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannelSpec<S> publishSubscribe(
Executor executor) {
return new PublishSubscribeChannelSpec<S>(executor);
}
public static PublishSubscribeChannelSpec publishSubscribe(String id, Executor executor) {
return publishSubscribe(executor).id(id);
public static <S extends PublishSubscribeChannelSpec<S>> PublishSubscribeChannelSpec<S> publishSubscribe(String id,
Executor executor) {
return MessageChannels.<S>publishSubscribe(executor).id(id);
}
private MessageChannels() {

28
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/PublishSubscribeChannelSpec.java
View File

@@ -24,40 +24,40 @@ import org.springframework.util.ErrorHandler;
/**
* @author Artem Bilan
*/
public class PublishSubscribeChannelSpec
extends MessageChannelSpec<PublishSubscribeChannelSpec, PublishSubscribeChannel> {
public class PublishSubscribeChannelSpec<S extends PublishSubscribeChannelSpec<S>>
extends MessageChannelSpec<S, PublishSubscribeChannel> {
PublishSubscribeChannelSpec() {
protected PublishSubscribeChannelSpec() {
this.channel = new PublishSubscribeChannel();
}
PublishSubscribeChannelSpec(Executor executor) {
protected PublishSubscribeChannelSpec(Executor executor) {
this.channel = new PublishSubscribeChannel(executor);
}
PublishSubscribeChannelSpec errorHandler(ErrorHandler errorHandler) {
public S errorHandler(ErrorHandler errorHandler) {
this.channel.setErrorHandler(errorHandler);
return this;
return _this();
}
public PublishSubscribeChannelSpec ignoreFailures(boolean ignoreFailures) {
public S ignoreFailures(boolean ignoreFailures) {
this.channel.setIgnoreFailures(ignoreFailures);
return this;
return _this();
}
public PublishSubscribeChannelSpec applySequence(boolean applySequence) {
public S applySequence(boolean applySequence) {
this.channel.setApplySequence(applySequence);
return this;
return _this();
}
public PublishSubscribeChannelSpec maxSubscribers(Integer maxSubscribers) {
public S maxSubscribers(Integer maxSubscribers) {
this.channel.setMaxSubscribers(maxSubscribers);
return this;
return _this();
}
public PublishSubscribeChannelSpec minSubscribers(int minSubscribers) {
public S minSubscribers(int minSubscribers) {
this.channel.setMinSubscribers(minSubscribers);
return this;
return _this();
}
}

27
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/channel/QueueChannelSpec.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.dsl.channel;
import java.lang.reflect.Constructor;
import java.util.Queue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.locks.Lock;
@@ -24,20 +26,21 @@ import org.springframework.integration.store.ChannelMessageStore;
import org.springframework.integration.store.MessageGroupQueue;
import org.springframework.integration.store.PriorityCapableChannelMessageStore;
import org.springframework.messaging.Message;
import org.springframework.util.ClassUtils;
/**
* @author Artem Bilan
*/
public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, QueueChannel> {
protected BlockingQueue<Message<?>> queue;
protected Queue<Message<?>> queue;
protected Integer capacity;
QueueChannelSpec() {
}
QueueChannelSpec(BlockingQueue<Message<?>> queue) {
QueueChannelSpec(Queue<Message<?>> queue) {
this.queue = queue;
}
@@ -48,7 +51,25 @@ public class QueueChannelSpec extends MessageChannelSpec<QueueChannelSpec, Queue
@Override
protected QueueChannel doGet() {
if (this.queue != null) {
this.channel = new QueueChannel(this.queue);
Constructor<?> queueConstructor =
ClassUtils.getConstructorIfAvailable(QueueChannel.class, Queue.class);
if (queueConstructor == null) {
if (!(this.queue instanceof BlockingQueue)) {
throw new IllegalArgumentException("The 'queue' must be an instance of BlockingQueue " +
"for Spring Integration versions less than 4.1");
}
else {
this.channel = new QueueChannel((BlockingQueue<Message<?>>) this.queue);
}
}
else {
try {
this.channel = (QueueChannel) queueConstructor.newInstance(this.queue);
}
catch (Exception e) {
throw new IllegalStateException(e);
}
}
}
else if (this.capacity != null) {
this.channel = new QueueChannel(this.capacity);

16
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/config/IntegrationFlowBeanPostProcessor.java
View File

@@ -36,8 +36,8 @@ import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec;
import org.springframework.integration.dsl.StandardIntegrationFlow;
import org.springframework.integration.dsl.core.ConsumerEndpointSpec;
import org.springframework.integration.dsl.core.IntegrationComponentSpec;
import org.springframework.integration.dsl.support.MessageChannelReference;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.messaging.MessageChannel;
@@ -69,8 +69,8 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof IntegrationFlowBuilder.StandardIntegrationFlow) {
return processStandardIntegrationFlow((IntegrationFlowBuilder.StandardIntegrationFlow) bean, beanName);
if (bean instanceof StandardIntegrationFlow) {
return processStandardIntegrationFlow((StandardIntegrationFlow) bean, beanName);
}
else if (bean instanceof IntegrationFlow) {
return processIntegrationFlowImpl((IntegrationFlow) bean, beanName);
@@ -78,9 +78,10 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
return bean;
}
private Object processStandardIntegrationFlow(IntegrationFlowBuilder.StandardIntegrationFlow flow,
private Object processStandardIntegrationFlow(StandardIntegrationFlow flow,
String beanName) {
String flowNamePrefix = beanName + ".";
int subFlowNameIndex = 0;
int channelNameIndex = 0;
for (Object component : flow.getIntegrationComponents()) {
if (component instanceof ConsumerEndpointSpec) {
@@ -147,7 +148,7 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
else if (component instanceof SourcePollingChannelAdapterSpec) {
SourcePollingChannelAdapterSpec spec = (SourcePollingChannelAdapterSpec) component;
SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1();
String id = ((IntegrationComponentSpec) spec).getId();
String id = spec.getId();
if (!StringUtils.hasText(id)) {
id = generateBeanName(pollingChannelAdapterFactoryBean);
}
@@ -166,6 +167,11 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
registerComponent(messageSource, messageSourceId);
}
}
else if (component instanceof StandardIntegrationFlow) {
String subFlowBeanName = flowNamePrefix + "subFlow" +
BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + subFlowNameIndex++;
registerComponent(component, subFlowBeanName);
}
else if (!this.beanFactory
.getBeansOfType(AopUtils.getTargetClass(component), false, false)
.values()

147
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/amqp/AmqpTests.java Normal file
View File

@@ -0,0 +1,147 @@
/*
* 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.test.amqp;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.AmqpTemplate;
import org.springframework.amqp.core.AnonymousQueue;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageProducers;
import org.springframework.integration.dsl.amqp.Amqp;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class AmqpTests {
@Autowired
private AmqpTemplate amqpTemplate;
@Autowired
@Qualifier("queue")
private Queue amqpQueue;
@Test
public void testAmqpInboundGatewayFlow() throws Exception {
Object result = this.amqpTemplate.convertSendAndReceive(this.amqpQueue.getName(), "world");
assertEquals("HELLO WORLD", result);
}
@Autowired
@Qualifier("amqpOutboundInput")
private MessageChannel amqpOutboundInput;
@Autowired
@Qualifier("amqpReplyChannel.channel")
private PollableChannel amqpReplyChannel;
@Test
public void testAmqpOutboundFlow() throws Exception {
this.amqpOutboundInput.send(MessageBuilder.withPayload("hello through the amqp")
.setHeader("routingKey", "foo")
.build());
Message<?> receive = null;
int i = 0;
do {
receive = this.amqpReplyChannel.receive();
if (receive != null) {
break;
}
Thread.sleep(100);
i++;
} while (i < 10);
assertNotNull(receive);
assertEquals("HELLO THROUGH THE AMQP", receive.getPayload());
}
@Configuration
@EnableAutoConfiguration
public static class ContextConfiguration {
@Autowired
private ConnectionFactory rabbitConnectionFactory;
@Autowired
private AmqpTemplate amqpTemplate;
@Bean
public Queue queue() {
return new AnonymousQueue();
}
@Bean
public IntegrationFlow amqpFlow() {
return IntegrationFlows.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue()))
.transform("hello "::concat)
.transform(String.class, String::toUpperCase)
.get();
}
@Bean
public IntegrationFlow amqpOutboundFlow() {
return IntegrationFlows.from(Amqp.channel("amqpOutboundInput", this.rabbitConnectionFactory))
.handle(Amqp.outboundAdapter(this.amqpTemplate).routingKeyExpression("headers.routingKey"))
.get();
}
@Bean
public Queue fooQueue() {
return new Queue("foo");
}
@Bean
public Queue amqpReplyChannel() {
return new Queue("amqpReplyChannel");
}
@Bean
public IntegrationFlow amqpInboundFlow() {
return IntegrationFlows.from((MessageProducers p) -> p.amqp(this.rabbitConnectionFactory, fooQueue()))
.transform(String.class, String::toUpperCase)
.channel(Amqp.pollableChannel(this.rabbitConnectionFactory)
.queueName("amqpReplyChannel")
.channelTransacted(true))
.get();
}
}
}

261
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/file/FileTests.java Normal file
View File

@@ -0,0 +1,261 @@
/*
* 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.test.file;
import static org.hamcrest.Matchers.endsWith;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.aop.TargetSource;
import org.springframework.aop.framework.Advised;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
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.core.Pollers;
import org.springframework.integration.dsl.support.Transformers;
import org.springframework.integration.file.DefaultFileNameGenerator;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.FileWritingMessageHandler;
import org.springframework.integration.file.tail.ApacheCommonsFileTailingMessageProducer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.StreamUtils;
/**
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class FileTests {
private static final File tmpDir = new File(System.getProperty("java.io.tmpdir"));
@Autowired
private ListableBeanFactory beanFactory;
@Autowired
private ControlBusGateway controlBus;
@Autowired
@Qualifier("fileFlow1Input")
private MessageChannel fileFlow1Input;
@Autowired
@Qualifier("fileWriting.handler")
private MessageHandler fileWritingMessageHandler;
@Autowired
@Qualifier("tailChannel")
private PollableChannel tailChannel;
@Autowired
private ApacheCommonsFileTailingMessageProducer tailer;
@Autowired
@Qualifier("fileReadingResultChannel")
private PollableChannel fileReadingResultChannel;
@Autowired
@Qualifier("fileWritingInput")
private MessageChannel fileWritingInput;
@Autowired
@Qualifier("fileWritingResultChannel")
private PollableChannel fileWritingResultChannel;
@Test
public void testFileHandler() throws Exception {
Message<?> message = MessageBuilder.withPayload("foo").setHeader(FileHeaders.FILENAME, "foo").build();
try {
this.fileFlow1Input.send(message);
fail("NullPointerException expected");
}
catch (Exception e) {
assertThat(e, instanceOf(MessageHandlingException.class));
assertThat(e.getCause(), instanceOf(NullPointerException.class));
}
DefaultFileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
fileNameGenerator.setBeanFactory(this.beanFactory);
Object targetFileWritingMessageHandler = this.fileWritingMessageHandler;
if (this.fileWritingMessageHandler instanceof Advised) {
TargetSource targetSource = ((Advised) this.fileWritingMessageHandler).getTargetSource();
if (targetSource != null) {
targetFileWritingMessageHandler = targetSource.getTarget();
}
}
DirectFieldAccessor dfa = new DirectFieldAccessor(targetFileWritingMessageHandler);
dfa.setPropertyValue("fileNameGenerator", fileNameGenerator);
this.fileFlow1Input.send(message);
assertTrue(new File(tmpDir, "foo").exists());
}
@Test
public void testMessageProducerFlow() throws Exception {
FileOutputStream file = new FileOutputStream(new File(tmpDir, "TailTest"));
for (int i = 0; i < 50; i++) {
file.write((i + "\n").getBytes());
}
this.tailer.start();
for (int i = 0; i < 50; i++) {
Message<?> message = this.tailChannel.receive(5000);
assertNotNull(message);
assertEquals("hello " + i, message.getPayload());
}
assertNull(this.tailChannel.receive(1));
this.controlBus.send("@tailer.stop()");
file.close();
}
@Test
public void testFileReadingFlow() throws Exception {
List<Integer> evens = new ArrayList<>(25);
for (int i = 0; i < 50; i++) {
boolean even = i % 2 == 0;
String extension = even ? ".sitest" : ".foofile";
if (even) {
evens.add(i);
}
FileOutputStream file = new FileOutputStream(new File(tmpDir, i + extension));
file.write(("" + i).getBytes());
file.flush();
file.close();
}
Message<?> message = fileReadingResultChannel.receive(10000);
assertNotNull(message);
Object payload = message.getPayload();
assertThat(payload, instanceOf(List.class));
@SuppressWarnings("unchecked")
List<String> result = (List<String>) payload;
assertEquals(25, result.size());
result.forEach(s -> assertTrue(evens.contains(Integer.parseInt(s))));
}
@Test
public void testFileWritingFlow() throws Exception {
String payload = "Spring Integration";
this.fileWritingInput.send(new GenericMessage<>(payload));
Message<?> receive = this.fileWritingResultChannel.receive(1000);
assertNotNull(receive);
assertThat(receive.getPayload(), instanceOf(File.class));
File resultFile = (File) receive.getPayload();
assertThat(resultFile.getAbsolutePath(),
endsWith(TestUtils.applySystemFileSeparator("fileWritingFlow/foo.sitest")));
String fileContent = StreamUtils.copyToString(new FileInputStream(resultFile), Charset.defaultCharset());
assertEquals(payload, fileContent);
}
@MessagingGateway(defaultRequestChannel = "controlBus.input")
private static interface ControlBusGateway {
void send(String command);
}
@Configuration
@EnableIntegration
@IntegrationComponentScan
public static class ContextConfiguration {
@Bean
public IntegrationFlow controlBus() {
return f -> f.controlBus();
}
@Bean
public IntegrationFlow fileFlow1() {
return IntegrationFlows.from("fileFlow1Input")
.<FileWritingMessageHandler>handleWithAdapter(h -> h.file(tmpDir).fileNameGenerator(message -> null)
, c -> c.id("fileWriting"))
.get();
}
@Bean
public IntegrationFlow tailFlow() {
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();
}
@Bean
public IntegrationFlow fileReadingFlow() {
return IntegrationFlows
.from(s -> s.file(tmpDir).patternFilter("*.sitest"),
e -> e.poller(Pollers.fixedDelay(100)))
.transform(Transformers.fileToString())
.aggregate(a -> a.correlationExpression("1")
.releaseStrategy(g -> g.size() == 25), null)
.channel(MessageChannels.queue("fileReadingResultChannel"))
.get();
}
@Bean
public IntegrationFlow fileWritingFlow() {
return IntegrationFlows.from("fileWritingInput")
.enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.sitest")
.header("directory", new File(tmpDir, "fileWritingFlow")))
.handleWithAdapter(a -> a.fileGateway(m -> m.getHeaders().get("directory")))
.channel(MessageChannels.queue("fileWritingResultChannel"))
.get();
}
}
}

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

File diff suppressed because it is too large Load Diff

259
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/ftp/FtpTests.java Normal file
View File

@@ -0,0 +1,259 @@
/*
* 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.test.ftp;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.isOneOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import java.io.File;
import java.util.List;
import java.util.regex.Matcher;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.apache.commons.net.ftp.FTPFile;
import org.hamcrest.Matchers;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.ConfigFileApplicationContextInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.channel.MessageChannels;
import org.springframework.integration.dsl.core.Pollers;
import org.springframework.integration.dsl.ftp.Ftp;
import org.springframework.integration.dsl.sftp.Sftp;
import org.springframework.integration.dsl.test.sftp.TestSftpServer;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.remote.RemoteFileTemplate;
import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
*/
@ContextConfiguration(initializers = ConfigFileApplicationContextInitializer.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class FtpTests {
@Autowired
private ControlBusGateway controlBus;
@Autowired
private MBeanServer mBeanServer;
@Autowired
private TestFtpServer ftpServer;
@Autowired
private DefaultFtpSessionFactory ftpSessionFactory;
@Autowired
@Qualifier("ftpInboundResultChannel")
private PollableChannel ftpInboundResultChannel;
@Autowired
@Qualifier("toFtpChannel")
private MessageChannel toFtpChannel;
@Autowired
@Qualifier("ftpMgetInputChannel")
private MessageChannel ftpMgetInputChannel;
@Autowired
@Qualifier("remoteFileOutputChannel")
private PollableChannel remoteFileOutputChannel;
@Before
@After
public void setupRemoteFileServers() {
this.ftpServer.recursiveDelete(this.ftpServer.getTargetLocalDirectory());
this.ftpServer.recursiveDelete(this.ftpServer.getTargetFtpDirectory());
}
@Test
public void testFtpInboundFlow() {
this.controlBus.send("@ftpInboundAdapter.start()");
Message<?> message = this.ftpInboundResultChannel.receive(1000);
assertNotNull(message);
Object payload = message.getPayload();
assertThat(payload, instanceOf(File.class));
File file = (File) payload;
assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
assertThat(file.getAbsolutePath(), containsString("ftpTest"));
message = this.ftpInboundResultChannel.receive(1000);
assertNotNull(message);
file = (File) message.getPayload();
assertThat(file.getName(), isOneOf("FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
assertThat(file.getAbsolutePath(), containsString("ftpTest"));
this.controlBus.send("@ftpInboundAdapter.stop()");
}
@Test
public void testFtpOutboundFlow() {
String fileName = "foo.file";
this.toFtpChannel.send(MessageBuilder.withPayload("foo")
.setHeader(FileHeaders.FILENAME, fileName)
.build());
RemoteFileTemplate<FTPFile> template = new RemoteFileTemplate<>(this.ftpSessionFactory);
FTPFile[] files = template.execute(session ->
session.list(this.ftpServer.getTargetFtpDirectory().getName() + "/" + fileName));
assertEquals(1, files.length);
assertEquals(3, files[0].getSize());
}
@Test
@SuppressWarnings("unchecked")
public void testFtpMgetFlow() {
String dir = "ftpSource/";
this.ftpMgetInputChannel.send(new GenericMessage<Object>(dir + "*"));
Message<?> result = this.remoteFileOutputChannel.receive(1000);
assertNotNull(result);
List<File> localFiles = (List<File>) result.getPayload();
// should have filtered ftpSource2.txt
assertEquals(2, localFiles.size());
for (File file : localFiles) {
assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
Matchers.containsString(dir));
}
assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
Matchers.containsString(dir + "subFtpSource"));
}
@Test
public void testMBeansForDSL() throws MalformedObjectNameException {
assertFalse(this.mBeanServer.queryMBeans(ObjectName.getInstance("org.springframework.integration:" +
"type=MessageHandler,name=ftpMgetInputChannel,bean=anonymous"), null).isEmpty());
}
@MessagingGateway(defaultRequestChannel = "controlBus.input")
private static interface ControlBusGateway {
void send(String command);
}
@Configuration
@Import(TestFtpServer.class)
@EnableAutoConfiguration
@IntegrationComponentScan
public static class ContextConfiguration {
@Autowired
private TestFtpServer ftpServer;
@Autowired
private DefaultFtpSessionFactory ftpSessionFactory;
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
return Pollers.fixedRate(500).maxMessagesPerPoll(1).get();
}
@Bean
public IntegrationFlow controlBus() {
return f -> f.controlBus();
}
@Bean
public IntegrationFlow ftpInboundFlow() {
return IntegrationFlows
.from(s -> s.ftp(this.ftpSessionFactory)
.preserveTimestamp(true)
.remoteDirectory("ftpSource")
.regexFilter(".*\\.txt$")
.localFilename(f -> f.toUpperCase() + ".a")
.localDirectory(this.ftpServer.getTargetLocalDirectory()),
e -> e.id("ftpInboundAdapter").autoStartup(false))
.channel(MessageChannels.queue("ftpInboundResultChannel"))
.get();
}
@Bean
public IntegrationFlow ftpOutboundFlow() {
return IntegrationFlows.from("toFtpChannel")
.handle(Ftp.outboundAdapter(this.ftpSessionFactory)
.useTemporaryFileName(false)
.fileNameExpression("headers['" + FileHeaders.FILENAME + "']")
.remoteDirectory(this.ftpServer.getTargetFtpDirectory().getName())
).get();
}
@Bean
public PollableChannel remoteFileOutputChannel() {
return new QueueChannel();
}
@Bean
public MessageHandler ftpOutboundGateway() {
return Ftp.outboundGateway(this.ftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
"payload")
.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
.regexFileNameFilter("(subFtpSource|.*1.txt)")
.localDirectoryExpression("@ftpServer.targetLocalDirectoryName + #remoteDirectory")
.localFilenameExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')")
.get();
}
@Bean
public IntegrationFlow ftpMGetFlow() {
return IntegrationFlows.from("ftpMgetInputChannel")
.handle(ftpOutboundGateway())
.channel(remoteFileOutputChannel())
.get();
}
}
}

2
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestFtpServer.java → spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/ftp/TestFtpServer.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.dsl.test;
package org.springframework.integration.dsl.test.ftp;
import java.io.File;
import java.io.FileOutputStream;

8
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jdbc/JdbcTests.java
View File

@@ -83,7 +83,7 @@ public class JdbcTests {
new ResultSetIterator<Foo>(ps.executeQuery(),
(rs1, rowNum) ->
new Foo(rs1.getInt(1), rs1.getString(2))))
, null)
, e -> e.applySequence(false))
.channel(c -> c.queue("splitResultsChannel"));
}
@@ -116,7 +116,11 @@ public class JdbcTests {
@Override
public boolean hasNext() {
try {
return !this.rs.isLast();
boolean hasNext = !this.rs.isLast();
if (!hasNext) {
this.rs.close();
}
return hasNext;
}
catch (SQLException e) {
throw new InvalidResultSetAccessException(e);

263
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jms/JmsTests.java Normal file
View File

@@ -0,0 +1,263 @@
/*
* 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.test.jms;
import static org.hamcrest.Matchers.instanceOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.ConnectionFactory;
import org.hamcrest.Matchers;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.channel.FixedSubscriberChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.GlobalChannelInterceptor;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.MessageSources;
import org.springframework.integration.dsl.MessagingGateways;
import org.springframework.integration.dsl.channel.MessageChannels;
import org.springframework.integration.dsl.core.Pollers;
import org.springframework.integration.dsl.jms.Jms;
import org.springframework.integration.endpoint.MethodInvokingMessageSource;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.stereotype.Component;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class JmsTests {
@Autowired
private ListableBeanFactory beanFactory;
@Autowired
private ControlBusGateway controlBus;
@Autowired
@Qualifier("flow1QueueChannel")
private PollableChannel outputChannel;
@Autowired
@Qualifier("jmsOutboundFlow.input")
private MessageChannel jmsOutboundInboundChannel;
@Autowired
@Qualifier("jmsOutboundInboundReplyChannel")
private PollableChannel jmsOutboundInboundReplyChannel;
@Autowired
@Qualifier("jmsOutboundGatewayFlow.input")
private MessageChannel jmsOutboundGatewayChannel;
@Autowired
private TestChannelInterceptor testChannelInterceptor;
@Test
public void testPollingFlow() {
this.controlBus.send("@integerEndpoint.start()");
assertThat(this.beanFactory.getBean("integerChannel"), instanceOf(FixedSubscriberChannel.class));
for (int i = 0; i < 5; i++) {
Message<?> message = this.outputChannel.receive(20000);
assertNotNull(message);
assertEquals("" + i, message.getPayload());
}
this.controlBus.send("@integerEndpoint.stop()");
assertTrue(((ChannelInterceptorAware) this.outputChannel).getChannelInterceptors()
.contains(this.testChannelInterceptor));
assertThat(this.testChannelInterceptor.invoked.get(), Matchers.greaterThanOrEqualTo(5));
}
@Test
public void testJmsOutboundInboundFlow() {
this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS")
.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsInbound")
.build());
Message<?> receive = this.jmsOutboundInboundReplyChannel.receive(5000);
assertNotNull(receive);
assertEquals("HELLO THROUGH THE JMS", receive.getPayload());
this.jmsOutboundInboundChannel.send(MessageBuilder.withPayload("hello THROUGH the JMS")
.setHeader(SimpMessageHeaderAccessor.DESTINATION_HEADER, "jmsMessageDriver")
.build());
receive = this.jmsOutboundInboundReplyChannel.receive(5000);
assertNotNull(receive);
assertEquals("hello through the jms", receive.getPayload());
}
@Test
public void testJmsPipelineFlow() {
PollableChannel replyChannel = new QueueChannel();
Message<String> message = MessageBuilder.withPayload("hello through the jms pipeline")
.setReplyChannel(replyChannel)
.setHeader("destination", "jmsPipelineTest")
.build();
this.jmsOutboundGatewayChannel.send(message);
Message<?> receive = replyChannel.receive(5000);
assertNotNull(receive);
assertEquals("HELLO THROUGH THE JMS PIPELINE", receive.getPayload());
}
@MessagingGateway(defaultRequestChannel = "controlBus.input")
private static interface ControlBusGateway {
void send(String command);
}
@Configuration
@EnableAutoConfiguration
@IntegrationComponentScan
@ComponentScan
public static class ContextConfiguration {
@Autowired
private ConnectionFactory jmsConnectionFactory;
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
return Pollers.fixedRate(500).maxMessagesPerPoll(1).get();
}
@Bean
public IntegrationFlow controlBus() {
return f -> f.controlBus();
}
@Bean
public MessageSource<?> integerMessageSource() {
MethodInvokingMessageSource source = new MethodInvokingMessageSource();
source.setObject(new AtomicInteger());
source.setMethodName("getAndIncrement");
return source;
}
@Bean
public IntegrationFlow flow1() {
return IntegrationFlows.from(integerMessageSource(),
c -> c.poller(p -> p.fixedRate(100))
.id("integerEndpoint")
.autoStartup(false))
.fixedSubscriberChannel("integerChannel")
.transform("payload.toString()")
.channel(Jms.pollableChannel("flow1QueueChannel", this.jmsConnectionFactory)
.destination("flow1QueueChannel"))
.get();
}
@Bean
public IntegrationFlow jmsOutboundFlow() {
return f -> f.handleWithAdapter(h -> h.jms(this.jmsConnectionFactory)
.destinationExpression("headers." + SimpMessageHeaderAccessor.DESTINATION_HEADER));
}
@Bean
public MessageChannel jmsOutboundInboundReplyChannel() {
return MessageChannels.queue().get();
}
@Bean
public IntegrationFlow jmsInboundFlow() {
return IntegrationFlows
.from((MessageSources s) -> s.jms(this.jmsConnectionFactory).destination("jmsInbound"))
.<String, String>transform(String::toUpperCase)
.channel(this.jmsOutboundInboundReplyChannel())
.get();
}
@Bean
public IntegrationFlow jmsMessageDriverFlow() {
return IntegrationFlows
.from(Jms.messageDriverChannelAdapter(this.jmsConnectionFactory)
.destination("jmsMessageDriver"))
.<String, String>transform(String::toLowerCase)
.channel(this.jmsOutboundInboundReplyChannel())
.get();
}
@Bean
public IntegrationFlow jmsOutboundGatewayFlow() {
return f -> f.handleWithAdapter(a ->
a.jmsGateway(this.jmsConnectionFactory)
.replyContainer()
.requestDestination("jmsPipelineTest"));
}
@Bean
public IntegrationFlow jmsInboundGatewayFlow() {
return IntegrationFlows.from((MessagingGateways g) ->
g.jms(this.jmsConnectionFactory)
.destination("jmsPipelineTest"))
.<String, String>transform(String::toUpperCase)
.get();
}
}
@Component
@GlobalChannelInterceptor(patterns = "flow1QueueChannel")
public static class TestChannelInterceptor extends ChannelInterceptorAdapter {
private final AtomicInteger invoked = new AtomicInteger();
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
this.invoked.incrementAndGet();
return message;
}
}
}

199
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mongodb/MongoDbTest.java Normal file
View File

@@ -0,0 +1,199 @@
/*
* 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.test.mongodb;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import java.io.IOException;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.MongoDbFactory;
import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.Channels;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.channel.MessageChannels;
import org.springframework.integration.dsl.core.Pollers;
import org.springframework.integration.mongodb.store.MongoDbChannelMessageStore;
import org.springframework.integration.store.PriorityCapableChannelMessageStore;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.mongodb.MongoClient;
import de.flapdoodle.embed.mongo.MongodExecutable;
import de.flapdoodle.embed.mongo.MongodStarter;
import de.flapdoodle.embed.mongo.config.MongodConfigBuilder;
import de.flapdoodle.embed.mongo.config.Net;
import de.flapdoodle.embed.mongo.distribution.Version;
import de.flapdoodle.embed.process.runtime.Network;
/**
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class MongoDbTest {
private static int mongoPort;
private static MongodExecutable mongodExe;
@BeforeClass
public static void setup() throws IOException {
mongoPort = Network.getFreeServerPort();
mongodExe = MongodStarter.getDefaultInstance()
.prepare(new MongodConfigBuilder()
.version(Version.Main.PRODUCTION)
.net(new Net(mongoPort, Network.localhostIsIPv6()))
.build());
mongodExe.start();
}
@AfterClass
public static void tearDown() {
mongodExe.stop();
}
@Autowired
private ControlBusGateway controlBus;
@Autowired
@Qualifier("priorityChannel")
private MessageChannel priorityChannel;
@Autowired
@Qualifier("priorityReplyChannel")
private PollableChannel priorityReplyChannel;
@Test
public void testPriority() throws InterruptedException {
Message<String> message = MessageBuilder.withPayload("1").setPriority(1).build();
this.priorityChannel.send(message);
message = MessageBuilder.withPayload("-1").setPriority(-1).build();
this.priorityChannel.send(message);
message = MessageBuilder.withPayload("3").setPriority(3).build();
this.priorityChannel.send(message);
message = MessageBuilder.withPayload("0").setPriority(0).build();
this.priorityChannel.send(message);
message = MessageBuilder.withPayload("2").setPriority(2).build();
this.priorityChannel.send(message);
message = MessageBuilder.withPayload("none").build();
this.priorityChannel.send(message);
message = MessageBuilder.withPayload("31").setPriority(3).build();
this.priorityChannel.send(message);
this.controlBus.send("@priorityChannelBridge.start()");
Message<?> receive = this.priorityReplyChannel.receive(2000);
assertNotNull(receive);
assertEquals("3", receive.getPayload());
receive = this.priorityReplyChannel.receive(2000);
assertNotNull(receive);
assertEquals("31", receive.getPayload());
receive = this.priorityReplyChannel.receive(2000);
assertNotNull(receive);
assertEquals("2", receive.getPayload());
receive = this.priorityReplyChannel.receive(2000);
assertNotNull(receive);
assertEquals("1", receive.getPayload());
receive = this.priorityReplyChannel.receive(2000);
assertNotNull(receive);
assertEquals("0", receive.getPayload());
receive = this.priorityReplyChannel.receive(2000);
assertNotNull(receive);
assertEquals("-1", receive.getPayload());
receive = this.priorityReplyChannel.receive(2000);
assertNotNull(receive);
assertEquals("none", receive.getPayload());
this.controlBus.send("@priorityChannelBridge.stop()");
}
@MessagingGateway(defaultRequestChannel = "controlBus.input")
private static interface ControlBusGateway {
void send(String command);
}
@Configuration
@EnableIntegration
@IntegrationComponentScan
public static class ContextConfiguration {
@Bean
public IntegrationFlow controlBus() {
return f -> f.controlBus();
}
@Bean
public MongoDbFactory mongoDbFactory() throws Exception {
return new SimpleMongoDbFactory(new MongoClient("localhost", mongoPort), "local");
}
@Bean
public MongoDbChannelMessageStore mongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) {
MongoDbChannelMessageStore mongoDbChannelMessageStore = new MongoDbChannelMessageStore(mongoDbFactory);
mongoDbChannelMessageStore.setPriorityEnabled(true);
return mongoDbChannelMessageStore;
}
@Bean
public IntegrationFlow priorityFlow(PriorityCapableChannelMessageStore mongoDbChannelMessageStore) {
return IntegrationFlows.from((Channels c) ->
c.priority("priorityChannel", mongoDbChannelMessageStore, "priorityGroup"))
.bridge(s -> s.poller(Pollers.fixedDelay(100))
.autoStartup(false)
.id("priorityChannelBridge"))
.channel(MessageChannels.queue("priorityReplyChannel"))
.get();
}
}
}

250
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/sftp/SftpTests.java Normal file
View File

@@ -0,0 +1,250 @@
/*
* 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.test.sftp;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.isOneOf;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import java.io.File;
import java.util.List;
import java.util.regex.Matcher;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.hamcrest.Matchers;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.test.ConfigFileApplicationContextInitializer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.dsl.channel.MessageChannels;
import org.springframework.integration.dsl.core.Pollers;
import org.springframework.integration.dsl.sftp.Sftp;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.remote.RemoteFileTemplate;
import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import com.jcraft.jsch.ChannelSftp;
/**
* @author Artem Bilan
*/
@ContextConfiguration(initializers = ConfigFileApplicationContextInitializer.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class SftpTests {
@Autowired
private ControlBusGateway controlBus;
@Autowired
private MBeanServer mBeanServer;
@Autowired
private TestSftpServer sftpServer;
@Autowired
private DefaultSftpSessionFactory sftpSessionFactory;
@Autowired
@Qualifier("sftpInboundResultChannel")
private PollableChannel sftpInboundResultChannel;
@Autowired
@Qualifier("toSftpChannel")
private MessageChannel toSftpChannel;
@Autowired
@Qualifier("remoteFileOutputChannel")
private PollableChannel remoteFileOutputChannel;
@Autowired
@Qualifier("sftpMgetInputChannel")
private MessageChannel sftpMgetInputChannel;
@Before
@After
public void setupRemoteFileServers() {
this.sftpServer.recursiveDelete(this.sftpServer.getTargetLocalDirectory());
this.sftpServer.recursiveDelete(this.sftpServer.getTargetSftpDirectory());
}
@Test
public void testSftpInboundFlow() {
this.controlBus.send("@sftpInboundAdapter.start()");
Message<?> message = this.sftpInboundResultChannel.receive(1000);
assertNotNull(message);
Object payload = message.getPayload();
assertThat(payload, instanceOf(File.class));
File file = (File) payload;
assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a"));
assertThat(file.getAbsolutePath(), containsString("sftpTest"));
message = this.sftpInboundResultChannel.receive(1000);
assertNotNull(message);
file = (File) message.getPayload();
assertThat(file.getName(), isOneOf("SFTPSOURCE1.TXT.a", "SFTPSOURCE2.TXT.a"));
assertThat(file.getAbsolutePath(), containsString("sftpTest"));
this.controlBus.send("@sftpInboundAdapter.stop()");
}
@Test
public void testSftpOutboundFlow() {
String fileName = "foo.file";
this.toSftpChannel.send(MessageBuilder.withPayload("foo")
.setHeader(FileHeaders.FILENAME, fileName)
.build());
RemoteFileTemplate<ChannelSftp.LsEntry> template = new RemoteFileTemplate<>(this.sftpSessionFactory);
ChannelSftp.LsEntry[] files = template.execute(session ->
session.list(this.sftpServer.getTargetSftpDirectory().getName() + "/" + fileName));
assertEquals(1, files.length);
assertEquals(3, files[0].getAttrs().getSize());
}
@Test
@SuppressWarnings("unchecked")
public void testSftpMgetFlow() {
String dir = "sftpSource/";
this.sftpMgetInputChannel.send(new GenericMessage<Object>(dir + "*"));
Message<?> result = this.remoteFileOutputChannel.receive(1000);
assertNotNull(result);
List<File> localFiles = (List<File>) result.getPayload();
// should have filtered sftpSource2.txt
assertEquals(2, localFiles.size());
for (File file : localFiles) {
assertThat(file.getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
Matchers.containsString(dir));
}
assertThat(localFiles.get(1).getPath().replaceAll(Matcher.quoteReplacement(File.separator), "/"),
Matchers.containsString(dir + "subSftpSource"));
}
@Test
public void testMBeansForDSL() throws MalformedObjectNameException {
assertFalse(this.mBeanServer.queryMBeans(ObjectName.getInstance("org.springframework.integration:" +
"bean=anonymous,name=sftpMgetInputChannel,type=MessageHandler"), null).isEmpty());
}
@MessagingGateway(defaultRequestChannel = "controlBus.input")
private static interface ControlBusGateway {
void send(String command);
}
@Configuration
@Import(TestSftpServer.class)
@EnableAutoConfiguration
@IntegrationComponentScan
public static class ContextConfiguration {
@Autowired
private TestSftpServer sftpServer;
@Autowired
private DefaultSftpSessionFactory sftpSessionFactory;
@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
return Pollers.fixedRate(500).maxMessagesPerPoll(1).get();
}
@Bean
public IntegrationFlow controlBus() {
return f -> f.controlBus();
}
@Bean
public IntegrationFlow sftpInboundFlow() {
return IntegrationFlows
.from(s -> s.sftp(this.sftpSessionFactory)
.preserveTimestamp(true)
.remoteDirectory("sftpSource")
.regexFilter(".*\\.txt$")
.localFilenameExpression("#this.toUpperCase() + '.a'")
.localDirectory(this.sftpServer.getTargetLocalDirectory()),
e -> e.id("sftpInboundAdapter").autoStartup(false))
.channel(MessageChannels.queue("sftpInboundResultChannel"))
.get();
}
@Bean
public IntegrationFlow sftpOutboundFlow() {
return IntegrationFlows.from("toSftpChannel")
.handle(Sftp.outboundAdapter(this.sftpSessionFactory)
.useTemporaryFileName(false)
.remoteDirectory(this.sftpServer.getTargetSftpDirectory().getName())
).get();
}
@Bean
public PollableChannel remoteFileOutputChannel() {
return new QueueChannel();
}
@Bean
public IntegrationFlow sftpMGetFlow() {
return IntegrationFlows.from("sftpMgetInputChannel")
.handleWithAdapter(h ->
h.sftpGateway(this.sftpSessionFactory, AbstractRemoteFileOutboundGateway.Command.MGET,
"payload")
.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
.regexFileNameFilter("(subSftpSource|.*1.txt)")
.localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory")
.localFilenameExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
.channel(remoteFileOutputChannel())
.get();
}
}
}

2
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/TestSftpServer.java → spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/sftp/TestSftpServer.java
View File

@@ -14,7 +14,7 @@
* limitations under the License.
*/
package org.springframework.integration.dsl.test;
package org.springframework.integration.dsl.test.sftp;
import java.io.File;
import java.io.FileOutputStream;