add test
This commit is contained in:
parent
aa532dda6d
commit
c2339044df
|
@ -27,12 +27,6 @@
|
|||
<version>${redisson.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>redis.clients</groupId>
|
||||
<artifactId>jedis</artifactId>
|
||||
<version>${jedis.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>cn.hutool</groupId>
|
||||
<artifactId>hutool-crypto</artifactId>
|
||||
|
|
|
@ -33,7 +33,7 @@ public class RClient {
|
|||
/**
|
||||
* get hashmap of the key
|
||||
*
|
||||
* @param key
|
||||
* @param key hash name
|
||||
* @return hashmap
|
||||
*/
|
||||
public Map<String, String> getMap(String key) {
|
||||
|
@ -53,8 +53,8 @@ public class RClient {
|
|||
/**
|
||||
* add listener of the key
|
||||
*
|
||||
* @param key
|
||||
* @param listener
|
||||
* @param key hash name
|
||||
* @param listener listener
|
||||
* @return listener id
|
||||
*/
|
||||
public int addListener(String key, MapEntryListener listener) {
|
||||
|
@ -65,8 +65,8 @@ public class RClient {
|
|||
/**
|
||||
* get all keys of hash
|
||||
*
|
||||
* @param key
|
||||
* @return
|
||||
* @param key hash name
|
||||
* @return keySet
|
||||
*/
|
||||
public Set<String> hkeys(String key) {
|
||||
RMap<String, String> map = redissonClient.getMap(key, new StringCodec());
|
||||
|
@ -76,9 +76,9 @@ public class RClient {
|
|||
/**
|
||||
* gey value of the key
|
||||
*
|
||||
* @param key
|
||||
* @param field
|
||||
* @return
|
||||
* @param key hash name
|
||||
* @param field hash field
|
||||
* @return hash value
|
||||
*/
|
||||
public String hget(String key, String field) {
|
||||
RMap<String, String> map = redissonClient.getMap(key, new StringCodec());
|
||||
|
@ -87,7 +87,7 @@ public class RClient {
|
|||
|
||||
/**
|
||||
* Loads Lua script into Redis scripts cache and returns its SHA-1 digest
|
||||
* @param luaScript
|
||||
* @param luaScript script
|
||||
* @return shaDigest
|
||||
*/
|
||||
public String scriptLoad(String luaScript) {
|
||||
|
@ -97,9 +97,9 @@ public class RClient {
|
|||
|
||||
/**
|
||||
* Executes Lua script stored in Redis scripts cache by SHA-1 digest
|
||||
* @param shaDigest
|
||||
* @param args
|
||||
* @return
|
||||
* @param shaDigest script cache by sha-1
|
||||
* @param args script args
|
||||
* @return string
|
||||
*/
|
||||
public String evalSha(String shaDigest, String... args){
|
||||
RScript script = redissonClient.getScript(new StringCodec());
|
||||
|
|
|
@ -1,13 +1,15 @@
|
|||
package com.yomahub.liteflow.parser.redis.mode;
|
||||
|
||||
import cn.hutool.core.collection.CollUtil;
|
||||
import cn.hutool.core.text.StrFormatter;
|
||||
import cn.hutool.core.util.ReUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.yomahub.liteflow.builder.LiteFlowNodeBuilder;
|
||||
import com.yomahub.liteflow.core.FlowExecutor;
|
||||
import com.yomahub.liteflow.enums.NodeTypeEnum;
|
||||
import com.yomahub.liteflow.log.LFLog;
|
||||
import com.yomahub.liteflow.log.LFLoggerManager;
|
||||
import com.yomahub.liteflow.parser.redis.vo.RedisParserVO;
|
||||
import org.redisson.config.Config;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
|
@ -15,7 +17,7 @@ import java.util.List;
|
|||
* Redis 解析器通用接口
|
||||
*
|
||||
* @author hxinyu
|
||||
* @since 2.11.0
|
||||
* @since 2.11.0
|
||||
*/
|
||||
|
||||
public interface RedisParserHelper {
|
||||
|
@ -39,10 +41,34 @@ public interface RedisParserHelper {
|
|||
void listenRedis();
|
||||
|
||||
|
||||
/**
|
||||
* 获取Redisson客户端的Config配置通用方法
|
||||
* @param redisParserVO redisParserVO
|
||||
* @param dataBase redisson连接的数据库号
|
||||
* @return redisson config
|
||||
*/
|
||||
default Config getRedissonConfig(RedisParserVO redisParserVO, Integer dataBase) {
|
||||
Config config = new Config();
|
||||
String redisAddress = StrFormatter.format(REDIS_URL_PATTERN, redisParserVO.getHost(), redisParserVO.getPort());
|
||||
//如果配置了密码
|
||||
if (StrUtil.isNotBlank(redisParserVO.getPassword())) {
|
||||
config.useSingleServer().setAddress(redisAddress)
|
||||
.setPassword(redisParserVO.getPassword())
|
||||
.setDatabase(dataBase);
|
||||
}
|
||||
//没有配置密码
|
||||
else {
|
||||
config.useSingleServer().setAddress(redisAddress)
|
||||
.setDatabase(dataBase);
|
||||
}
|
||||
return config;
|
||||
}
|
||||
|
||||
/**
|
||||
* script节点的修改/添加
|
||||
*
|
||||
* @param scriptFieldValue 新的script名
|
||||
* @param newValue 新的script值
|
||||
* @param newValue 新的script值
|
||||
*/
|
||||
static void changeScriptNode(String scriptFieldValue, String newValue) {
|
||||
NodeSimpleVO nodeSimpleVO = convert(scriptFieldValue);
|
||||
|
|
|
@ -5,8 +5,8 @@ import cn.hutool.crypto.digest.DigestUtil;
|
|||
import com.yomahub.liteflow.builder.el.LiteFlowChainELBuilder;
|
||||
import com.yomahub.liteflow.flow.FlowBus;
|
||||
import com.yomahub.liteflow.log.LFLog;
|
||||
import com.yomahub.liteflow.parser.redis.mode.RClient;
|
||||
import com.yomahub.liteflow.parser.redis.vo.RedisParserVO;
|
||||
import redis.clients.jedis.Jedis;
|
||||
|
||||
import java.util.*;
|
||||
|
||||
|
@ -20,7 +20,7 @@ public class ChainPollingTask {
|
|||
|
||||
private RedisParserVO redisParserVO;
|
||||
|
||||
private Jedis chainJedis;
|
||||
private RClient chainClient;
|
||||
|
||||
private Integer chainNum;
|
||||
|
||||
|
@ -28,9 +28,9 @@ public class ChainPollingTask {
|
|||
|
||||
LFLog LOG;
|
||||
|
||||
public ChainPollingTask(RedisParserVO redisParserVO, Jedis chainJedis, Integer chainNum, Map<String, String> chainSHAMap, LFLog LOG) {
|
||||
public ChainPollingTask(RedisParserVO redisParserVO, RClient chainClient, Integer chainNum, Map<String, String> chainSHAMap, LFLog LOG) {
|
||||
this.redisParserVO = redisParserVO;
|
||||
this.chainJedis = chainJedis;
|
||||
this.chainClient = chainClient;
|
||||
this.chainNum = chainNum;
|
||||
this.chainSHAMap = chainSHAMap;
|
||||
this.LOG = LOG;
|
||||
|
@ -46,7 +46,7 @@ public class ChainPollingTask {
|
|||
Runnable r = () -> {
|
||||
String chainKey = redisParserVO.getChainKey();
|
||||
//Lua获取chainKey中最新的chain数量
|
||||
String keyNum = chainJedis.evalsha(keyLua, 1, chainKey).toString();
|
||||
String keyNum = chainClient.evalSha(keyLua, chainKey);
|
||||
//修改chainNum为最新chain数量
|
||||
chainNum = Integer.parseInt(keyNum);
|
||||
|
||||
|
@ -56,7 +56,7 @@ public class ChainPollingTask {
|
|||
String chainId = entry.getKey();
|
||||
String oldSHA = entry.getValue();
|
||||
//在redis服务端通过Lua脚本计算SHA值
|
||||
String newSHA = chainJedis.evalsha(valueLua, 2, chainKey, chainId).toString();
|
||||
String newSHA = chainClient.evalSha(valueLua, chainKey, chainId);
|
||||
if (StrUtil.equals(newSHA, "nil")) {
|
||||
//新SHA值为nil, 即未获取到该chain,表示该chain已被删除
|
||||
FlowBus.removeChain(chainId);
|
||||
|
@ -68,7 +68,7 @@ public class ChainPollingTask {
|
|||
}
|
||||
else if (!StrUtil.equals(newSHA, oldSHA)) {
|
||||
//SHA值发生变化,表示该chain的值已被修改,重新拉取变化的chain
|
||||
String chainData = chainJedis.hget(chainKey, chainId);
|
||||
String chainData = chainClient.hget(chainKey, chainId);
|
||||
LiteFlowChainELBuilder.createChain().setChainId(chainId).setEL(chainData).build();
|
||||
LOG.info("starting reload flow config... update key={} new value={},", chainId, chainData);
|
||||
|
||||
|
@ -90,11 +90,11 @@ public class ChainPollingTask {
|
|||
// 2、修改了chainId:因为遍历到旧的id时会取到nil,SHAMap会把原来的chainId删掉,但没有机会添加新的chainId
|
||||
// 3、上述两者结合
|
||||
//在此处重新拉取所有chainId集合,补充添加新chain
|
||||
Set<String> newChainSet = chainJedis.hkeys(chainKey);
|
||||
Set<String> newChainSet = chainClient.hkeys(chainKey);
|
||||
for (String chainId : newChainSet) {
|
||||
if (!chainSHAMap.containsKey(chainId)) {
|
||||
//将新chainId添加到LiteFlowChainELBuilder和SHAMap
|
||||
String chainData = chainJedis.hget(chainKey, chainId);
|
||||
String chainData = chainClient.hget(chainKey, chainId);
|
||||
LiteFlowChainELBuilder.createChain().setChainId(chainId).setEL(chainData).build();
|
||||
LOG.info("starting reload flow config... create key={} new value={},", chainId, chainData);
|
||||
chainSHAMap.put(chainId, DigestUtil.sha1Hex(chainData));
|
||||
|
|
|
@ -6,10 +6,12 @@ import cn.hutool.core.util.ObjectUtil;
|
|||
import cn.hutool.core.util.StrUtil;
|
||||
import cn.hutool.crypto.digest.DigestUtil;
|
||||
import com.yomahub.liteflow.parser.redis.exception.RedisException;
|
||||
import com.yomahub.liteflow.parser.redis.mode.RClient;
|
||||
import com.yomahub.liteflow.parser.redis.mode.RedisParserHelper;
|
||||
import com.yomahub.liteflow.parser.redis.vo.RedisParserVO;
|
||||
import com.yomahub.liteflow.spi.holder.ContextAwareHolder;
|
||||
import redis.clients.jedis.Jedis;
|
||||
import org.redisson.Redisson;
|
||||
import org.redisson.config.Config;
|
||||
|
||||
import java.util.*;
|
||||
import java.util.concurrent.*;
|
||||
|
@ -25,9 +27,9 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
|
||||
private final RedisParserVO redisParserVO;
|
||||
|
||||
private Jedis chainJedis;
|
||||
private RClient chainClient;
|
||||
|
||||
private Jedis scriptJedis;
|
||||
private RClient scriptClient;
|
||||
|
||||
//chainKey中chain总数
|
||||
private Integer chainNum = 0;
|
||||
|
@ -63,25 +65,18 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
|
||||
try{
|
||||
try{
|
||||
this.chainJedis = ContextAwareHolder.loadContextAware().getBean("chainJedis");
|
||||
this.scriptJedis = ContextAwareHolder.loadContextAware().getBean("scriptJedis");
|
||||
this.chainClient = ContextAwareHolder.loadContextAware().getBean("chainClient");
|
||||
this.scriptClient = ContextAwareHolder.loadContextAware().getBean("scriptClient");
|
||||
}
|
||||
catch (Exception ignored) {
|
||||
}
|
||||
if (ObjectUtil.isNull(chainJedis)) {
|
||||
chainJedis = new Jedis(redisParserVO.getHost(), redisParserVO.getPort());
|
||||
//如果配置了密码
|
||||
if (StrUtil.isNotBlank(redisParserVO.getPassword())) {
|
||||
chainJedis.auth(redisParserVO.getPassword());
|
||||
}
|
||||
chainJedis.select(redisParserVO.getChainDataBase());
|
||||
if (ObjectUtil.isNull(chainClient)) {
|
||||
Config config = getRedissonConfig(redisParserVO, redisParserVO.getChainDataBase());
|
||||
this.chainClient = new RClient(Redisson.create(config));
|
||||
//如果有脚本数据
|
||||
if (ObjectUtil.isNotNull(redisParserVO.getScriptDataBase())) {
|
||||
scriptJedis = new Jedis(redisParserVO.getHost(), redisParserVO.getPort());
|
||||
if (StrUtil.isNotBlank(redisParserVO.getPassword())) {
|
||||
scriptJedis.auth(redisParserVO.getPassword());
|
||||
}
|
||||
scriptJedis.select(redisParserVO.getScriptDataBase());
|
||||
config = getRedissonConfig(redisParserVO, redisParserVO.getScriptDataBase());
|
||||
this.scriptClient = new RClient(Redisson.create(config));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -95,7 +90,7 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
try {
|
||||
// 检查chainKey下有没有子节点
|
||||
String chainKey = redisParserVO.getChainKey();
|
||||
Set<String> chainNameSet = chainJedis.hkeys(chainKey);
|
||||
Set<String> chainNameSet = chainClient.hkeys(chainKey);
|
||||
if (CollectionUtil.isEmpty(chainNameSet)) {
|
||||
throw new RedisException(StrUtil.format("There are no chains in key [{}]", chainKey));
|
||||
}
|
||||
|
@ -103,7 +98,7 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
// 获取chainKey下的所有子节点内容List
|
||||
List<String> chainItemContentList = new ArrayList<>();
|
||||
for (String chainName : chainNameSet) {
|
||||
String chainData = chainJedis.hget(chainKey, chainName);
|
||||
String chainData = chainClient.hget(chainKey, chainName);
|
||||
if (StrUtil.isNotBlank(chainData)) {
|
||||
chainItemContentList.add(StrUtil.format(CHAIN_XML_PATTERN, chainName, chainData));
|
||||
}
|
||||
|
@ -119,7 +114,7 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
String scriptAllContent = StrUtil.EMPTY;
|
||||
if (hasScript()) {
|
||||
String scriptKey = redisParserVO.getScriptKey();
|
||||
Set<String> scriptFieldSet = scriptJedis.hkeys(scriptKey);
|
||||
Set<String> scriptFieldSet = scriptClient.hkeys(scriptKey);
|
||||
scriptNum = scriptFieldSet.size();
|
||||
|
||||
List<String> scriptItemContentList = new ArrayList<>();
|
||||
|
@ -130,7 +125,7 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
StrUtil.format("The name of the redis field [{}] in scriptKey [{}] is invalid",
|
||||
scriptFieldValue, scriptKey));
|
||||
}
|
||||
String scriptData = scriptJedis.hget(scriptKey, scriptFieldValue);
|
||||
String scriptData = scriptClient.hget(scriptKey, scriptFieldValue);
|
||||
|
||||
// 有语言类型
|
||||
if (StrUtil.isNotBlank(nodeSimpleVO.getLanguage())) {
|
||||
|
@ -161,7 +156,7 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
}
|
||||
|
||||
public boolean hasScript() {
|
||||
if (ObjectUtil.isNull(scriptJedis) || ObjectUtil.isNull(redisParserVO.getScriptDataBase())) {
|
||||
if (ObjectUtil.isNull(scriptClient) || ObjectUtil.isNull(redisParserVO.getScriptDataBase())) {
|
||||
return false;
|
||||
}
|
||||
try{
|
||||
|
@ -169,7 +164,7 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
if (StrUtil.isBlank(scriptKey)) {
|
||||
return false;
|
||||
}
|
||||
Set<String> scriptKeySet = scriptJedis.hkeys(scriptKey);
|
||||
Set<String> scriptKeySet = scriptClient.hkeys(scriptKey);
|
||||
return !CollUtil.isEmpty(scriptKeySet);
|
||||
}
|
||||
catch (Exception e) {
|
||||
|
@ -183,8 +178,8 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
@Override
|
||||
public void listenRedis() {
|
||||
//将lua脚本添加到chainJedis脚本缓存
|
||||
String keyLuaOfChain = chainJedis.scriptLoad(luaOfKey);
|
||||
String valueLuaOfChain = chainJedis.scriptLoad(luaOfValue);
|
||||
String keyLuaOfChain = chainClient.scriptLoad(luaOfKey);
|
||||
String valueLuaOfChain = chainClient.scriptLoad(luaOfValue);
|
||||
|
||||
//定时任务线程池
|
||||
ScheduledThreadPoolExecutor pollExecutor = new ScheduledThreadPoolExecutor(
|
||||
|
@ -192,19 +187,19 @@ public class RedisParserPollingMode implements RedisParserHelper {
|
|||
new ThreadPoolExecutor.DiscardOldestPolicy());
|
||||
|
||||
//添加轮询chain的定时任务
|
||||
ChainPollingTask chainTask = new ChainPollingTask(redisParserVO, chainJedis, chainNum, chainSHAMap, LOG);
|
||||
ChainPollingTask chainTask = new ChainPollingTask(redisParserVO, chainClient, chainNum, chainSHAMap, LOG);
|
||||
pollExecutor.scheduleAtFixedRate(chainTask.pollChainTask(keyLuaOfChain, valueLuaOfChain),
|
||||
60, redisParserVO.getPollingInterval().longValue(), TimeUnit.SECONDS);
|
||||
|
||||
//如果有脚本
|
||||
if (ObjectUtil.isNotNull(scriptJedis) && ObjectUtil.isNotNull(redisParserVO.getScriptDataBase())
|
||||
if (ObjectUtil.isNotNull(scriptClient) && ObjectUtil.isNotNull(redisParserVO.getScriptDataBase())
|
||||
&& StrUtil.isNotBlank(redisParserVO.getScriptKey())) {
|
||||
//将lua脚本添加到scriptJedis脚本缓存
|
||||
String keyLuaOfScript = scriptJedis.scriptLoad(luaOfKey);
|
||||
String valueLuaOfScript = scriptJedis.scriptLoad(luaOfValue);
|
||||
String keyLuaOfScript = scriptClient.scriptLoad(luaOfKey);
|
||||
String valueLuaOfScript = scriptClient.scriptLoad(luaOfValue);
|
||||
|
||||
//添加轮询script的定时任务
|
||||
ScriptPollingTask scriptTask = new ScriptPollingTask(redisParserVO, scriptJedis, scriptNum, scriptSHAMap, LOG);
|
||||
ScriptPollingTask scriptTask = new ScriptPollingTask(redisParserVO, scriptClient, scriptNum, scriptSHAMap, LOG);
|
||||
pollExecutor.scheduleAtFixedRate(scriptTask.pollScriptTask(keyLuaOfScript, valueLuaOfScript),
|
||||
60, redisParserVO.getPollingInterval().longValue(), TimeUnit.SECONDS);
|
||||
}
|
||||
|
|
|
@ -4,9 +4,9 @@ import cn.hutool.core.util.StrUtil;
|
|||
import cn.hutool.crypto.digest.DigestUtil;
|
||||
import com.yomahub.liteflow.flow.FlowBus;
|
||||
import com.yomahub.liteflow.log.LFLog;
|
||||
import com.yomahub.liteflow.parser.redis.mode.RClient;
|
||||
import com.yomahub.liteflow.parser.redis.mode.RedisParserHelper;
|
||||
import com.yomahub.liteflow.parser.redis.vo.RedisParserVO;
|
||||
import redis.clients.jedis.Jedis;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
@ -23,7 +23,7 @@ public class ScriptPollingTask {
|
|||
|
||||
private RedisParserVO redisParserVO;
|
||||
|
||||
private Jedis scriptJedis;
|
||||
private RClient scriptClient;
|
||||
|
||||
private Integer scriptNum;
|
||||
|
||||
|
@ -31,9 +31,9 @@ public class ScriptPollingTask {
|
|||
|
||||
LFLog LOG;
|
||||
|
||||
public ScriptPollingTask(RedisParserVO redisParserVO, Jedis scriptJedis, Integer scriptNum, Map<String, String> scriptSHAMap, LFLog LOG) {
|
||||
public ScriptPollingTask(RedisParserVO redisParserVO, RClient scriptClient, Integer scriptNum, Map<String, String> scriptSHAMap, LFLog LOG) {
|
||||
this.redisParserVO = redisParserVO;
|
||||
this.scriptJedis = scriptJedis;
|
||||
this.scriptClient = scriptClient;
|
||||
this.scriptNum = scriptNum;
|
||||
this.scriptSHAMap = scriptSHAMap;
|
||||
this.LOG = LOG;
|
||||
|
@ -49,7 +49,7 @@ public class ScriptPollingTask {
|
|||
Runnable r = () -> {
|
||||
String scriptKey = redisParserVO.getScriptKey();
|
||||
//Lua获取scriptKey中最新的script数量
|
||||
String keyNum = scriptJedis.evalsha(keyLua, 1, scriptKey).toString();
|
||||
String keyNum = scriptClient.evalSha(keyLua, scriptKey);
|
||||
//修改scriptNum为最新script数量
|
||||
scriptNum = Integer.parseInt(keyNum);
|
||||
|
||||
|
@ -59,7 +59,7 @@ public class ScriptPollingTask {
|
|||
String scriptFieldValue = entry.getKey();
|
||||
String oldSHA = entry.getValue();
|
||||
//在redis服务端通过Lua脚本计算SHA值
|
||||
String newSHA = scriptJedis.evalsha(valueLua, 2, scriptKey, scriptFieldValue).toString();
|
||||
String newSHA = scriptClient.evalSha(valueLua, scriptKey, scriptFieldValue);
|
||||
if (StrUtil.equals(newSHA, "nil")) {
|
||||
//新SHA值为nil, 即未获取到该script,表示该script已被删除
|
||||
RedisParserHelper.NodeSimpleVO nodeSimpleVO = RedisParserHelper.convert(scriptFieldValue);
|
||||
|
@ -72,7 +72,7 @@ public class ScriptPollingTask {
|
|||
}
|
||||
else if (!StrUtil.equals(newSHA, oldSHA)) {
|
||||
//SHA值发生变化,表示该script的值已被修改,重新拉取变化的script
|
||||
String scriptData = scriptJedis.hget(scriptKey, scriptFieldValue);
|
||||
String scriptData = scriptClient.hget(scriptKey, scriptFieldValue);
|
||||
RedisParserHelper.changeScriptNode(scriptFieldValue, scriptData);
|
||||
LOG.info("starting reload flow config... update key={} new value={},", scriptFieldValue, scriptData);
|
||||
|
||||
|
@ -94,11 +94,11 @@ public class ScriptPollingTask {
|
|||
// 2、修改了script名:因为遍历到旧的id时会取到nil,SHAMap会把原来的script删掉,但没有机会添加新的script
|
||||
// 3、上述两者结合
|
||||
//在此处重新拉取所有script名集合,补充添加新script
|
||||
Set<String> newScriptSet = scriptJedis.hkeys(scriptKey);
|
||||
Set<String> newScriptSet = scriptClient.hkeys(scriptKey);
|
||||
for (String scriptFieldValue : newScriptSet) {
|
||||
if (!scriptSHAMap.containsKey(scriptFieldValue)) {
|
||||
//将新script添加到LiteFlowChainELBuilder和SHAMap
|
||||
String scriptData = scriptJedis.hget(scriptKey, scriptFieldValue);
|
||||
String scriptData = scriptClient.hget(scriptKey, scriptFieldValue);
|
||||
RedisParserHelper.changeScriptNode(scriptFieldValue, scriptData);
|
||||
LOG.info("starting reload flow config... create key={} new value={},", scriptFieldValue, scriptData);
|
||||
scriptSHAMap.put(scriptFieldValue, DigestUtil.sha1Hex(scriptData));
|
||||
|
|
|
@ -2,7 +2,6 @@ package com.yomahub.liteflow.parser.redis.mode.subscribe;
|
|||
|
||||
import cn.hutool.core.collection.CollUtil;
|
||||
import cn.hutool.core.collection.CollectionUtil;
|
||||
import cn.hutool.core.text.StrFormatter;
|
||||
import cn.hutool.core.util.ObjectUtil;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import com.yomahub.liteflow.builder.el.LiteFlowChainELBuilder;
|
||||
|
@ -64,23 +63,6 @@ public class RedisParserSubscribeMode implements RedisParserHelper {
|
|||
|
||||
}
|
||||
|
||||
private Config getRedissonConfig(RedisParserVO redisParserVO, Integer dataBase) {
|
||||
Config config = new Config();
|
||||
String redisAddress = StrFormatter.format(REDIS_URL_PATTERN, redisParserVO.getHost(), redisParserVO.getPort());
|
||||
//如果配置了密码
|
||||
if (StrUtil.isNotBlank(redisParserVO.getPassword())) {
|
||||
config.useSingleServer().setAddress(redisAddress)
|
||||
.setPassword(redisParserVO.getPassword())
|
||||
.setDatabase(dataBase);
|
||||
}
|
||||
//没有配置密码
|
||||
else {
|
||||
config.useSingleServer().setAddress(redisAddress)
|
||||
.setDatabase(dataBase);
|
||||
}
|
||||
return config;
|
||||
}
|
||||
|
||||
@Override
|
||||
public String getContent() {
|
||||
try {
|
||||
|
|
|
@ -4,6 +4,7 @@ import cn.hutool.crypto.digest.DigestUtil;
|
|||
import com.yomahub.liteflow.core.FlowExecutor;
|
||||
import com.yomahub.liteflow.flow.FlowBus;
|
||||
import com.yomahub.liteflow.flow.LiteflowResponse;
|
||||
import com.yomahub.liteflow.parser.redis.mode.RClient;
|
||||
import com.yomahub.liteflow.slot.DefaultContext;
|
||||
import com.yomahub.liteflow.test.BaseTest;
|
||||
import org.junit.jupiter.api.*;
|
||||
|
@ -37,11 +38,11 @@ import static org.mockito.Mockito.when;
|
|||
@ComponentScan({"com.yomahub.liteflow.test.redis.cmp"})
|
||||
public class RedisWithXmlELPollSpringbootTest extends BaseTest {
|
||||
|
||||
@MockBean(name = "chainJedis")
|
||||
private static Jedis chainJedis;
|
||||
@MockBean(name = "chainClient")
|
||||
private static RClient chainClient;
|
||||
|
||||
@MockBean(name = "scriptJedis")
|
||||
private static Jedis scriptJedis;
|
||||
@MockBean(name = "scriptClient")
|
||||
private static RClient scriptClient;
|
||||
|
||||
@Resource
|
||||
private FlowExecutor flowExecutor;
|
||||
|
@ -65,14 +66,14 @@ public class RedisWithXmlELPollSpringbootTest extends BaseTest {
|
|||
chainNameSet.add("chain11");
|
||||
String chainValue = "THEN(a, b, c);";
|
||||
//SHA值用于测试修改chain的轮询刷新功能
|
||||
Object chainSHA = DigestUtil.sha1Hex(chainValue);
|
||||
String chainSHA = DigestUtil.sha1Hex(chainValue);
|
||||
|
||||
//修改chain并更新SHA值
|
||||
String changeChainValue = "THEN(a, c);";
|
||||
Object changeChainSHA = DigestUtil.sha1Hex(changeChainValue);
|
||||
when(chainJedis.hkeys("pollChainKey")).thenReturn(chainNameSet);
|
||||
when(chainJedis.hget("pollChainKey", "chain11")).thenReturn(chainValue).thenReturn(changeChainValue);
|
||||
when(chainJedis.evalsha(anyString(), anyInt(), anyString())).thenReturn(chainSHA).thenReturn(changeChainSHA);
|
||||
String changeChainSHA = DigestUtil.sha1Hex(changeChainValue);
|
||||
when(chainClient.hkeys("pollChainKey")).thenReturn(chainNameSet);
|
||||
when(chainClient.hget("pollChainKey", "chain11")).thenReturn(chainValue).thenReturn(changeChainValue);
|
||||
when(chainClient.evalSha(anyString(), anyString())).thenReturn(chainSHA).thenReturn(changeChainSHA);
|
||||
|
||||
//测试修改前的chain
|
||||
LiteflowResponse response = flowExecutor.execute2Resp("chain11", "arg");
|
||||
|
@ -95,8 +96,8 @@ public class RedisWithXmlELPollSpringbootTest extends BaseTest {
|
|||
Set<String> chainNameSet = new HashSet<>();
|
||||
chainNameSet.add("chain22");
|
||||
String chainValue = "THEN(a, b, c, s11, s22, s33);";
|
||||
when(chainJedis.hkeys("pollChainKey")).thenReturn(chainNameSet);
|
||||
when(chainJedis.hget("pollChainKey", "chain22")).thenReturn(chainValue);
|
||||
when(chainClient.hkeys("pollChainKey")).thenReturn(chainNameSet);
|
||||
when(chainClient.hget("pollChainKey", "chain22")).thenReturn(chainValue);
|
||||
|
||||
Set<String> scriptFieldSet = new HashSet<>();
|
||||
scriptFieldSet.add("s11:script:脚本s11:groovy");
|
||||
|
@ -106,21 +107,21 @@ public class RedisWithXmlELPollSpringbootTest extends BaseTest {
|
|||
String s22 = "defaultContext.setData(\"test22\",\"hello s22\");";
|
||||
String s33 = "defaultContext.setData(\"test33\",\"hello s33\");";
|
||||
//SHA值用于测试修改script的轮询刷新功能
|
||||
Object s11SHA = DigestUtil.sha1Hex(s11);
|
||||
Object s22SHA = DigestUtil.sha1Hex(s22);
|
||||
Object s33SHA = DigestUtil.sha1Hex(s33);
|
||||
String s11SHA = DigestUtil.sha1Hex(s11);
|
||||
String s22SHA = DigestUtil.sha1Hex(s22);
|
||||
String s33SHA = DigestUtil.sha1Hex(s33);
|
||||
//修改script值并更新SHA值
|
||||
String changeS11 = "defaultContext.setData(\"test11\",\"hello world\");";
|
||||
Object changeS11SHA = DigestUtil.sha1Hex(changeS11);
|
||||
String changeS11SHA = DigestUtil.sha1Hex(changeS11);
|
||||
|
||||
when(scriptJedis.hkeys("pollScriptKey")).thenReturn(scriptFieldSet);
|
||||
when(scriptJedis.hget("pollScriptKey", "s11:script:脚本s11:groovy")).thenReturn(s11).thenReturn(changeS11);
|
||||
when(scriptJedis.hget("pollScriptKey", "s22:script:脚本s22:js")).thenReturn(s22);
|
||||
when(scriptJedis.hget("pollScriptKey", "s33:script:脚本s33")).thenReturn(s33);
|
||||
when(scriptClient.hkeys("pollScriptKey")).thenReturn(scriptFieldSet);
|
||||
when(scriptClient.hget("pollScriptKey", "s11:script:脚本s11:groovy")).thenReturn(s11).thenReturn(changeS11);
|
||||
when(scriptClient.hget("pollScriptKey", "s22:script:脚本s22:js")).thenReturn(s22);
|
||||
when(scriptClient.hget("pollScriptKey", "s33:script:脚本s33")).thenReturn(s33);
|
||||
//分别模拟三个script的evalsha指纹值计算的返回值, 其中s11脚本修改 指纹值变化
|
||||
when(scriptJedis.evalsha(anyString(), eq(2), eq("pollScriptKey"), eq("s11:script:脚本s11:groovy"))).thenReturn(s11SHA).thenReturn(changeS11SHA);
|
||||
when(scriptJedis.evalsha(anyString(), eq(2), eq("pollScriptKey"), eq("s22:script:脚本s22:js"))).thenReturn(s22SHA);
|
||||
when(scriptJedis.evalsha(anyString(), eq(2), eq("pollScriptKey"), eq("s33:script:脚本s33"))).thenReturn(s33SHA);
|
||||
when(scriptClient.evalSha(anyString(), eq("pollScriptKey"), eq("s11:script:脚本s11:groovy"))).thenReturn(s11SHA).thenReturn(changeS11SHA);
|
||||
when(scriptClient.evalSha(anyString(), eq("pollScriptKey"), eq("s22:script:脚本s22:js"))).thenReturn(s22SHA);
|
||||
when(scriptClient.evalSha(anyString(), eq("pollScriptKey"), eq("s33:script:脚本s33"))).thenReturn(s33SHA);
|
||||
|
||||
//测试修改前的script
|
||||
LiteflowResponse response = flowExecutor.execute2Resp("chain22", "arg");
|
||||
|
|
Loading…
Reference in New Issue