优化上下文

hsweb-core/src/main/java/org/hswebframework/web/logger/ReactiveLogger.java

@@ -2,9 +2,7 @@ package org.hswebframework.web.logger;
 
 import lombok.extern.slf4j.Slf4j;
 import org.slf4j.MDC;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.Signal;
-import reactor.core.publisher.SignalType;
+import reactor.core.publisher.*;
 import reactor.util.context.Context;
 
 import java.util.*;
@@ -89,6 +87,14 @@ public class ReactiveLogger {
                 .then();
     }
 
+    public static <T, R> BiConsumer<T, SynchronousSink<R>> handle(BiConsumer<T, SynchronousSink<R>> logger) {
+        return (t, rFluxSink) -> {
+            log(rFluxSink.currentContext(), context -> {
+                logger.accept(t, rFluxSink);
+            });
+        };
+    }
+
     public static <T> Consumer<Signal<T>> onNext(Consumer<T> logger) {
         return on(SignalType.ON_NEXT, (ctx, signal) -> {
             logger.accept(signal.get());

hsweb-core/src/test/java/org/hswebframework/web/logger/ReactiveLoggerTest.java

@@ -22,7 +22,7 @@ public class ReactiveLoggerTest {
                 .doOnEach(ReactiveLogger.onNext(v -> {
                     log.info("test:{}", v);
                 }))
-                .subscriberContext(ReactiveLogger.start("r","1"))
+                .subscriberContext(ReactiveLogger.start("r", "1"))
                 .as(StepVerifier::create)
                 .expectNextCount(5)
                 .verifyComplete();
@@ -30,4 +30,18 @@ public class ReactiveLoggerTest {
 
     }
 
+    @Test
+    public void testHandle() {
+        Flux.range(0, 5)
+                .delayElements(Duration.ofSeconds(2))
+                .flatMap(i -> ReactiveLogger.mdc("requestId", "test").thenReturn(i))
+                .handle(ReactiveLogger.handle((o, fluxSink) -> {
+                    log.info("test:{}", fluxSink.currentContext());
+                    fluxSink.next(o);
+                })).subscriberContext(ReactiveLogger.start("r", "1"))
+                .as(StepVerifier::create)
+                .expectNextCount(5)
+                .verifyComplete();
+
+    }
 }

hsweb-logging/hsweb-access-logging-aop/src/main/java/org/hswebframework/web/loggin/aop/ReactiveAopAccessLoggerSupport.java

@@ -3,6 +3,7 @@ package org.hswebframework.web.loggin.aop;
 import org.aopalliance.intercept.MethodInterceptor;
 import org.hswebframework.web.aop.MethodInterceptorHolder;
 import org.hswebframework.web.id.IDGenerator;
+import org.hswebframework.web.logger.ReactiveLogger;
 import org.hswebframework.web.logging.RequestInfo;
 import org.hswebframework.web.logging.AccessLoggerInfo;
 import org.hswebframework.web.logging.AccessLoggerListener;
@@ -65,12 +66,13 @@ public class ReactiveAopAccessLoggerSupport extends StaticMethodMatcherPointcutA
                 .doFinally(f -> {
                     loggerInfo.setResponseTime(System.currentTimeMillis());
                     eventPublisher.publishEvent(new AccessLoggerAfterEvent(loggerInfo));
-                });
+                }).subscriberContext(ReactiveLogger.start("accessLogId",loggerInfo.getId()));
     }
 
     protected Mono<?> wrapMonoResponse(Mono<?> mono, AccessLoggerInfo loggerInfo) {
         return Mono.subscriberContext()
-                .<RequestInfo>flatMap(ctx -> Mono.justOrEmpty(ctx.getOrEmpty(RequestInfo.class)))
+                .<RequestInfo>flatMap(ctx -> Mono.<RequestInfo>justOrEmpty(ctx.getOrEmpty(RequestInfo.class))
+                        .doOnNext(info -> ReactiveLogger.log(ctx, info::setContext)))
                 .doOnNext(loggerInfo::putAccessInfo)
                 .then(mono)
                 .doOnError(loggerInfo::setException)
@@ -78,7 +80,7 @@ public class ReactiveAopAccessLoggerSupport extends StaticMethodMatcherPointcutA
                 .doFinally(f -> {
                     loggerInfo.setResponseTime(System.currentTimeMillis());
                     eventPublisher.publishEvent(new AccessLoggerAfterEvent(loggerInfo));
-                });
+                }).subscriberContext(ReactiveLogger.start("accessLogId",loggerInfo.getId()));
     }
 
     @SuppressWarnings("all")

hsweb-logging/hsweb-access-logging-api/src/main/java/org/hswebframework/web/logging/AccessLoggerInfo.java

@@ -72,6 +72,11 @@ public class AccessLoggerInfo {
      */
     private Map<String, String> httpHeaders;
 
+    /**
+     * 上下文
+     */
+    private Map<String, String> context;
+
     /**
      * http 请求方法, GET,POST...
      */
@@ -130,7 +135,7 @@ public class AccessLoggerInfo {
         map.put("response", objectFilter.apply(response));
         map.put("requestTime", requestTime);
         map.put("responseTime", responseTime);
-        map.put("id",id);
+        map.put("id", id);
         map.put("useTime", responseTime - requestTime);
         if (exception != null) {
             StringWriter writer = new StringWriter();
@@ -141,10 +146,11 @@ public class AccessLoggerInfo {
     }
 
 
-    public void putAccessInfo(RequestInfo info){
+    public void putAccessInfo(RequestInfo info) {
         setIp(info.getIpAddr());
         setHttpMethod(info.getRequestMethod());
         setHttpHeaders(info.getHeaders());
         setUrl(info.getPath());
+        setContext(info.getContext());
     }
 }

hsweb-logging/hsweb-access-logging-api/src/main/java/org/hswebframework/web/logging/RequestInfo.java

@@ -19,7 +19,13 @@ public class RequestInfo {
 
     private String requestMethod;
 
+    private String userId;
+
+    private String username;
+
     private Map<String,String> headers;
 
+    private Map<String,String> context;
+
 
 }