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

INT-4497: Add RateLimiterRequestHandlerAdvice (#2781)

Browse Source 
* INT-4497: Add RateLimiterRequestHandlerAdvice

JIRA: https://jira.spring.io/browse/INT-4497

* * Remove unused property
This commit is contained in:
Artem Bilan
2019-03-06 16:38:04 -05:00
committed by Gary Russell
parent e32f87731a
commit e9216287c2
5 changed files with 289 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
build.gradle
View File

@@ -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"

147
spring-integration-core/src/main/java/org/springframework/integration/handler/advice/RateLimiterRequestHandlerAdvice.java Normal file
View File

@@ -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
* <a href="https://github.com/resilience4j/resilience4j#ratelimiter">Resilience4j</a>.
*
* @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<Object> 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);
}
}
}

100
spring-integration-core/src/test/java/org/springframework/integration/handler/advice/RateLimiterRequestHandlerAdviceTests.java Normal file
View File

@@ -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;
}
}
}

34
src/reference/asciidoc/handler-advice.adoc
View File

@@ -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 <<retry-advice>>)
* `RequestHandlerCircuitBreakerAdvice` (described in <<circuit-breaker-advice>>)
* `ExpressionEvaluatingRequestHandlerAdvice` (described in <<expression-advice>>)
* `RateLimiterRequestHandlerAdvice` (described in <<rate-limiter-advice>>)
[[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

6
src/reference/asciidoc/whats-new.adoc
View File

@@ -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 <<rate-limiter-advice>> for more information.
[[x5.2-general]]
=== General Changes