Merge pull request #67 from ghillert/INTSAMPLES-23
INTSAMPLES-23 - Add Splitter-Aggregator example
This commit is contained in:
Gunnar Hillert
@@ -10,7 +10,7 @@ Welcome to the Spring Integration Samples. To simplify your experience, Spring I
|
||||
* Advanced
|
||||
* Applications
|
||||
|
||||
Inside of each category you'll find a 'README.md' which will contain a more detailed description of that category's specifics. Each sample also comes with its own 'readme.txt' file explaining details.
|
||||
Inside of each category you'll find a 'README.md' which will contain a more detailed description of that category's specifics. Each sample also comes with its own 'readme.txt' file explaining details.
|
||||
|
||||
*Happy Integration!*
|
||||
|
||||
@@ -70,6 +70,7 @@ This category targets developers who are already more familiar with the Spring I
|
||||
* **stored-procedures-oracle** Provides an example of the stored procedure Outbound Gateway using *ORACLE XE*
|
||||
* **stored-procedures-postgresql** Provides an example of the stored procedure Outbound Gateway using *[PostgreSQL](http://www.postgresql.org/)*
|
||||
* **rest-http** - This sample demonstrates how to send an HTTP request to a Spring Integration's HTTP service while utilizing Spring Integration's new HTTP Path usage. This sample also uses Spring Security for HTTP Basic authentication. With HTTP Path facility, the client program can send requests with URL Variables.
|
||||
* **splitter-aggregator-reaper** A demonstration of implementing the Splitter and Aggregator *Enterprise Integration Patterns* (EIP) together. This sample also provides a concrete example of a [message store reaper][] in action.
|
||||
* **stored-procedures-derby** Provides an example of the stored procedure Outbound Gateway using *[Apache Derby](http://db.apache.org/derby/)*
|
||||
* **stored-procedures-oracle** Provides an example of the stored procedure Outbound Gateway using *ORACLE XE*
|
||||
* **monitoring** The project used in the *[Spring Integration Management and Monitoring Webinar](http://www.springsource.org/node/3598)* Also available on the *[SpringSourceDev YouTube Channel](http://www.youtube.com/SpringSourceDev)*
|
||||
@@ -94,8 +95,11 @@ This category targets developers and architects who have a good understanding of
|
||||
|
||||
For more information, please visit the Spring Integration website at: [http://www.springsource.org/spring-integration](http://www.springsource.org/spring-integration)
|
||||
|
||||
[Spring Integration]: https://github.com/SpringSource/spring-integration
|
||||
[Spring Integration Extensions]: https://github.com/SpringSource/spring-integration-extensions
|
||||
[Spring Integration Templates]: https://github.com/SpringSource/spring-integration-templates/tree/master/si-sts-templates
|
||||
[Spring Integration Dsl Groovy]: https://github.com/SpringSource/spring-integration-dsl-groovy
|
||||
[Spring Integration Dsl Scala]: https://github.com/SpringSource/spring-integration-dsl-scala
|
||||
[Spring Integration Pattern Catalog]: https://github.com/SpringSource/spring-integration-pattern-catalog
|
||||
|
||||
[message store reaper]: http://static.springsource.org/spring-integration/api/org/springframework/integration/store/MessageGroupStoreReaper.html
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
<module>errorhandling</module>
|
||||
<module>file-processing</module>
|
||||
<module>multipart-http</module>
|
||||
<module>splitter-aggregator-reaper</module>
|
||||
<module>stored-procedures-derby</module>
|
||||
<module>tcp-client-server-multiplex</module>
|
||||
<module>travel</module>
|
||||
|
||||
50
intermediate/splitter-aggregator-reaper/README.md
Normal file
50
intermediate/splitter-aggregator-reaper/README.md
Normal file
@@ -0,0 +1,50 @@
|
||||
Splitter/Aggregator with Reaper Sample
|
||||
======================================
|
||||
|
||||
# Overview
|
||||
|
||||
Demonstration of how to implement the Splitter/Aggregator [Enterprise Integration Patterns][] (EIP) using *Spring Integration*. This sample provides a demonstration of request-reply, [splitting][] a message, and then [aggregating][] the replies. Furthermore, this sample is processing the split messages concurrently and also deals with timeout conditions. Lastly, this sample provides a concrete example of a [message store reaper][] in action.
|
||||
|
||||
# Run the Sample
|
||||
|
||||
* running the "Main" class from within STS (Right-click on Main class --> Run As --> Java Application)
|
||||
* or from the command line:
|
||||
- mvn package
|
||||
- mvn exec:java
|
||||
|
||||
You should see the following output:
|
||||
|
||||
13:47:54.886 INFO [main][org.springframework.integration.samples.splitteraggregator.Main]
|
||||
=========================================================
|
||||
|
||||
Welcome to Spring Integration!
|
||||
|
||||
For more information please visit:
|
||||
http://www.springsource.org/spring-integration
|
||||
|
||||
=========================================================
|
||||
13:47:56.039 INFO [main][org.springframework.integration.store.MessageGroupStoreReaper] started org.springframework.integration.store.MessageGroupStoreReaper@4b85c17
|
||||
Please enter a choice and press <enter>:
|
||||
1. Submit 2 search queries, 2 results returned.
|
||||
2. Submit 2 search queries, 1 search query takes too long, 1 results returned.
|
||||
3. Submit 2 search queries, 2 search queries take too long, 0 results returned.
|
||||
q. Quit the application
|
||||
Enter you choice: 1
|
||||
13:48:01.036 INFO [searchRequestExecutor-2][org.springframework.integration.samples.splitteraggregator.SearchA] This search will take 1000ms.
|
||||
13:48:01.036 INFO [searchRequestExecutor-1][org.springframework.integration.samples.splitteraggregator.SearchB] This search will take 1000ms.
|
||||
Number of Search Results: 2
|
||||
|
||||
# Credits
|
||||
|
||||
We would like to thank Christopher Hunt ([@huntchr](http://twitter.com/huntchr)) for contributing this sample.
|
||||
|
||||
--------------------------------------------------------------------------------
|
||||
|
||||
For help please take a look at the Spring Integration documentation:
|
||||
|
||||
http://www.springsource.org/spring-integration
|
||||
|
||||
[aggregating]: http://static.springsource.org/spring-integration/reference/html/messaging-routing-chapter.html#aggregator
|
||||
[Enterprise Integration Patterns]: http://www.eaipatterns.com/
|
||||
[message store reaper]: http://static.springsource.org/spring-integration/reference/html/messaging-routing-chapter.html#aggregator-config
|
||||
[splitting]: http://static.springsource.org/spring-integration/reference/html/messaging-routing-chapter.html#splitter
|
||||
101
intermediate/splitter-aggregator-reaper/pom.xml
Normal file
101
intermediate/splitter-aggregator-reaper/pom.xml
Normal file
@@ -0,0 +1,101 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<groupId>org.springframework.integration.samples</groupId>
|
||||
<artifactId>splitter-aggregator-reaper</artifactId>
|
||||
<version>2.2.0.BUILD-SNAPSHOT</version>
|
||||
<packaging>jar</packaging>
|
||||
|
||||
<name>Samples (Intermediate) - Splitter-Aggregator-Reaper</name>
|
||||
<url>http://www.springsource.org/spring-integration</url>
|
||||
<description>
|
||||
A demonstration of implementing the Splitter and Aggregator Enterprise
|
||||
Integration Patterns (EIP) together
|
||||
</description>
|
||||
|
||||
<prerequisites>
|
||||
<maven>2.2.1</maven>
|
||||
</prerequisites>
|
||||
|
||||
<properties>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
<spring.integration.version>2.2.0.RC2</spring.integration.version>
|
||||
<log4j.version>1.2.16</log4j.version>
|
||||
<junit.version>4.10</junit.version>
|
||||
<inject.version>1</inject.version>
|
||||
<spring.version>3.1.2.RELEASE</spring.version>
|
||||
</properties>
|
||||
|
||||
<repositories>
|
||||
<repository>
|
||||
<id>repo.springsource.org.milestone</id>
|
||||
<name>Spring Framework Maven Milestone Repository</name>
|
||||
<url>https://repo.springsource.org/libs-milestone</url>
|
||||
</repository>
|
||||
</repositories>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-compiler-plugin</artifactId>
|
||||
<version>2.3.2</version>
|
||||
<configuration>
|
||||
<source>1.6</source>
|
||||
<target>1.6</target>
|
||||
<compilerArgument>-Xlint:all</compilerArgument>
|
||||
<showWarnings>true</showWarnings>
|
||||
<showDeprecation>true</showDeprecation>
|
||||
</configuration>
|
||||
</plugin>
|
||||
<plugin>
|
||||
<groupId>org.codehaus.mojo</groupId>
|
||||
<artifactId>exec-maven-plugin</artifactId>
|
||||
<version>1.2.1</version>
|
||||
<configuration>
|
||||
<mainClass>org.springframework.integration.samples.splitteraggregator.Main</mainClass>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
<dependencies>
|
||||
|
||||
<!-- Compile dependencies -->
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.integration</groupId>
|
||||
<artifactId>spring-integration-core</artifactId>
|
||||
<version>${spring.integration.version}</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>javax.inject</groupId>
|
||||
<artifactId>javax.inject</artifactId>
|
||||
<version>${inject.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Logging -->
|
||||
|
||||
<dependency>
|
||||
<groupId>log4j</groupId>
|
||||
<artifactId>log4j</artifactId>
|
||||
<version>${log4j.version}</version>
|
||||
</dependency>
|
||||
|
||||
<!-- Test dependencies -->
|
||||
|
||||
<dependency>
|
||||
<groupId>junit</groupId>
|
||||
<artifactId>junit</artifactId>
|
||||
<version>${junit.version}</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework</groupId>
|
||||
<artifactId>spring-test</artifactId>
|
||||
<version>${spring.version}</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1,26 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
/**
|
||||
* A class that represents all criteria.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public abstract class AbstractCriteria {
|
||||
|
||||
}
|
||||
@@ -0,0 +1,34 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
|
||||
/**
|
||||
* Criteria that contains other criteria.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class CompositeCriteria extends AbstractCriteria {
|
||||
|
||||
private final Collection<AbstractCriteria> criteria = new ArrayList<AbstractCriteria>();
|
||||
|
||||
public Collection<AbstractCriteria> getCriteria() {
|
||||
return criteria;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,33 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
|
||||
/**
|
||||
* A result that can contain other results.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class CompositeResult extends Result {
|
||||
private Collection<Result> results = new ArrayList<Result>();
|
||||
|
||||
public Collection<Result> getResults() {
|
||||
return results;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
/**
|
||||
* One type of criteria.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class CriteriaA extends AbstractCriteria {
|
||||
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
/**
|
||||
* Another type of criteria.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class CriteriaB extends AbstractCriteria {
|
||||
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
import java.util.Scanner;
|
||||
|
||||
import org.apache.log4j.Logger;
|
||||
import org.springframework.context.support.AbstractApplicationContext;
|
||||
import org.springframework.context.support.ClassPathXmlApplicationContext;
|
||||
import org.springframework.integration.samples.splitteraggregator.support.TestUtils;
|
||||
|
||||
|
||||
/**
|
||||
* Starts the Spring Context and will initialize the Spring Integration routes.
|
||||
*
|
||||
* @author Gunnar Hillert
|
||||
* @version 1.0
|
||||
*
|
||||
*/
|
||||
public final class Main {
|
||||
|
||||
private static final Logger LOGGER = Logger.getLogger(Main.class);
|
||||
|
||||
private Main() { }
|
||||
|
||||
/**
|
||||
* Load the Spring Integration Application Context
|
||||
*
|
||||
* @param args - command line arguments
|
||||
*/
|
||||
public static void main(final String... args) {
|
||||
|
||||
LOGGER.info("\n========================================================="
|
||||
+ "\n "
|
||||
+ "\n Welcome to Spring Integration! "
|
||||
+ "\n "
|
||||
+ "\n For more information please visit: "
|
||||
+ "\n http://www.springsource.org/spring-integration "
|
||||
+ "\n "
|
||||
+ "\n=========================================================" );
|
||||
|
||||
final AbstractApplicationContext context =
|
||||
new ClassPathXmlApplicationContext("classpath:META-INF/spring/integration/*-context.xml");
|
||||
|
||||
context.registerShutdownHook();
|
||||
|
||||
final SearchRequestor searchRequestor = context.getBean(SearchRequestor.class);
|
||||
final SearchA searchA = context.getBean(SearchA.class);
|
||||
final SearchB searchB = context.getBean(SearchB.class);
|
||||
|
||||
final Scanner scanner = new Scanner(System.in);
|
||||
|
||||
System.out.println("Please enter a choice and press <enter>: ");
|
||||
System.out.println("\t1. Submit 2 search queries, 2 results returned.");
|
||||
System.out.println("\t2. Submit 2 search queries, 1 search query takes too long, 1 result returned.");
|
||||
System.out.println("\t3. Submit 2 search queries, 2 search queries take too long, 0 results returned.");
|
||||
|
||||
System.out.println("\tq. Quit the application");
|
||||
System.out.print("Enter your choice: ");
|
||||
|
||||
while (true) {
|
||||
final String input = scanner.nextLine();
|
||||
|
||||
if("1".equals(input.trim())) {
|
||||
searchA.setExecutionTime(1000L);
|
||||
searchB.setExecutionTime(1000L);
|
||||
final CompositeResult result = searchRequestor.search(TestUtils.getCompositeCriteria());
|
||||
System.out.println("Number of Search Results: " + result.getResults().size());
|
||||
} else if("2".equals(input.trim())) {
|
||||
searchA.setExecutionTime(6000L);
|
||||
searchB.setExecutionTime(1000L);
|
||||
final CompositeResult result = searchRequestor.search(TestUtils.getCompositeCriteria());
|
||||
System.out.println("Number of Search Results: " + result.getResults().size());
|
||||
} else if("3".equals(input.trim())) {
|
||||
searchA.setExecutionTime(6000L);
|
||||
searchB.setExecutionTime(6000L);
|
||||
final CompositeResult result = searchRequestor.search(TestUtils.getCompositeCriteria());
|
||||
System.out.println("Result is null: " + (result == null));
|
||||
} else if("q".equals(input.trim())) {
|
||||
break;
|
||||
} else {
|
||||
System.out.println("Invalid choice\n\n");
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
System.out.println("Exiting application...bye.");
|
||||
System.exit(0);
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
/**
|
||||
* The result for a search query.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class Result {
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
import org.apache.log4j.Logger;
|
||||
|
||||
/**
|
||||
* One type of search.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class SearchA {
|
||||
|
||||
private static final Logger LOGGER = Logger.getLogger(SearchA.class);
|
||||
|
||||
private long executionTime = 1000L;
|
||||
|
||||
public Result search(CriteriaA criteria) {
|
||||
|
||||
LOGGER.info(String.format("This search will take %sms.", executionTime));
|
||||
|
||||
try {
|
||||
Thread.sleep(executionTime);
|
||||
} catch (InterruptedException e) {
|
||||
}
|
||||
return new Result();
|
||||
}
|
||||
|
||||
public long getExecutionTime() {
|
||||
return executionTime;
|
||||
}
|
||||
|
||||
public void setExecutionTime(long executionTime) {
|
||||
this.executionTime = executionTime;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,50 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
import org.apache.log4j.Logger;
|
||||
|
||||
/**
|
||||
* Another type of search.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class SearchB {
|
||||
|
||||
private static final Logger LOGGER = Logger.getLogger(SearchB.class);
|
||||
|
||||
private long executionTime = 1000L;
|
||||
|
||||
public Result search(CriteriaB criteria) {
|
||||
|
||||
LOGGER.info(String.format("This search will take %sms.", executionTime));
|
||||
|
||||
try {
|
||||
Thread.sleep(executionTime);
|
||||
} catch (InterruptedException e) {
|
||||
}
|
||||
return new Result();
|
||||
}
|
||||
|
||||
public long getExecutionTime() {
|
||||
return executionTime;
|
||||
}
|
||||
|
||||
public void setExecutionTime(long executionTime) {
|
||||
this.executionTime = executionTime;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,31 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
import java.util.Collection;
|
||||
|
||||
/**
|
||||
* Given CompositeCriteria, return a collection of the individual criterion.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class SearchRequestSplitter {
|
||||
|
||||
public Collection<AbstractCriteria> split(CompositeCriteria criteria) {
|
||||
return criteria.getCriteria();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
/**
|
||||
* A service interface responsible for performing a search and returning a
|
||||
* result synchronously.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public interface SearchRequestor {
|
||||
|
||||
CompositeResult search(CompositeCriteria criteria);
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
import java.util.Collection;
|
||||
|
||||
/**
|
||||
* Aggregate a collection of results into a composite result object.
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
*
|
||||
*/
|
||||
public class SearchResultAggregator {
|
||||
public Result aggregate(Collection<Result> results) {
|
||||
CompositeResult result = new CompositeResult();
|
||||
result.getResults().addAll(results);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,44 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator.support;
|
||||
|
||||
import org.springframework.integration.samples.splitteraggregator.CompositeCriteria;
|
||||
import org.springframework.integration.samples.splitteraggregator.CriteriaA;
|
||||
import org.springframework.integration.samples.splitteraggregator.CriteriaB;
|
||||
|
||||
|
||||
/**
|
||||
* @author Gunnar Hillert
|
||||
* @since 1.0
|
||||
*
|
||||
*/
|
||||
public final class TestUtils {
|
||||
|
||||
/**
|
||||
* Private constructor to prevent instantiation.
|
||||
*/
|
||||
private TestUtils() {
|
||||
}
|
||||
|
||||
public static CompositeCriteria getCompositeCriteria() {
|
||||
|
||||
final CompositeCriteria generalCriteria = new CompositeCriteria();
|
||||
generalCriteria.getCriteria().add(new CriteriaA());
|
||||
generalCriteria.getCriteria().add(new CriteriaB());
|
||||
return generalCriteria;
|
||||
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<beans xmlns="http://www.springframework.org/schema/beans"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration"
|
||||
xmlns:context="http://www.springframework.org/schema/context"
|
||||
xmlns:task="http://www.springframework.org/schema/task"
|
||||
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
|
||||
http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd
|
||||
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
|
||||
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
|
||||
|
||||
<!-- Integration -->
|
||||
|
||||
<int:gateway id="searchRequestor" default-reply-channel="search-aggregated-reply"
|
||||
default-request-channel="search-request" default-reply-timeout="5000"
|
||||
service-interface="org.springframework.integration.samples.splitteraggregator.SearchRequestor">
|
||||
</int:gateway>
|
||||
|
||||
<int:channel id="search-request"></int:channel>
|
||||
<int:channel id="search-aggregated-reply"></int:channel>
|
||||
|
||||
<int:splitter input-channel="search-request" ref="searchRequestSplitter"
|
||||
output-channel="search-requests">
|
||||
</int:splitter>
|
||||
|
||||
<int:channel id="search-requests"></int:channel>
|
||||
|
||||
<int:payload-type-router input-channel="search-requests">
|
||||
<int:mapping
|
||||
type="org.springframework.integration.samples.splitteraggregator.CriteriaA"
|
||||
channel="search-request-a" />
|
||||
<int:mapping
|
||||
type="org.springframework.integration.samples.splitteraggregator.CriteriaB"
|
||||
channel="search-request-b" />
|
||||
</int:payload-type-router>
|
||||
|
||||
<int:channel id="search-request-a">
|
||||
<int:queue capacity="10" />
|
||||
</int:channel>
|
||||
<int:service-activator input-channel="search-request-a"
|
||||
method="search" output-channel="search-reply" ref="searchA">
|
||||
<int:poller max-messages-per-poll="1" fixed-delay="10"
|
||||
receive-timeout="30000" task-executor="searchRequestExecutor" />
|
||||
</int:service-activator>
|
||||
|
||||
<int:channel id="search-request-b">
|
||||
<int:queue capacity="10" />
|
||||
</int:channel>
|
||||
<int:service-activator input-channel="search-request-b"
|
||||
method="search" output-channel="search-reply" ref="searchB">
|
||||
<int:poller max-messages-per-poll="1" fixed-delay="10"
|
||||
receive-timeout="30000" task-executor="searchRequestExecutor" />
|
||||
</int:service-activator>
|
||||
|
||||
<int:publish-subscribe-channel id="search-reply" />
|
||||
<int:aggregator input-channel="search-reply" method="aggregate"
|
||||
ref="searchResultAggregator" output-channel="search-aggregated-reply"
|
||||
message-store="searchResultMessageStore"
|
||||
send-partial-result-on-expiry="true">
|
||||
</int:aggregator>
|
||||
|
||||
<!-- Define a store for our search results and set up a reaper that will
|
||||
periodically expire those results. -->
|
||||
<bean id="searchResultMessageStore" class="org.springframework.integration.store.SimpleMessageStore" />
|
||||
|
||||
<bean id="searchResultMessageStoreReaper" class="org.springframework.integration.store.MessageGroupStoreReaper">
|
||||
<property name="messageGroupStore" ref="searchResultMessageStore" />
|
||||
<property name="timeout" value="2000" />
|
||||
</bean>
|
||||
|
||||
<task:scheduled-tasks>
|
||||
<task:scheduled ref="searchResultMessageStoreReaper" method="run" fixed-rate="1000" />
|
||||
</task:scheduled-tasks>
|
||||
|
||||
<!-- Execution of search requests. We are stating here that we only ever
|
||||
let three searches of any type run concurrently (2 in the pool, one in the
|
||||
calling thread if we already have two in operation). The queue size of 0
|
||||
means that we don't allow the executor to queue things up (we take care of
|
||||
queues further up the chain). -->
|
||||
<task:executor id="searchRequestExecutor" pool-size="2" queue-capacity="0"
|
||||
rejection-policy="CALLER_RUNS" />
|
||||
|
||||
<!-- Enable @Inject -->
|
||||
<context:annotation-config />
|
||||
|
||||
<!-- Application beans -->
|
||||
<bean id="searchRequestSplitter" class="org.springframework.integration.samples.splitteraggregator.SearchRequestSplitter" />
|
||||
<bean id="searchA" class="org.springframework.integration.samples.splitteraggregator.SearchA" />
|
||||
<bean id="searchB" class="org.springframework.integration.samples.splitteraggregator.SearchB" />
|
||||
<bean id="searchResultAggregator" class="org.springframework.integration.samples.splitteraggregator.SearchResultAggregator" />
|
||||
|
||||
</beans>
|
||||
@@ -0,0 +1,36 @@
|
||||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
|
||||
<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
|
||||
|
||||
<!-- Appenders -->
|
||||
<appender name="console" class="org.apache.log4j.ConsoleAppender">
|
||||
<param name="Target" value="System.out" />
|
||||
<layout class="org.apache.log4j.PatternLayout">
|
||||
<param name="ConversionPattern" value="%d{HH:mm:ss.SSS} %-5p [%t][%c] %m%n" />
|
||||
</layout>
|
||||
</appender>
|
||||
|
||||
<!-- Loggers -->
|
||||
<logger name="org.springframework.integration">
|
||||
<level value="warn" />
|
||||
</logger>
|
||||
|
||||
<logger name="org.springframework.integration.store">
|
||||
<level value="info" />
|
||||
</logger>
|
||||
|
||||
<logger name="org.springframework.integration.aggregator">
|
||||
<level value="info" />
|
||||
</logger>
|
||||
|
||||
<logger name="org.springframework.integration.samples">
|
||||
<level value="info" />
|
||||
</logger>
|
||||
|
||||
<!-- Root Logger -->
|
||||
<root>
|
||||
<priority value="warn" />
|
||||
<appender-ref ref="console" />
|
||||
</root>
|
||||
|
||||
</log4j:configuration>
|
||||
@@ -0,0 +1,81 @@
|
||||
/*
|
||||
* Copyright 2002-2012 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.samples.splitteraggregator;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNull;
|
||||
|
||||
import javax.inject.Inject;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.springframework.integration.samples.splitteraggregator.support.TestUtils;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
|
||||
/**
|
||||
* Integration tests for the Scatter-Gather
|
||||
*
|
||||
* @author Christopher Hunt
|
||||
* @author Gunnar Hillert
|
||||
*
|
||||
*/
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@ContextConfiguration({"classpath:META-INF/spring/integration/spring-integration-context.xml"})
|
||||
public class TestSplitterAggregator {
|
||||
|
||||
@Inject
|
||||
SearchRequestor searchRequestor;
|
||||
|
||||
@Inject
|
||||
SearchA searchA;
|
||||
|
||||
@Inject
|
||||
SearchB searchB;
|
||||
|
||||
/**
|
||||
* Test the happy path.
|
||||
*
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
@Test
|
||||
public void testSearch() throws InterruptedException {
|
||||
CompositeResult result = searchRequestor.search(TestUtils.getCompositeCriteria());
|
||||
assertEquals(2, result.getResults().size());
|
||||
}
|
||||
|
||||
/**
|
||||
* Test searchA taking longer than we're expecting.
|
||||
*/
|
||||
@Test
|
||||
public void testSearchNoSearchA() {
|
||||
searchA.setExecutionTime(6000L);
|
||||
CompositeResult result = searchRequestor.search(TestUtils.getCompositeCriteria());
|
||||
assertEquals(1, result.getResults().size());
|
||||
}
|
||||
|
||||
/**
|
||||
* Test both searchA and searchB taking longer than we're expecting.
|
||||
*/
|
||||
@Test
|
||||
public void testSearchNoSearchB() {
|
||||
searchA.setExecutionTime(6000L);
|
||||
searchB.setExecutionTime(6000L);
|
||||
CompositeResult result = searchRequestor.search(TestUtils.getCompositeCriteria());
|
||||
assertNull(result);
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user