优化国际化逻辑

hsweb-commons/hsweb-commons-crud/src/main/java/org/hswebframework/web/crud/events/ValidateEventListener.java

@@ -35,9 +35,10 @@ public class ValidateEventListener implements EventListener {
             resultHolder
                     .ifPresent(holder -> holder
                             .invoke(LocaleUtils
-                                            .currentReactive()
-                                            .doOnNext(locale -> LocaleUtils.doWith(locale, (l) -> tryValidate(type, context)))
-                                            .then()
+                                            .doInReactive(() -> {
+                                                tryValidate(type, context);
+                                                return null;
+                                            })
                             ));
         } else {
             tryValidate(type, context);

hsweb-core/src/main/java/org/hswebframework/web/exception/TraceSourceException.java

@@ -35,7 +35,7 @@ public class TraceSourceException extends RuntimeException {
     }
 
     public TraceSourceException(Throwable e) {
-        super(e);
+        super(e.getMessage(),e);
     }
 
     public TraceSourceException(String message, Throwable e) {

hsweb-core/src/main/java/org/hswebframework/web/i18n/LocaleUtils.java

@@ -1,15 +1,17 @@
 package org.hswebframework.web.i18n;
 
+import lombok.AllArgsConstructor;
 import org.hswebframework.web.exception.I18nSupportException;
 import org.reactivestreams.Publisher;
+import org.reactivestreams.Subscription;
 import org.springframework.context.MessageSource;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.core.publisher.Signal;
-import reactor.core.publisher.SignalType;
+import reactor.core.CoreSubscriber;
+import reactor.core.publisher.*;
 import reactor.util.context.Context;
 
+import javax.annotation.Nonnull;
 import java.util.Locale;
+import java.util.concurrent.Callable;
 import java.util.function.BiConsumer;
 import java.util.function.BiFunction;
 import java.util.function.Consumer;
@@ -23,7 +25,6 @@ import java.util.function.Function;
  *  <li>{@link LocaleUtils#current()} </li>
  *  <li>{@link LocaleUtils#currentReactive()}</li>
  *  <li>{@link LocaleUtils#resolveMessageReactive(String, Object...)}</li>
- *  <li>{@link LocaleUtils#doOnNext(BiConsumer)}</li>
  * </ul>
  *
  * @author zhouhao
@@ -63,11 +64,12 @@ public final class LocaleUtils {
      * @return 返回值
      */
     public static <T, R> R doWith(T data, Locale locale, BiFunction<T, Locale, R> mapper) {
+        Locale old = CONTEXT_THREAD_LOCAL.get();
         try {
             CONTEXT_THREAD_LOCAL.set(locale);
             return mapper.apply(data, locale);
         } finally {
-            CONTEXT_THREAD_LOCAL.remove();
+            CONTEXT_THREAD_LOCAL.set(old);
         }
     }
 
@@ -78,11 +80,12 @@ public final class LocaleUtils {
      * @param consumer 任务
      */
     public static void doWith(Locale locale, Consumer<Locale> consumer) {
+        Locale old = CONTEXT_THREAD_LOCAL.get();
         try {
             CONTEXT_THREAD_LOCAL.set(locale);
             consumer.accept(locale);
         } finally {
-            CONTEXT_THREAD_LOCAL.remove();
+            CONTEXT_THREAD_LOCAL.set(old);
         }
     }
 
@@ -112,6 +115,23 @@ public final class LocaleUtils {
                 .subscriberContext()
                 .map(ctx -> ctx.getOrDefault(Locale.class, DEFAULT_LOCALE));
     }
+    public static <T> Mono<T> doInReactive(Callable<T> call) {
+        return currentReactive()
+                .handle((locale, sink) -> {
+                    Locale old = CONTEXT_THREAD_LOCAL.get();
+                    try {
+                        CONTEXT_THREAD_LOCAL.set(locale);
+                        T data = call.call();
+                        if (data != null) {
+                            sink.next(data);
+                        }
+                    } catch (Throwable e) {
+                        sink.error(e);
+                    } finally {
+                        CONTEXT_THREAD_LOCAL.set(old);
+                    }
+                });
+    }
 
     /**
      * 响应式方式解析出异常的区域消息，并进行结果转换.
@@ -450,4 +470,86 @@ public final class LocaleUtils {
         return doOn(SignalType.ON_ERROR, (s, l) -> operation.accept(s.getThrowable(), l));
     }
 
+    public static <T> Flux<T> transform(Flux<T> flux) {
+        return new LocaleFlux<>(flux);
+    }
+
+    public static <T> Mono<T> transform(Mono<T> mono) {
+        return new LocaleMono<>(mono);
+    }
+
+    @AllArgsConstructor
+    static class LocaleMono<T> extends Mono<T> {
+        private final Mono<T> source;
+
+        @Override
+        public void subscribe(@Nonnull CoreSubscriber<? super T> actual) {
+            doWith(actual,
+                   actual.currentContext().getOrDefault(Locale.class, DEFAULT_LOCALE),
+                   (a, l) -> {
+                       source.subscribe(
+                               new LocaleSwitchSubscriber<>(a)
+                       );
+                       return null;
+                   }
+            );
+        }
+    }
+
+    @AllArgsConstructor
+    static class LocaleFlux<T> extends Flux<T> {
+        private final Flux<T> source;
+
+        @Override
+        public void subscribe(@Nonnull CoreSubscriber<? super T> actual) {
+            source.subscribe(
+                    new LocaleSwitchSubscriber<>(actual)
+            );
+        }
+    }
+
+    @AllArgsConstructor
+    static class LocaleSwitchSubscriber<T> extends BaseSubscriber<T> {
+        private final CoreSubscriber<T> actual;
+
+        @Override
+        @Nonnull
+        public Context currentContext() {
+            return actual
+                    .currentContext();
+        }
+
+        @Override
+        protected void hookOnSubscribe(@Nonnull Subscription subscription) {
+            actual.onSubscribe(this);
+        }
+
+        private Locale current() {
+            return currentContext()
+                    .getOrDefault(Locale.class, DEFAULT_LOCALE);
+        }
+
+        @Override
+        protected void hookOnComplete() {
+            doWith(current(), (l) -> actual.onComplete());
+        }
+
+        @Override
+        protected void hookOnError(@Nonnull Throwable error) {
+
+            doWith(error, current(), (v, l) -> {
+                actual.onError(v);
+                return null;
+            });
+        }
+
+        @Override
+        protected void hookOnNext(@Nonnull T value) {
+
+            doWith(value, current(), (v, l) -> {
+                actual.onNext(v);
+                return null;
+            });
+        }
+    }
 }

hsweb-core/src/main/java/org/hswebframework/web/i18n/WebFluxLocaleFilter.java

@@ -15,6 +15,7 @@ public class WebFluxLocaleFilter implements WebFilter {
     public Mono<Void> filter(@NonNull ServerWebExchange exchange, WebFilterChain chain) {
         return chain
                 .filter(exchange)
+                .as(LocaleUtils::transform)
                 .subscriberContext(LocaleUtils.useLocale(getLocaleContext(exchange)));
     }
 

hsweb-core/src/test/java/org/hswebframework/web/i18n/LocaleUtilsTest.java

@@ -2,6 +2,8 @@ package org.hswebframework.web.i18n;
 
 import org.junit.Test;
 import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
 
 import java.util.Locale;
 
@@ -11,15 +13,35 @@ public class LocaleUtilsTest {
 
 
     @Test
-    public void testOnNext() {
+    public void testFlux() {
         Flux.just(1)
-            .as(LocaleUtils.doOnNext((i, l) -> {
+            .as(LocaleUtils::transform)
+            .doOnNext(i -> {
                 assertEquals(i.intValue(), 1);
-                assertEquals(l, Locale.CHINA);
-            }))
-            .subscriberContext(LocaleUtils.useLocale(Locale.CHINA))
+                assertEquals(LocaleUtils.current(), Locale.ENGLISH);
+            })
+            .subscriberContext(LocaleUtils.useLocale(Locale.ENGLISH))
             .blockLast();
     }
 
+    @Test
+    public void testMono() {
+        Mono.just(1)
+            .doOnNext(i -> {
+                assertEquals(i.intValue(), 1);
+                assertEquals(LocaleUtils.current(), Locale.ENGLISH);
+            })
+            .as(LocaleUtils::transform)
+            .subscriberContext(LocaleUtils.useLocale(Locale.ENGLISH))
+            .block();
+
+        LocaleUtils
+                .doInReactive(()->{
+                    assertEquals(LocaleUtils.current(), Locale.ENGLISH);
+                    return null;
+                })
+                .subscriberContext(LocaleUtils.useLocale(Locale.ENGLISH))
+                .block();
+    }
 
 }

hsweb-starter/src/main/java/org/hswebframework/web/starter/jackson/CustomJackson2JsonDecoder.java

@@ -28,12 +28,10 @@ import org.springframework.util.MimeType;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
-import javax.annotation.Nonnull;
 import java.io.IOException;
 import java.lang.annotation.Annotation;
 import java.lang.reflect.ParameterizedType;
 import java.lang.reflect.Type;
-import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 
@@ -70,23 +68,19 @@ public class CustomJackson2JsonDecoder extends Jackson2CodecSupport implements H
 
         ObjectReader reader = getObjectReader(elementType, hints);
 
-        return LocaleUtils
-                .currentReactive()
-                .flatMapMany(locale -> tokens
-                        .handle((tokenBuffer, sink) -> {
-                            LocaleUtils.doWith(locale, l -> {
-                                try {
-                                    Object value = reader.readValue(tokenBuffer.asParser(getObjectMapper()));
-                                    logValue(value, hints);
-                                    if (value != null) {
-                                        sink.next(value);
-                                    }
-                                } catch (IOException ex) {
-                                    sink.error(processException(ex));
-                                }
-                            });
-
-                        }));
+        return tokens
+                .as(LocaleUtils::transform)
+                .handle((tokenBuffer, sink) -> {
+                    try {
+                        Object value = reader.readValue(tokenBuffer.asParser(getObjectMapper()));
+                        logValue(value, hints);
+                        if (value != null) {
+                            sink.next(value);
+                        }
+                    } catch (IOException ex) {
+                        sink.error(processException(ex));
+                    }
+                });
     }
 
     @Override
@@ -94,15 +88,10 @@ public class CustomJackson2JsonDecoder extends Jackson2CodecSupport implements H
     public Mono<Object> decodeToMono(@NonNull Publisher<DataBuffer> input, @NonNull ResolvableType elementType,
                                      @Nullable MimeType mimeType, @Nullable Map<String, Object> hints) {
 
-        return LocaleUtils
-                .currentReactive()
-                .flatMap(locale -> DataBufferUtils
-                        .join(input)
-                        .map(dataBuffer -> LocaleUtils
-                                .doWith(dataBuffer,
-                                        locale,
-                                        (buf, l) -> decode(buf, elementType, mimeType, hints)))
-                );
+        return DataBufferUtils
+                .join(input)
+                .as(LocaleUtils::transform)
+                .map(dataBuffer -> decode(dataBuffer, elementType, mimeType, hints));
     }
 
     @Override

hsweb-starter/src/main/java/org/hswebframework/web/starter/jackson/CustomJackson2jsonEncoder.java

@@ -116,72 +116,55 @@ public class CustomJackson2jsonEncoder extends Jackson2CodecSupport implements H
         Assert.notNull(bufferFactory, "'bufferFactory' must not be null");
         Assert.notNull(elementType, "'elementType' must not be null");
 
-        return LocaleUtils
-                .currentReactive()
-                .flatMapMany(locale -> {
-                    if (inputStream instanceof Mono) {
-                        return Mono.from(inputStream)
-                                   .map(value -> LocaleUtils
-                                           .doWith(value, locale,
-                                                   ((val, loc) ->
-                                                           encodeValue(val, bufferFactory, elementType, mimeType, hints)
-                                                   )
-                                           ))
-                                   .flux();
-                    } else {
-                        byte[] separator = streamSeparator(mimeType);
-                        if (separator != null) { // streaming
-                            try {
-                                ObjectWriter writer = createObjectWriter(elementType, mimeType, hints);
-                                ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer
-                                                                                            .getFactory()
-                                                                                            ._getBufferRecycler());
-                                JsonEncoding encoding = getJsonEncoding(mimeType);
-                                JsonGenerator generator = getObjectMapper()
-                                        .getFactory()
-                                        .createGenerator(byteBuilder, encoding);
-                                SequenceWriter sequenceWriter = writer.writeValues(generator);
-
-                                return Flux
-                                        .from(inputStream)
-                                        .map(value -> LocaleUtils
-                                                .doWith(value,
-                                                        locale,
-                                                        ((val, loc) -> this
-                                                                .encodeStreamingValue(val,
-                                                                                      bufferFactory,
-                                                                                      hints,
-                                                                                      sequenceWriter,
-                                                                                      byteBuilder,
-                                                                                      separator)
-                                                        )
-                                                ))
-                                        .doAfterTerminate(() -> {
-                                            try {
-                                                byteBuilder.release();
-                                                generator.close();
-                                            } catch (IOException ex) {
-                                                logger.error("Could not close Encoder resources", ex);
-                                            }
-                                        });
-                            } catch (IOException ex) {
-                                return Flux.error(ex);
-                            }
-                        } else { // non-streaming
-                            ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType);
-                            return Flux.from(inputStream)
-                                       .collectList()
-                                       .map(value -> LocaleUtils
-                                               .doWith(value, locale,
-                                                       ((val, loc) ->
-                                                               encodeValue(val, bufferFactory, listType, mimeType, hints)
-                                                       )
-                                               ))
-                                       .flux();
-                        }
-
-                    }
-                });
+        if (inputStream instanceof Mono) {
+            return Mono.from(inputStream)
+                       .as(LocaleUtils::transform)
+                       .map(value -> encodeValue(value, bufferFactory, elementType, mimeType, hints))
+                       .flux();
+        } else {
+            byte[] separator = streamSeparator(mimeType);
+            if (separator != null) { // streaming
+                try {
+                    ObjectWriter writer = createObjectWriter(elementType, mimeType, hints);
+                    ByteArrayBuilder byteBuilder = new ByteArrayBuilder(writer
+                                                                                .getFactory()
+                                                                                ._getBufferRecycler());
+                    JsonEncoding encoding = getJsonEncoding(mimeType);
+                    JsonGenerator generator = getObjectMapper()
+                            .getFactory()
+                            .createGenerator(byteBuilder, encoding);
+                    SequenceWriter sequenceWriter = writer.writeValues(generator);
+
+                    return Flux
+                            .from(inputStream)
+                            .as(LocaleUtils::transform)
+                            .map(value -> this.encodeStreamingValue(value,
+                                                                    bufferFactory,
+                                                                    hints,
+                                                                    sequenceWriter,
+                                                                    byteBuilder,
+                                                                    separator))
+                            .doAfterTerminate(() -> {
+                                try {
+                                    byteBuilder.release();
+                                    generator.close();
+                                } catch (IOException ex) {
+                                    logger.error("Could not close Encoder resources", ex);
+                                }
+                            });
+                } catch (IOException ex) {
+                    return Flux.error(ex);
+                }
+            } else { // non-streaming
+                ResolvableType listType = ResolvableType.forClassWithGenerics(List.class, elementType);
+                return Flux.from(inputStream)
+                           .collectList()
+                           .as(LocaleUtils::transform)
+                           .map(value -> encodeValue(value, bufferFactory, listType, mimeType, hints))
+                           .flux();
+            }
+
+        }
     }
 
     @Override