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

Java DSL: Introduce FunctionExpression

Browse Source 
Add support for `FunctionExpression` and apply it alongside with `expression`, where it is possible
For example:
```
.enrich(e -> e.requestChannel("enrichChannel")
		.requestPayload(Message::getPayload)
		.shouldClonePayload(false)
		.<Map<String, String>>headerFunction("foo", m -> m.getPayload().get("name")))
```
Remove redundant functional interfaces in favor of `Function`

FunctionExpression: Add JDBC Splitter sample

Minor Polishing to FunctionExpression

FunctionExpression: JavaDocs
This commit is contained in:
Artem Bilan
2014-10-08 00:19:04 +03:00
committed by Gary Russell
parent cab77b5e76
commit 4005ff874d
34 changed files with 660 additions and 172 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -38,11 +38,13 @@ 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.dsl.support.Function;
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 org.springframework.messaging.Message;
import com.jcraft.jsch.ChannelSftp;
@@ -67,6 +69,10 @@ public class Adapters {
return Files.outboundAdapter(directoryExpression);
}
public <P> FileWritingMessageHandlerSpec file(Function<Message<P>, ?> directoryFunction) {
return Files.outboundAdapter(directoryFunction);
}
public FileWritingMessageHandlerSpec fileGateway(File destinationDirectory) {
return Files.outboundGateway(destinationDirectory);
}
@@ -75,6 +81,10 @@ public class Adapters {
return Files.outboundGateway(directoryExpression);
}
public <P> FileWritingMessageHandlerSpec fileGateway(Function<Message<P>, ?> directoryFunction) {
return Files.outboundGateway(directoryFunction);
}
public FtpMessageHandlerSpec ftp(SessionFactory<FTPFile> sessionFactory) {
return Ftp.outboundAdapter(sessionFactory);
}

18
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java
View File

@@ -25,7 +25,10 @@ import org.springframework.integration.aggregator.ReleaseStrategy;
import org.springframework.integration.config.CorrelationStrategyFactoryBean;
import org.springframework.integration.config.ReleaseStrategyFactoryBean;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.store.MessageGroup;
import org.springframework.integration.store.MessageGroupStore;
import org.springframework.messaging.MessageChannel;
import org.springframework.scheduling.TaskScheduler;
@@ -76,8 +79,13 @@ public abstract class
return _this();
}
public S groupTimeoutExpression(String expression) {
this.groupTimeoutExpression = PARSER.parseExpression(expression);
public S groupTimeoutExpression(String groupTimeoutExpression) {
this.groupTimeoutExpression = PARSER.parseExpression(groupTimeoutExpression);
return _this();
}
public S groupTimeout(Function<MessageGroup, Long> groupTimeoutFunction) {
this.groupTimeoutExpression = new FunctionExpression<MessageGroup>(groupTimeoutFunction);
return _this();
}
@@ -98,7 +106,7 @@ public abstract class
public S processor(Object target, String methodName) {
try {
return this.correlationStrategy(new CorrelationStrategyFactoryBean(target, methodName).getObject())
return correlationStrategy(new CorrelationStrategyFactoryBean(target, methodName).getObject())
.releaseStrategy(new ReleaseStrategyFactoryBean(target, methodName).getObject());
}
catch (Exception e) {
@@ -107,7 +115,7 @@ public abstract class
}
public S correlationExpression(String correlationExpression) {
return this.correlationStrategy(new ExpressionEvaluatingCorrelationStrategy(correlationExpression));
return correlationStrategy(new ExpressionEvaluatingCorrelationStrategy(correlationExpression));
}
public S correlationStrategy(Object target, String methodName) {
@@ -125,7 +133,7 @@ public abstract class
}
public S releaseExpression(String releaseExpression) {
return this.releaseStrategy(new ExpressionEvaluatingReleaseStrategy(releaseExpression));
return releaseStrategy(new ExpressionEvaluatingReleaseStrategy(releaseExpression));
}
public S releaseStrategy(Object target, String methodName) {

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

@@ -20,14 +20,16 @@ import java.util.HashMap;
import java.util.Map;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.transformer.ContentEnricher;
import org.springframework.integration.transformer.support.AbstractHeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.HeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.StaticHeaderValueMessageProcessor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
@@ -37,13 +39,12 @@ import org.springframework.util.Assert;
*/
public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnricher> {
private final static SpelExpressionParser PARSER = new SpelExpressionParser();
private final ContentEnricher enricher = new ContentEnricher();
private final Map<String, Expression> propertyExpressions = new HashMap<String, Expression>();
private final Map<String, HeaderValueMessageProcessor<?>> headerExpressions = new HashMap<String, HeaderValueMessageProcessor<?>>();
private final Map<String, HeaderValueMessageProcessor<?>> headerExpressions =
new HashMap<String, HeaderValueMessageProcessor<?>>();
EnricherSpec() {
}
@@ -83,6 +84,11 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
return _this();
}
public <P> EnricherSpec requestPayload(Function<Message<P>, ?> requestPayloadFunction) {
this.enricher.setRequestPayloadExpression(new FunctionExpression<Message<P>>(requestPayloadFunction));
return _this();
}
public EnricherSpec shouldClonePayload(boolean shouldClonePayload) {
this.enricher.setShouldClonePayload(shouldClonePayload);
return _this();
@@ -99,6 +105,12 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
return _this();
}
public <P> EnricherSpec propertyFunction(String key, Function<Message<P>, Object> function) {
Assert.notNull(key);
this.propertyExpressions.put(key, new FunctionExpression<Message<P>>(function));
return _this();
}
public <V> EnricherSpec header(String name, V value) {
return this.header(name, value, null);
}
@@ -107,22 +119,36 @@ public class EnricherSpec extends MessageHandlerSpec<EnricherSpec, ContentEnrich
AbstractHeaderValueMessageProcessor<V> headerValueMessageProcessor =
new StaticHeaderValueMessageProcessor<V>(value);
headerValueMessageProcessor.setOverwrite(overwrite);
return this.header(name, headerValueMessageProcessor);
return header(name, headerValueMessageProcessor);
}
public EnricherSpec headerExpression(String name, String expression) {
return this.headerExpression(name, expression, null);
return headerExpression(name, expression, null);
}
public EnricherSpec headerExpression(String name, String expression, Boolean overwrite) {
Assert.hasText(expression);
return headerExpression(name, PARSER.parseExpression(expression), overwrite);
}
public <P> EnricherSpec headerFunction(String name, Function<Message<P>, Object> function) {
return headerFunction(name, function, null);
}
public <P> EnricherSpec headerFunction(String name, Function<Message<P>, Object> function, Boolean overwrite) {
Assert.notNull(function);
return headerExpression(name, new FunctionExpression<Message<P>>(function), overwrite);
}
private EnricherSpec headerExpression(String name, Expression expression, Boolean overwrite) {
AbstractHeaderValueMessageProcessor<?> headerValueMessageProcessor =
new ExpressionEvaluatingHeaderValueMessageProcessor<Object>(expression, null);
headerValueMessageProcessor.setOverwrite(overwrite);
return this.header(name, headerValueMessageProcessor);
return header(name, headerValueMessageProcessor);
}
public <V> EnricherSpec header(String name, HeaderValueMessageProcessor<V> headerValueMessageProcessor) {
Assert.notNull(name);
Assert.hasText(name);
this.headerExpressions.put(name, headerValueMessageProcessor);
return _this();
}

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

@@ -21,12 +21,13 @@ import java.util.Map;
import java.util.Map.Entry;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.dsl.core.IntegrationComponentSpec;
import org.springframework.integration.dsl.support.BeanNameMessageProcessor;
import org.springframework.integration.dsl.support.Consumer;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.dsl.support.MapBuilder;
import org.springframework.integration.dsl.support.MapBuilder.MapBuilderConfigurer;
import org.springframework.integration.dsl.support.StringStringMapBuilder;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.handler.MessageProcessor;
@@ -35,6 +36,7 @@ import org.springframework.integration.transformer.support.AbstractHeaderValueMe
import org.springframework.integration.transformer.support.ExpressionEvaluatingHeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.HeaderValueMessageProcessor;
import org.springframework.integration.transformer.support.StaticHeaderValueMessageProcessor;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
@@ -43,8 +45,6 @@ import org.springframework.util.Assert;
*/
public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherSpec, HeaderEnricher> {
private final static SpelExpressionParser PARSER = new SpelExpressionParser();
private final Map<String, HeaderValueMessageProcessor<?>> headerToAdd = new HashMap<String, HeaderValueMessageProcessor<?>>();
private final HeaderEnricher headerEnricher = new HeaderEnricher(headerToAdd);
@@ -62,19 +62,17 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
return _this();
}
public HeaderEnricherSpec messageProcessor(MessageProcessor<?> messageProcessor) {
this.headerEnricher.setMessageProcessor(messageProcessor);
return _this();
}
public HeaderEnricherSpec messageProcessor(String expression) {
return this.messageProcessor(new ExpressionEvaluatingMessageProcessor<Object>(
PARSER.parseExpression(expression)));
return messageProcessor(new ExpressionEvaluatingMessageProcessor<Object>(PARSER.parseExpression(expression)));
}
public HeaderEnricherSpec messageProcessor(String beanName, String methodName) {
return this.messageProcessor(new BeanNameMessageProcessor<Object>(beanName, methodName));
return messageProcessor(new BeanNameMessageProcessor<Object>(beanName, methodName));
}
public HeaderEnricherSpec headers(MapBuilder<?, String, Object> headers) {
@@ -97,13 +95,14 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
}
public HeaderEnricherSpec headerExpressions(MapBuilder<?, String, String> headers) {
Assert.notNull(headers);
return headerExpressions(headers.get());
}
public HeaderEnricherSpec headerExpressions(
MapBuilderConfigurer<StringStringMapBuilder, String, String> configurer) {
public HeaderEnricherSpec headerExpressions(Consumer<StringStringMapBuilder> configurer) {
Assert.notNull(configurer);
StringStringMapBuilder builder = new StringStringMapBuilder();
configurer.configure(builder);
configurer.accept(builder);
return headerExpressions(builder.get());
}
@@ -116,29 +115,44 @@ public class HeaderEnricherSpec extends IntegrationComponentSpec<HeaderEnricherS
}
public <V> HeaderEnricherSpec header(String name, V value) {
return this.header(name, value, null);
return header(name, value, null);
}
public <V> HeaderEnricherSpec header(String name, V value, Boolean overwrite) {
AbstractHeaderValueMessageProcessor<V> headerValueMessageProcessor =
new StaticHeaderValueMessageProcessor<V>(value);
headerValueMessageProcessor.setOverwrite(overwrite);
return this.header(name, headerValueMessageProcessor);
return header(name, headerValueMessageProcessor);
}
public HeaderEnricherSpec headerExpression(String name, String expression) {
return this.headerExpression(name, expression, null);
return headerExpression(name, expression, null);
}
public HeaderEnricherSpec headerExpression(String name, String expression, Boolean overwrite) {
Assert.hasText(expression);
return headerExpression(name, PARSER.parseExpression(expression), overwrite);
}
public <P> HeaderEnricherSpec headerFunction(String name, Function<Message<P>, Object> function) {
return headerFunction(name, function, null);
}
public <P> HeaderEnricherSpec headerFunction(String name, Function<Message<P>, Object> function,
Boolean overwrite) {
Assert.notNull(function);
return headerExpression(name, new FunctionExpression<Message<P>>(function), overwrite);
}
private HeaderEnricherSpec headerExpression(String name, Expression expression, Boolean overwrite) {
AbstractHeaderValueMessageProcessor<?> headerValueMessageProcessor =
new ExpressionEvaluatingHeaderValueMessageProcessor<Object>(expression, null);
headerValueMessageProcessor.setOverwrite(overwrite);
return this.header(name, headerValueMessageProcessor);
return header(name, headerValueMessageProcessor);
}
public <V> HeaderEnricherSpec header(String name, HeaderValueMessageProcessor<V> headerValueMessageProcessor) {
Assert.notNull(name);
Assert.hasText(name);
this.headerToAdd.put(name, headerValueMessageProcessor);
return _this();
}

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

@@ -16,11 +16,10 @@
package org.springframework.integration.dsl;
import org.springframework.integration.dsl.support.Consumer;
/**
* @author Artem Bilan
*/
public interface IntegrationFlow {
void define(IntegrationFlowDefinition<?> flow);
public interface IntegrationFlow extends Consumer<IntegrationFlowDefinition<?>> {
}

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

@@ -62,7 +62,7 @@ public final class IntegrationFlowBuilder extends IntegrationFlowDefinition<Inte
}
@Override
public void define(IntegrationFlowDefinition<?> flow) {
public void accept(IntegrationFlowDefinition<?> flow) {
throw new UnsupportedOperationException();
}

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

@@ -24,6 +24,7 @@ import java.util.Set;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.factory.BeanCreationException;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler;
import org.springframework.integration.aggregator.AggregatingMessageHandler;
@@ -40,9 +41,8 @@ import org.springframework.integration.dsl.support.BeanNameMessageProcessor;
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.FunctionExpression;
import org.springframework.integration.dsl.support.GenericHandler;
import org.springframework.integration.dsl.support.GenericRouter;
import org.springframework.integration.dsl.support.GenericSplitter;
import org.springframework.integration.dsl.support.MapBuilder;
import org.springframework.integration.dsl.support.MessageChannelReference;
import org.springframework.integration.expression.ControlBusMethodFilter;
@@ -73,6 +73,7 @@ import org.springframework.integration.transformer.HeaderFilter;
import org.springframework.integration.transformer.MessageTransformingHandler;
import org.springframework.integration.transformer.MethodInvokingTransformer;
import org.springframework.integration.transformer.Transformer;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
@@ -151,7 +152,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
public B controlBus(Consumer<GenericEndpointSpec<ServiceActivatingHandler>> endpointConfigurer) {
return this.handle(new ServiceActivatingHandler(new ExpressionCommandMessageProcessor(
new ControlBusMethodFilter())), endpointConfigurer);
new ControlBusMethodFilter())), endpointConfigurer);
}
public B transform(String expression) {
@@ -241,7 +242,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
public <P> B handle(GenericHandler<P> handler) {
return this.handle(null, handler);
return handle(null, handler);
}
public <P> B handle(GenericHandler<P> handler,
@@ -281,15 +282,38 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
return this.register(new GenericEndpointSpec<BridgeHandler>(new BridgeHandler()), endpointConfigurer);
}
public B delay(String groupId) {
return this.delay(groupId, (String) null);
}
public B delay(String groupId, Consumer<DelayerEndpointSpec> endpointConfigurer) {
return this.delay(groupId, (String) null, endpointConfigurer);
}
public B delay(String groupId, String expression) {
return this.delay(groupId, expression, null);
}
public B delay(String groupId, String expression,
public <P> B delay(String groupId, Function<Message<P>, Object> function) {
return this.delay(groupId, function, null);
}
public <P> B delay(String groupId, Function<Message<P>, Object> function,
Consumer<DelayerEndpointSpec> endpointConfigurer) {
Assert.notNull(function);
return this.delay(groupId, new FunctionExpression<Message<P>>(function), endpointConfigurer);
}
public B delay(String groupId, String expression, Consumer<DelayerEndpointSpec> endpointConfigurer) {
return delay(groupId,
StringUtils.hasText(expression) ? PARSER.parseExpression(expression) : null,
endpointConfigurer);
}
private B delay(String groupId, Expression expression, Consumer<DelayerEndpointSpec> endpointConfigurer) {
DelayHandler delayHandler = new DelayHandler(groupId);
if (StringUtils.hasText(expression)) {
delayHandler.setDelayExpression(PARSER.parseExpression(expression));
if (expression != null) {
delayHandler.setDelayExpression(expression);
}
return this.register(new DelayerEndpointSpec(delayHandler), endpointConfigurer);
}
@@ -351,6 +375,10 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
return transform(headerEnricherSpec.get(), endpointConfigurer);
}
public B split() {
return this.split((Consumer<SplitterEndpointSpec<DefaultMessageSplitter>>) null);
}
public B split(Consumer<SplitterEndpointSpec<DefaultMessageSplitter>> endpointConfigurer) {
return this.split(new DefaultMessageSplitter(), endpointConfigurer);
}
@@ -366,24 +394,24 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
public B split(String beanName, String methodName,
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
return this.split(new MethodInvokingSplitter(new BeanNameMessageProcessor<Collection<?>>(beanName, methodName)),
return this.split(new MethodInvokingSplitter(new BeanNameMessageProcessor<Object>(beanName, methodName)),
endpointConfigurer);
}
public <P> B split(Class<P> payloadType, GenericSplitter<P> splitter) {
return this.split(payloadType, splitter, null);
public <P> B split(Class<P> payloadType, Function<P, ?> splitter) {
return split(payloadType, splitter, null);
}
public <T> B split(GenericSplitter<T> splitter,
public <P> B split(Function<P, ?> splitter,
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
return split(null, splitter, endpointConfigurer);
}
public <P> B split(Class<P> payloadType, GenericSplitter<P> splitter,
public <P> B split(Class<P> payloadType, Function<P, ?> splitter,
Consumer<SplitterEndpointSpec<MethodInvokingSplitter>> endpointConfigurer) {
MethodInvokingSplitter split = isLambda(splitter)
? new MethodInvokingSplitter(new LambdaMessageProcessor(splitter, payloadType))
: new MethodInvokingSplitter(splitter, "split");
: new MethodInvokingSplitter(splitter);
return this.split(split, endpointConfigurer);
}
@@ -509,31 +537,31 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
endpointConfigurer);
}
public <S, T> B route(GenericRouter<S, T> router) {
public <S, T> B route(Function<S, T> router) {
return this.route(null, router);
}
public <S, T> B route(GenericRouter<S, T> router,
public <S, T> B route(Function<S, T> router,
Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
return this.route(null, router, routerConfigurer);
}
public <P, T> B route(Class<P> payloadType, GenericRouter<P, T> router) {
public <P, T> B route(Class<P> payloadType, Function<P, T> router) {
return this.route(payloadType, router, null, null);
}
public <P, T> B route(Class<P> payloadType, GenericRouter<P, T> router,
public <P, T> B route(Class<P> payloadType, Function<P, T> router,
Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer) {
return this.route(payloadType, router, routerConfigurer, null);
}
public <S, T> B route(GenericRouter<S, T> router,
public <S, T> B route(Function<S, T> router,
Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
return route(null, router, routerConfigurer, endpointConfigurer);
}
public <P, T> B route(Class<P> payloadType, GenericRouter<P, T> router,
public <P, T> B route(Class<P> payloadType, Function<P, T> router,
Consumer<RouterSpec<MethodInvokingRouter>> routerConfigurer,
Consumer<GenericEndpointSpec<MethodInvokingRouter>> endpointConfigurer) {
MethodInvokingRouter methodInvokingRouter = isLambda(router)

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

@@ -177,7 +177,7 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
private Object processIntegrationFlowImpl(IntegrationFlow flow, String beanName) {
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(beanName + ".input");
flow.define(flowBuilder);
flow.accept(flowBuilder);
return processStandardIntegrationFlow(flowBuilder.get(), beanName);
}

17
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileTransferringMessageHandlerSpec.java
View File

@@ -24,12 +24,15 @@ import java.util.Collections;
import org.springframework.expression.common.LiteralExpression;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.file.DefaultFileNameGenerator;
import org.springframework.integration.file.FileNameGenerator;
import org.springframework.integration.file.remote.RemoteFileTemplate;
import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
@@ -95,6 +98,11 @@ public abstract class FileTransferringMessageHandlerSpec<F, S extends FileTransf
return _this();
}
public <P> S remoteDirectory(Function<Message<P>, String> remoteDirectoryFunction) {
this.target.setRemoteDirectoryExpression(new FunctionExpression<Message<P>>(remoteDirectoryFunction));
return _this();
}
public S temporaryRemoteDirectory(String temporaryRemoteDirectory) {
this.target.setTemporaryRemoteDirectoryExpression(new LiteralExpression(temporaryRemoteDirectory));
return _this();
@@ -105,17 +113,24 @@ public abstract class FileTransferringMessageHandlerSpec<F, S extends FileTransf
return _this();
}
public <P> S temporaryRemoteDirectory(Function<Message<P>, String> temporaryRemoteDirectoryFunction) {
this.target.setTemporaryRemoteDirectoryExpression(
new FunctionExpression<Message<P>>(temporaryRemoteDirectoryFunction));
return _this();
}
public S useTemporaryFileName(boolean useTemporaryFileName) {
this.target.setUseTemporaryFileName(useTemporaryFileName);
return _this();
}
public S fileNameGenerator(FileNameGenerator fileNameGenerator) {
this.fileNameGenerator = fileNameGenerator;
this.target.setFileNameGenerator(fileNameGenerator);
return _this();
}
public S fileNameGeneratorExpression(String fileNameGeneratorExpression) {
public S fileNameExpression(String fileNameGeneratorExpression) {
Assert.isNull(this.fileNameGenerator,
"'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive.");
this.defaultFileNameGenerator = new DefaultFileNameGenerator();

14
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/FileWritingMessageHandlerSpec.java
View File

@@ -16,15 +16,19 @@
package org.springframework.integration.dsl.file;
import java.io.File;
import java.util.Collection;
import java.util.Collections;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.file.DefaultFileNameGenerator;
import org.springframework.integration.file.FileNameGenerator;
import org.springframework.integration.file.FileWritingMessageHandler;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
@@ -38,7 +42,7 @@ public class FileWritingMessageHandlerSpec
private DefaultFileNameGenerator defaultFileNameGenerator;
FileWritingMessageHandlerSpec(java.io.File destinationDirectory) {
FileWritingMessageHandlerSpec(File destinationDirectory) {
this.target = new FileWritingMessageHandler(destinationDirectory);
}
@@ -46,6 +50,10 @@ public class FileWritingMessageHandlerSpec
this.target = new FileWritingMessageHandler(PARSER.parseExpression(directoryExpression));
}
<P> FileWritingMessageHandlerSpec(Function<Message<P>, ?> directoryFunction) {
this.target = new FileWritingMessageHandler(new FunctionExpression<Message<P>>(directoryFunction));
}
FileWritingMessageHandlerSpec expectReply(boolean expectReply) {
target.setExpectReply(expectReply);
return _this();
@@ -72,11 +80,11 @@ public class FileWritingMessageHandlerSpec
return _this();
}
public FileWritingMessageHandlerSpec fileNameGeneratorExpression(String fileNameGeneratorExpression) {
public FileWritingMessageHandlerSpec fileNameExpression(String fileNameExpression) {
Assert.isNull(this.fileNameGenerator,
"'fileNameGenerator' and 'fileNameGeneratorExpression' are mutually exclusive.");
this.defaultFileNameGenerator = new DefaultFileNameGenerator();
this.defaultFileNameGenerator.setExpression(fileNameGeneratorExpression);
this.defaultFileNameGenerator.setExpression(fileNameExpression);
return fileNameGenerator(this.defaultFileNameGenerator);
}

11
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/Files.java
View File

@@ -19,6 +19,9 @@ package org.springframework.integration.dsl.file;
import java.io.File;
import java.util.Comparator;
import org.springframework.integration.dsl.support.Function;
import org.springframework.messaging.Message;
/**
* @author Artem Bilan
*/
@@ -41,6 +44,10 @@ public abstract class Files {
return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(false);
}
public static <P> FileWritingMessageHandlerSpec outboundAdapter(Function<Message<P>, ?> directoryFunction) {
return new FileWritingMessageHandlerSpec(directoryFunction).expectReply(false);
}
public static FileWritingMessageHandlerSpec outboundGateway(File destinationDirectory) {
return new FileWritingMessageHandlerSpec(destinationDirectory).expectReply(true);
}
@@ -49,6 +56,10 @@ public abstract class Files {
return new FileWritingMessageHandlerSpec(directoryExpression).expectReply(true);
}
public static <P> FileWritingMessageHandlerSpec outboundGateway(Function<Message<P>, ?> directoryFunction) {
return new FileWritingMessageHandlerSpec(directoryFunction).expectReply(true);
}
public static TailAdapterSpec tailAdapter(File file) {
return new TailAdapterSpec().file(file);
}

11
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileInboundChannelAdapterSpec.java
View File

@@ -22,6 +22,8 @@ import java.util.Collections;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.dsl.core.MessageSourceSpec;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource;
@@ -62,8 +64,13 @@ public abstract class RemoteFileInboundChannelAdapterSpec<F, S extends RemoteFil
return _this();
}
public S localFilenameGeneratorExpression(String localFilenameGeneratorExpression) {
this.synchronizer.setLocalFilenameGeneratorExpression(PARSER.parseExpression(localFilenameGeneratorExpression));
public S localFilenameExpression(String localFilenameExpression) {
this.synchronizer.setLocalFilenameGeneratorExpression(PARSER.parseExpression(localFilenameExpression));
return _this();
}
public S localFilename(Function<String, String> localFilenameFunction) {
this.synchronizer.setLocalFilenameGeneratorExpression(new FunctionExpression<String>(localFilenameFunction));
return _this();
}

19
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/RemoteFileOutboundGatewaySpec.java
View File

@@ -19,10 +19,13 @@ package org.springframework.integration.dsl.file;
import java.io.File;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.filters.RegexPatternFileListFilter;
import org.springframework.integration.file.filters.SimplePatternFileListFilter;
import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
@@ -59,7 +62,7 @@ public abstract class RemoteFileOutboundGatewaySpec<F, S extends RemoteFileOutbo
return _this();
}
public S localDirectory(java.io.File localDirectory) {
public S localDirectory(File localDirectory) {
this.target.setLocalDirectory(localDirectory);
return _this();
}
@@ -69,6 +72,11 @@ public abstract class RemoteFileOutboundGatewaySpec<F, S extends RemoteFileOutbo
return _this();
}
public <P> S localDirectory(Function<Message<P>, String> localDirectoryFunction) {
this.target.setLocalDirectoryExpression(new FunctionExpression<Message<P>>(localDirectoryFunction));
return _this();
}
public S autoCreateLocalDirectory(boolean autoCreateLocalDirectory) {
this.target.setAutoCreateLocalDirectory(autoCreateLocalDirectory);
return _this();
@@ -114,8 +122,13 @@ public abstract class RemoteFileOutboundGatewaySpec<F, S extends RemoteFileOutbo
return _this();
}
public S localFilenameGeneratorExpression(String localFilenameGeneratorExpression) {
this.target.setLocalFilenameGeneratorExpression(PARSER.parseExpression(localFilenameGeneratorExpression));
public S localFilenameExpression(String localFilenameExpression) {
this.target.setLocalFilenameGeneratorExpression(PARSER.parseExpression(localFilenameExpression));
return _this();
}
public <P> S localFilename(Function<Message<P>, String> localFilenameFunction) {
this.target.setLocalFilenameGeneratorExpression(new FunctionExpression<Message<P>>(localFilenameFunction));
return _this();
}

4
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/file/TailAdapterSpec.java
View File

@@ -16,6 +16,8 @@
package org.springframework.integration.dsl.file;
import java.io.File;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.core.task.TaskExecutor;
import org.springframework.integration.channel.NullChannel;
@@ -42,7 +44,7 @@ public class TailAdapterSpec extends MessageProducerSpec<TailAdapterSpec, FileTa
this.factoryBean.setBeanFactory(new DefaultListableBeanFactory());
}
TailAdapterSpec file(java.io.File file) {
TailAdapterSpec file(File file) {
Assert.notNull(file);
this.factoryBean.setFile(file);
return _this();

8
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundChannelAdapterSpec.java
View File

@@ -21,9 +21,12 @@ import javax.jms.Destination;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.Consumer;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.jms.JmsHeaderMapper;
import org.springframework.integration.jms.JmsSendingMessageHandler;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
@@ -67,6 +70,11 @@ public class JmsOutboundChannelAdapterSpec<S extends JmsOutboundChannelAdapterSp
return _this();
}
public <P> S destination(Function<Message<P>, ?> destinationFunction) {
this.target.setDestinationExpression(new FunctionExpression<Message<P>>(destinationFunction));
return _this();
}
@Override
protected JmsSendingMessageHandler doGet() {
return null;

13
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/jms/JmsOutboundGatewaySpec.java
View File

@@ -24,10 +24,13 @@ import javax.jms.Destination;
import org.springframework.integration.dsl.core.IntegrationComponentSpec;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.Consumer;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.jms.JmsHeaderMapper;
import org.springframework.integration.jms.JmsOutboundGateway;
import org.springframework.jms.support.converter.MessageConverter;
import org.springframework.jms.support.destination.DestinationResolver;
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
/**
@@ -70,6 +73,11 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec<JmsOutboundGatewa
return _this();
}
public <P> JmsOutboundGatewaySpec requestDestination(Function<Message<P>, ?> destinationFunction) {
this.target.setRequestDestinationExpression(new FunctionExpression<Message<P>>(destinationFunction));
return _this();
}
public JmsOutboundGatewaySpec replyDestination(Destination destination) {
this.target.setReplyDestination(destination);
return _this();
@@ -85,6 +93,11 @@ public class JmsOutboundGatewaySpec extends MessageHandlerSpec<JmsOutboundGatewa
return _this();
}
public <P> JmsOutboundGatewaySpec replyDestination(Function<Message<P>, ?> destinationFunction) {
this.target.setReplyDestinationExpression(new FunctionExpression<Message<P>>(destinationFunction));
return _this();
}
public JmsOutboundGatewaySpec destinationResolver(DestinationResolver destinationResolver) {
this.target.setDestinationResolver(destinationResolver);
return _this();

15
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/ImapIdleChannelAdapterSpec.java
View File

@@ -23,13 +23,16 @@ import java.util.concurrent.Executor;
import javax.mail.Authenticator;
import javax.mail.Session;
import javax.mail.internet.MimeMessage;
import org.aopalliance.aop.Advice;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.dsl.core.MessageProducerSpec;
import org.springframework.integration.dsl.support.Consumer;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.dsl.support.PropertiesBuilder;
import org.springframework.integration.dsl.support.PropertiesBuilder.PropertiesConfigurer;
import org.springframework.integration.mail.ImapIdleChannelAdapter;
import org.springframework.integration.mail.ImapMailReceiver;
import org.springframework.integration.mail.SearchTermStrategy;
@@ -37,6 +40,7 @@ import org.springframework.integration.transaction.TransactionSynchronizationFac
/**
* @author Gary Russell
* @author Artem Bilan
*/
public class ImapIdleChannelAdapterSpec
extends MessageProducerSpec<ImapIdleChannelAdapterSpec, ImapIdleChannelAdapter>
@@ -54,6 +58,11 @@ public class ImapIdleChannelAdapterSpec
return this;
}
public ImapIdleChannelAdapterSpec selector(Function<MimeMessage, Boolean> selectorFunction) {
this.receiver.setSelectorExpression(new FunctionExpression<MimeMessage>(selectorFunction));
return this;
}
public ImapIdleChannelAdapterSpec session(Session session) {
this.receiver.setSession(session);
return this;
@@ -64,9 +73,9 @@ public class ImapIdleChannelAdapterSpec
return this;
}
public ImapIdleChannelAdapterSpec javaMailProperties(PropertiesConfigurer configurer) {
public ImapIdleChannelAdapterSpec javaMailProperties(Consumer<PropertiesBuilder> configurer) {
PropertiesBuilder properties = new PropertiesBuilder();
configurer.configure(properties);
configurer.accept(properties);
return javaMailProperties(properties.get());
}

51
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/MailHeadersBuilder.java
View File

@@ -16,8 +16,11 @@
package org.springframework.integration.dsl.mail;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.dsl.support.MapBuilder;
import org.springframework.integration.mail.MailHeaders;
import org.springframework.messaging.Message;
/**
* @author Artem Bilan
@@ -25,9 +28,6 @@ import org.springframework.integration.mail.MailHeaders;
*/
public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, Object> {
MailHeadersBuilder() {
}
public MailHeadersBuilder subject(String subject) {
return put(MailHeaders.SUBJECT, subject);
}
@@ -36,7 +36,11 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.SUBJECT, subject);
}
public MailHeadersBuilder to(String to) {
public <P> MailHeadersBuilder subjectFunction(Function<Message<P>, String> subject) {
return put(MailHeaders.SUBJECT, new FunctionExpression<Message<P>>(subject));
}
public MailHeadersBuilder to(String... to) {
return put(MailHeaders.TO, to);
}
@@ -44,7 +48,11 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.TO, to);
}
public MailHeadersBuilder cc(String cc) {
public <P> MailHeadersBuilder toFunction(Function<Message<P>, String[]> to) {
return put(MailHeaders.TO, new FunctionExpression<Message<P>>(to));
}
public MailHeadersBuilder cc(String... cc) {
return put(MailHeaders.CC, cc);
}
@@ -52,7 +60,11 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.CC, cc);
}
public MailHeadersBuilder bcc(String bcc) {
public <P> MailHeadersBuilder ccFunction(Function<Message<P>, String[]> cc) {
return put(MailHeaders.CC, new FunctionExpression<Message<P>>(cc));
}
public MailHeadersBuilder bcc(String... bcc) {
return put(MailHeaders.BCC, bcc);
}
@@ -60,6 +72,10 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.BCC, bcc);
}
public <P> MailHeadersBuilder bccFunction(Function<Message<P>, String[]> bcc) {
return put(MailHeaders.BCC, new FunctionExpression<Message<P>>(bcc));
}
public MailHeadersBuilder from(String from) {
return put(MailHeaders.FROM, from);
}
@@ -68,6 +84,10 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.FROM, from);
}
public <P> MailHeadersBuilder fromFunction(Function<Message<P>, String> from) {
return put(MailHeaders.FROM, new FunctionExpression<Message<P>>(from));
}
public MailHeadersBuilder replyTo(String replyTo) {
return put(MailHeaders.REPLY_TO, replyTo);
}
@@ -76,6 +96,10 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.REPLY_TO, replyTo);
}
public <P> MailHeadersBuilder replyToFunction(Function<Message<P>, String> replyTo) {
return put(MailHeaders.REPLY_TO, new FunctionExpression<Message<P>>(replyTo));
}
/**
* @param multipartMode header value
* @return this
@@ -89,6 +113,10 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.MULTIPART_MODE, multipartMode);
}
public <P> MailHeadersBuilder multipartModeFunction(Function<Message<P>, Integer> multipartMode) {
return put(MailHeaders.MULTIPART_MODE, new FunctionExpression<Message<P>>(multipartMode));
}
public MailHeadersBuilder attachmentFilename(String attachmentFilename) {
return put(MailHeaders.ATTACHMENT_FILENAME, attachmentFilename);
}
@@ -97,6 +125,10 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.ATTACHMENT_FILENAME, attachmentFilename);
}
public <P> MailHeadersBuilder attachmentFilenameFunction(Function<Message<P>, String> attachmentFilename) {
return put(MailHeaders.ATTACHMENT_FILENAME, new FunctionExpression<Message<P>>(attachmentFilename));
}
public MailHeadersBuilder contentType(String contentType) {
return put(MailHeaders.CONTENT_TYPE, contentType);
}
@@ -105,8 +137,15 @@ public class MailHeadersBuilder extends MapBuilder<MailHeadersBuilder, String, O
return putExpression(MailHeaders.CONTENT_TYPE, contentType);
}
public <P> MailHeadersBuilder contentTypeFunction(Function<Message<P>, String> contentType) {
return put(MailHeaders.CONTENT_TYPE, new FunctionExpression<Message<P>>(contentType));
}
private MailHeadersBuilder putExpression(String key, String expression) {
return put(key, PARSER.parseExpression(expression));
}
MailHeadersBuilder() {
}
}

16
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/MailInboundChannelAdapterSpec.java
View File

@@ -21,17 +21,20 @@ import java.util.Properties;
import javax.mail.Authenticator;
import javax.mail.Session;
import javax.mail.internet.MimeMessage;
import org.springframework.integration.dsl.core.ComponentsRegistration;
import org.springframework.integration.dsl.core.MessageSourceSpec;
import org.springframework.integration.dsl.support.Consumer;
import org.springframework.integration.dsl.support.Function;
import org.springframework.integration.dsl.support.FunctionExpression;
import org.springframework.integration.dsl.support.PropertiesBuilder;
import org.springframework.integration.dsl.support.PropertiesBuilder.PropertiesConfigurer;
import org.springframework.integration.mail.AbstractMailReceiver;
import org.springframework.integration.mail.MailReceivingMessageSource;
/**
* @author Gary Russell
*
* @author Artem Bilan
*/
public abstract class MailInboundChannelAdapterSpec<S extends MailInboundChannelAdapterSpec<S, R>,
R extends AbstractMailReceiver>
@@ -45,6 +48,11 @@ public abstract class MailInboundChannelAdapterSpec<S extends MailInboundChannel
return _this();
}
public S selector(Function<MimeMessage, Boolean> selectorFunction) {
this.receiver.setSelectorExpression(new FunctionExpression<MimeMessage>(selectorFunction));
return _this();
}
public S session(Session session) {
this.receiver.setSession(session);
return _this();
@@ -55,9 +63,9 @@ public abstract class MailInboundChannelAdapterSpec<S extends MailInboundChannel
return _this();
}
public S javaMailProperties(PropertiesConfigurer configurer) {
public S javaMailProperties(Consumer<PropertiesBuilder> configurer) {
PropertiesBuilder properties = new PropertiesBuilder();
configurer.configure(properties);
configurer.accept(properties);
return javaMailProperties(properties.get());
}

6
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/mail/MailSendingMessageHandlerSpec.java
View File

@@ -20,8 +20,8 @@ import java.util.Properties;
import javax.activation.FileTypeMap;
import org.springframework.integration.dsl.core.MessageHandlerSpec;
import org.springframework.integration.dsl.support.Consumer;
import org.springframework.integration.dsl.support.PropertiesBuilder;
import org.springframework.integration.dsl.support.PropertiesBuilder.PropertiesConfigurer;
import org.springframework.integration.mail.MailSendingMessageHandler;
import org.springframework.mail.javamail.JavaMailSenderImpl;
@@ -44,9 +44,9 @@ public class MailSendingMessageHandlerSpec
return this;
}
public MailSendingMessageHandlerSpec javaMailProperties(PropertiesConfigurer propertiesConfigurer) {
public MailSendingMessageHandlerSpec javaMailProperties(Consumer<PropertiesBuilder> propertiesConfigurer) {
PropertiesBuilder properties = new PropertiesBuilder();
propertiesConfigurer.configure(properties);
propertiesConfigurer.accept(properties);
return javaMailProperties(properties.get());
}

2
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Consumer.java
View File

@@ -19,6 +19,8 @@ package org.springframework.integration.dsl.support;
/**
* Implementations accept a given value and perform work on the argument.
*
* <p>This is a copy of Java 8 {@code Consumer} interface.
*
* @param <T> the type of values to accept
*
* @author Jon Brisbin

2
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Function.java
View File

@@ -20,6 +20,8 @@ package org.springframework.integration.dsl.support;
* Implementations of this class perform work on the given parameter
* and return a result of an optionally different type.
*
* <p>This is a copy of Java 8 {@code Function} interface.
*
* @param <T> The type of the input to the apply operation
* @param <R> The type of the result of the apply operation
*

177
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FunctionExpression.java Normal file
View File

@@ -0,0 +1,177 @@
/*
* 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.support;
import org.springframework.core.convert.TypeDescriptor;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.EvaluationException;
import org.springframework.expression.Expression;
import org.springframework.expression.TypedValue;
import org.springframework.expression.common.ExpressionUtils;
import org.springframework.expression.spel.support.StandardEvaluationContext;
/**
* An {@link Expression} that simply invokes {@link Function#apply(Object)} on its
* provided {@link Function}.
* <p>
* This is a powerful alternative to the SpEL, when Java 8 and its Lambda support is in use.
* <p>
* If the target component has support for an {@link Expression} property,
* a {@link FunctionExpression} can be specified instead of a
* {@link org.springframework.expression.spel.standard.SpelExpression}
* as an alternative to evaluate the value from the Lambda, rather than runtime SpEL resolution.
* <p>
* The {@link FunctionExpression} is 'read-only', hence only {@link #getValue} operations
* are allowed.
* Any {@link #setValue} operations and {@link #getValueType} related operations
* throw {@link EvaluationException}.
*
* @author Artem Bilan
*/
public class FunctionExpression<S> implements Expression {
private final Function<S, ?> function;
private final EvaluationContext defaultContext = new StandardEvaluationContext();
private final EvaluationException readOnlyException;
public FunctionExpression(Function<S, ?> function) {
this.function = function;
this.readOnlyException = new EvaluationException(getExpressionString(),
"FunctionExpression is a 'read only' Expression implementation");
}
@Override
public Object getValue() throws EvaluationException {
return this.function.apply(null);
}
@Override
@SuppressWarnings("unchecked")
public Object getValue(Object rootObject) throws EvaluationException {
return this.function.apply((S) rootObject);
}
@Override
public <T> T getValue(Class<T> desiredResultType) throws EvaluationException {
return getValue(this.defaultContext, desiredResultType);
}
@Override
public <T> T getValue(Object rootObject, Class<T> desiredResultType) throws EvaluationException {
return getValue(this.defaultContext, rootObject, desiredResultType);
}
@Override
public Object getValue(EvaluationContext context) throws EvaluationException {
return getValue();
}
@Override
public Object getValue(EvaluationContext context, Object rootObject) throws EvaluationException {
return getValue(rootObject);
}
@Override
public <T> T getValue(EvaluationContext context, Class<T> desiredResultType) throws EvaluationException {
return ExpressionUtils.convertTypedValue(context, new TypedValue(getValue()), desiredResultType);
}
@Override
public <T> T getValue(EvaluationContext context, Object rootObject, Class<T> desiredResultType)
throws EvaluationException {
return ExpressionUtils.convertTypedValue(context, new TypedValue(getValue(rootObject)), desiredResultType);
}
@Override
public Class<?> getValueType() throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(EvaluationContext context) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public Class<?> getValueType(EvaluationContext context, Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor() throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(Object rootObject) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(EvaluationContext context) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public TypeDescriptor getValueTypeDescriptor(EvaluationContext context, Object rootObject)
throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(EvaluationContext context, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(Object rootObject, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public void setValue(EvaluationContext context, Object rootObject, Object value) throws EvaluationException {
throw this.readOnlyException;
}
@Override
public boolean isWritable(EvaluationContext context) throws EvaluationException {
return false;
}
@Override
public boolean isWritable(EvaluationContext context, Object rootObject) throws EvaluationException {
return false;
}
@Override
public boolean isWritable(Object rootObject) throws EvaluationException {
return false;
}
@Override
public String getExpressionString() {
return this.function.toString();
}
}

26
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/GenericRouter.java
View File

@@ -1,26 +0,0 @@
/*
* 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.support;
/**
* @author Artem Bilan
*/
public interface GenericRouter<S, T> {
T route(S source);
}

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

@@ -1,28 +0,0 @@
/*
* 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.support;
import java.util.Collection;
/**
* @author Artem Bilan
*/
public interface GenericSplitter<T> {
Collection<?> split(T target);
}

6
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MapBuilder.java
View File

@@ -44,10 +44,4 @@ public class MapBuilder<B extends MapBuilder<B, K, V>, K, V> {
return (B) this;
}
public interface MapBuilderConfigurer<B extends MapBuilder<B, K, V>, K, V> {
void configure(MapBuilder<B, K, V> builder);
}
}

6
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/PropertiesBuilder.java
View File

@@ -36,10 +36,4 @@ public class PropertiesBuilder {
return this.properties;
}
public interface PropertiesConfigurer {
void configure(PropertiesBuilder propertiesBuilder);
}
}

11
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/Transformers.java
View File

@@ -26,7 +26,6 @@ import org.springframework.core.io.Resource;
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.support.tuple.Tuple2;
import org.springframework.integration.file.transformer.FileToByteArrayTransformer;
import org.springframework.integration.file.transformer.FileToStringTransformer;
@@ -60,8 +59,6 @@ import org.springframework.xml.xpath.NodeMapper;
*/
public abstract class Transformers {
private final static SpelExpressionParser PARSER = new SpelExpressionParser();
public static ObjectToStringTransformer objectToString() {
return objectToString(null);
}
@@ -325,12 +322,14 @@ public abstract class Transformers {
return transformer;
}
public static XsltPayloadTransformer xslt(Resource xsltTemplate, Tuple2<String, String>... xslParameterMappings) {
@SuppressWarnings("unchecked")
public static XsltPayloadTransformer xslt(Resource xsltTemplate,
Tuple2<String, Expression>... xslParameterMappings) {
XsltPayloadTransformer transformer = new XsltPayloadTransformer(xsltTemplate);
if (xslParameterMappings != null) {
Map<String, Expression> params = new HashMap<String, Expression>(xslParameterMappings.length);
for (Tuple2<String, String> mapping : xslParameterMappings) {
params.put(mapping.getT1(), PARSER.parseExpression(mapping.getT2()));
for (Tuple2<String, Expression> mapping : xslParameterMappings) {
params.put(mapping.getT1(), mapping.getT2());
}
transformer.setXslParameterMappings(params);
}

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

@@ -110,11 +110,10 @@ 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;
import org.springframework.integration.dsl.file.Files;
import org.springframework.integration.dsl.core.Pollers;
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.core.Pollers;
import org.springframework.integration.dsl.support.Transformers;
import org.springframework.integration.dsl.test.TestFtpServer;
import org.springframework.integration.dsl.test.TestSftpServer;
@@ -1422,7 +1421,7 @@ public class IntegrationFlowTests {
.preserveTimestamp(true)
.remoteDirectory("ftpSource")
.regexFilter(".*\\.txt$")
.localFilenameGeneratorExpression("#this.toUpperCase() + '.a'")
.localFilename(f -> f.toUpperCase() + ".a")
.localDirectory(this.ftpServer.getTargetLocalDirectory()),
e -> e.id("ftpInboundAdapter"))
.channel(MessageChannels.queue("ftpInboundResultChannel"))
@@ -1436,7 +1435,7 @@ public class IntegrationFlowTests {
.preserveTimestamp(true)
.remoteDirectory("sftpSource")
.regexFilter(".*\\.txt$")
.localFilenameGeneratorExpression("#this.toUpperCase() + '.a'")
.localFilenameExpression("#this.toUpperCase() + '.a'")
.localDirectory(this.sftpServer.getTargetLocalDirectory()),
e -> e.id("sftpInboundAdapter"))
.channel(MessageChannels.queue("sftpInboundResultChannel"))
@@ -1473,8 +1472,8 @@ public class IntegrationFlowTests {
.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
.regexFileNameFilter("(subFtpSource|.*1.txt)")
.localDirectoryExpression("@ftpServer.targetLocalDirectoryName + #remoteDirectory")
.localFilenameGeneratorExpression("#remoteFileName.replaceFirst('ftpSource', " +
"'localTarget')").get();
.localFilenameExpression("#remoteFileName.replaceFirst('ftpSource', 'localTarget')")
.get();
}
@Bean
@@ -1494,8 +1493,7 @@ public class IntegrationFlowTests {
.options(AbstractRemoteFileOutboundGateway.Option.RECURSIVE)
.regexFileNameFilter("(subSftpSource|.*1.txt)")
.localDirectoryExpression("@sftpServer.targetLocalDirectoryName + #remoteDirectory")
.localFilenameGeneratorExpression(
"#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
.localFilenameExpression("#remoteFileName.replaceFirst('sftpSource', 'localTarget')"))
.channel(remoteFileOutputChannel())
.get();
}
@@ -1733,7 +1731,7 @@ public class IntegrationFlowTests {
.requestPayloadExpression("payload")
.shouldClonePayload(false)
.propertyExpression("name", "payload['name']")
.propertyExpression("date", "new java.util.Date()")
.propertyFunction("date", m -> new Date())
.headerExpression("foo", "payload['name']")
)
.get();
@@ -1757,10 +1755,9 @@ public class IntegrationFlowTests {
public IntegrationFlow enricherFlow3() {
return IntegrationFlows.from("enricherInput3", true)
.enrich(e -> e.requestChannel("enrichChannel")
.requestPayloadExpression("payload")
.shouldClonePayload(false)
.headerExpression("foo", "payload['name']")
)
.requestPayload(Message::getPayload)
.shouldClonePayload(false)
.<Map<String, String>>headerFunction("foo", m -> m.getPayload().get("name")))
.get();
}
@@ -1799,7 +1796,8 @@ public class IntegrationFlowTests {
.split(s -> s.applySequence(false).get().getT2().setDelimiters(","))
.channel(c -> c.executor(this.taskExecutor()))
.<String, Integer>transform(Integer::parseInt)
.enrichHeaders(s -> s.headerExpression(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, "payload"))
.enrichHeaders(h ->
h.headerFunction(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, Message::getPayload))
.resequence(r -> r.releasePartialSequences(true).correlationExpression("'foo'"), null)
.headerFilter("foo", false);
}
@@ -1807,7 +1805,7 @@ public class IntegrationFlowTests {
@Bean
public IntegrationFlow splitAggregateFlow() {
return IntegrationFlows.from("splitAggregateInput", true)
.split(null)
.split()
.channel(MessageChannels.executor(this.taskExecutor()))
.resequence()
.aggregate()
@@ -1967,7 +1965,7 @@ public class IntegrationFlowTests {
e -> e.poller(Pollers.fixedDelay(100)))
.transform(Transformers.fileToString())
.aggregate(a -> a.correlationExpression("1")
.releaseExpression("size() == 25"), null)
.releaseStrategy(g -> g.size() == 25), null)
.channel(MessageChannels.queue("fileReadingResultChannel"))
.get();
}
@@ -1977,7 +1975,7 @@ public class IntegrationFlowTests {
return IntegrationFlows.from("fileWritingInput")
.enrichHeaders(h -> h.header(FileHeaders.FILENAME, "foo.sitest")
.header("directory", new File(tmpDir, "fileWritingFlow")))
.handle(Files.outboundGateway("headers[directory]"))
.handleWithAdapter(a -> a.fileGateway(m -> m.getHeaders().get("directory")))
.channel(MessageChannels.queue("fileWritingResultChannel"))
.get();
}

139
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/jdbc/JdbcTests.java Normal file
View File

@@ -0,0 +1,139 @@
/*
* 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.jdbc;
import static org.junit.Assert.assertNotNull;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Iterator;
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.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.jdbc.InvalidResultSetAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
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;
/**
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class JdbcTests {
@Autowired
@Qualifier("jdbcSplitter.input")
private MessageChannel jdbcSplitterChannel;
@Autowired
private PollableChannel splitResultsChannel;
@Test
public void testJdbcSplitter() {
this.jdbcSplitterChannel.send(new GenericMessage<>("foo"));
for (int i = 0; i < 10; i++) {
Message<?> result = this.splitResultsChannel.receive(1000);
assertNotNull(result);
}
}
@Configuration
@EnableAutoConfiguration
public static class ContextConfiguration {
@Autowired
private JdbcTemplate jdbcTemplate;
@Bean
public IntegrationFlow jdbcSplitter() {
return f ->
f.<String>split(p ->
jdbcTemplate.execute("SELECT * from FOO",
(PreparedStatement ps) ->
new ResultSetIterator<Foo>(ps.executeQuery(),
(rs1, rowNum) ->
new Foo(rs1.getInt(1), rs1.getString(2))))
, null)
.channel(c -> c.queue("splitResultsChannel"));
}
}
private static class Foo {
private final int id;
private final String name;
private Foo(int id, String name) {
this.id = id;
this.name = name;
}
}
private static class ResultSetIterator<T> implements Iterator<T> {
private final ResultSet rs;
private final RowMapper<T> rowMapper;
private ResultSetIterator(ResultSet rs, RowMapper<T> rowMapper) {
this.rs = rs;
this.rowMapper = rowMapper;
}
@Override
public boolean hasNext() {
try {
return !this.rs.isLast();
}
catch (SQLException e) {
throw new InvalidResultSetAccessException(e);
}
}
@Override
public T next() {
try {
this.rs.next();
return this.rowMapper.mapRow(this.rs, this.rs.getRow());
}
catch (SQLException e) {
throw new InvalidResultSetAccessException(e);
}
}
}
}

7
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/mail/MailTests.java
View File

@@ -46,12 +46,12 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.HeaderEnricherSpec;
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.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 +197,10 @@ public class MailTests {
@Bean
public IntegrationFlow sendMailFlow() {
return IntegrationFlows.from("sendMailChannel")
.enrichHeaders(Mail.headers().subject("foo").from("foo@bar").to("bar@baz"))
.enrichHeaders(Mail.headers()
.subjectFunction(m -> "foo")
.from("foo@bar")
.toFunction(m -> new String[] {"bar@baz"}))
.handleWithAdapter(h -> h.mail("localhost")
.port(smtpPort)
.credentials("user", "pw")

10
spring-integration-java-dsl/src/test/resources/data.sql Normal file
View File

@@ -0,0 +1,10 @@
INSERT INTO FOO VALUES (1, 'foo1');
INSERT INTO FOO VALUES (2, 'foo2');
INSERT INTO FOO VALUES (3, 'foo3');
INSERT INTO FOO VALUES (4, 'foo4');
INSERT INTO FOO VALUES (5, 'foo5');
INSERT INTO FOO VALUES (6, 'foo6');
INSERT INTO FOO VALUES (7, 'foo7');
INSERT INTO FOO VALUES (8, 'foo8');
INSERT INTO FOO VALUES (9, 'foo9');
INSERT INTO FOO VALUES (10, 'foo10');

4
spring-integration-java-dsl/src/test/resources/schema.sql Normal file
View File

@@ -0,0 +1,4 @@
CREATE TABLE FOO (
id INTEGER IDENTITY PRIMARY KEY,
name VARCHAR(30),
);