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

INTSAMPLES-136 Async GW Promise/ListenableFuture

Browse Source 
JIRA: https://jira.spring.io/browse/INTSAMPLES-136

Polishing - PR Comments

* Add `compile "org.projectreactor.spring:reactor-spring-context:$reactorSpringVersion"` dependency
to show a usage of `@EnableReactor`
* Make `async-gateway` project Java 8 compatible and replace inline implementations for callbacks to Lambdas
This commit is contained in:
Gary Russell
2014-08-30 10:42:44 +03:00
committed by Artem Bilan
parent 3296463fb8
commit b20a275a6a
8 changed files with 374 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
build.gradle
View File

@@ -138,6 +138,8 @@ subprojects { subproject ->
}
repositories {
//TODO on release
maven { url 'http://repo.spring.io/libs-snapshot' }
maven { url 'http://repo.spring.io/libs-milestone' }
// maven { url 'http://repo.spring.io/libs-staging-local' }
}
@@ -185,10 +187,12 @@ subprojects { subproject ->
mockitoVersion = '1.9.5'
openJpaVersion = '2.3.0'
oracleDriverVersion = '11.2.0.3'
reactorSpringVersion = '1.1.3.RELEASE'
postgresVersion = '9.1-901-1.jdbc4'
subethasmtpVersion = '1.2'
slf4jVersion = '1.7.6'
springIntegrationVersion = '4.0.3.RELEASE'
springIntegration41Version = '4.1.0.BUILD-SNAPSHOT'
springIntegrationDslVersion = '1.0.0.M2'
springVersion = '4.0.5.RELEASE'
springSecurityVersion = '3.2.4.RELEASE'
@@ -787,8 +791,11 @@ project('xmpp') {
project('async-gateway') {
description = 'Async Gateway Sample'
sourceCompatibility = 1.8
dependencies {
compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion"
compile "org.springframework.integration:spring-integration-core:$springIntegration41Version"
compile "org.projectreactor.spring:reactor-spring-context:$reactorSpringVersion"
}
}

1
intermediate/async-gateway/.gitignore vendored Normal file
View File

@@ -0,0 +1 @@
/bin/

9
intermediate/async-gateway/README.md
View File

@@ -5,7 +5,7 @@ Gateways provide a convenient way to expose a Proxy over a service-interface thu
But what about the cases where you can't (e.g, message was filtered out and discarded or routed into a unidirectional sub-flow)?
Starting with Spring Integration 2.0, we are introducing support for an Asynchronous Gateway, which is a convenient way to initiate flows, where you may not know, if a reply is expected or how long will it take for it to arrive. A natural way to handle these types of scenarios in Java would be to rely upon **java.util.concurrent.Future** instances. That is exactly what Spring Integration uses to support Asynchronous Gateways.
Starting with Spring Integration 2.0, we introduced support for an Asynchronous Gateway, which is a convenient way to initiate flows, where you may not know, if a reply is expected or how long will it take for it to arrive. A natural way to handle these types of scenarios in Java would be to rely upon **java.util.concurrent.Future** instances. That is exactly what Spring Integration uses to support Asynchronous Gateways.
This example demonstrates how you can apply an Asynchronous Gateway based on the following simple use case:
@@ -24,3 +24,10 @@ You should see the following output:
INFO : org.springframework.integration.samples.async.gateway.AsyncGatewayTest - Multiplication of 39 by 2 is can not be accomplished in 20 seconds
INFO : org.springframework.integration.samples.async.gateway.AsyncGatewayTest - Multiplication of 36 by 2 is can not be accomplished in 20 seconds
INFO : org.springframework.integration.samples.async.gateway.AsyncGatewayTest - Multiplication of 37 by 2 is can not be accomplished in 20 seconds
Spring Integration 4.0 provided the capability to more easily configure Messaging Gateways with Java configuration.
Spring Integration 4.1 added support for **ListenableFuture** and **Promise** (from project reactor) return types.
The **ListenableFutureTest** and **PromiseTest** test classes replicate the above test case, using those return types, and showing the use of **@MessagingGateway** java configuration.

39
intermediate/async-gateway/src/main/resources/META-INF/reactor/default.properties Normal file
View File

@@ -0,0 +1,39 @@
##
# Dispatcher configuration
#
# Each dispatcher must be configured with a type:
#
# reactor.dispatchers.<name>.type = <type>
#
# Legal values for <type> are eventLoop, ringBuffer, synchronous, and threadPoolExecutor.
# Depending on the type, further configuration is be possible:
#
# reactor.dispatchers.<name>.size: eventLoop and threadPoolExecutor Dispatchers
# reactor.dispatchers.<name>.backlog: eventLoop, ringBuffer, and threadPoolExecutor Dispatchers
#
# A size less than 1 may be specified to indicate that the size should be the same as the number
# of CPUs.
# A thread pool executor dispatcher, named threadPoolExecutor
reactor.dispatchers.threadPoolExecutor.type = threadPoolExecutor
reactor.dispatchers.threadPoolExecutor.size = 100
# Backlog is how many Task objects to warm up internally
reactor.dispatchers.threadPoolExecutor.backlog = 100
# An event loop dispatcher, named eventLoop
reactor.dispatchers.eventLoop.type =
reactor.dispatchers.eventLoop.size = 0
reactor.dispatchers.eventLoop.backlog = 2048
# A ring buffer dispatcher, named ringBuffer
reactor.dispatchers.ringBuffer.type =
reactor.dispatchers.ringBuffer.backlog = 2048
# A work queue dispatcher, named workQueue
reactor.dispatchers.workQueue.type =
reactor.dispatchers.workQueue.size = 0
reactor.dispatchers.workQueue.backlog = 2048
# The dispatcher named ringBuffer should be the default dispatcher
reactor.dispatchers.default = threadPoolExecutor

19
intermediate/async-gateway/src/main/resources/META-INF/spring/integration/math-service-config.xml
View File

@@ -11,23 +11,18 @@
<int:gateway id="mathService"
service-interface="org.springframework.integration.samples.async.gateway.MathServiceGateway"
default-request-channel="requestChannel"/>
default-request-channel="requestChannel"
async-executor="executor" default-reply-timeout="0"/>
<int:channel id="requestChannel">
<int:queue/>
</int:channel>
<int:channel id="requestChannel" />
<int:filter input-channel="requestChannel" output-channel="calculatingChannel" expression="payload &gt; 100">
<int:poller fixed-rate="100" max-messages-per-poll="10" task-executor="executor"/>
</int:filter>
<int:channel id="calculatingChannel" >
<int:dispatcher task-executor="executor"/>
</int:channel>
<int:filter input-channel="requestChannel" output-channel="calculatingChannel" expression="payload gt 100" />
<int:channel id="calculatingChannel" />
<int:service-activator input-channel="calculatingChannel">
<bean class="org.springframework.integration.samples.async.gateway.MathService"/>
</int:service-activator>
<task:executor id="executor" pool-size="10"/>
<task:executor id="executor" pool-size="100" />
</beans>

36
intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/AsyncGatewayTest.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-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.
@@ -27,20 +27,23 @@ import java.util.concurrent.TimeoutException;
import org.apache.log4j.Logger;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.support.FileSystemXmlApplicationContext;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
*
*/
public class AsyncGatewayTest {
private static Logger logger = Logger.getLogger(AsyncGatewayTest.class);
private static ExecutorService executor = Executors.newFixedThreadPool(100);
private static int timeout = 20;
@Test
public void testAsyncGateway() throws Exception{
ApplicationContext ac = new FileSystemXmlApplicationContext("src/main/resources/META-INF/spring/integration/*.xml");
ConfigurableApplicationContext ac =
new FileSystemXmlApplicationContext("src/main/resources/META-INF/spring/integration/*.xml");
MathServiceGateway mathService = ac.getBean("mathService", MathServiceGateway.class);
Map<Integer, Future<Integer>> results = new HashMap<Integer, Future<Integer>>();
Random random = new Random();
@@ -50,25 +53,24 @@ public class AsyncGatewayTest {
results.put(number, result);
}
for (final Map.Entry<Integer, Future<Integer>> resultEntry : results.entrySet()) {
executor.execute(new Runnable() {
public void run() {
int[] result = processFuture(resultEntry);
if (result[1] == -1){
logger.info("Multiplying " + result[0] + " should be easy. You should be able to multiply any number < 100 by 2 in your head");
} else if (result[1] == -2){
logger.info("Multiplication of " + result[0] + " by 2 is can not be accomplished in " + timeout + " seconds");
} else {
logger.info("Result of multiplication of " + result[0] + " by 2 is " + result[1]);
}
executor.execute(() -> {
int[] result = processFuture(resultEntry);
if (result[1] == -1){
logger.info("Multiplying " + result[0] + " should be easy. You should be able to multiply any number < 100 by 2 in your head");
} else if (result[1] == -2){
logger.info("Multiplication of " + result[0] + " by 2 is can not be accomplished in " + timeout + " seconds");
} else {
logger.info("Result of multiplication of " + result[0] + " by 2 is " + result[1]);
}
});
});
}
executor.shutdown();
executor.awaitTermination(60, TimeUnit.SECONDS);
logger.info("Finished");
ac.close();
}
public static int[] processFuture(Map.Entry<Integer, Future<Integer>> resultEntry){
int originalNumber = resultEntry.getKey();
Future<Integer> result = resultEntry.getValue();

148
intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/ListenableFutureTest.java Normal file
View File

@@ -0,0 +1,148 @@
/*
* Copyright 2002-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.samples.async.gateway;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.annotation.Filter;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
*
*/
@ContextConfiguration(classes = ListenableFutureTest.TestConfig.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class ListenableFutureTest {
private static Logger logger = Logger.getLogger(ListenableFutureTest.class);
@Autowired
private MathGateway gateway;
@Test
public void testAsyncGateway() throws Exception{
Random random = new Random();
int[] numbers = new int[100];
int expectedResults = 0;
for (int i = 0; i < 100; i++) {
numbers[i] = random.nextInt(200);
if (numbers[i] > 100) {
expectedResults++;
}
}
final CountDownLatch latch = new CountDownLatch(expectedResults);
final AtomicInteger failures = new AtomicInteger();
for (int i = 0; i < 100; i++) {
final int number = numbers[i];
ListenableFuture<Integer> result = gateway.multiplyByTwo(number);
ListenableFutureCallback<Integer> callback = new ListenableFutureCallback<Integer>() {
@Override
public void onSuccess(Integer result) {
logger.info("Result of multiplication of " + number + " by 2 is " + result);
latch.countDown();
}
@Override
public void onFailure(Throwable t) {
failures.incrementAndGet();
logger.error("Unexpected exception for " + number, t);
latch.countDown();
}
};
result.addCallback(callback);
}
assertTrue(latch.await(60, TimeUnit.SECONDS));
assertEquals(0, failures.get());
logger.info("Finished");
}
@Configuration
@ComponentScan
@EnableIntegration
@IntegrationComponentScan
public static class TestConfig {
@Bean
public MessageChannel gatewayChannel() {
return new DirectChannel();
}
@Bean
@ServiceActivator(inputChannel="mathServiceChannel")
public MathService mathService() {
return new MathService();
}
@Bean
public AsyncTaskExecutor exec() {
SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
simpleAsyncTaskExecutor.setThreadNamePrefix("exec-");
return simpleAsyncTaskExecutor;
}
}
@MessagingGateway(defaultReplyTimeout = 0)
public interface MathGateway {
@Gateway(requestChannel = "gatewayChannel")
ListenableFuture<Integer> multiplyByTwo(int number);
}
@MessageEndpoint
public static class Gt100Filter {
@Filter(inputChannel="gatewayChannel", outputChannel="mathServiceChannel")
public boolean filter(int i) {
return i > 100;
}
}
}

144
intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/PromiseTest.java Normal file
View File

@@ -0,0 +1,144 @@
/*
* Copyright 2002-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.samples.async.gateway;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;
import org.junit.Test;
import org.junit.runner.RunWith;
import reactor.core.composable.Promise;
import reactor.spring.context.config.EnableReactor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.AsyncTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.integration.annotation.Filter;
import org.springframework.integration.annotation.Gateway;
import org.springframework.integration.annotation.IntegrationComponentScan;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.MessagingGateway;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.messaging.MessageChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Oleg Zhurakousky
* @author Gary Russell
*
*/
@ContextConfiguration(classes = PromiseTest.TestConfig.class)
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class PromiseTest {
private static Logger logger = Logger.getLogger(PromiseTest.class);
@Autowired
private MathGateway gateway;
@Test
public void testPromiseGateway() throws Exception {
Random random = new Random();
int[] numbers = new int[100];
int expectedResults = 0;
for (int i = 0; i < 100; i++) {
numbers[i] = random.nextInt(200);
if (numbers[i] > 100) {
expectedResults++;
}
}
final CountDownLatch latch = new CountDownLatch(expectedResults);
final AtomicInteger failures = new AtomicInteger();
for (int i = 0; i < 100; i++) {
final int number = numbers[i];
gateway.multiplyByTwo(number)
.onSuccess(result1 -> {
logger.info("Result of multiplication of " + number + " by 2 is " + result1);
latch.countDown();
})
.onError(t -> {
failures.incrementAndGet();
logger.error("Unexpected exception for " + number, t);
latch.countDown();
})
.flush();
}
assertTrue(latch.await(60, TimeUnit.SECONDS));
assertEquals(0, failures.get());
logger.info("Finished");
}
@Configuration
@EnableIntegration
@EnableReactor
@ComponentScan
@IntegrationComponentScan
public static class TestConfig {
@Bean
public MessageChannel gatewayChannel() {
return new DirectChannel();
}
@Bean
@ServiceActivator(inputChannel = "mathServiceChannel")
public MathService mathService() {
return new MathService();
}
@Bean
public AsyncTaskExecutor exec() {
SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = new SimpleAsyncTaskExecutor();
simpleAsyncTaskExecutor.setThreadNamePrefix("exec-");
return simpleAsyncTaskExecutor;
}
}
@MessagingGateway(defaultReplyTimeout = 0, reactorEnvironment = "reactorEnv")
public interface MathGateway {
@Gateway(requestChannel = "gatewayChannel")
Promise<Integer> multiplyByTwo(int number);
}
@MessageEndpoint
public static class Gt100Filter {
@Filter(inputChannel = "gatewayChannel", outputChannel = "mathServiceChannel")
public boolean filter(int i) {
return i > 100;
}
}
}