diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java new file mode 100644 index 0000000..a9573db --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/AggregatorSpec.java @@ -0,0 +1,64 @@ +/* + * 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 org.springframework.integration.aggregator.AggregatingMessageHandler; +import org.springframework.integration.aggregator.ExpressionEvaluatingMessageGroupProcessor; +import org.springframework.integration.aggregator.MessageGroupProcessor; +import org.springframework.integration.aggregator.MethodInvokingMessageGroupProcessor; + +/** + * @author Artem Bilan + */ +public class AggregatorSpec extends CorrelationHandlerSpec { + + private MessageGroupProcessor outputProcessor; + + private boolean expireGroupsUponCompletion; + + AggregatorSpec() { + } + + public AggregatorSpec processor(Object target, String methodName) { + super.processor(target, methodName); + return this.outputProcessor(methodName != null + ? new MethodInvokingMessageGroupProcessor(target, methodName) + : new MethodInvokingMessageGroupProcessor(target)); + } + + public AggregatorSpec outputExpression(String expression) { + return this.outputProcessor(new ExpressionEvaluatingMessageGroupProcessor(expression)); + } + + public AggregatorSpec outputProcessor(MessageGroupProcessor outputProcessor) { + this.outputProcessor = outputProcessor; + return _this(); + } + + public AggregatorSpec expireGroupsUponCompletion(boolean expireGroupsUponCompletion) { + this.expireGroupsUponCompletion = expireGroupsUponCompletion; + return _this(); + } + + @Override + protected AggregatingMessageHandler doGet() { + AggregatingMessageHandler handler = new AggregatingMessageHandler(this.outputProcessor); + handler.setExpireGroupsUponCompletion(this.expireGroupsUponCompletion); + return this.configure(handler); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java new file mode 100644 index 0000000..f814215 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java @@ -0,0 +1,143 @@ +/* + * 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 org.springframework.integration.aggregator.AbstractCorrelatingMessageHandler; +import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.aggregator.ExpressionEvaluatingCorrelationStrategy; +import org.springframework.integration.aggregator.ExpressionEvaluatingReleaseStrategy; +import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.config.CorrelationStrategyFactoryBean; +import org.springframework.integration.config.ReleaseStrategyFactoryBean; +import org.springframework.integration.dsl.core.IntegrationComponentSpec; +import org.springframework.integration.store.MessageGroupStore; +import org.springframework.messaging.MessageChannel; + +/** + * @author Artem Bilan + */ +public abstract class CorrelationHandlerSpec, H extends AbstractCorrelatingMessageHandler> + extends IntegrationComponentSpec { + + protected MessageGroupStore messageStore; + + protected boolean sendPartialResultOnExpiry; + + private long minimumTimeoutForEmptyGroups; + + private MessageChannel discardChannel; + + private String discardChannelName; + + private CorrelationStrategy correlationStrategy; + + private ReleaseStrategy releaseStrategy; + + public S messageStore(MessageGroupStore messageStore) { + this.messageStore = messageStore; + return _this(); + } + + public S sendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) { + this.sendPartialResultOnExpiry = sendPartialResultOnExpiry; + return _this(); + } + + public S minimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups) { + this.minimumTimeoutForEmptyGroups = minimumTimeoutForEmptyGroups; + return _this(); + } + + public S discardChannel(MessageChannel discardChannel) { + this.discardChannel = discardChannel; + return _this(); + } + + public S discardChannel(String discardChannelName) { + this.discardChannelName = discardChannelName; + return _this(); + } + + public S processor(Object target, String methodName) { + try { + return this.correlationStrategy(new CorrelationStrategyFactoryBean(target, methodName).getObject()) + .releaseStrategy(new ReleaseStrategyFactoryBean(target, methodName).getObject()); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + public S correlationExpression(String correlationExpression) { + return this.correlationStrategy(new ExpressionEvaluatingCorrelationStrategy(correlationExpression)); + } + + public S correlationStrategy(Object target, String methodName) { + try { + return correlationStrategy(new CorrelationStrategyFactoryBean(target, methodName).getObject()); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + public S correlationStrategy(CorrelationStrategy correlationStrategy) { + this.correlationStrategy = correlationStrategy; + return _this(); + } + + public S releaseExpression(String releaseExpression) { + return this.releaseStrategy(new ExpressionEvaluatingReleaseStrategy(releaseExpression)); + } + + public S releaseStrategy(Object target, String methodName) { + try { + return releaseStrategy(new ReleaseStrategyFactoryBean(target, methodName).getObject()); + } + catch (Exception e) { + throw new IllegalStateException(e); + } + } + + public S releaseStrategy(ReleaseStrategy releaseStrategy) { + this.releaseStrategy = releaseStrategy; + return _this(); + } + + protected H configure(H handler) { + if (this.discardChannel != null) { + handler.setDiscardChannel(this.discardChannel); + } + handler.setDiscardChannelName(this.discardChannelName); + if (this.messageStore != null) { + handler.setMessageStore(this.messageStore); + } + handler.setMinimumTimeoutForEmptyGroups(this.minimumTimeoutForEmptyGroups); + handler.setSendPartialResultOnExpiry(this.sendPartialResultOnExpiry); + if (this.correlationStrategy != null) { + handler.setCorrelationStrategy(this.correlationStrategy); + } + if (this.releaseStrategy != null) { + handler.setReleaseStrategy(this.releaseStrategy); + } + return handler; + } + + CorrelationHandlerSpec() { + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java index 4ecd3bf..da87845 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/EnricherSpec.java @@ -1,3 +1,19 @@ +/* + * 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.HashMap; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java index 99c3fda..7a2a172 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java @@ -39,6 +39,11 @@ public final class FilterEndpointSpec extends ConsumerEndpointSpec> endpointConfigurer) { + public IntegrationFlowBuilder handle(String beanName, String methodName, + EndpointConfigurer> endpointConfigurer) { return this.handle(new ServiceActivatingHandler(new BeanNameMessageProcessor(beanName, methodName)), endpointConfigurer); } - public IntegrationFlowBuilder handle(H messageHandler, EndpointConfigurer> endpointConfigurer) { + public IntegrationFlowBuilder handle(H messageHandler, + EndpointConfigurer> endpointConfigurer) { Assert.notNull(messageHandler); return this.register(new GenericEndpointSpec(messageHandler), endpointConfigurer); } @@ -179,7 +186,8 @@ public final class IntegrationFlowBuilder { return this.enrich(enricherConfigurer, null); } - public IntegrationFlowBuilder enrich(ComponentConfigurer enricherConfigurer, EndpointConfigurer> endpointConfigurer) { + public IntegrationFlowBuilder enrich(ComponentConfigurer enricherConfigurer, + EndpointConfigurer> endpointConfigurer) { Assert.notNull(enricherConfigurer); EnricherSpec enricherSpec = new EnricherSpec(); enricherConfigurer.configure(enricherSpec); @@ -210,23 +218,24 @@ public final class IntegrationFlowBuilder { return this.enrichHeaders(headerEnricher, null); } - public IntegrationFlowBuilder enrichHeaders(HeaderEnricher headerEnricher, EndpointConfigurer> endpointConfigurer) { - return this.transform(headerEnricher, endpointConfigurer); + public IntegrationFlowBuilder enrichHeaders(HeaderEnricher headerEnricher, + EndpointConfigurer> endpointConfigurer) { + return this.addComponent(headerEnricher).transform(headerEnricher, endpointConfigurer); } public IntegrationFlowBuilder split() { - return this.split((EndpointConfigurer>) null); + return this.split((EndpointConfigurer>) null); } - public IntegrationFlowBuilder split(EndpointConfigurer> endpointConfigurer) { + public IntegrationFlowBuilder split(EndpointConfigurer> endpointConfigurer) { return this.split(new DefaultMessageSplitter(), endpointConfigurer); } public IntegrationFlowBuilder split(String expression) { - return this.split(expression, (EndpointConfigurer>) null); + return this.split(expression, (EndpointConfigurer>) null); } - public IntegrationFlowBuilder split(String expression, EndpointConfigurer> endpointConfigurer) { + public IntegrationFlowBuilder split(String expression, EndpointConfigurer> endpointConfigurer) { return this.split(new ExpressionEvaluatingSplitter(PARSER.parseExpression(expression)), endpointConfigurer); } @@ -235,7 +244,7 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder split(String beanName, String methodName, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.split(new MethodInvokingSplitter(new BeanNameMessageProcessor>(beanName, methodName)), endpointConfigurer); } @@ -249,12 +258,13 @@ public final class IntegrationFlowBuilder { } public IntegrationFlowBuilder split(GenericSplitter splitter, - EndpointConfigurer> endpointConfigurer) { + EndpointConfigurer> endpointConfigurer) { return this.split(new MethodInvokingSplitter(splitter, "split"), endpointConfigurer); } - public IntegrationFlowBuilder split(T splitter, EndpointConfigurer> endpointConfigurer) { - return this.handle(splitter, endpointConfigurer); + public IntegrationFlowBuilder split(S splitter, EndpointConfigurer> endpointConfigurer) { + Assert.notNull(splitter); + return this.register(new SplitterEndpointSpec(splitter), endpointConfigurer); } /** @@ -281,11 +291,71 @@ public final class IntegrationFlowBuilder { return this.headerFilter(headerFilter, null); } - public IntegrationFlowBuilder headerFilter(HeaderFilter headerFilter, EndpointConfigurer> endpointConfigurer) { + public IntegrationFlowBuilder headerFilter(HeaderFilter headerFilter, + EndpointConfigurer> endpointConfigurer) { return this.transform(headerFilter, endpointConfigurer); } - private > IntegrationFlowBuilder register(S endpointSpec, EndpointConfigurer endpointConfigurer) { + public IntegrationFlowBuilder resequence() { + return this.resequence((EndpointConfigurer>) null); + } + + public IntegrationFlowBuilder resequence(EndpointConfigurer> endpointConfigurer) { + return this.resequence(new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()), endpointConfigurer); + } + + public IntegrationFlowBuilder resequence(ComponentConfigurer resequencerConfigurer) { + return this.resequence(resequencerConfigurer, null); + } + + public IntegrationFlowBuilder resequence(ComponentConfigurer resequencerConfigurer, + EndpointConfigurer> endpointConfigurer) { + Assert.notNull(resequencerConfigurer); + ResequencerSpec spec = new ResequencerSpec(); + resequencerConfigurer.configure(spec); + return this.resequence(spec.get(), endpointConfigurer); + } + + public IntegrationFlowBuilder resequence(ResequencingMessageHandler resequencer) { + return this.resequence(resequencer, null); + } + + public IntegrationFlowBuilder resequence(ResequencingMessageHandler resequencer, + EndpointConfigurer> endpointConfigurer) { + return this.handle(resequencer, endpointConfigurer); + } + + public IntegrationFlowBuilder aggregate() { + return this.aggregate((EndpointConfigurer>) null); + } + + public IntegrationFlowBuilder aggregate(EndpointConfigurer> endpointConfigurer) { + return this.aggregate(new AggregatingMessageHandler(new DefaultAggregatingMessageGroupProcessor()), endpointConfigurer); + } + + public IntegrationFlowBuilder aggregate(ComponentConfigurer aggregatorConfigurer) { + return this.aggregate(aggregatorConfigurer, null); + } + + public IntegrationFlowBuilder aggregate(ComponentConfigurer aggregatorConfigurer, + EndpointConfigurer> endpointConfigurer) { + Assert.notNull(aggregatorConfigurer); + AggregatorSpec spec = new AggregatorSpec(); + aggregatorConfigurer.configure(spec); + return this.aggregate(spec.get(), endpointConfigurer); + } + + public IntegrationFlowBuilder aggregate(AggregatingMessageHandler aggregator) { + return this.aggregate(aggregator, null); + } + + public IntegrationFlowBuilder aggregate(AggregatingMessageHandler aggregator, + EndpointConfigurer> endpointConfigurer) { + return this.handle(aggregator, endpointConfigurer); + } + + + private > IntegrationFlowBuilder register(S endpointSpec, EndpointConfigurer endpointConfigurer) { if (endpointConfigurer != null) { endpointConfigurer.configure(endpointSpec); } @@ -340,6 +410,15 @@ public final class IntegrationFlowBuilder { pollingChannelAdapterFactoryBean.setOutputChannel(outputChannel); } } + else if (this.currentComponent instanceof AbstractCorrelatingMessageHandler) { + AbstractCorrelatingMessageHandler messageProducer = (AbstractCorrelatingMessageHandler) this.currentComponent; + if (channelName != null) { + messageProducer.setOutputChannelName(channelName); + } + else { + messageProducer.setOutputChannel(outputChannel); + } + } else { throw new BeanCreationException("The 'currentComponent' (" + this.currentComponent + ") is a one-way 'MessageHandler'" + " and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow."); diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ResequencerSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ResequencerSpec.java new file mode 100644 index 0000000..e05b6e6 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/ResequencerSpec.java @@ -0,0 +1,42 @@ +/* + * 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 org.springframework.integration.aggregator.ResequencingMessageGroupProcessor; +import org.springframework.integration.aggregator.ResequencingMessageHandler; + +/** + * @author Artem Bilan + */ +public class ResequencerSpec extends CorrelationHandlerSpec { + + private final ResequencingMessageHandler resequencingMessageHandler = new ResequencingMessageHandler(new ResequencingMessageGroupProcessor()); + + ResequencerSpec() { + } + + public ResequencerSpec releasePartialSequences(boolean releasePartialSequences) { + this.resequencingMessageHandler.setReleasePartialSequences(releasePartialSequences); + return _this(); + } + + @Override + protected ResequencingMessageHandler doGet() { + return this.configure(this.resequencingMessageHandler); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java index 25217e4..8c5bbc6 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java @@ -1,3 +1,19 @@ +/* + * 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 org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SplitterEndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SplitterEndpointSpec.java new file mode 100644 index 0000000..c5c2233 --- /dev/null +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/SplitterEndpointSpec.java @@ -0,0 +1,36 @@ +/* + * 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 org.springframework.integration.dsl.core.ConsumerEndpointSpec; +import org.springframework.integration.splitter.AbstractMessageSplitter; + +/** + * @author Artem Bilan + */ +public final class SplitterEndpointSpec extends ConsumerEndpointSpec, S> { + + SplitterEndpointSpec(S splitter) { + super(splitter); + } + + public SplitterEndpointSpec applySequence(boolean applySequence) { + this.target.getT2().setApplySequence(applySequence); + return _this(); + } + +} diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java index 2d23795..bd39af4 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/core/EndpointSpec.java @@ -1,3 +1,19 @@ +/* + * Copyright 2014 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + package org.springframework.integration.dsl.core; import org.springframework.beans.factory.BeanNameAware; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/BeanNameMessageProcessor.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/BeanNameMessageProcessor.java index 75a1113..eab72ff 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/BeanNameMessageProcessor.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/BeanNameMessageProcessor.java @@ -1,3 +1,19 @@ +/* + * 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.beans.BeansException; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java index dd7b64a..cef6258 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/FixedSubscriberChannelPrototype.java @@ -1,3 +1,19 @@ +/* + * 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.messaging.Message; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/GenericSplitter.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/GenericSplitter.java index 76b6406..18a2b6e 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/GenericSplitter.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/GenericSplitter.java @@ -1,3 +1,19 @@ +/* + * 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; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java index fac20b0..99b95bf 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/support/MessageChannelReference.java @@ -1,3 +1,19 @@ +/* + * 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.messaging.Message; diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple.java index e76ebc3..cfcc63e 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple.java @@ -1,11 +1,11 @@ /* - * Copyright (c) 2011-2013 GoPivotal, Inc. All Rights Reserved. + * 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 + * 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, diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple1.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple1.java index 9bb3c07..3d284aa 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple1.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple1.java @@ -1,11 +1,11 @@ /* - * Copyright (c) 2011-2013 GoPivotal, Inc. All Rights Reserved. + * 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 + * 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, diff --git a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple2.java b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple2.java index fa844a8..43a3cd3 100644 --- a/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple2.java +++ b/spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/tuple/Tuple2.java @@ -1,11 +1,11 @@ /* - * Copyright (c) 2011-2013 GoPivotal, Inc. All Rights Reserved. + * 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 + * 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, diff --git a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java index e2f2011..af22a9e 100644 --- a/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java +++ b/spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java @@ -55,6 +55,7 @@ import org.springframework.context.annotation.AnnotationConfigApplicationContext import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.MessageDispatchingException; import org.springframework.integration.annotation.MessageEndpoint; import org.springframework.integration.annotation.ServiceActivator; @@ -64,9 +65,10 @@ import org.springframework.integration.channel.PublishSubscribeChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; import org.springframework.integration.core.MessageSource; -import org.springframework.integration.dsl.GenericEndpointSpec; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.ResequencerSpec; +import org.springframework.integration.dsl.SplitterEndpointSpec; import org.springframework.integration.dsl.channel.DirectChannelSpec; import org.springframework.integration.dsl.channel.MessageChannels; import org.springframework.integration.dsl.support.GenericSplitter; @@ -84,7 +86,7 @@ import org.springframework.integration.store.SimpleMessageStore; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.transformer.PayloadDeserializingTransformer; import org.springframework.integration.transformer.PayloadSerializingTransformer; -import org.springframework.integration.xml.transformer.XPathHeaderEnricher; +import org.springframework.integration.xml.transformer.support.XPathExpressionEvaluatingHeaderValueMessageProcessor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; @@ -175,6 +177,10 @@ public class IntegrationFlowTests { @Qualifier("xpathHeaderEnricherInput") private DirectChannel xpathHeaderEnricherInput; + @Autowired + @Qualifier("splitAggregateInput") + private MessageChannel splitAggregateInput; + @Test public void testPollingFlow() { @@ -336,27 +342,40 @@ public class IntegrationFlowTests { TestPojo result = (TestPojo) payload; assertEquals("Bar Bar", result.getName()); assertNotNull(result.getDate()); - assertThat(new Date(), Matchers.greaterThan(result.getDate())); + assertThat(new Date(), Matchers.greaterThanOrEqualTo(result.getDate())); } @Test - public void testSplitter() { + public void testSplitterResequencer() { QueueChannel replyChannel = new QueueChannel(); this.splitInput.send(MessageBuilder.withPayload("").setReplyChannel(replyChannel).setHeader("foo", "bar").build()); - List results = new ArrayList(); - for (int i = 0; i < 12; i++) { Message receive = replyChannel.receive(2000); assertNotNull(receive); assertFalse(receive.getHeaders().containsKey("foo")); assertTrue(receive.getHeaders().containsKey("FOO")); assertEquals("BAR", receive.getHeaders().get("FOO")); - results.add(receive.getPayload()); + assertEquals(new Integer(i + 1), receive.getPayload()); } + } - assertTrue(results.containsAll(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12))); + @Test + public void testSplitterAggregator() { + List payload = Arrays.asList('a', 'b', 'c', 'd', 'e'); + + QueueChannel replyChannel = new QueueChannel(); + this.splitAggregateInput.send(MessageBuilder.withPayload(payload).setReplyChannel(replyChannel).build()); + + Message receive = replyChannel.receive(2000); + assertNotNull(receive); + assertThat(receive.getPayload(), Matchers.instanceOf(List.class)); + @SuppressWarnings("unchecked") + List result = (List) receive.getPayload(); + for (int i = 0; i < payload.size(); i++) { + assertEquals(payload.get(i), result.get(i)); + } } @Test @@ -609,10 +628,10 @@ public class IntegrationFlowTests { } @Bean - public IntegrationFlow splitFlow() { + public IntegrationFlow splitResequenceFlow() { return IntegrationFlows.from("splitInput") .enrichHeaders(s -> s.header("FOO", "BAR")) - .split("testSplitterData", "buildList", c -> c.get().getT2().setApplySequence(true)) + .split("testSplitterData", "buildList", c -> c.applySequence(false)) .channel(MessageChannels.executor(this.taskExecutor())) .split(new GenericSplitter>>() { @@ -620,21 +639,33 @@ public class IntegrationFlowTests { public Collection split(Message> target) { return target.getPayload(); } - }) + }, c -> c.applySequence(false)) .channel(MessageChannels.executor(this.taskExecutor())) - .split((GenericEndpointSpec s) -> s.get().getT2().setDelimiters(",")) + .split((SplitterEndpointSpec s) -> s.applySequence(false).get().getT2().setDelimiters(",")) .channel(MessageChannels.executor(this.taskExecutor())) .transform(Integer::parseInt) + .enrichHeaders(s -> s.headerExpression(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER, "payload")) + .resequence((ResequencerSpec r) -> r.releasePartialSequences(true).correlationExpression("'foo'")) .headerFilter("foo", false) .get(); } + @Bean + public IntegrationFlow splitAggregateFlow() { + return IntegrationFlows.fromFixedMessageChannel("splitAggregateInput") + .split() + .channel(MessageChannels.executor(this.taskExecutor())) + .resequence() + .aggregate() + .get(); + } + @Bean public IntegrationFlow xpathHeaderEnricherFlow() { return IntegrationFlows.from("xpathHeaderEnricherInput") .enrichHeaders( - s -> s.header("one", new XPathHeaderEnricher.XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne")) - .header("two", new XPathHeaderEnricher.XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo")), + s -> s.header("one", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementOne")) + .header("two", new XPathExpressionEvaluatingHeaderValueMessageProcessor("/root/elementTwo")), c -> c.autoStartup(false).id("xpathHeaderEnricher") ) .get();