优化异步事件

hsweb-authorization/hsweb-authorization-api/src/main/java/org/hswebframework/web/authorization/events/AbstractAuthorizationEvent.java

@@ -19,7 +19,7 @@
 package org.hswebframework.web.authorization.events;
 
 
-import org.springframework.context.ApplicationEvent;
+import org.hswebframework.web.event.DefaultAsyncEvent;
 
 import java.util.Optional;
 import java.util.function.Function;
@@ -30,14 +30,14 @@ import java.util.function.Function;
  * @author zhouhao
  * @since 3.0
  */
-public abstract class AbstractAuthorizationEvent extends ApplicationEvent implements AuthorizationEvent {
+public abstract class AbstractAuthorizationEvent extends DefaultAsyncEvent implements AuthorizationEvent {
     private static final long serialVersionUID = -3027505108916079214L;
 
     protected String username;
 
     protected String password;
 
-    private transient Function<String, Object> parameterGetter;
+    private final transient Function<String, Object> parameterGetter;
 
     /**
      * 所有参数不能为null
@@ -47,7 +47,6 @@ public abstract class AbstractAuthorizationEvent extends ApplicationEvent implem
      * @param parameterGetter 参数获取函数,用户获取授权时传入的参数
      */
     public AbstractAuthorizationEvent(String username, String password, Function<String, Object> parameterGetter) {
-        super(username + "/" + password);
         if (username == null || password == null || parameterGetter == null) {
             throw new NullPointerException();
         }
@@ -57,7 +56,7 @@ public abstract class AbstractAuthorizationEvent extends ApplicationEvent implem
     }
 
     @SuppressWarnings("unchecked")
-    public  <T> Optional<T> getParameter(String name) {
+    public <T> Optional<T> getParameter(String name) {
         return Optional.ofNullable((T) parameterGetter.apply(name));
     }
 

hsweb-authorization/hsweb-authorization-api/src/main/java/org/hswebframework/web/authorization/events/AuthorizationSuccessEvent.java

@@ -19,7 +19,7 @@
 package org.hswebframework.web.authorization.events;
 
 import org.hswebframework.web.authorization.Authentication;
-import org.springframework.context.ApplicationEvent;
+import org.hswebframework.web.event.DefaultAsyncEvent;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -33,16 +33,15 @@ import java.util.function.Function;
  * @see Authentication
  * @since 3.0
  */
-public class AuthorizationSuccessEvent extends ApplicationEvent implements AuthorizationEvent {
+public class AuthorizationSuccessEvent extends DefaultAsyncEvent implements AuthorizationEvent {
     private static final long serialVersionUID = -2452116314154155058L;
-    private Authentication authentication;
+    private final Authentication authentication;
 
-    private transient Function<String, Object> parameterGetter;
+    private final transient Function<String, Object> parameterGetter;
 
     private Map<String, Object> result = new HashMap<>();
 
     public AuthorizationSuccessEvent(Authentication authentication, Function<String, Object> parameterGetter) {
-        super(authentication);
         this.authentication = authentication;
         this.parameterGetter = parameterGetter;
     }

hsweb-authorization/hsweb-authorization-basic/src/main/java/org/hswebframework/web/authorization/basic/web/AuthorizationController.java

@@ -94,31 +94,34 @@ public class AuthorizationController {
             AuthorizationFailedEvent.Reason reason = AuthorizationFailedEvent.Reason.OTHER;
             Function<String, Object> parameterGetter = parameters::get;
             return Mono.defer(() -> {
-                String username = username_;
-                String password = password_;
-
-                AuthorizationDecodeEvent decodeEvent = new AuthorizationDecodeEvent(username, password, parameterGetter);
-                eventPublisher.publishEvent(decodeEvent);
-                username = decodeEvent.getUsername();
-                password = decodeEvent.getPassword();
-                AuthorizationBeforeEvent beforeEvent = new AuthorizationBeforeEvent(username, password, parameterGetter);
-                eventPublisher.publishEvent(beforeEvent);
-                // 验证通过
-                return authenticationManager
-                        .authenticate(Mono.just(new PlainTextUsernamePasswordAuthenticationRequest(username, password)))
-                        .switchIfEmpty(Mono.error(() -> new IllegalArgumentException("密码错误")))
-                        .map(auth -> {
-                            //触发授权成功事件
-                            AuthorizationSuccessEvent event = new AuthorizationSuccessEvent(auth, parameterGetter);
-                            event.getResult().put("userId", auth.getUser().getId());
-                            eventPublisher.publishEvent(event);
-                            return event.getResult();
-                        });
+                AuthorizationDecodeEvent decodeEvent = new AuthorizationDecodeEvent(username_, password_, parameterGetter);
+                return decodeEvent
+                        .publish(eventPublisher)
+                        .then(Mono.defer(() -> {
+                            String username = decodeEvent.getUsername();
+                            String password = decodeEvent.getPassword();
+                            AuthorizationBeforeEvent beforeEvent = new AuthorizationBeforeEvent(username, password, parameterGetter);
+                            return beforeEvent
+                                    .publish(eventPublisher)
+                                    .then(authenticationManager
+                                            .authenticate(Mono.just(new PlainTextUsernamePasswordAuthenticationRequest(username, password)))
+                                            .switchIfEmpty(Mono.error(() -> new IllegalArgumentException("密码错误")))
+                                            .flatMap(auth -> {
+                                                //触发授权成功事件
+                                                AuthorizationSuccessEvent event = new AuthorizationSuccessEvent(auth, parameterGetter);
+                                                event.getResult().put("userId", auth.getUser().getId());
+                                                return event
+                                                        .publish(eventPublisher)
+                                                        .then(Mono.fromCallable(event::getResult));
+                                            }));
+
+                        }));
             }).onErrorResume(err -> {
                 AuthorizationFailedEvent failedEvent = new AuthorizationFailedEvent(username_, password_, parameterGetter, reason);
                 failedEvent.setException(err);
-                eventPublisher.publishEvent(failedEvent);
-                return Mono.error(failedEvent.getException());
+                return failedEvent
+                        .publish(eventPublisher)
+                        .then(Mono.error(failedEvent.getException()));
             });
         });
 

hsweb-authorization/hsweb-authorization-basic/src/main/java/org/hswebframework/web/authorization/basic/web/UserOnSignIn.java

@@ -7,6 +7,7 @@ import org.hswebframework.web.authorization.token.UserTokenHolder;
 import org.hswebframework.web.authorization.token.UserTokenManager;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.EventListener;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -21,10 +22,11 @@ import java.util.List;
  * @see UserTokenGenerator
  * @since 3.0
  */
-public class UserOnSignIn implements ApplicationListener<AuthorizationSuccessEvent> {
+public class UserOnSignIn {
 
     /**
      * 默认到令牌类型
+     *
      * @see UserToken#getType()
      * @see SessionIdUserTokenGenerator#getSupportTokenType()
      */
@@ -50,14 +52,14 @@ public class UserOnSignIn implements ApplicationListener<AuthorizationSuccessEve
         this.userTokenGenerators = userTokenGenerators;
     }
 
-    @Override
+    @EventListener
     public void onApplicationEvent(AuthorizationSuccessEvent event) {
         UserToken token = UserTokenHolder.currentToken();
         String tokenType = (String) event.getParameter("token_type").orElse(defaultTokenType);
 
         if (token != null) {
             //先退出已登陆的用户
-            userTokenManager.signOutByToken(token.getToken()).block();
+            event.async(userTokenManager.signOutByToken(token.getToken()));
         }
         //创建token
         GeneratedToken newToken = userTokenGenerators.stream()
@@ -66,8 +68,7 @@ public class UserOnSignIn implements ApplicationListener<AuthorizationSuccessEve
                 .orElseThrow(() -> new UnsupportedOperationException(tokenType))
                 .generate(event.getAuthentication());
         //登入
-        userTokenManager.signIn(newToken.getToken(), newToken.getType(), event.getAuthentication().getUser().getId(), newToken.getTimeout())
-        .block();
+        event.async(userTokenManager.signIn(newToken.getToken(), newToken.getType(), event.getAuthentication().getUser().getId(), newToken.getTimeout()).then());
 
         //响应结果
         event.getResult().putAll(newToken.getResponse());

hsweb-authorization/hsweb-authorization-basic/src/main/java/org/hswebframework/web/authorization/basic/web/UserTokenWebFilter.java

@@ -28,9 +28,9 @@ import java.util.Map;
 @Slf4j
 public class UserTokenWebFilter implements WebFilter, BeanPostProcessor {
 
-    private List<ReactiveUserTokenParser> parsers = new ArrayList<>();
+    private final List<ReactiveUserTokenParser> parsers = new ArrayList<>();
 
-    private Map<String, ReactiveUserTokenGenerator> tokenGeneratorMap = new HashMap<>();
+    private final Map<String, ReactiveUserTokenGenerator> tokenGeneratorMap = new HashMap<>();
 
     @Autowired
     private UserTokenManager userTokenManager;
@@ -62,12 +62,12 @@ public class UserTokenWebFilter implements WebFilter, BeanPostProcessor {
                     .map(String::valueOf)
                     .map(Long::parseLong)
                     .orElse(token.getTimeout());
+            event.getResult().put("expires", expires);
 
-            userTokenManager
+            event.async(userTokenManager
                     .signIn(token.getToken(), token.getType(), event.getAuthentication().getUser().getId(), expires)
-                    .subscribe(t -> {
-                        log.debug("user [{}] sign in", t.getUserId());
-                    });
+                    .doOnNext(t -> log.debug("user [{}] sign in", t.getUserId()))
+                    .then());
         }
 
     }

hsweb-core/src/main/java/org/hswebframework/web/event/AsyncEvent.java

@@ -1,16 +1,29 @@
 package org.hswebframework.web.event;
 
-import lombok.Getter;
 import org.reactivestreams.Publisher;
+import org.springframework.context.ApplicationEventPublisher;
 import reactor.core.publisher.Mono;
 
-@Getter
-public class AsyncEvent {
+/**
+ * 异步事件,使用响应式编程进行事件监听时,请使用此事件接口
+ *
+ * @author zhouhao
+ * @since 4.0.5
+ */
+public interface AsyncEvent {
 
-    private Mono<Void> async = Mono.empty();
-
-    public synchronized void async(Publisher<?> publisher) {
-        this.async = async.then(Mono.from(publisher).then());
-    }
+    /**
+     * 注册一个异步任务
+     *
+     * @param publisher 异步任务
+     */
+    void async(Publisher<?> publisher);
 
+    /**
+     * 推送事件到 ApplicationEventPublisher
+     *
+     * @param eventPublisher ApplicationEventPublisher
+     * @return async void
+     */
+    Mono<Void> publish(ApplicationEventPublisher eventPublisher);
 }

hsweb-core/src/main/java/org/hswebframework/web/event/DefaultAsyncEvent.java

@@ -0,0 +1,22 @@
+package org.hswebframework.web.event;
+
+import org.reactivestreams.Publisher;
+import org.springframework.context.ApplicationEventPublisher;
+import reactor.core.publisher.Mono;
+
+public class DefaultAsyncEvent implements AsyncEvent {
+
+    private Mono<Void> async = Mono.empty();
+
+    public synchronized void async(Publisher<?> publisher) {
+        this.async = async.then(Mono.from(publisher).then());
+    }
+
+    @Override
+    public Mono<Void> publish(ApplicationEventPublisher eventPublisher) {
+
+        eventPublisher.publishEvent(this);
+
+        return this.async;
+    }
+}