Add adapter module and implementation for Spring WebFlux

- The implementation leverages Sentinel Reactor Adapter. Two main components: SentinelWebFluxFilter and SentinelBlockExceptionHandler

Signed-off-by: Eric Zhao <sczyh16@gmail.com>

sentinel-adapter/pom.xml

@@ -20,6 +20,7 @@
         <module>sentinel-grpc-adapter</module>
         <module>sentinel-zuul-adapter</module>
         <module>sentinel-reactor-adapter</module>
+        <module>sentinel-spring-webflux-adapter</module>
     </modules>
 
     <dependencyManagement>
@@ -39,6 +40,11 @@
                 <artifactId>sentinel-web-servlet</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.alibaba.csp</groupId>
+                <artifactId>sentinel-reactor-adapter</artifactId>
+                <version>${project.version}</version>
+            </dependency>
 
             <dependency>
                 <groupId>junit</groupId>

sentinel-adapter/sentinel-spring-webflux-adapter/pom.xml

@@ -0,0 +1,55 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<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">
+    <parent>
+        <artifactId>sentinel-adapter</artifactId>
+        <groupId>com.alibaba.csp</groupId>
+        <version>1.5.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>sentinel-spring-webflux-adapter</artifactId>
+
+    <properties>
+        <java.source.version>1.8</java.source.version>
+        <java.target.version>1.8</java.target.version>
+        <spring.version>5.1.5.RELEASE</spring.version>
+        <spring.boot.version>2.1.3.RELEASE</spring.boot.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-reactor-adapter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-webflux</artifactId>
+            <version>${spring.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-webflux</artifactId>
+            <version>${spring.boot.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <version>${spring.boot.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

sentinel-adapter/sentinel-spring-webflux-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/spring/webflux/SentinelWebFluxFilter.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright 1999-2019 Alibaba Group Holding Ltd.
+ *
+ * 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 com.alibaba.csp.sentinel.adapter.spring.webflux;
+
+import java.util.Optional;
+
+import com.alibaba.csp.sentinel.EntryType;
+import com.alibaba.csp.sentinel.adapter.reactor.ContextConfig;
+import com.alibaba.csp.sentinel.adapter.reactor.EntryConfig;
+import com.alibaba.csp.sentinel.adapter.reactor.SentinelReactorTransformer;
+import com.alibaba.csp.sentinel.adapter.spring.webflux.callback.WebFluxCallbackManager;
+
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.WebFilter;
+import org.springframework.web.server.WebFilterChain;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author Eric Zhao
+ * @since 1.5.0
+ */
+public class SentinelWebFluxFilter implements WebFilter {
+
+    @Override
+    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
+        return chain.filter(exchange)
+            .transform(buildSentinelTransformer(exchange));
+    }
+
+    private SentinelReactorTransformer<Void> buildSentinelTransformer(ServerWebExchange exchange) {
+        // Maybe we can get the URL pattern elsewhere via:
+        // exchange.getAttributeOrDefault(HandlerMapping.BEST_MATCHING_PATTERN_ATTRIBUTE, path)
+
+        String path = exchange.getRequest().getPath().value();
+        String finalPath = Optional.ofNullable(WebFluxCallbackManager.getUrlCleaner())
+            .map(f -> f.apply(exchange, path))
+            .orElse(path);
+        String origin = Optional.ofNullable(WebFluxCallbackManager.getRequestOriginParser())
+            .map(f -> f.apply(exchange))
+            .orElse(EMPTY_ORIGIN);
+
+        return new SentinelReactorTransformer<>(
+            new EntryConfig(finalPath, EntryType.IN, new ContextConfig(finalPath, origin)));
+    }
+
+    private static final String EMPTY_ORIGIN = "";
+}

sentinel-adapter/sentinel-spring-webflux-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/spring/webflux/callback/BlockRequestHandler.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 1999-2019 Alibaba Group Holding Ltd.
+ *
+ * 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 com.alibaba.csp.sentinel.adapter.spring.webflux.callback;
+
+import org.springframework.web.reactive.function.server.ServerResponse;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Mono;
+
+/**
+ * Reactive handler for the blocked request.
+ *
+ * @author Eric Zhao
+ * @since 1.5.0
+ */
+@FunctionalInterface
+public interface BlockRequestHandler {
+
+    /**
+     * Handle the blocked request.
+     *
+     * @param exchange server exchange object
+     * @param t block exception
+     * @return server response to return
+     */
+    Mono<ServerResponse> handleRequest(ServerWebExchange exchange, Throwable t);
+}

sentinel-adapter/sentinel-spring-webflux-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/spring/webflux/callback/DefaultBlockRequestHandler.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright 1999-2019 Alibaba Group Holding Ltd.
+ *
+ * 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 com.alibaba.csp.sentinel.adapter.spring.webflux.callback;
+
+import java.util.List;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.InvalidMediaTypeException;
+import org.springframework.http.MediaType;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Mono;
+
+import static org.springframework.web.reactive.function.BodyInserters.fromObject;
+
+/**
+ * The default implementation of {@link BlockRequestHandler}.
+ *
+ * @author Eric Zhao
+ * @since 1.5.0
+ */
+public class DefaultBlockRequestHandler implements BlockRequestHandler {
+
+    private static final String DEFAULT_BLOCK_MSG_PREFIX = "Blocked by Sentinel: ";
+
+    @Override
+    public Mono<ServerResponse> handleRequest(ServerWebExchange exchange, Throwable ex) {
+        if (acceptsHtml(exchange)) {
+            return htmlErrorResponse(ex);
+        }
+        // JSON result by default.
+        return ServerResponse.status(HttpStatus.TOO_MANY_REQUESTS)
+            .contentType(MediaType.APPLICATION_JSON_UTF8)
+            .body(fromObject(buildErrorResult(ex)));
+    }
+
+    private Mono<ServerResponse> htmlErrorResponse(Throwable ex) {
+        return ServerResponse.status(HttpStatus.TOO_MANY_REQUESTS)
+            .contentType(MediaType.TEXT_PLAIN)
+            .syncBody(DEFAULT_BLOCK_MSG_PREFIX + ex.getClass().getSimpleName());
+    }
+
+    private ErrorResult buildErrorResult(Throwable ex) {
+        return new ErrorResult(HttpStatus.TOO_MANY_REQUESTS.value(),
+            DEFAULT_BLOCK_MSG_PREFIX + ex.getClass().getSimpleName());
+    }
+
+    /**
+     * Reference from {@code DefaultErrorWebExceptionHandler} of Spring Boot.
+     */
+    private boolean acceptsHtml(ServerWebExchange exchange) {
+        try {
+            List<MediaType> acceptedMediaTypes = exchange.getRequest().getHeaders().getAccept();
+            acceptedMediaTypes.remove(MediaType.ALL);
+            MediaType.sortBySpecificityAndQuality(acceptedMediaTypes);
+            return acceptedMediaTypes.stream()
+                .anyMatch(MediaType.TEXT_HTML::isCompatibleWith);
+        } catch (InvalidMediaTypeException ex) {
+            return false;
+        }
+    }
+
+    private static class ErrorResult {
+        private final int code;
+        private final String message;
+
+        ErrorResult(int code, String message) {
+            this.code = code;
+            this.message = message;
+        }
+
+        public int getCode() {
+            return code;
+        }
+
+        public String getMessage() {
+            return message;
+        }
+    }
+}

sentinel-adapter/sentinel-spring-webflux-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/spring/webflux/callback/WebFluxCallbackManager.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright 1999-2019 Alibaba Group Holding Ltd.
+ *
+ * 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 com.alibaba.csp.sentinel.adapter.spring.webflux.callback;
+
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+import com.alibaba.csp.sentinel.util.AssertUtil;
+
+import org.springframework.web.server.ServerWebExchange;
+
+/**
+ * @author Eric Zhao
+ * @since 1.5.0
+ */
+public final class WebFluxCallbackManager {
+
+    private static final BiFunction<ServerWebExchange, String, String> DEFAULT_URL_CLEANER = (w, url) -> url;
+    private static final Function<ServerWebExchange, String> DEFAULT_ORIGIN_PARSER = (w) -> "";
+
+    /**
+     * BlockRequestHandler: (serverExchange, exception) -> response
+     */
+    private static volatile BlockRequestHandler blockHandler = new DefaultBlockRequestHandler();
+    /**
+     * UrlCleaner: (serverExchange, originalUrl) -> finalUrl
+     */
+    private static volatile BiFunction<ServerWebExchange, String, String> urlCleaner = DEFAULT_URL_CLEANER;
+    /**
+     * RequestOriginParser: (serverExchange) -> origin
+     */
+    private static volatile Function<ServerWebExchange, String> requestOriginParser = DEFAULT_ORIGIN_PARSER;
+
+    public static /*@NonNull*/ BlockRequestHandler getBlockHandler() {
+        return blockHandler;
+    }
+
+    public static void resetBlockHandler() {
+        WebFluxCallbackManager.blockHandler = new DefaultBlockRequestHandler();
+    }
+
+    public static void setBlockHandler(BlockRequestHandler blockHandler) {
+        AssertUtil.notNull(blockHandler, "blockHandler cannot be null");
+        WebFluxCallbackManager.blockHandler = blockHandler;
+    }
+
+    public static /*@NonNull*/ BiFunction<ServerWebExchange, String, String> getUrlCleaner() {
+        return urlCleaner;
+    }
+
+    public static void resetUrlCleaner() {
+        WebFluxCallbackManager.urlCleaner = DEFAULT_URL_CLEANER;
+    }
+
+    public static void setUrlCleaner(BiFunction<ServerWebExchange, String, String> urlCleaner) {
+        AssertUtil.notNull(urlCleaner, "urlCleaner cannot be null");
+        WebFluxCallbackManager.urlCleaner = urlCleaner;
+    }
+
+    public static /*@NonNull*/ Function<ServerWebExchange, String> getRequestOriginParser() {
+        return requestOriginParser;
+    }
+
+    public static void resetRequestOriginParser() {
+        WebFluxCallbackManager.requestOriginParser = DEFAULT_ORIGIN_PARSER;
+    }
+
+    public static void setRequestOriginParser(Function<ServerWebExchange, String> requestOriginParser) {
+        AssertUtil.notNull(requestOriginParser, "requestOriginParser cannot be null");
+        WebFluxCallbackManager.requestOriginParser = requestOriginParser;
+    }
+
+    private WebFluxCallbackManager() {}
+}

sentinel-adapter/sentinel-spring-webflux-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/spring/webflux/exception/SentinelBlockExceptionHandler.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 1999-2019 Alibaba Group Holding Ltd.
+ *
+ * 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 com.alibaba.csp.sentinel.adapter.spring.webflux.exception;
+
+import java.util.List;
+
+import com.alibaba.csp.sentinel.adapter.spring.webflux.callback.WebFluxCallbackManager;
+import com.alibaba.csp.sentinel.slots.block.BlockException;
+import com.alibaba.csp.sentinel.util.function.Supplier;
+
+import org.springframework.http.codec.HttpMessageWriter;
+import org.springframework.http.codec.ServerCodecConfigurer;
+import org.springframework.web.reactive.function.server.ServerResponse;
+import org.springframework.web.reactive.result.view.ViewResolver;
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.WebExceptionHandler;
+import reactor.core.publisher.Mono;
+
+/**
+ * @author Eric Zhao
+ * @since 1.5.0
+ */
+public class SentinelBlockExceptionHandler implements WebExceptionHandler {
+
+    private List<ViewResolver> viewResolvers;
+    private List<HttpMessageWriter<?>> messageWriters;
+
+    public SentinelBlockExceptionHandler(List<ViewResolver> viewResolvers, ServerCodecConfigurer serverCodecConfigurer) {
+        this.viewResolvers = viewResolvers;
+        this.messageWriters = serverCodecConfigurer.getWriters();
+    }
+
+    private Mono<Void> writeResponse(ServerResponse response, ServerWebExchange exchange) {
+        return response.writeTo(exchange, contextSupplier.get());
+    }
+
+    @Override
+    public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
+        if (exchange.getResponse().isCommitted()) {
+            return Mono.error(ex);
+        }
+        // This exception handler only handles rejection by Sentinel.
+        if (!BlockException.isBlockException(ex)) {
+            return Mono.error(ex);
+        }
+        return handleBlockedRequest(exchange, ex)
+            .flatMap(response -> writeResponse(response, exchange));
+    }
+
+    private Mono<ServerResponse> handleBlockedRequest(ServerWebExchange exchange, Throwable throwable) {
+        return WebFluxCallbackManager.getBlockHandler().handleRequest(exchange, throwable);
+    }
+
+    private final Supplier<ServerResponse.Context> contextSupplier = () -> new ServerResponse.Context() {
+        @Override
+        public List<HttpMessageWriter<?>> messageWriters() {
+            return SentinelBlockExceptionHandler.this.messageWriters;
+        }
+
+        @Override
+        public List<ViewResolver> viewResolvers() {
+            return SentinelBlockExceptionHandler.this.viewResolvers;
+        }
+    };
+}