From 26121b41722022d773327f07ecef35bc7d5088a4 Mon Sep 17 00:00:00 2001
From: Administrator <yf1033327537@163.com>
Date: Tue, 18 Jul 2023 22:47:27 +0800
Subject: [PATCH] seata for webflux support

---
 ...ReactiveSeataHandlerAutoConfiguration.java | 19 +++++++
 .../reactive/ReactiveSeataHandlerFilter.java  | 54 +++++++++++++++++++
 ...ot.autoconfigure.AutoConfiguration.imports |  1 +
 3 files changed, 74 insertions(+)
 create mode 100644 spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/web/reactive/ReactiveSeataHandlerAutoConfiguration.java
 create mode 100644 spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/web/reactive/ReactiveSeataHandlerFilter.java

diff --git a/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/web/reactive/ReactiveSeataHandlerAutoConfiguration.java b/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/web/reactive/ReactiveSeataHandlerAutoConfiguration.java
new file mode 100644
index 0000000000..40dc6c8e38
--- /dev/null
+++ b/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/web/reactive/ReactiveSeataHandlerAutoConfiguration.java
@@ -0,0 +1,19 @@
+package com.alibaba.cloud.seata.web.reactive;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author yangfengwei
+ */
+@Configuration
+@ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.REACTIVE)
+public class ReactiveSeataHandlerAutoConfiguration {
+
+    @Bean
+    public ReactiveSeataHandlerFilter seataHandlerReactiveFilter() {
+        return new ReactiveSeataHandlerFilter();
+    }
+
+}
diff --git a/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/web/reactive/ReactiveSeataHandlerFilter.java b/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/web/reactive/ReactiveSeataHandlerFilter.java
new file mode 100644
index 0000000000..bfa35fadfc
--- /dev/null
+++ b/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/java/com/alibaba/cloud/seata/web/reactive/ReactiveSeataHandlerFilter.java
@@ -0,0 +1,54 @@
+package com.alibaba.cloud.seata.web.reactive;
+
+import io.seata.common.util.StringUtils;
+import io.seata.core.context.RootContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.web.server.ServerWebExchange;
+import org.springframework.web.server.WebFilter;
+import org.springframework.web.server.WebFilterChain;
+import reactor.core.publisher.Mono;
+
+/**
+ *
+ * @author yangfengwei
+ */
+public class ReactiveSeataHandlerFilter implements WebFilter {
+
+    private static final Logger log = LoggerFactory.getLogger(ReactiveSeataHandlerFilter.class);
+
+    @Override
+    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
+        String xid = RootContext.getXID();
+        String rpcXid = exchange.getRequest().getHeaders().getFirst(RootContext.KEY_XID);
+        if (log.isDebugEnabled()) {
+            log.debug("xid in RootContext {} xid in RpcContext {}", xid, rpcXid);
+        }
+        if (StringUtils.isBlank(xid) && rpcXid != null) {
+            RootContext.bind(rpcXid);
+            if (log.isDebugEnabled()) {
+                log.debug("bind {} to RootContext", rpcXid);
+            }
+        }
+        Mono<Void> mono = chain.filter(exchange);
+        return mono.then(Mono.defer(() -> {
+            if (StringUtils.isNotBlank(RootContext.getXID())) {
+                if (StringUtils.isNotEmpty(rpcXid)) {
+                    String unbindXid = RootContext.unbind();
+                    if (log.isDebugEnabled()) {
+                        log.debug("unbind {} from RootContext", unbindXid);
+                    }
+                    if (!rpcXid.equalsIgnoreCase(unbindXid)) {
+                        log.warn("xid in change during RPC from {} to {}", rpcXid, unbindXid);
+                        if (unbindXid != null) {
+                            RootContext.bind(unbindXid);
+                            log.warn("bind {} back to RootContext", unbindXid);
+                        }
+                    }
+                }
+            }
+            return Mono.empty();
+        }));
+
+    }
+}
diff --git a/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
index 814ee6afe9..0c7141ca88 100644
--- a/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
+++ b/spring-cloud-alibaba-starters/spring-cloud-starter-alibaba-seata/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports
@@ -1,3 +1,4 @@
 com.alibaba.cloud.seata.rest.SeataRestTemplateAutoConfiguration
 com.alibaba.cloud.seata.web.SeataHandlerInterceptorConfiguration
 com.alibaba.cloud.seata.feign.SeataFeignClientAutoConfiguration
+com.alibaba.cloud.seata.web.reactive.ReactiveSeataHandlerAutoConfiguration