redis token manager 支持事件

hsweb-authorization/hsweb-authorization-api/src/main/java/org/hswebframework/web/authorization/token/event/UserTokenChangedEvent.java

@@ -5,7 +5,7 @@ import org.hswebframework.web.authorization.token.UserToken;
 import org.springframework.context.ApplicationEvent;
 
 public class UserTokenChangedEvent extends ApplicationEvent implements AuthorizationEvent {
-    private UserToken before, after;
+    private final UserToken before, after;
 
     public UserTokenChangedEvent(UserToken before, UserToken after) {
         super(after);

hsweb-authorization/hsweb-authorization-api/src/main/java/org/hswebframework/web/authorization/token/redis/RedisUserTokenManager.java

@@ -2,21 +2,24 @@ package org.hswebframework.web.authorization.token.redis;
 
 import lombok.Getter;
 import lombok.Setter;
-import org.apache.commons.collections.CollectionUtils;
 import org.hswebframework.web.authorization.exception.AccessDenyException;
 import org.hswebframework.web.authorization.token.AllopatricLoginMode;
 import org.hswebframework.web.authorization.token.TokenState;
 import org.hswebframework.web.authorization.token.UserToken;
 import org.hswebframework.web.authorization.token.UserTokenManager;
-import org.springframework.data.redis.core.ReactiveHashOperations;
-import org.springframework.data.redis.core.ReactiveRedisOperations;
-import org.springframework.data.redis.core.ReactiveSetOperations;
-import org.springframework.data.redis.core.ScanOptions;
+import org.hswebframework.web.authorization.token.event.UserTokenChangedEvent;
+import org.hswebframework.web.authorization.token.event.UserTokenCreatedEvent;
+import org.hswebframework.web.authorization.token.event.UserTokenRemovedEvent;
+import org.hswebframework.web.bean.FastBeanCopier;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
+import org.springframework.data.redis.core.*;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.RedisSerializer;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import java.time.Duration;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -35,6 +38,18 @@ public class RedisUserTokenManager implements UserTokenManager {
         this.userTokenMapping = operations.opsForSet();
     }
 
+    @SuppressWarnings("all")
+    public RedisUserTokenManager(ReactiveRedisConnectionFactory connectionFactory) {
+        this(new ReactiveRedisTemplate<>(connectionFactory,
+                RedisSerializationContext.newSerializationContext()
+                        .key((RedisSerializer) RedisSerializer.string())
+                        .value(RedisSerializer.java())
+                        .hashKey(RedisSerializer.string())
+                        .hashValue(RedisSerializer.java())
+                        .build()
+        ));
+    }
+
     @Getter
     @Setter
     private Map<String, AllopatricLoginMode> allopatricLoginModes = new HashMap<>();
@@ -44,6 +59,9 @@ public class RedisUserTokenManager implements UserTokenManager {
     //异地登录模式，默认允许异地登录
     private AllopatricLoginMode allopatricLoginMode = AllopatricLoginMode.allow;
 
+    @Setter
+    private ApplicationEventPublisher eventPublisher;
+
     private String getTokenRedisKey(String key) {
         return "user-token:".concat(key);
     }
@@ -114,20 +132,24 @@ public class RedisUserTokenManager implements UserTokenManager {
     public Mono<Void> signOutByUserId(String userId) {
         String key = getUserRedisKey(userId);
         return getByUserId(key)
-                .map(UserToken::getToken)
-                .map(this::getTokenRedisKey)
-                .concatWithValues(key)
-                .as(operations::delete)
+                .flatMap(userToken -> operations
+                        .delete(getTokenRedisKey(userToken.getToken()))
+                        .then(onTokenRemoved(userToken)))
+                .then(operations.delete(key))
                 .then();
     }
 
     @Override
     public Mono<Void> signOutByToken(String token) {
         //delete token
-        // srem user token
+        //srem user token
         return getByToken(token)
-                .flatMap(t -> operations.delete(getTokenRedisKey(t.getToken()))
-                        .then(userTokenMapping.remove(getUserRedisKey(t.getToken()),token))).then();
+                .flatMap(t -> operations
+                        .delete(getTokenRedisKey(t.getToken()))
+                        .then(userTokenMapping.remove(getUserRedisKey(t.getToken()), token))
+                        .then(onTokenRemoved(t))
+                )
+                .then();
     }
 
     @Override
@@ -140,60 +162,68 @@ public class RedisUserTokenManager implements UserTokenManager {
 
     @Override
     public Mono<Void> changeTokenState(String token, TokenState state) {
-        return userTokenStore
-                .put(getTokenRedisKey(token), "state", state.getValue())
-                .then();
+
+        return getByToken(token)
+                .flatMap(old -> {
+                    SimpleUserToken newToken = FastBeanCopier.copy(old, new SimpleUserToken());
+                    newToken.setState(state);
+                    return userTokenStore
+                            .put(getTokenRedisKey(token), "state", state.getValue())
+                            .then(onTokenChanged(old, newToken));
+                });
     }
 
     @Override
     public Mono<UserToken> signIn(String token, String type, String userId, long maxInactiveInterval) {
-        return Mono.defer(() -> {
-            Mono<UserToken> doSign = Mono.defer(() -> {
-                Map<String, Object> map = new HashMap<>();
-                map.put("token", token);
-                map.put("type", type);
-                map.put("userId", userId);
-                map.put("maxInactiveInterval", maxInactiveInterval);
-                map.put("state", TokenState.normal.getValue());
-                map.put("signInTime", System.currentTimeMillis());
-                map.put("lastRequestTime", System.currentTimeMillis());
-
-                String key = getTokenRedisKey(token);
-                return userTokenStore
-                        .putAll(key, map)
-                        .then(Mono.defer(() -> {
-                            if (maxInactiveInterval > 0) {
-                                return operations.expire(key, Duration.ofMillis(maxInactiveInterval));
-                            }
-                            return Mono.empty();
-                        }))
-                        .then(userTokenMapping.add(getUserRedisKey(userId), token))
-                        .thenReturn(SimpleUserToken.of(map));
-            });
-
-            AllopatricLoginMode mode = allopatricLoginModes.getOrDefault(type, allopatricLoginMode);
-            if (mode == AllopatricLoginMode.deny) {
-                return userIsLoggedIn(userId)
-                        .flatMap(r -> {
-                            if (r) {
-                                return Mono.error(new AccessDenyException("已在其他地方登录", TokenState.deny.getValue(), null));
-                            }
-                            return doSign;
-                        });
-
-            } else if (mode == AllopatricLoginMode.offlineOther) {
-                return getByUserId(userId)
-                        .flatMap(userToken -> {
-                            if (type.equals(userToken.getType())) {
-                                return this.changeTokenState(userToken.getToken(), TokenState.offline);
-                            }
-                            return Mono.empty();
-                        })
-                        .then(doSign);
-            }
-
-            return doSign;
-        });
+        return Mono
+                .defer(() -> {
+                    Mono<UserToken> doSign = Mono.defer(() -> {
+                        Map<String, Object> map = new HashMap<>();
+                        map.put("token", token);
+                        map.put("type", type);
+                        map.put("userId", userId);
+                        map.put("maxInactiveInterval", maxInactiveInterval);
+                        map.put("state", TokenState.normal.getValue());
+                        map.put("signInTime", System.currentTimeMillis());
+                        map.put("lastRequestTime", System.currentTimeMillis());
+
+                        String key = getTokenRedisKey(token);
+                        return userTokenStore
+                                .putAll(key, map)
+                                .then(Mono.defer(() -> {
+                                    if (maxInactiveInterval > 0) {
+                                        return operations.expire(key, Duration.ofMillis(maxInactiveInterval));
+                                    }
+                                    return Mono.empty();
+                                }))
+                                .then(userTokenMapping.add(getUserRedisKey(userId), token))
+                                .thenReturn(SimpleUserToken.of(map));
+                    });
+
+                    AllopatricLoginMode mode = allopatricLoginModes.getOrDefault(type, allopatricLoginMode);
+                    if (mode == AllopatricLoginMode.deny) {
+                        return userIsLoggedIn(userId)
+                                .flatMap(r -> {
+                                    if (r) {
+                                        return Mono.error(new AccessDenyException("已在其他地方登录", TokenState.deny.getValue(), null));
+                                    }
+                                    return doSign;
+                                });
+
+                    } else if (mode == AllopatricLoginMode.offlineOther) {
+                        return getByUserId(userId)
+                                .flatMap(userToken -> {
+                                    if (type.equals(userToken.getType())) {
+                                        return this.changeTokenState(userToken.getToken(), TokenState.offline);
+                                    }
+                                    return Mono.empty();
+                                })
+                                .then(doSign);
+                    }
+
+                    return doSign;
+                })
+                .flatMap(this::onUserTokenCreated);
     }
 
 
@@ -213,9 +243,8 @@ public class RedisUserTokenManager implements UserTokenManager {
     @Override
     public Mono<Void> checkExpiredToken() {
 
-        return operations.scan(ScanOptions
-                .scanOptions()
-                .match("user-token-user:*").build())
+        return operations
+                .scan(ScanOptions.scanOptions().match("user-token-user:*").build())
                 .map(String::valueOf)
                 .flatMap(key -> userTokenMapping.members(key)
                         .map(String::valueOf)
@@ -228,4 +257,28 @@ public class RedisUserTokenManager implements UserTokenManager {
                                 })))
                 .then();
     }
+
+    private Mono<Void> onTokenRemoved(UserToken token) {
+        if (eventPublisher == null) {
+            return Mono.empty();
+        }
+        return Mono.fromRunnable(() -> eventPublisher.publishEvent(new UserTokenRemovedEvent(token)));
+    }
+
+    private Mono<Void> onTokenChanged(UserToken old, UserToken newToken) {
+        if (eventPublisher == null) {
+            return Mono.empty();
+        }
+        return Mono.fromRunnable(() -> eventPublisher.publishEvent(new UserTokenChangedEvent(old, newToken)));
+    }
+
+    private Mono<UserToken> onUserTokenCreated(UserToken token) {
+        if (eventPublisher == null) {
+            return Mono.just(token);
+        }
+        return Mono
+                .fromRunnable(() -> eventPublisher.publishEvent(new UserTokenCreatedEvent(token)))
+                .thenReturn(token);
+    }
+
 }