diff --git a/build.gradle b/build.gradle index 7408b2b67f..a38cabdeeb 100644 --- a/build.gradle +++ b/build.gradle @@ -1,4 +1,4 @@ -buildscript { +buildscript { ext.kotlinVersion = '1.3.21' repositories { maven { url 'https://repo.spring.io/plugins-release' } @@ -131,6 +131,7 @@ subprojects { subproject -> postgresVersion = '42.2.5' reactorNettyVersion = '0.8.5.RELEASE' reactorVersion = '3.2.6.RELEASE' + resilience4jVersion = '0.13.2' romeToolsVersion = '1.12.0' servletApiVersion = '4.0.1' smackVersion = '4.3.1' @@ -381,6 +382,7 @@ project('spring-integration-core') { compile("io.fastjson:boon:$boonVersion", optional) compile("com.esotericsoftware:kryo-shaded:$kryoShadedVersion", optional) compile("io.micrometer:micrometer-core:$micrometerVersion", optional) + compile("io.github.resilience4j:resilience4j-ratelimiter:$resilience4jVersion", optional) testCompile ("org.aspectj:aspectjweaver:$aspectjVersion") testCompile "io.projectreactor:reactor-test:$reactorVersion" diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RateLimiterRequestHandlerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RateLimiterRequestHandlerAdvice.java new file mode 100644 index 0000000000..7270f2a82f --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RateLimiterRequestHandlerAdvice.java @@ -0,0 +1,147 @@ +/* + * Copyright 2019 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.handler.advice; + +import java.time.Duration; + +import org.springframework.messaging.Message; +import org.springframework.messaging.MessagingException; +import org.springframework.util.Assert; + +import io.github.resilience4j.ratelimiter.RateLimiter; +import io.github.resilience4j.ratelimiter.RateLimiterConfig; +import io.github.resilience4j.ratelimiter.RequestNotPermitted; +import io.vavr.CheckedFunction0; +import io.vavr.control.Try; + +/** + * An {@link AbstractRequestHandlerAdvice} extension for a rate limiting to service method calls. + * The implementation is based on the + * Resilience4j. + * + * @author Artem Bilan + * + * @since 5.2 + */ +public class RateLimiterRequestHandlerAdvice extends AbstractRequestHandlerAdvice { + + public static final String DEFAULT_NAME = "RateLimiterRequestHandlerAdvice"; + + private final RateLimiter rateLimiter; + + /** + * Construct an instance based on default rate limiter options + * and {@value #DEFAULT_NAME} as a rate limiter name. + * @see RateLimiter#ofDefaults + */ + public RateLimiterRequestHandlerAdvice() { + this(RateLimiter.ofDefaults(DEFAULT_NAME)); + } + + /** + * Construct an instance based on default rate limiter options and provided name. + * @param name the name for the rate limiter. + */ + public RateLimiterRequestHandlerAdvice(String name) { + this(RateLimiter.ofDefaults(name)); + Assert.hasText(name, "'name' must not be empty"); + } + + /** + * Construct an instance based on the provided {@link RateLimiter}. + * @param rateLimiter the {@link RateLimiter} to use. + */ + public RateLimiterRequestHandlerAdvice(RateLimiter rateLimiter) { + Assert.notNull(rateLimiter, "'rateLimiter' must not be null"); + this.rateLimiter = rateLimiter; + } + + /** + * Construct an instance based on the provided {@link RateLimiterConfig} + * and {@value #DEFAULT_NAME} as a rate limiter name. + * @param rateLimiterConfig the {@link RateLimiterConfig} to use. + */ + public RateLimiterRequestHandlerAdvice(RateLimiterConfig rateLimiterConfig) { + this(rateLimiterConfig, DEFAULT_NAME); + } + + /** + * Construct an instance based on the provided {@link RateLimiterConfig} and name. + * @param rateLimiterConfig the {@link RateLimiterConfig} to use. + * @param name the name for the rate limiter. + */ + public RateLimiterRequestHandlerAdvice(RateLimiterConfig rateLimiterConfig, String name) { + Assert.notNull(rateLimiterConfig, "'rateLimiterConfig' must not be null"); + Assert.hasText(name, "'name' must not be empty"); + this.rateLimiter = RateLimiter.of(name, rateLimiterConfig); + } + + /** + * Change the {@code limitForPeriod} option of the {@link #rateLimiter}. + * @param limitForPeriod the {@code limitForPeriod} to use. + * @see RateLimiter#changeLimitForPeriod(int) + */ + public void setLimitForPeriod(int limitForPeriod) { + this.rateLimiter.changeLimitForPeriod(limitForPeriod); + } + + /** + * Change the {@code timeoutDuration} option of the {@link #rateLimiter}. + * @param timeoutDuration the {@code timeoutDuration} to use. + * @see RateLimiter#changeTimeoutDuration(Duration) + */ + public void setTimeoutDuration(Duration timeoutDuration) { + this.rateLimiter.changeTimeoutDuration(timeoutDuration); + } + + /** + * Obtain the metrics from the rate limiter. + * @return the {@link RateLimiter.Metrics} from rate limiter. + * @see RateLimiter#getMetrics() + */ + public RateLimiter.Metrics getMetrics() { + return this.rateLimiter.getMetrics(); + } + + @Override + protected Object doInvoke(ExecutionCallback callback, Object target, Message message) throws Exception { + CheckedFunction0 restrictedCall = + RateLimiter.decorateCheckedSupplier(this.rateLimiter, callback::execute); + try { + return Try.of(restrictedCall).get(); + } + catch (RequestNotPermitted ex) { + throw new RateLimitExceededException(message, "Rate limit exceeded for: " + target, ex); + } + } + + + /** + * A {@link MessagingException} wrapper for the {@link RequestNotPermitted} + * with the {@code requestMessage} and {@code target} context. + */ + public static class RateLimitExceededException extends MessagingException { + + private static final long serialVersionUID = 1L; + + RateLimitExceededException(Message message, String description, RequestNotPermitted cause) { + super(message, description, cause); + } + + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/RateLimiterRequestHandlerAdviceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/RateLimiterRequestHandlerAdviceTests.java new file mode 100644 index 0000000000..570ca050cf --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/RateLimiterRequestHandlerAdviceTests.java @@ -0,0 +1,100 @@ +/* + * Copyright 2019 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.handler.advice; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; + +import java.time.Duration; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import io.github.resilience4j.ratelimiter.RateLimiterConfig; +import io.github.resilience4j.ratelimiter.RequestNotPermitted; + +/** + * @author Artem Bilan + * + * @since 5.2 + */ +@SpringJUnitConfig +public class RateLimiterRequestHandlerAdviceTests { + + @Autowired + private MessageChannel requestChannel; + + @Autowired + private PollableChannel resultChannel; + + @Test + void testRateLimiter() throws InterruptedException { + Message testMessage = new GenericMessage<>("test"); + this.requestChannel.send(testMessage); + + assertThatExceptionOfType(MessagingException.class) + .isThrownBy(() -> this.requestChannel.send(testMessage)) + .withCauseInstanceOf(RequestNotPermitted.class) + .withMessageContaining("Rate limit exceeded for: "); + + Thread.sleep(200); + + this.requestChannel.send(testMessage); + + assertThat(this.resultChannel.receive(10_000)).isNotNull(); + assertThat(this.resultChannel.receive(10_000)).isNotNull(); + } + + @Configuration + @EnableIntegration + public static class ContextConfiguration { + + @Bean + public RateLimiterRequestHandlerAdvice rateLimiterRequestHandlerAdvice() { + return new RateLimiterRequestHandlerAdvice(RateLimiterConfig.custom() + .timeoutDuration(Duration.ofMillis(100)) + .limitRefreshPeriod(Duration.ofMillis(500)) + .limitForPeriod(1) + .build()); + } + + @Bean + public PollableChannel resultChannel() { + return new QueueChannel(); + } + + @ServiceActivator(inputChannel = "requestChannel", outputChannel = "resultChannel", + adviceChain = "rateLimiterRequestHandlerAdvice") + public String handleRequest(String payload) { + return payload; + } + + } + +} diff --git a/src/reference/asciidoc/handler-advice.adoc b/src/reference/asciidoc/handler-advice.adoc index 4870155ff3..803d7e5c60 100644 --- a/src/reference/asciidoc/handler-advice.adoc +++ b/src/reference/asciidoc/handler-advice.adoc @@ -49,11 +49,12 @@ For chains that produce a reply, every child element can be advised. [[advice-classes]] ==== Provided Advice Classes -In addition to providing the general mechanism to apply AOP advice classes, Spring Integration provides three standard advice classes: +In addition to providing the general mechanism to apply AOP advice classes, Spring Integration provides these out-of-the-box advice implementations: * `RequestHandlerRetryAdvice` (described in <>) * `RequestHandlerCircuitBreakerAdvice` (described in <>) * `ExpressionEvaluatingRequestHandlerAdvice` (described in <>) +* `RateLimiterRequestHandlerAdvice` (described in <>) [[retry-advice]] ===== Retry Advice @@ -464,6 +465,37 @@ public class EerhaApplication { ---- ==== +[[rate-limiter-advice]] +===== Rate Limiter Advice + +The Rate Limiter advice (`RateLimiterRequestHandlerAdvice`) allows to ensure that an endpoint does not get overloaded with requests. +When the rate limit is breached the request will go in a blocked state. + +A typical use case for this advice might be an external service provider not allowing more than `n` number of request per minute. + +The `RateLimiterRequestHandlerAdvice` implementation is fully based on the https://github.com/resilience4j/resilience4j#ratelimiter[Resilience4j] project and requires either `RateLimiter` or `RateLimiterConfig` injections. +Can also be configured with defaults and/or custom name. + +The following example configures a rate limiter advice with one request per 1 second: +==== +[source, java] +---- +@Bean +public RateLimiterRequestHandlerAdvice rateLimiterRequestHandlerAdvice() { + return new RateLimiterRequestHandlerAdvice(RateLimiterConfig.custom() + .limitRefreshPeriod(Duration.ofSeconds(1)) + .limitForPeriod(1) + .build()); +} + +@ServiceActivator(inputChannel = "requestChannel", outputChannel = "resultChannel", + adviceChain = "rateLimiterRequestHandlerAdvice") +public String handleRequest(String payload) { + ... +} +---- +==== + [[custom-advice]] ==== Custom Advice Classes diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 645a4b65e9..ad87b341e2 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -7,6 +7,12 @@ If you are interested in more details, see the Issue Tracker tickets that were r [[x5.2-new-components]] === New Components +[[x5.2-rateLimitAdvice]] +=== Rate Limit Advice Support + +The `RateLimiterRequestHandlerAdvice` is now available for limiting requests rate on handlers. +See <> for more information. + [[x5.2-general]] === General Changes