Reactive: add Sentinel Reactor adapter module (#545)

* Add reactive adapter implementation for Project Reactor (Mono/Flux) including a reactor transformer and an experimental `ReactorSphU`
* Add an `InheritableBaseSubscriber` that derives from the original BaseSubscriber of reactor-core
* Add basic test cases for reactor adapter
* Add Sentinel context enter support for EntryConfig in reactor transformer

Signed-off-by: Eric Zhao <[email protected]>

sentinel-adapter/pom.xml

@@ -19,6 +19,7 @@
         <module>sentinel-dubbo-adapter</module>
         <module>sentinel-grpc-adapter</module>
         <module>sentinel-zuul-adapter</module>
+        <module>sentinel-reactor-adapter</module>
     </modules>
 
     <dependencyManagement>
@@ -38,6 +39,7 @@
                 <artifactId>sentinel-web-servlet</artifactId>
                 <version>${project.version}</version>
             </dependency>
+
             <dependency>
                 <groupId>junit</groupId>
                 <artifactId>junit</artifactId>

sentinel-adapter/sentinel-reactor-adapter/pom.xml

@@ -0,0 +1,44 @@
+<?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-reactor-adapter</artifactId>
+
+    <properties>
+        <java.source.version>1.8</java.source.version>
+        <java.target.version>1.8</java.target.version>
+        <reactor.version>3.2.6.RELEASE</reactor.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.alibaba.csp</groupId>
+            <artifactId>sentinel-core</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+            <version>${reactor.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <version>${reactor.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

sentinel-adapter/sentinel-reactor-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/reactor/ContextConfig.java

@@ -0,0 +1,57 @@
+/*
+ * 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.reactor;
+
+import com.alibaba.csp.sentinel.util.AssertUtil;
+import com.alibaba.csp.sentinel.util.StringUtil;
+
+/**
+ * @author Eric Zhao
+ */
+public class ContextConfig {
+
+    private final String contextName;
+    private final String origin;
+
+    public ContextConfig(String contextName) {
+        this(contextName, "");
+    }
+
+    public ContextConfig(String contextName, String origin) {
+        AssertUtil.assertNotBlank(contextName, "contextName cannot be blank");
+        this.contextName = contextName;
+        if (StringUtil.isBlank(origin)) {
+            origin = "";
+        }
+        this.origin = origin;
+    }
+
+    public String getContextName() {
+        return contextName;
+    }
+
+    public String getOrigin() {
+        return origin;
+    }
+
+    @Override
+    public String toString() {
+        return "ContextConfig{" +
+            "contextName='" + contextName + '\'' +
+            ", origin='" + origin + '\'' +
+            '}';
+    }
+}

sentinel-adapter/sentinel-reactor-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/reactor/EntryConfig.java

@@ -0,0 +1,77 @@
+/*
+ * 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.reactor;
+
+import com.alibaba.csp.sentinel.EntryType;
+import com.alibaba.csp.sentinel.util.AssertUtil;
+
+/**
+ * @author Eric Zhao
+ * @since 1.5.0
+ */
+public class EntryConfig {
+
+    private final String resourceName;
+    private final EntryType entryType;
+    private final ContextConfig contextConfig;
+
+    public EntryConfig(String resourceName) {
+        this(resourceName, EntryType.OUT);
+    }
+
+    public EntryConfig(String resourceName, EntryType entryType) {
+        this(resourceName, entryType, null);
+    }
+
+    public EntryConfig(String resourceName, EntryType entryType, ContextConfig contextConfig) {
+        checkParams(resourceName, entryType);
+        this.resourceName = resourceName;
+        this.entryType = entryType;
+        // Constructed ContextConfig should be valid here. Null is allowed here.
+        this.contextConfig = contextConfig;
+    }
+
+    public String getResourceName() {
+        return resourceName;
+    }
+
+    public EntryType getEntryType() {
+        return entryType;
+    }
+
+    public ContextConfig getContextConfig() {
+        return contextConfig;
+    }
+
+    public static void assertValid(EntryConfig config) {
+        AssertUtil.notNull(config, "entry config cannot be null");
+        checkParams(config.resourceName, config.entryType);
+    }
+
+    private static void checkParams(String resourceName, EntryType entryType) {
+        AssertUtil.assertNotBlank(resourceName, "resourceName cannot be blank");
+        AssertUtil.notNull(entryType, "entryType cannot be null");
+    }
+
+    @Override
+    public String toString() {
+        return "EntryConfig{" +
+            "resourceName='" + resourceName + '\'' +
+            ", entryType=" + entryType +
+            ", contextConfig=" + contextConfig +
+            '}';
+    }
+}

sentinel-adapter/sentinel-reactor-adapter/src/main/java/com/alibaba/csp/sentinel/adapter/reactor/FluxSentinelOperator.java

@@ -0,0 +1,40 @@
+/*
+ * 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.reactor;
+
+import reactor.core.CoreSubscriber;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.FluxOperator;
+
+/**
+ * @author Eric Zhao
+ * @since 1.5.0
+ */
+public class FluxSentinelOperator<T> extends FluxOperator<T, T> {
+
+    private final EntryConfig entryConfig;
+
+    public FluxSentinelOperator(Flux<? extends T> source, EntryConfig entryConfig) {
+        super(source);
+        EntryConfig.assertValid(entryConfig);
+        this.entryConfig = entryConfig;
+    }
+
+    @Override
+    public void subscribe(CoreSubscriber<? super T> actual) {
+        source.subscribe(new SentinelReactorSubscriber<>(entryConfig, actual, false));
+    }
+}