From b20a275a6a0a32168f37b63789d0042ff516dcf4 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Sat, 30 Aug 2014 10:42:44 +0300 Subject: [PATCH] INTSAMPLES-136 Async GW Promise/ListenableFuture 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 --- build.gradle | 9 +- intermediate/async-gateway/.gitignore | 1 + intermediate/async-gateway/README.md | 9 +- .../META-INF/reactor/default.properties | 39 +++++ .../integration/math-service-config.xml | 19 +-- .../async/gateway/AsyncGatewayTest.java | 36 +++-- .../async/gateway/ListenableFutureTest.java | 148 ++++++++++++++++++ .../samples/async/gateway/PromiseTest.java | 144 +++++++++++++++++ 8 files changed, 374 insertions(+), 31 deletions(-) create mode 100644 intermediate/async-gateway/.gitignore create mode 100644 intermediate/async-gateway/src/main/resources/META-INF/reactor/default.properties create mode 100644 intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/ListenableFutureTest.java create mode 100644 intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/PromiseTest.java diff --git a/build.gradle b/build.gradle index 424a45e1..b5757782 100644 --- a/build.gradle +++ b/build.gradle @@ -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" } } diff --git a/intermediate/async-gateway/.gitignore b/intermediate/async-gateway/.gitignore new file mode 100644 index 00000000..ae3c1726 --- /dev/null +++ b/intermediate/async-gateway/.gitignore @@ -0,0 +1 @@ +/bin/ diff --git a/intermediate/async-gateway/README.md b/intermediate/async-gateway/README.md index 360b29a7..93518209 100644 --- a/intermediate/async-gateway/README.md +++ b/intermediate/async-gateway/README.md @@ -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. \ No newline at end of file diff --git a/intermediate/async-gateway/src/main/resources/META-INF/reactor/default.properties b/intermediate/async-gateway/src/main/resources/META-INF/reactor/default.properties new file mode 100644 index 00000000..fbc1192c --- /dev/null +++ b/intermediate/async-gateway/src/main/resources/META-INF/reactor/default.properties @@ -0,0 +1,39 @@ +## +# Dispatcher configuration +# +# Each dispatcher must be configured with a type: +# +# reactor.dispatchers..type = +# +# Legal values for are eventLoop, ringBuffer, synchronous, and threadPoolExecutor. + +# Depending on the type, further configuration is be possible: +# +# reactor.dispatchers..size: eventLoop and threadPoolExecutor Dispatchers +# reactor.dispatchers..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 diff --git a/intermediate/async-gateway/src/main/resources/META-INF/spring/integration/math-service-config.xml b/intermediate/async-gateway/src/main/resources/META-INF/spring/integration/math-service-config.xml index 75f3e3b3..9a082795 100644 --- a/intermediate/async-gateway/src/main/resources/META-INF/spring/integration/math-service-config.xml +++ b/intermediate/async-gateway/src/main/resources/META-INF/spring/integration/math-service-config.xml @@ -11,23 +11,18 @@ + default-request-channel="requestChannel" + async-executor="executor" default-reply-timeout="0"/> - - - + - - - - - - - + + + - + diff --git a/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/AsyncGatewayTest.java b/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/AsyncGatewayTest.java index 08edcbc6..43cb9e4f 100644 --- a/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/AsyncGatewayTest.java +++ b/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/AsyncGatewayTest.java @@ -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> results = new HashMap>(); Random random = new Random(); @@ -50,25 +53,24 @@ public class AsyncGatewayTest { results.put(number, result); } for (final Map.Entry> 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> resultEntry){ int originalNumber = resultEntry.getKey(); Future result = resultEntry.getValue(); diff --git a/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/ListenableFutureTest.java b/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/ListenableFutureTest.java new file mode 100644 index 00000000..0ffb8366 --- /dev/null +++ b/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/ListenableFutureTest.java @@ -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 result = gateway.multiplyByTwo(number); + ListenableFutureCallback callback = new ListenableFutureCallback() { + + @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 multiplyByTwo(int number); + + } + + @MessageEndpoint + public static class Gt100Filter { + + @Filter(inputChannel="gatewayChannel", outputChannel="mathServiceChannel") + public boolean filter(int i) { + return i > 100; + } + + } + +} diff --git a/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/PromiseTest.java b/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/PromiseTest.java new file mode 100644 index 00000000..3206951d --- /dev/null +++ b/intermediate/async-gateway/src/test/java/org/springframework/integration/samples/async/gateway/PromiseTest.java @@ -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 multiplyByTwo(int number); + + } + + @MessageEndpoint + public static class Gt100Filter { + + @Filter(inputChannel = "gatewayChannel", outputChannel = "mathServiceChannel") + public boolean filter(int i) { + return i > 100; + } + + } + +}