From 0ab33845f0cb172ee0ad55fb5b58180eae185095 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E7=BB=9D=E5=B0=98?= <237809796@qq.com>
Date: Thu, 25 Jan 2024 02:03:46 +0800
Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=EF=BC=9Aredkale-2.2=20?=
=?UTF-8?q?=E7=89=88=E6=9C=AC=E6=94=AF=E6=8C=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
---
pom.xml | 4 +-
src/main/java/net/tccn/AbstractConsumer.java | 12 +-
src/main/java/net/tccn/ZhubAgentProvider.java | 64 -
src/main/java/net/tccn/ZhubListener.java | 50 +
src/main/java/net/tccn/ZhubProvider.java | 21 -
src/main/java/net/tccn/zhub/ZHubClient.java | 64 +-
.../cache/redis/AbstractRedisSource.java | 234 --
.../cache/redis/MyRedisCacheSource.java | 504 +--
.../org/redkalex/cache/redis/RedisCRC16.java | 79 -
.../cache/redis/RedisCacheClient.java | 58 -
.../redkalex/cache/redis/RedisCacheCodec.java | 255 --
.../cache/redis/RedisCacheConnection.java | 64 -
.../cache/redis/RedisCacheReqAuth.java | 44 -
.../cache/redis/RedisCacheReqClientName.java | 45 -
.../cache/redis/RedisCacheReqClose.java | 43 -
.../redkalex/cache/redis/RedisCacheReqDB.java | 42 -
.../cache/redis/RedisCacheReqPing.java | 38 -
.../cache/redis/RedisCacheRequest.java | 116 -
.../cache/redis/RedisCacheResult.java | 262 --
.../cache/redis/RedisCacheSource.java | 3019 +++++++++++------
.../cache/redis/RedisCacheSourceProvider.java | 30 -
.../redkalex/cache/redis/RedisCommand.java | 124 -
.../org/redkalex/cache/redis/RedisConfig.java | 190 --
.../redkalex/cache/redis/RedisCryptor.java | 49 -
.../redkalex/cache/redis}/RedisTest.java | 166 +-
.../cache/redis/RedissionCacheSource.java | 1944 +++++++++++
26 files changed, 4268 insertions(+), 3253 deletions(-)
delete mode 100644 src/main/java/net/tccn/ZhubAgentProvider.java
create mode 100644 src/main/java/net/tccn/ZhubListener.java
delete mode 100644 src/main/java/net/tccn/ZhubProvider.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCRC16.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheClient.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheConnection.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheResult.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCacheSourceProvider.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCommand.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisConfig.java
delete mode 100644 src/main/java/org/redkalex/cache/redis/RedisCryptor.java
rename src/main/java/{net/tccn => org/redkalex/cache/redis}/RedisTest.java (68%)
create mode 100644 src/main/java/org/redkalex/cache/redis/RedissionCacheSource.java
diff --git a/pom.xml b/pom.xml
index cdc1e51..47346e5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -6,7 +6,7 @@
net.tccn
zhub-client-redkale
- 0.1.1.dev
+ x.22.0
17
@@ -18,7 +18,7 @@
org.redkale
redkale
- 2.8.0.dev
+ 2.2.0
compile
diff --git a/src/main/java/net/tccn/AbstractConsumer.java b/src/main/java/net/tccn/AbstractConsumer.java
index f126411..0844712 100644
--- a/src/main/java/net/tccn/AbstractConsumer.java
+++ b/src/main/java/net/tccn/AbstractConsumer.java
@@ -1,9 +1,9 @@
package net.tccn;
import org.redkale.convert.json.JsonConvert;
-import org.redkale.util.Resourcable;
import org.redkale.util.TypeToken;
+import javax.annotation.Resource;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
@@ -13,11 +13,12 @@ import java.util.function.Consumer;
* @author Liang
* @data 2020-09-05 23:18
*/
-public abstract class AbstractConsumer extends ZhubAgentProvider implements IConsumer, Resourcable {
+public abstract class AbstractConsumer implements IConsumer {
protected JsonConvert convert = JsonConvert.root();
- protected static String APP_NAME = "";
+ @Resource(name = "APP_NAME")
+ protected String APP_NAME = "";
private Map eventMap = new ConcurrentHashMap<>();
@@ -72,9 +73,4 @@ public abstract class AbstractConsumer extends ZhubAgentProvider implements ICon
}
// --------------
-
- @Override
- public String resourceName() {
- return super.getName();
- }
}
diff --git a/src/main/java/net/tccn/ZhubAgentProvider.java b/src/main/java/net/tccn/ZhubAgentProvider.java
deleted file mode 100644
index 5eff32b..0000000
--- a/src/main/java/net/tccn/ZhubAgentProvider.java
+++ /dev/null
@@ -1,64 +0,0 @@
-package net.tccn;
-
-import org.redkale.boot.Application;
-import org.redkale.boot.NodeServer;
-import org.redkale.cluster.CacheClusterAgent;
-import org.redkale.cluster.ClusterAgent;
-import org.redkale.service.Service;
-import org.redkale.util.ResourceEvent;
-
-import java.net.InetSocketAddress;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-
-public abstract class ZhubAgentProvider extends ClusterAgent {
-
- @Override
- public void onResourceChange(ResourceEvent[] events) {
-
- }
-
- @Override
- public void register(Application application) {
-
- }
-
- @Override
- public void deregister(Application application) {
-
- }
-
- @Override
- public CompletableFuture> queryHttpAddress(String protocol, String module, String resname) {
- return null;
- }
-
- @Override
- public CompletableFuture> querySncpAddress(String protocol, String restype, String resname) {
- return null;
- }
-
- @Override
- protected CompletableFuture> queryAddress(ClusterEntry entry) {
- return null;
- }
-
- @Override
- protected ClusterEntry register(NodeServer ns, String protocol, Service service) {
- deregister(ns, protocol, service);
- ClusterEntry clusterEntry = new ClusterEntry(ns, protocol, service);
- CacheClusterAgent.AddressEntry entry = new CacheClusterAgent.AddressEntry();
- entry.addr = clusterEntry.address;
- entry.resname = clusterEntry.resourceName;
- entry.nodeid = this.nodeid;
- entry.time = System.currentTimeMillis();
- //source.hset(clusterEntry.serviceName, clusterEntry.serviceid, CacheClusterAgent.AddressEntry.class, entry);
- return clusterEntry;
- }
-
- @Override
- protected void deregister(NodeServer ns, String protocol, Service service) {
-
- }
-}
diff --git a/src/main/java/net/tccn/ZhubListener.java b/src/main/java/net/tccn/ZhubListener.java
new file mode 100644
index 0000000..dabda0e
--- /dev/null
+++ b/src/main/java/net/tccn/ZhubListener.java
@@ -0,0 +1,50 @@
+package net.tccn;
+
+import net.tccn.zhub.ZHubClient;
+import org.redkale.boot.Application;
+import org.redkale.boot.ApplicationListener;
+import org.redkale.service.Service;
+import org.redkale.util.AnyValue;
+import org.redkale.util.RedkaleClassLoader;
+import org.redkale.util.ResourceFactory;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * 服务监听
+ *
+ * @author: liangxy.
+ */
+public class ZhubListener implements ApplicationListener {
+
+ @Override
+ public void preStart(Application application) {
+
+ CompletableFuture.runAsync(() -> {
+ ResourceFactory resourceFactory = application.getResourceFactory();
+ RedkaleClassLoader classLoader = application.getClassLoader();
+
+ AnyValue appConfig = application.getAppConfig();
+ AnyValue zhubs = appConfig.getAnyValue("zhubs");
+ AnyValue[] values = zhubs.getAnyValues("zhub");
+ for (AnyValue zhub : values) {
+ String className = zhub.getValue("value", ZHubClient.class.getCanonicalName());
+ try {
+ Class clazz = classLoader.loadClass(className);
+ Service obj = (Service) clazz.getDeclaredConstructor().newInstance();
+ application.getResourceFactory().inject(obj);
+ obj.init(zhub);
+ resourceFactory.register(zhub.get("name"), clazz, obj);
+ } catch (InstantiationException | IllegalAccessException | InvocationTargetException | NoSuchMethodException | ClassNotFoundException e) {
+ e.printStackTrace();
+ }
+ }
+ });
+ }
+
+ @Override
+ public void preShutdown(Application application) {
+
+ }
+}
diff --git a/src/main/java/net/tccn/ZhubProvider.java b/src/main/java/net/tccn/ZhubProvider.java
deleted file mode 100644
index 737f557..0000000
--- a/src/main/java/net/tccn/ZhubProvider.java
+++ /dev/null
@@ -1,21 +0,0 @@
-package net.tccn;
-
-import net.tccn.zhub.ZHubClient;
-import org.redkale.annotation.Priority;
-import org.redkale.cluster.ClusterAgent;
-import org.redkale.cluster.ClusterAgentProvider;
-import org.redkale.util.AnyValue;
-
-@Priority(1)
-public class ZhubProvider implements ClusterAgentProvider {
-
- @Override
- public boolean acceptsConf(AnyValue config) {
- return new ZHubClient().acceptsConf(config);
- }
-
- @Override
- public ClusterAgent createInstance() {
- return new ZHubClient();
- }
-}
diff --git a/src/main/java/net/tccn/zhub/ZHubClient.java b/src/main/java/net/tccn/zhub/ZHubClient.java
index bd8c86f..fea63eb 100644
--- a/src/main/java/net/tccn/zhub/ZHubClient.java
+++ b/src/main/java/net/tccn/zhub/ZHubClient.java
@@ -2,14 +2,9 @@ package net.tccn.zhub;
import net.tccn.*;
import net.tccn.timer.Timers;
-import org.redkale.annotation.AutoLoad;
-import org.redkale.annotation.ResourceType;
import org.redkale.service.Local;
import org.redkale.service.Service;
-import org.redkale.util.AnyValue;
-import org.redkale.util.Comment;
-import org.redkale.util.TypeToken;
-import org.redkale.util.Utility;
+import org.redkale.util.*;
import java.io.BufferedReader;
import java.io.IOException;
@@ -56,38 +51,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
private static Map mainHub = new HashMap<>(); // 127.0.0.1:1216 - ZHubClient
- public ZHubClient() {
-
- }
-
- public ZHubClient(String name, Map attr) {
- this.APP_NAME = name;
- this.addr = attr.get("addr");
- this.groupid = attr.get("groupid");
- this.auth = attr.get("auth");
-
- this.initClient(null);
- }
-
@Override
public void init(AnyValue config) {
- APP_NAME = application.getName();
- /*if (!preInit()) {
- return;
- }*/
-
- if (config == null) {
- initClient(null);
- } else {
- Map nodes = getNodes(config);
- for (String rsName : nodes.keySet()) {
- ZHubClient client = new ZHubClient().initClient(nodes.get(rsName));
- application.getResourceFactory().register(rsName, client);
- }
- }
- }
-
- private ZHubClient initClient(AnyValue config) {
// 自动注入
if (config != null) {
addr = config.getValue("addr", addr);
@@ -103,8 +68,10 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
}
// 设置第一个启动的 实例为主实例
- if (!mainHub.containsKey(addr)) { // 确保同步执行此 init 逻辑
- mainHub.put(addr, this);
+ synchronized (ZHubClient.class) {
+ if (!mainHub.containsKey(addr)) { // 确保同步执行此 init 逻辑
+ mainHub.put(addr, this);
+ }
}
CompletableFuture.runAsync(() -> {
@@ -272,18 +239,17 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
while (true) {
Event event = null;
try {
- event = rpcCallQueue.take();
- logger.info(String.format("rpc-call:[%s] %s", event.topic, event.value));
+ event = rpcBackQueue.take();
+ logger.info(String.format("rpc-back:[%s]", event.value));
- String topic = event.topic;
String value = event.value;
- executor.submit(() -> accept(topic, value)).get(5, TimeUnit.SECONDS);
+ executor.submit(() -> rpcAccept(value)).get(5, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
if (e instanceof TimeoutException) {
executor = Executors.newSingleThreadExecutor();
- logger.log(Level.WARNING, "rpc-call TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
+ logger.log(Level.WARNING, "rpc-back TimeoutException, topic[" + event.topic + "], value[" + event.value + "]", e);
} else if (event != null) {
- logger.log(Level.WARNING, "rpc-call[" + event.topic + "] event accept error :" + event.value, e);
+ logger.log(Level.WARNING, "rpc-back[" + event.value + "] event accept error :" + event.value, e);
}
}
}
@@ -335,8 +301,6 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
}
}).start();
});
-
- return this;
}
public boolean acceptsConf(AnyValue config) {
@@ -621,8 +585,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
private static Map rpcRetType = new ConcurrentHashMap<>();
@Comment("rpc call")
- public RpcResult rpc(String topic, Object v) {
- return rpc(topic, v, null);
+ public RpcResult rpc(String topic, Object v) {
+ return rpc(topic, v, IType.STRING);
}
@Comment("rpc call")
@@ -671,8 +635,8 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
return rpc.getRpcResult();
}
- public CompletableFuture> rpcAsync(String topic, T v) {
- return CompletableFuture.supplyAsync(() -> rpc(topic, v, null));
+ public CompletableFuture> rpcAsync(String topic, T v) {
+ return CompletableFuture.supplyAsync(() -> rpc(topic, v, IType.STRING));
}
public CompletableFuture> rpcAsync(String topic, T v, TypeToken typeToken) {
diff --git a/src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java b/src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java
deleted file mode 100644
index 87c1f1b..0000000
--- a/src/main/java/org/redkalex/cache/redis/AbstractRedisSource.java
+++ /dev/null
@@ -1,234 +0,0 @@
-/*
- * Click nbfs://nbhost/SystemFileSystem/Templates/Licenses/license-default.txt to change this license
- * Click nbfs://nbhost/SystemFileSystem/Templates/Classes/Class.java to edit this template
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.annotation.Resource;
-import org.redkale.convert.Convert;
-import org.redkale.convert.json.JsonConvert;
-import org.redkale.source.AbstractCacheSource;
-import org.redkale.util.*;
-
-import java.lang.reflect.Type;
-import java.nio.charset.StandardCharsets;
-import java.util.Arrays;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.function.Function;
-
-import static org.redkale.boot.Application.RESNAME_APP_EXECUTOR;
-import static org.redkale.boot.Application.RESNAME_APP_NAME;
-
-/**
- *
- * @author zhangjx
- *
- * @since 2.8.0
- */
-public abstract class AbstractRedisSource extends AbstractCacheSource {
-
- public static final String CACHE_SOURCE_CRYPTOR = "cryptor";
-
- protected String name;
-
- @Resource(name = RESNAME_APP_NAME, required = false)
- protected String appName = "";
-
- @Resource(required = false)
- protected ResourceFactory resourceFactory;
-
- @Resource(required = false)
- protected JsonConvert defaultConvert;
-
- @Resource(name = Resource.PARENT_NAME + "_convert", required = false)
- protected JsonConvert convert;
-
- protected int db;
-
- protected RedisCryptor cryptor;
-
- protected AnyValue conf;
-
- private ExecutorService subExecutor;
-
- private final ReentrantLock subExecutorLock = new ReentrantLock();
-
- @Resource(name = RESNAME_APP_EXECUTOR, required = false)
- protected ExecutorService workExecutor;
-
- @Override
- public void init(AnyValue conf) {
- this.conf = conf;
- super.init(conf);
- this.name = conf.getValue("name", "");
- if (this.convert == null) {
- this.convert = this.defaultConvert;
- }
- if (conf != null) {
- String cryptStr = conf.getValue(CACHE_SOURCE_CRYPTOR, "").trim();
- if (!cryptStr.isEmpty()) {
- try {
- Class cryptClass = (Class) getClass().getClassLoader().loadClass(cryptStr);
- RedkaleClassLoader.putReflectionPublicConstructors(cryptClass, cryptClass.getName());
- this.cryptor = cryptClass.getConstructor().newInstance();
- } catch (ReflectiveOperationException e) {
- throw new RedkaleException(e);
- }
- }
- }
- if (cryptor != null) {
- if (resourceFactory != null) {
- resourceFactory.inject(cryptor);
- }
- cryptor.init(conf);
- }
- }
-
- @Override
- public void destroy(AnyValue conf) {
- super.destroy(conf);
- if (cryptor != null) {
- cryptor.destroy(conf);
- }
- }
-
- public boolean acceptsConf(AnyValue config) {
- if (config == null) {
- return false;
- }
- return "redis".equalsIgnoreCase(config.getValue(CACHE_SOURCE_TYPE))
- || getClass().getName().equalsIgnoreCase(config.getValue(CACHE_SOURCE_TYPE))
- || config.getValue(CACHE_SOURCE_NODES, config.getValue("url", "")).startsWith("redis://")
- || config.getValue(CACHE_SOURCE_NODES, config.getValue("url", "")).startsWith("rediss://");
- }
-
- protected ExecutorService subExecutor() {
- ExecutorService executor = subExecutor;
- if (executor != null) {
- return executor;
- }
- subExecutorLock.lock();
- try {
- if (subExecutor == null) {
- String threadNameFormat = "CacheSource-" + resourceName() + "-SubThread-%s";
- Function func = Utility.virtualExecutorFunction();
- final AtomicInteger counter = new AtomicInteger();
- subExecutor = func == null ? Executors.newFixedThreadPool(Utility.cpus(), r -> {
- Thread t = new Thread(r);
- t.setDaemon(true);
- int c = counter.incrementAndGet();
- t.setName(String.format(threadNameFormat, "Virtual-" + (c < 10 ? ("00" + c) : (c < 100 ? ("0" + c) : c))));
- return t;
- }) : func.apply(threadNameFormat);
- }
- executor = subExecutor;
- } finally {
- subExecutorLock.unlock();
- }
- return executor;
- }
-
- protected String getNodes(AnyValue config) {
- return config.getValue(CACHE_SOURCE_NODES, config.getValue("url", ""));
- }
-
- @Override
- public void close() throws Exception { //在 Application 关闭时调用
- destroy(null);
- }
-
- @Override
- public String resourceName() {
- return name;
- }
-
- protected String decryptValue(String key, RedisCryptor cryptor, String value) {
- return cryptor != null ? cryptor.decrypt(key, value) : value;
- }
-
- protected T decryptValue(String key, RedisCryptor cryptor, Type type, byte[] bs) {
- return decryptValue(key, cryptor, convert, type, bs);
- }
-
- protected T decryptValue(String key, RedisCryptor cryptor, Convert c, Type type, byte[] bs) {
- if (bs == null) {
- return null;
- }
- if (type == byte[].class) {
- return (T) bs;
- }
- if (cryptor == null && type == String.class) {
- return (T) new String(bs, StandardCharsets.UTF_8);
- }
- if (cryptor == null || (type instanceof Class && (((Class) type).isPrimitive() || Number.class.isAssignableFrom((Class) type)))) {
- return (T) (c == null ? this.convert : c).convertFrom(type, bs);
- }
- String deval = cryptor.decrypt(key, new String(bs, StandardCharsets.UTF_8));
- if (type == String.class) {
- return (T) deval;
- }
- return deval == null ? null : (T) (c == null ? this.convert : c).convertFrom(type, deval.getBytes(StandardCharsets.UTF_8));
- }
-
- protected String encryptValue(String key, RedisCryptor cryptor, String value) {
- return cryptor != null ? cryptor.encrypt(key, value) : value;
- }
-
- protected byte[] encryptValue(String key, RedisCryptor cryptor, Convert c, T value) {
- return encryptValue(key, cryptor, null, c, value);
- }
-
- protected byte[] encryptValue(String key, RedisCryptor cryptor, Type type, Convert c, T value) {
- if (value == null) {
- return null;
- }
- Type t = type == null ? value.getClass() : type;
- if (cryptor == null && t == String.class) {
- return value.toString().getBytes(StandardCharsets.UTF_8);
- }
- byte[] bs = (c == null ? this.convert : c).convertToBytes(t, value);
- if (bs.length > 1 && t instanceof Class && !CharSequence.class.isAssignableFrom((Class) t)) {
- if (bs[0] == '"' && bs[bs.length - 1] == '"') {
- bs = Arrays.copyOfRange(bs, 1, bs.length - 1);
- }
- }
- return encryptValue(key, cryptor, t, bs);
- }
-
- protected byte[] encryptValue(String key, RedisCryptor cryptor, Type type, byte[] bs) {
- if (bs == null) {
- return null;
- }
- if (cryptor == null || (type instanceof Class && (((Class) type).isPrimitive() || Number.class.isAssignableFrom((Class) type)))) {
- return bs;
- }
- String enval = cryptor.encrypt(key, new String(bs, StandardCharsets.UTF_8));
- return enval == null ? null : enval.getBytes(StandardCharsets.UTF_8);
- }
-
- protected T decryptScore(Class scoreType, Double score) {
- if (score == null) {
- return null;
- }
- if (scoreType == int.class || scoreType == Integer.class) {
- return (T) (Number) score.intValue();
- } else if (scoreType == long.class || scoreType == Long.class) {
- return (T) (Number) score.longValue();
- } else if (scoreType == float.class || scoreType == Float.class) {
- return (T) (Number) score.floatValue();
- } else if (scoreType == double.class || scoreType == Double.class) {
- return (T) (Number) score;
- } else {
- return JsonConvert.root().convertFrom(scoreType, score.toString());
- }
- }
-
- protected CompletableFuture returnFutureSize(List> futures) {
- return futures == null || futures.isEmpty() ? CompletableFuture.completedFuture(0) : Utility.allOfFutures(futures).thenApply(v -> futures.size());
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java b/src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java
index a81859e..a8cf91c 100644
--- a/src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java
+++ b/src/main/java/org/redkalex/cache/redis/MyRedisCacheSource.java
@@ -1,27 +1,299 @@
package org.redkalex.cache.redis;
-import org.redkale.annotation.AutoLoad;
-import org.redkale.annotation.ResourceType;
+
+import org.redkale.convert.Convert;
import org.redkale.service.Local;
import org.redkale.source.CacheSource;
-import org.redkale.util.AnyValue;
+import org.redkale.util.AutoLoad;
+import org.redkale.util.ResourceType;
import java.io.Serializable;
+import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.CompletableFuture;
+import java.util.stream.Stream;
@Local
@AutoLoad(false)
@ResourceType(CacheSource.class)
-public class MyRedisCacheSource extends RedisCacheSource {
-
- @Override
- public void init(AnyValue conf) {
- super.init(conf);
+public class MyRedisCacheSource extends RedisCacheSource {
+ //--------------------- oth ------------------------------
+ public boolean setnx(String key, Object v) {
+ byte[][] bytes = Stream.of(key, v).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ Serializable rs = send("SETNX", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ return rs == null ? false : (long) rs == 1;
+ }
+ //--------------------- oth ------------------------------
+
+ //--------------------- bit ------------------------------
+ public boolean getBit(String key, int offset) {
+ byte[][] bytes = Stream.of(key, offset).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ Serializable v = send("GETBIT", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ return v == null ? false : (long) v == 1;
+ }
+
+ public void setBit(String key, int offset, boolean bool) {
+ byte[][] bytes = Stream.of(key, offset, bool ? 1 : 0).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ send("SETBIT", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ }
+
+ //--------------------- bit ------------------------------
+ //--------------------- lock ------------------------------
+ // 尝试加锁，成功返回0，否则返回上一锁剩余毫秒值
+ public int tryLock(String key, int millis) {
+ byte[][] bytes = Stream.of("" +
+ "if (redis.call('exists',KEYS[1]) == 0) then " +
+ "redis.call('psetex', KEYS[1], ARGV[1], 1) " +
+ "return 0; " +
+ "else " +
+ "return redis.call('PTTL', KEYS[1]); " +
+ "end; ", 1, key, millis).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ int n = (int) send("EVAL", CacheEntryType.OBJECT, (Type) null, null, bytes).join();
+ return n;
+ }
+
+ // 加锁
+ public void lock(String key, int millis) {
+ int i;
+ do {
+ i = tryLock(key, millis);
+ try {
+ Thread.sleep(10);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ } while (i > 0);
+ }
+
+ // 解锁
+ public void unlock(String key) {
+ remove(key);
+ }
+
+
+ //--------------------- key ------------------------------
+
+ public long getTtl(String key) {
+ return (long) send("TTL", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8)).join();
+ }
+
+ public long getPttl(String key) {
+ return (long) send("PTTL", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8)).join();
+ }
+
+ public int remove(String... keys) {
+ if (keys == null || keys.length == 0) {
+ return 0;
+ }
+ List para = new ArrayList<>();
+ para.add("" +
+ " local args = ARGV;" +
+ " local x = 0;" +
+ " for i,v in ipairs(args) do" +
+ " local inx = redis.call('del', v);" +
+ " if(inx > 0) then" +
+ " x = x + 1;" +
+ " end" +
+ " end" +
+ " return x;");
+
+ para.add("0");
+ for (Object field : keys) {
+ para.add(String.valueOf(field));
+ }
+ byte[][] bytes = para.stream().map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ return (int) send("EVAL", CacheEntryType.OBJECT, (Type) null, null, bytes).join();
+ }
+
+ //--------------------- hmget ------------------------------
+ public V getHm(String key, T field) {
+ // return (V) send("HMGET", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8)).join();
+ Map map = getHms(key, field);
+ return map.get(field);
+ }
+
+ public Map getHms(String key, T... field) {
+ if (field == null || field.length == 0) {
+ return new HashMap<>();
+ }
+ byte[][] bytes = Stream.concat(Stream.of(key), Stream.of(field)).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ Map result = new HashMap<>();
+
+ List vs = (List) send("HMGET", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ for (int i = 0; i < field.length; i++) { // /*vs != null && vs.size() > i &&*/
+ if (vs.get(i) == null) {
+ continue;
+ }
+ result.put(field[i], vs.get(i));
+ }
+
+ return result;
+ }
+
+ public Map getHmall(String key) {
+ List vs = (List) send("HGETALL", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8)).join();
+ Map result = new HashMap<>(vs.size() / 2);
+ for (int i = 0; i < vs.size(); i += 2) {
+ result.put(String.valueOf(vs.get(i)), vs.get(i + 1));
+ }
+
+ return result;
+ }
+
+ //--------------------- hmset、hmdel、incr ------------------------------
+ public void setHm(String key, T field, V value) {
+ byte[][] bytes = Stream.of(key, field, value).map(x -> x.toString().getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ send("HMSET", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ }
+
+ public void setHms(String key, Map kv) {
+ List args = new ArrayList();
+ args.add(key);
+
+ kv.forEach((k, v) -> {
+ args.add(String.valueOf(k));
+ args.add(String.valueOf(v));
+ });
+
+ byte[][] bytes = args.stream().map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ send("HMSET", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ }
+
+ public Long incrHm(String key, T field, long n) {
+ byte[][] bytes = Stream.of(key, String.valueOf(field), String.valueOf(n)).map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ return (Long) send("HINCRBY", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ }
+
+ public Double incrHm(String key, T field, double n) {
+ byte[][] bytes = Stream.of(key, String.valueOf(field), String.valueOf(n)).map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ Serializable v = send("HINCRBYFLOAT", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ if (v == null) {
+ return null;
+ }
+ return Double.parseDouble(String.valueOf(v));
+ }
+
+ public void hdel(String key, T... field) {
+ byte[][] bytes = Stream.concat(Stream.of(key), Stream.of(field)).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ send("HDEL", null, (Type) null, key, bytes).join();
+ }
+
+ public List zexists(String key, T... fields) {
+ if (fields == null || fields.length == 0) {
+ return new ArrayList<>();
+ }
+ List para = new ArrayList<>();
+ para.add("" +
+ " local key = KEYS[1];" +
+ " local args = ARGV;" +
+ " local result = {};" +
+ " for i,v in ipairs(args) do" +
+ " local inx = redis.call('ZREVRANK', key, v);" +
+ " if(inx) then" +
+ " table.insert(result,1,v);" +
+ " end" +
+ " end" +
+ " return result;");
+ para.add("1");
+ para.add(key);
+ for (Object field : fields) {
+ para.add(String.valueOf(field));
+ }
+ byte[][] bytes = para.stream().map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ return (List) send("EVAL", CacheEntryType.OBJECT, (Type) null, null, bytes).join();
+ }
+
+ //--------------------- set ------------------------------
+ public T srandomItem(String key) {
+ byte[][] bytes = Stream.of(key, 1).map(x -> formatValue(CacheEntryType.OBJECT, (Convert) null, (Type) null, x)).toArray(byte[][]::new);
+ List list = (List) send("SRANDMEMBER", null, (Type) null, key, bytes).join();
+ return list != null && !list.isEmpty() ? list.get(0) : null;
+ }
+
+ public List srandomItems(String key, int n) {
+ byte[][] bytes = Stream.of(key, n).map(x -> formatValue(CacheEntryType.OBJECT, (Convert) null, (Type) null, x)).toArray(byte[][]::new);
+ return (List) send("SRANDMEMBER", null, (Type) null, key, bytes).join();
+ }
+
+ //--------------------- list ------------------------------
+ public CompletableFuture appendListItemsAsync(String key, V... values) {
+ byte[][] bytes = Stream.concat(Stream.of(key), Stream.of(values)).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ return (CompletableFuture) send("RPUSH", null, (Type) null, key, bytes);
+ }
+
+ public CompletableFuture lpushListItemAsync(String key, V value) {
+ return (CompletableFuture) send("LPUSH", null, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.OBJECT, (Convert) null, (Type) null, value));
+ }
+
+ public void lpushListItem(String key, V value) {
+ lpushListItemAsync(key, value).join();
+ }
+
+ public void appendListItems(String key, V... values) {
+ appendListItemsAsync(key, values).join();
+ }
+
+ public void appendSetItems(String key, V... values) {
+ // todo:
+ for (V v : values) {
+ appendSetItem(key, v);
+ }
+ }
+
+ // 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
+ public CompletableFuture> getCollectionAsync(String key, int offset, int limit) {
+ return (CompletableFuture) send("OBJECT", null, (Type) null, key, "ENCODING".getBytes(StandardCharsets.UTF_8), key.getBytes(StandardCharsets.UTF_8)).thenCompose(t -> {
+ if (t == null) return CompletableFuture.completedFuture(null);
+ if (new String((byte[]) t).contains("list")) { //list
+ return send("LRANGE", CacheEntryType.OBJECT, (Type) null, false, key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(offset).getBytes(StandardCharsets.UTF_8), String.valueOf(offset + limit - 1).getBytes(StandardCharsets.UTF_8));
+ } else {
+ return send("SMEMBERS", CacheEntryType.OBJECT, (Type) null, true, key, key.getBytes(StandardCharsets.UTF_8));
+ }
+ });
+ }
+
+ public Collection getCollection(String key, int offset, int limit) {
+ return getCollectionAsync(key, offset, limit).join();
+ }
+
+ public V brpop(String key, int seconds) {
+ byte[][] bytes = Stream.concat(Stream.of(key), Stream.of(seconds)).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ return (V) send("BRPOP", null, (Type) null, key, bytes).join();
+ }
+
+ //--------------------- zset ------------------------------
+ public void zadd(String key, Map kv) {
+ if (kv == null || kv.isEmpty()) {
+ return;
+ }
+ List args = new ArrayList();
+ args.add(key);
+
+ kv.forEach((k, v) -> {
+ args.add(String.valueOf(v));
+ args.add(String.valueOf(k));
+ });
+
+ byte[][] bytes = args.stream().map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ send("ZADD", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ }
+
+ public double zincr(String key, Object number, N n) {
+ byte[][] bytes = Stream.of(key, n, number).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ Serializable v = send("ZINCRBY", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
+ return Double.parseDouble(String.valueOf(v));
+ }
+
+ public void zrem(String key, V... vs) {
+ List args = new ArrayList();
+ args.add(key);
+ for (V v : vs) {
+ args.add(String.valueOf(v));
+ }
+ byte[][] bytes = args.stream().map(x -> x.getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
+ send("ZREM", CacheEntryType.OBJECT, (Type) null, key, bytes).join();
}
-/*
-//--------------------- zset ------------------------------
public int getZrank(String key, V v) {
byte[][] bytes = Stream.of(key, v).map(x -> String.valueOf(x).getBytes(StandardCharsets.UTF_8)).toArray(byte[][]::new);
@@ -142,209 +414,17 @@ public class MyRedisCacheSource extends RedisCacheSource {
}
return map;
}
-* */
- // --------------------
- /*
- supper had support
- public void zadd(String key, Map kv) {
- if (kv == null || kv.isEmpty()) {
- return;
- }
- List args = new ArrayList();
- kv.forEach((k, v) -> {
- args.add(k);
- args.add(v);
- });
+ // ----------
+ protected byte[] formatValue(CacheEntryType cacheType, Convert convert0, Type resultType, Object value) {
+ if (value == null) return "null".getBytes(StandardCharsets.UTF_8);
+ if (convert0 == null) convert0 = convert;
+ if (cacheType == CacheEntryType.LONG || cacheType == CacheEntryType.ATOMIC)
+ return String.valueOf(value).getBytes(StandardCharsets.UTF_8);
+ if (cacheType == CacheEntryType.STRING) return convert0.convertToBytes(String.class, value);
- sendAsync(RedisCommand.ZADD, key, args.toArray(Serializable[]::new)).join();
+ if (value instanceof String) return String.valueOf(value).getBytes(StandardCharsets.UTF_8);
+ if (value instanceof Number) return String.valueOf(value).getBytes(StandardCharsets.UTF_8);
+ return convert0.convertToBytes(resultType == null ? objValueType : resultType, value);
}
-
- public double zincr(String key, Serializable number, N n) {
- return sendAsync(RedisCommand.ZINCRBY, key, number, n).thenApply(x -> x.getDoubleValue(0d)).join();
- }
-
- @Override
- public long zrem(String key, String... vs) {
- return sendAsync(RedisCommand.ZREM, key, keysArgs(key, vs)).thenApply(x -> x.getLongValue(0L)).join();
- }*/
-
- /*public List zexists(String key, T... fields) {
- if (fields == null || fields.length == 0) {
- return new ArrayList<>();
- }
- List para = new ArrayList<>();
- para.add("" +
- " local key = KEYS[1];" +
- " local args = ARGV;" +
- " local result = {};" +
- " for i,v in ipairs(args) do" +
- " local inx = redis.call('ZREVRANK', key, v);" +
- " if(inx) then" +
- " table.insert(result,1,v);" +
- " end" +
- " end" +
- " return result;");
- para.add("1");
- para.add(key);
- for (Object field : fields) {
- para.add(String.valueOf(field));
- }
-
- // todo:
- //sendAsync("EVAL", null, para.toArray(Serializable[]::new)).thenApply(x -> x.).join();
-
- return null;
- }*/
-
- //--------------------- bit ------------------------------
- public boolean getBit(String key, int offset) {
- return sendAsync(RedisCommand.GETBIT, key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(offset).getBytes(StandardCharsets.UTF_8)).thenApply(v -> v.getIntValue(0) > 0).join();
- }
-
- public void setBit(String key, int offset, boolean bool) {
- sendAsync(RedisCommand.SETBIT, key, keysArgs(key, offset + "", bool ? "1" : "0")).join();
- }
- //--------------------- bit ------------------------------
-
- //--------------------- lock ------------------------------
- // 尝试加锁，成功返回0，否则返回上一锁剩余毫秒值
- public long tryLock(String key, int millis) {
- String[] obj = {"" +
- "if (redis.call('EXISTS',KEYS[1]) == 0) then " +
- "redis.call('PSETEX',KEYS[1],ARGV[1],1); " +
- "return 0; " +
- "else " +
- "return redis.call('PTTL',KEYS[1]); " +
- "end;", 1 + "", key, millis + ""
- };
-
- return sendAsync(RedisCommand.EVAL, null, keysArgs(null, obj)).thenApply(v -> v.getIntValue(1)).join();
- }
-
- // 加锁
- public void lock(String key, int millis) {
- long i;
- do {
- i = tryLock(key, millis);
- try {
- Thread.sleep(10);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- } while (i > 0);
- }
-
- // 解锁
- public void unlock(String key) {
- remove(key);
- }
-
- //--------------------- key ------------------------------
-
- public String get(String key) {
- return get(key, String.class);
- }
-
- public void set(String key, Serializable value) {
- sendAsync(RedisCommand.SET, key, keysArgs(key, value + "")).join();
- }
-
- //--------------------- set ------------------------------
- /*public void sadd(String key, Collection args) {
- saddAsync(key, args.toArray(T[]::new)).join();
- }*/
-
- /*public void sadd(String key, Serializable... args) {
- String[] arr = new String[args.length];
- for (int i = 0; i < args.length; i++) {
- arr[i] = args[i] + "";
- }
- saddAsync(key, arr).join();
- }
-
- public void srem(String key, String... args) {
- sremAsync(key, args).join();
- }
-
- public CompletableFuture saddAsync(String key, Serializable... args) {
- return sendAsync(RedisCommand.SADD, key, keysArgs(key, args));
- }
-
- public CompletableFuture sremAsync(String key, String... args) {
- return sendAsync(RedisCommand.SREM, key, keysArgs(key, args));
- }*/
-
- //--------------------- hm ------------------------------
-
- /*public Long incrHm(String key, String field, int value) {
- return sendAsync("HINCRBY", key, field, value).thenApply(x -> x.getLongValue(0l)).join();
- }
-
- public Double incrHm(String key, String field, double value) {
- return sendAsync("HINCRBYFLOAT", key, field, value).thenApply(x -> x.getDoubleValue(0d)).join();
- }*/
-
- public void setHm(String key, String field, Serializable value) {
- setHmsAsync(key, Map.of(field, value)).join();
- }
-
- public void setHms(String key, Map kv) {
- setHmsAsync(key, kv).join();
- }
-
- public CompletableFuture setHmsAsync(String key, Map kv) {
- List args = new ArrayList();
- kv.forEach((k, v) -> {
- args.add(k);
- args.add(v + "");
- });
-
- return sendAsync(RedisCommand.HMSET, key, keysArgs(key, args.toArray(String[]::new)));
- }
-
- public String getHm(String key, String field) {
- return getHm(key, String.class, field);
- }
-
- public T getHm(String key, Class type, String field) {
- List list = super.hmget(key, type, field);
- if (list == null && list.isEmpty()) {
- return null;
- }
- return (T) list.get(0);
- }
-
- public Map getHms(String key, String... field) {
- return getHms(key, String.class, field);
- }
-
- public Map getHms(String key, Class type, String... field) {
- List list = super.hmget(key, type, field);
- if (list == null && list.isEmpty()) {
- return null;
- }
- Map map = new HashMap<>(field.length);
-
- for (int i = 0; i < field.length; i++) {
- if (list.get(i) == null) {
- continue;
- }
- map.put(field[i], (T) list.get(i));
- }
- return map;
- }
-
- /*public Map getHmall(String key) {
- List list = null; // TODO:
- Map map = new HashMap<>();
- if (list.isEmpty()) {
- return map;
- }
-
- for (int i = 0; i + 1 < list.size(); i += 2) {
- map.put((String) list.get(i), list.get(i + 1));
- }
- return map;
- }*/
}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCRC16.java b/src/main/java/org/redkalex/cache/redis/RedisCRC16.java
deleted file mode 100644
index a0e74db..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCRC16.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- *
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.util.Utility;
-
-import java.util.Arrays;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCRC16 {
-
- private static final int MAX_SLOT = 16384;
-
- private static final int[] LOOKUP_TABLE = {0x0000, 0x1021, 0x2042, 0x3063, 0x4084, 0x50A5, 0x60C6,
- 0x70E7, 0x8108, 0x9129, 0xA14A, 0xB16B, 0xC18C, 0xD1AD, 0xE1CE, 0xF1EF, 0x1231, 0x0210, 0x3273,
- 0x2252, 0x52B5, 0x4294, 0x72F7, 0x62D6, 0x9339, 0x8318, 0xB37B, 0xA35A, 0xD3BD, 0xC39C, 0xF3FF,
- 0xE3DE, 0x2462, 0x3443, 0x0420, 0x1401, 0x64E6, 0x74C7, 0x44A4, 0x5485, 0xA56A, 0xB54B, 0x8528,
- 0x9509, 0xE5EE, 0xF5CF, 0xC5AC, 0xD58D, 0x3653, 0x2672, 0x1611, 0x0630, 0x76D7, 0x66F6, 0x5695,
- 0x46B4, 0xB75B, 0xA77A, 0x9719, 0x8738, 0xF7DF, 0xE7FE, 0xD79D, 0xC7BC, 0x48C4, 0x58E5, 0x6886,
- 0x78A7, 0x0840, 0x1861, 0x2802, 0x3823, 0xC9CC, 0xD9ED, 0xE98E, 0xF9AF, 0x8948, 0x9969, 0xA90A,
- 0xB92B, 0x5AF5, 0x4AD4, 0x7AB7, 0x6A96, 0x1A71, 0x0A50, 0x3A33, 0x2A12, 0xDBFD, 0xCBDC, 0xFBBF,
- 0xEB9E, 0x9B79, 0x8B58, 0xBB3B, 0xAB1A, 0x6CA6, 0x7C87, 0x4CE4, 0x5CC5, 0x2C22, 0x3C03, 0x0C60,
- 0x1C41, 0xEDAE, 0xFD8F, 0xCDEC, 0xDDCD, 0xAD2A, 0xBD0B, 0x8D68, 0x9D49, 0x7E97, 0x6EB6, 0x5ED5,
- 0x4EF4, 0x3E13, 0x2E32, 0x1E51, 0x0E70, 0xFF9F, 0xEFBE, 0xDFDD, 0xCFFC, 0xBF1B, 0xAF3A, 0x9F59,
- 0x8F78, 0x9188, 0x81A9, 0xB1CA, 0xA1EB, 0xD10C, 0xC12D, 0xF14E, 0xE16F, 0x1080, 0x00A1, 0x30C2,
- 0x20E3, 0x5004, 0x4025, 0x7046, 0x6067, 0x83B9, 0x9398, 0xA3FB, 0xB3DA, 0xC33D, 0xD31C, 0xE37F,
- 0xF35E, 0x02B1, 0x1290, 0x22F3, 0x32D2, 0x4235, 0x5214, 0x6277, 0x7256, 0xB5EA, 0xA5CB, 0x95A8,
- 0x8589, 0xF56E, 0xE54F, 0xD52C, 0xC50D, 0x34E2, 0x24C3, 0x14A0, 0x0481, 0x7466, 0x6447, 0x5424,
- 0x4405, 0xA7DB, 0xB7FA, 0x8799, 0x97B8, 0xE75F, 0xF77E, 0xC71D, 0xD73C, 0x26D3, 0x36F2, 0x0691,
- 0x16B0, 0x6657, 0x7676, 0x4615, 0x5634, 0xD94C, 0xC96D, 0xF90E, 0xE92F, 0x99C8, 0x89E9, 0xB98A,
- 0xA9AB, 0x5844, 0x4865, 0x7806, 0x6827, 0x18C0, 0x08E1, 0x3882, 0x28A3, 0xCB7D, 0xDB5C, 0xEB3F,
- 0xFB1E, 0x8BF9, 0x9BD8, 0xABBB, 0xBB9A, 0x4A75, 0x5A54, 0x6A37, 0x7A16, 0x0AF1, 0x1AD0, 0x2AB3,
- 0x3A92, 0xFD2E, 0xED0F, 0xDD6C, 0xCD4D, 0xBDAA, 0xAD8B, 0x9DE8, 0x8DC9, 0x7C26, 0x6C07, 0x5C64,
- 0x4C45, 0x3CA2, 0x2C83, 0x1CE0, 0x0CC1, 0xEF1F, 0xFF3E, 0xCF5D, 0xDF7C, 0xAF9B, 0xBFBA, 0x8FD9,
- 0x9FF8, 0x6E17, 0x7E36, 0x4E55, 0x5E74, 0x2E93, 0x3EB2, 0x0ED1, 0x1EF0};
-
- private RedisCRC16() {
- }
-
- public static int crc16(byte[] bytes) {
- int crc = 0x0000;
- for (byte b : bytes) {
- crc = (crc << 8) ^ LOOKUP_TABLE[((crc >>> 8) ^ (b & 0xFF)) & 0xFF];
- }
- return crc & 0xFFFF;
- }
-
- public static int calcSlot(byte[] key) {
- if (key == null) {
- return 0;
- }
- int start = Utility.indexOf(key, (byte) '{');
- if (start != -1) {
- int end = Utility.indexOf(key, start + 1, (byte) '}');
- if (end != -1) {
- key = Arrays.copyOfRange(key, start + 1, end);
- }
- }
- return crc16(key) % MAX_SLOT;
- }
-
- public static int calcSlot(String key) {
- if (key == null) {
- return 0;
- }
- int start = key.indexOf('{');
- if (start != -1) {
- int end = key.indexOf('}');
- if (end != -1 && start + 1 < end) {
- key = key.substring(start + 1, end);
- }
- }
- return crc16(key.getBytes()) % MAX_SLOT;
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheClient.java b/src/main/java/org/redkalex/cache/redis/RedisCacheClient.java
deleted file mode 100644
index d089cf3..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheClient.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.AsyncConnection;
-import org.redkale.net.AsyncGroup;
-import org.redkale.net.client.Client;
-import org.redkale.net.client.ClientAddress;
-import org.redkale.util.Traces;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheClient extends Client {
-
- public RedisCacheClient(String appName, String name, AsyncGroup group, String key,
- ClientAddress address, int maxConns, int maxPipelines, RedisCacheReqAuth authReq, RedisCacheReqDB dbReq) {
- super(name, group, true, address, maxConns, maxPipelines, () -> new RedisCacheReqPing(), () -> new RedisCacheReqClose(), null); //maxConns
- RedisCacheReqClientName clientNameReq = new RedisCacheReqClientName(appName, name);
- if (authReq != null && dbReq != null) {
- this.authenticate = traceid -> {
- Traces.currentTraceid(traceid);
- return conn -> writeChannelBatch(conn, authReq.createTime(), dbReq.createTime(), clientNameReq.createTime())
- .thenApply(v -> conn);
- };
- } else if (authReq != null) {
- this.authenticate = traceid -> {
- Traces.currentTraceid(traceid);
- return conn -> writeChannelBatch(conn, authReq.createTime(), clientNameReq.createTime())
- .thenApply(v -> conn);
- };
- } else if (dbReq != null) {
- this.authenticate = traceid -> {
- Traces.currentTraceid(traceid);
- return conn -> writeChannelBatch(conn, dbReq.createTime(), clientNameReq.createTime())
- .thenApply(v -> conn);
- };
- } else {
- this.authenticate = traceid -> {
- Traces.currentTraceid(traceid);
- return conn -> writeChannel(conn, clientNameReq.createTime())
- .thenApply(v -> conn);
- };
- }
- this.readTimeoutSeconds = 3;
- this.writeTimeoutSeconds = 3;
- }
-
- @Override
- protected RedisCacheConnection createClientConnection(final int index, AsyncConnection channel) {
- return new RedisCacheConnection(this, index, channel);
- }
-
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java b/src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java
deleted file mode 100644
index 793df77..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheCodec.java
+++ /dev/null
@@ -1,255 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientCodec;
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-import org.redkale.util.RedkaleException;
-
-import java.nio.ByteBuffer;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.logging.Logger;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheCodec extends ClientCodec {
-
- protected static final byte TYPE_BULK = '$'; //字符串块类型, 例如：$6\r\n\abcdef\r\n，NULL字符串：$-1\r\n
-
- protected static final byte TYPE_MULTI = '*'; //数组，紧接的数字为数组长度
-
- protected static final byte TYPE_STRING = '+'; //字符串值类型，字符串以\r\n结尾, 例如：+OK\r\n
-
- protected static final byte TYPE_ERROR = '-'; //错误字符串类型，字符串以\r\n结尾, 例如：-ERR unknown command 'red'\r\n
-
- protected static final byte TYPE_NUMBER = ':'; //整型， 例如：:2\r\n
-
- private static final Logger logger = Logger.getLogger(RedisCacheCodec.class.getSimpleName());
-
- private ByteArray halfFrameBytes;
-
- private int halfFrameBulkLength = Integer.MIN_VALUE;
-
- private int halfFrameMultiSize = Integer.MIN_VALUE;
-
- private int halfFrameMultiItemIndex; //从0开始
-
- private byte halfFrameMultiItemType;
-
- private int halfFrameMultiItemLength = Integer.MIN_VALUE;
-
- private byte frameType;
-
- private byte[] frameCursor;
-
- private byte[] frameValue;
-
- private List frameList;
-
- private ByteArray recyclableArray;
-
- public RedisCacheCodec(ClientConnection connection) {
- super(connection);
- }
-
- private ByteArray pollArray(ByteArray array) {
- if (recyclableArray == null) {
- recyclableArray = new ByteArray();
- }
- recyclableArray.clear();
- if (array != null) {
- recyclableArray.put(array);
- }
- return recyclableArray;
- }
-
- private boolean readFrames(RedisCacheConnection conn, ByteBuffer buffer, ByteArray array) {
-// byte[] dbs = new byte[buffer.remaining()];
-// for (int i = 0; i < dbs.length; i++) {
-// dbs[i] = buffer.get(buffer.position() + i);
-// }
-// (System. out).println("[" + Utility.nowMillis() + "] [" + Thread.currentThread().getName() + "]: " + conn + ", 原始数据: " + new String(dbs).replace("\r\n", " "));
-
- array.clear();
- if (this.frameType == 0) {
- this.frameType = buffer.get();
- } else if (halfFrameBytes != null) {
- array.put(halfFrameBytes);
- halfFrameBytes = null;
- }
- if (frameType == TYPE_STRING || frameType == TYPE_ERROR || frameType == TYPE_NUMBER) {
- if (!readComplete(buffer, array)) {
- halfFrameBytes = pollArray(array);
- return false;
- }
- frameValue = array.getBytes();
- } else if (frameType == TYPE_BULK) {
- if (halfFrameBulkLength == Integer.MIN_VALUE) {
- if (!readComplete(buffer, array)) { //没有读到bulkLength
- halfFrameBytes = pollArray(array);
- return false;
- }
- halfFrameBulkLength = readInt(array);
- array.clear();
- }
- if (halfFrameBulkLength == -1) {
- frameValue = null;
- } else {
- int expect = halfFrameBulkLength + 2 - array.length();
- if (buffer.remaining() < expect) {
- array.put(buffer);
- halfFrameBytes = pollArray(array);
- return false;
- }
- array.put(buffer, expect);
- array.removeLastByte(); //移除\n
- array.removeLastByte(); //移除\r
- frameValue = array.getBytes();
- }
- } else if (frameType == TYPE_MULTI) {
- int size = halfFrameMultiSize;
- if (size == Integer.MIN_VALUE) {
- if (!readComplete(buffer, array)) { //没有读到bulkLength
- halfFrameBytes = pollArray(array);
- return false;
- }
- size = readInt(array);
- halfFrameMultiSize = size;
- array.clear();
- frameValue = null;
- }
- if (frameList == null) {
- frameList = new ArrayList<>();
- }
- if (size > 0) {
- int index = halfFrameMultiItemIndex;
- for (int i = index; i < size; i++) {
- if (!buffer.hasRemaining()) {
- return false;
- }
- if (halfFrameMultiItemType == 0) {
- halfFrameMultiItemType = buffer.get();
- }
- halfFrameMultiItemIndex = i;
- final byte itemType = halfFrameMultiItemType;
- if (itemType == TYPE_STRING || itemType == TYPE_ERROR || itemType == TYPE_NUMBER) {
- if (!readComplete(buffer, array)) {
- halfFrameBytes = pollArray(array);
- return false;
- }
- frameList.add(array.getBytes());
- } else if (itemType == TYPE_BULK) {
- if (halfFrameMultiItemLength == Integer.MIN_VALUE) {
- if (!readComplete(buffer, array)) { //没有读到bulkLength
- halfFrameBytes = pollArray(array);
- return false;
- }
- halfFrameMultiItemLength = readInt(array);
- array.clear();
- }
- if (halfFrameMultiItemLength == -1) {
- frameList.add(null);
- } else {
- int expect = halfFrameMultiItemLength + 2 - array.length();
- if (buffer.remaining() < expect) {
- array.put(buffer);
- halfFrameBytes = pollArray(array);
- return false;
- }
- array.put(buffer, expect);
- array.removeLastByte(); //移除\n
- array.removeLastByte(); //移除\r
- frameList.add(array.getBytes());
- }
- } else if (itemType == TYPE_MULTI) { //数组中嵌套数组，例如: SCAN、HSCAN
- if (size == 2 && frameList != null && frameList.size() == 1) {
- //读游标 数据例如: *2 $1 0 *4 $4 key1 $2 10 $4 key2 $2 30
- frameCursor = frameList.get(0);
- frameList.clear();
- clearHalfFrame();
- return readFrames(conn, buffer, array);
- } else {
- throw new RedkaleException("Not support multi type in array data");
- }
- }
- halfFrameMultiItemType = 0;
- halfFrameMultiItemLength = Integer.MIN_VALUE;
- array.clear();
- }
- }
- }
- return true;
- }
-
- private void clearHalfFrame() {
- halfFrameBytes = null;
- halfFrameBulkLength = Integer.MIN_VALUE;
- halfFrameMultiSize = Integer.MIN_VALUE;
- halfFrameMultiItemLength = Integer.MIN_VALUE;
- halfFrameMultiItemIndex = 0; //从0开始
- halfFrameMultiItemType = 0;
- }
-
- @Override
- public void decodeMessages(ByteBuffer realbuf, ByteArray array) {
- RedisCacheConnection conn = (RedisCacheConnection) connection;
- if (!realbuf.hasRemaining()) {
- return;
- }
- ByteBuffer buffer = realbuf;
-
- while (buffer.hasRemaining()) {
- if (!readFrames(conn, buffer, array)) {
- break;
- }
- RedisCacheRequest request = nextRequest();
- if (frameType == TYPE_ERROR) {
- addMessage(request, new RedkaleException(new String(frameValue, StandardCharsets.UTF_8)));
- } else {
- addMessage(request, conn.pollResultSet(request).prepare(frameType, frameCursor, frameValue, frameList));
- }
- frameType = 0;
- frameCursor = null;
- frameValue = null;
- frameList = null;
- clearHalfFrame();
- buffer = realbuf;
- }
- }
-
- protected RedisCacheRequest nextRequest() {
- return super.nextRequest();
- }
-
- private boolean readComplete(ByteBuffer buffer, ByteArray array) {
- while (buffer.hasRemaining()) {
- byte b = buffer.get();
- if (b == '\n') {
- array.removeLastByte(); //移除 \r
- return true;
- }
- array.put(b);
- }
- return false;
- }
-
- private int readInt(ByteArray array) {
- String val = array.toString(StandardCharsets.ISO_8859_1);
- if (val.length() == 1 && val.charAt(0) == '0') {
- return 0;
- }
- if (val.length() == 2 && val.charAt(0) == '-' && val.charAt(1) == '1') {
- return -1;
- }
- return Integer.parseInt(val);
- }
-
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheConnection.java b/src/main/java/org/redkalex/cache/redis/RedisCacheConnection.java
deleted file mode 100644
index a0ae7b7..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheConnection.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.AsyncConnection;
-import org.redkale.net.WorkThread;
-import org.redkale.net.client.Client;
-import org.redkale.net.client.ClientCodec;
-import org.redkale.net.client.ClientConnection;
-import org.redkale.net.client.ClientFuture;
-
-import java.io.Serializable;
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.function.Function;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheConnection extends ClientConnection {
-
- public RedisCacheConnection(Client client, int index, AsyncConnection channel) {
- super(client, index, channel);
- }
-
- @Override
- protected ClientCodec createCodec() {
- return new RedisCacheCodec(this);
- }
-
- protected CompletableFuture writeRequest(RedisCacheRequest request) {
- return super.writeChannel(request);
- }
-
- protected CompletableFuture> writeRequest(RedisCacheRequest[] requests) {
- return super.writeChannel(requests);
- }
-
- protected CompletableFuture writeRequest(RedisCacheRequest request, Function respTransfer) {
- return super.writeChannel(request, respTransfer);
- }
-
- protected CompletableFuture> writeRequest(RedisCacheRequest[] requests, Function respTransfer) {
- return super.writeChannel(requests, respTransfer);
- }
-
- public RedisCacheResult pollResultSet(RedisCacheRequest request) {
- RedisCacheResult rs = new RedisCacheResult();
- return rs;
- }
-
- public RedisCacheRequest pollRequest(WorkThread workThread, String traceid) {
- RedisCacheRequest rs = new RedisCacheRequest().workThread(workThread).traceid(traceid);
- return rs;
- }
-
- protected ClientFuture pollRespFuture(Serializable requestid) {
- return super.pollRespFuture(requestid);
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java
deleted file mode 100644
index b83f70d..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqAuth.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqAuth extends RedisCacheRequest {
-
- private static final byte[] PS = "AUTH\r\n".getBytes(StandardCharsets.UTF_8);
-
- protected String password;
-
- public RedisCacheReqAuth(String password) {
- this.password = password;
- }
-
- @Override
- public void writeTo(ClientConnection conn, ByteArray writer) {
- byte[] pwd = password.getBytes();
- writer.put(mutliLengthBytes(2));
- writer.put(bulkLengthBytes(4));
- writer.put(PS);
-
- writer.put(bulkLengthBytes(pwd.length));
- writer.put(pwd);
- writer.put(CRLF);
-
- }
-
- @Override
- public String toString() {
- return getClass().getSimpleName() + "{AUTH " + password + "}";
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java
deleted file mode 100644
index 1407344..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqClientName.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- *
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-import org.redkale.util.Utility;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqClientName extends RedisCacheRequest {
-
- private final String clientName;
-
- public RedisCacheReqClientName(String appName, String resourceName) {
- this.clientName = "redkalex" + (Utility.isEmpty(appName) ? "" : ("-" + appName))
- + (Utility.isEmpty(resourceName) ? "" : (":" + resourceName));
- }
-
- @Override
- public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put(mutliLengthBytes(3));
-
- writer.put(bulkLengthBytes(6));
- writer.put("CLIENT\r\n".getBytes(StandardCharsets.UTF_8));
-
- writer.put(bulkLengthBytes(7));
- writer.put("SETNAME\r\n".getBytes(StandardCharsets.UTF_8));
-
- byte[] ns = clientName.getBytes(StandardCharsets.UTF_8);
- writer.put(bulkLengthBytes(ns.length));
- writer.put(ns);
- writer.put(CRLF);
- }
-
- @Override
- public String toString() {
- return getClass().getSimpleName() + "{CLIENT SETNAME " + clientName + "}";
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java
deleted file mode 100644
index f1a304d..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqClose.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqClose extends RedisCacheRequest {
-
- private static final byte[] BYTES = new ByteArray()
- .put((byte) '*')
- .put((byte) '1')
- .put((byte) '\r', (byte) '\n')
- .put((byte) '$')
- .put((byte) '4')
- .put((byte) '\r', (byte) '\n')
- .put("QUIT".getBytes(StandardCharsets.UTF_8))
- .put((byte) '\r', (byte) '\n').getBytes();
-
- @Override
- public final boolean isCloseType() {
- return true;
- }
-
- @Override
- public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put(BYTES);
- }
-
- @Override
- public String toString() {
- return getClass().getSimpleName() + "{QUIT}";
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java
deleted file mode 100644
index d9848e0..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqDB.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqDB extends RedisCacheRequest {
-
- protected int db;
-
- public RedisCacheReqDB(int db) {
- this.db = db;
- }
-
- @Override
- public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put(mutliLengthBytes(2));
- writer.put(bulkLengthBytes(6));
- writer.put("SELECT\r\n".getBytes(StandardCharsets.UTF_8));
-
- byte[] dbs = String.valueOf(db).getBytes(StandardCharsets.UTF_8);
- writer.put(bulkLengthBytes(dbs.length));
- writer.put(dbs);
- writer.put(CRLF);
-
- }
-
- @Override
- public String toString() {
- return getClass().getSimpleName() + "{SELECT " + db + "}";
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java b/src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java
deleted file mode 100644
index b6d469a..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheReqPing.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheReqPing extends RedisCacheRequest {
-
- private static final byte[] BYTES = new ByteArray()
- .put((byte) '*')
- .put((byte) '1')
- .put((byte) '\r', (byte) '\n')
- .put((byte) '$')
- .put((byte) '4')
- .put((byte) '\r', (byte) '\n')
- .put("PING".getBytes(StandardCharsets.UTF_8))
- .put((byte) '\r', (byte) '\n').getBytes();
-
- @Override
- public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put(BYTES);
- }
-
- @Override
- public String toString() {
- return getClass().getSimpleName() + "{PING}";
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java b/src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java
deleted file mode 100644
index c32029a..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheRequest.java
+++ /dev/null
@@ -1,116 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.net.client.ClientConnection;
-import org.redkale.net.client.ClientRequest;
-import org.redkale.util.ByteArray;
-
-import java.nio.charset.StandardCharsets;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheRequest extends ClientRequest {
-
- static final byte[] BYTES_TRUE = new byte[]{'t'};
-
- static final byte[] BYTES_FALSE = new byte[]{'f'};
-
- static final byte[] BYTES_MATCH = "MATCH".getBytes(StandardCharsets.UTF_8);
-
- static final byte[] BYTES_COUNT = "COUNT".getBytes(StandardCharsets.UTF_8);
-
- protected static final byte[] CRLF = new byte[]{'\r', '\n'};
-
- private static final byte[][] starLengthBytes;
-
- private static final byte[][] dollarLengthBytes;
-
- static {
- starLengthBytes = new byte[1024][];
- dollarLengthBytes = new byte[1024][];
- for (int i = 0; i < dollarLengthBytes.length; i++) {
- starLengthBytes[i] = ("*" + i + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
- dollarLengthBytes[i] = ("$" + i + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
- }
- }
-
- protected RedisCommand command;
-
- protected String key;
-
- protected byte[][] args;
-
- public static RedisCacheRequest create(RedisCommand command, String key, String... args) {
- return new RedisCacheRequest().prepare(command, key, RedisCacheSource.keysArgs(key, args));
- }
-
- public static RedisCacheRequest create(RedisCommand command, String key, byte[]... args) {
- return new RedisCacheRequest().prepare(command, key, args);
- }
-
- public RedisCacheRequest prepare(RedisCommand command, String key, byte[]... args) {
- super.prepare();
- this.command = command;
- this.key = key;
- this.args = args;
- return this;
- }
-
- public RedisCacheRequest createTime() {
- this.createTime = System.currentTimeMillis();
- return this;
- }
-
- @Override
- public void writeTo(ClientConnection conn, ByteArray writer) {
- writer.put(mutliLengthBytes(args.length + 1));
- writer.put(command.getBytes());
-
- for (final byte[] arg : args) {
- putArgBytes(writer, arg);
- }
- }
-
- protected void putArgBytes(ByteArray writer, byte[] arg) {
- writer.put(bulkLengthBytes(arg.length));
- writer.put(arg);
- writer.put(CRLF);
- }
-
- protected static byte[] mutliLengthBytes(int length) {
- if (length >= 0 && length < starLengthBytes.length) {
- return starLengthBytes[length];
- } else {
- return ("*" + length + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
- }
- }
-
- protected static byte[] bulkLengthBytes(int length) {
- if (length >= 0 && length < dollarLengthBytes.length) {
- return dollarLengthBytes[length];
- } else {
- return ("$" + length + "\r\n").getBytes(StandardCharsets.ISO_8859_1);
- }
- }
-
- @Override
- public String toString() {
- if (args == null || args.length == 0) {
- return getClass().getSimpleName() + "{" + command + " " + key + "}";
- } else {
- StringBuilder sb = new StringBuilder();
- sb.append(command);
- sb.append(" ").append(key);
- for (final byte[] arg : args) {
- sb.append(" ").append(arg == null ? null : new String(arg, StandardCharsets.UTF_8));
- }
- return sb.toString();
- }
- }
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheResult.java b/src/main/java/org/redkalex/cache/redis/RedisCacheResult.java
deleted file mode 100644
index 330569a..0000000
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheResult.java
+++ /dev/null
@@ -1,262 +0,0 @@
-/*
- * To change this license header, choose License Headers in Project Properties.
- * To change this template file, choose Tools | Templates
- * and open the template in the editor.
- */
-package org.redkalex.cache.redis;
-
-import org.redkale.convert.json.JsonConvert;
-import org.redkale.net.client.ClientResult;
-import org.redkale.source.CacheScoredValue;
-
-import java.lang.reflect.Type;
-import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.nio.charset.StandardCharsets;
-import java.util.*;
-
-/**
- *
- * @author zhangjx
- */
-public class RedisCacheResult implements ClientResult {
-
- //$ 块字符串类型
- //* 数组
- //+ 简单字符串类型
- //- 错误类型
- //: 整型
- protected byte frameType;
-
- protected byte[] frameCursor;
-
- protected byte[] frameValue;
-
- protected List frameList;
-
- public RedisCacheResult prepare(byte byteType, byte[] frameCursor, byte[] frameValue, List frameList) {
- this.frameType = byteType;
- this.frameCursor = frameCursor;
- this.frameValue = frameValue;
- this.frameList = frameList;
- return this;
- }
-
- @Override
- public boolean isKeepAlive() {
- return true;
- }
-
- public Void getVoidValue() {
- return null;
- }
-
- public byte[] getFrameValue() {
- return frameValue;
- }
-
- public int getCursor() {
- if (frameCursor == null || frameCursor.length < 1) {
- return -1;
- } else {
- return Integer.parseInt(new String(frameCursor));
- }
- }
-
- public Boolean getBoolValue() {
- if (frameValue == null) {
- return false;
- }
- String val = new String(frameValue, StandardCharsets.UTF_8);
- if ("OK".equals(val)) {
- return true;
- }
- if (val.isEmpty()) {
- return false;
- }
- for (char ch : val.toCharArray()) {
- if (!Character.isDigit(ch)) {
- return false;
- }
- }
- return Integer.parseInt(val) > 0;
- }
-
- public Double getDoubleValue(Double defValue) {
- if (frameValue == null) {
- return defValue;
- }
- String val = new String(frameValue, StandardCharsets.UTF_8);
- if ("nan".equalsIgnoreCase(val) || "-nan".equalsIgnoreCase(val)) {
- return Double.NaN;
- } else if ("inf".equalsIgnoreCase(val)) {
- return Double.POSITIVE_INFINITY;
- } else if ("-inf".equalsIgnoreCase(val)) {
- return Double.NEGATIVE_INFINITY;
- } else if ("-1".equalsIgnoreCase(val)) {
- return -1.0;
- } else if ("0".equalsIgnoreCase(val)) {
- return 0.0;
- } else if ("1".equalsIgnoreCase(val)) {
- return 1.0;
- } else {
- return Double.parseDouble(val);
- }
- }
-
- public Long getLongValue(Long defValue) {
- if (frameValue == null) {
- return defValue;
- }
- String val = new String(frameValue, StandardCharsets.UTF_8);
- if ("-1".equalsIgnoreCase(val)) {
- return -1L;
- } else if ("0".equalsIgnoreCase(val)) {
- return 0L;
- } else if ("1".equalsIgnoreCase(val)) {
- return 1L;
- } else {
- return Long.parseLong(val);
- }
- }
-
- public Integer getIntValue(Integer defValue) {
- if (frameValue == null) {
- return defValue;
- }
- String val = new String(frameValue, StandardCharsets.UTF_8);
- if ("-1".equalsIgnoreCase(val)) {
- return -1;
- } else if ("0".equalsIgnoreCase(val)) {
- return 0;
- } else if ("1".equalsIgnoreCase(val)) {
- return 1;
- } else {
- return Integer.parseInt(val);
- }
- }
-
- public T getObjectValue(String key, RedisCryptor cryptor, Type type) {
- return decodeValue(key, cryptor, frameValue, type);
- }
-
- protected Set getSetValue(String key, RedisCryptor cryptor, Type type) {
- if (frameList == null || frameList.isEmpty()) {
- return new LinkedHashSet<>();
- }
- Set set = new LinkedHashSet<>();
- for (byte[] bs : frameList) {
- set.add(decodeValue(key, cryptor, bs, type));
- }
- return set;
- }
-
- protected List getScoreListValue(String key, RedisCryptor cryptor, Type scoreType) {
- if (frameList == null || frameList.isEmpty()) {
- return new ArrayList<>();
- }
- List set = new ArrayList<>();
- for (int i = 0; i < frameList.size(); i += 2) {
- byte[] bs1 = frameList.get(i);
- byte[] bs2 = frameList.get(i + 1);
- Number val = decodeValue(key, cryptor, bs2, scoreType);
- if (val != null) {
- set.add(CacheScoredValue.create(val, new String(bs1, StandardCharsets.UTF_8)));
- }
- }
- return set;
- }
-
- protected List getListValue(String key, RedisCryptor cryptor, Type type) {
- if (frameList == null || frameList.isEmpty()) {
- return new ArrayList<>();
- }
- List list = new ArrayList<>();
- for (byte[] bs : frameList) {
- list.add(decodeValue(key, cryptor, bs, type));
- }
- return list;
- }
-
- protected Map getMapValue(String key, RedisCryptor cryptor, Type type) {
- if (frameList == null || frameList.isEmpty()) {
- return new LinkedHashMap<>();
- }
- Map map = new LinkedHashMap<>();
- for (int i = 0; i < frameList.size(); i += 2) {
- byte[] bs1 = frameList.get(i);
- byte[] bs2 = frameList.get(i + 1);
- T val = decodeValue(key, cryptor, bs2, type);
- if (val != null) {
- map.put(decodeValue(key, cryptor, bs1, String.class).toString(), val);
- }
- }
- return map;
- }
-
- protected static T decodeValue(String key, RedisCryptor cryptor, byte[] frames, Type type) {
- if (frames == null) {
- return null;
- }
- if (type == byte[].class) {
- return (T) frames;
- }
- if (type == String.class) {
- String val = new String(frames, StandardCharsets.UTF_8);
- if (cryptor != null) {
- val = cryptor.decrypt(key, val);
- }
- return (T) val;
- }
- if (type == int.class || type == Integer.class) {
- return (T) (Integer) Integer.parseInt(new String(frames, StandardCharsets.UTF_8));
- }
- if (type == long.class || type == Long.class) {
- return (T) (Long) Long.parseLong(new String(frames, StandardCharsets.UTF_8));
- }
- if (type == float.class || type == Float.class) {
- return (T) (Float) Float.parseFloat(new String(frames, StandardCharsets.UTF_8));
- }
- if (type == BigInteger.class) {
- return (T) new BigInteger(new String(frames, StandardCharsets.UTF_8));
- }
- if (type == BigDecimal.class) {
- return (T) new BigDecimal(new String(frames, StandardCharsets.UTF_8));
- }
- if (type == boolean.class || type == Boolean.class) {
- String v = new String(frames, StandardCharsets.UTF_8);
- return (T) (Boolean) ("t".equalsIgnoreCase(v) || "1".equals(v));
- }
- if (type == double.class || type == Double.class) {
- return (T) (Double) Double.parseDouble(new String(frames, StandardCharsets.UTF_8));
- }
- if (cryptor != null) {
- String val = cryptor.decrypt(key, new String(frames, StandardCharsets.UTF_8));
- return (T) JsonConvert.root().convertFrom(type, val);
- }
- return (T) JsonConvert.root().convertFrom(type, frames);
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("{type: ").append(frameType);
- if (frameValue != null) {
- sb.append(", value: ").append(new String(frameValue, StandardCharsets.UTF_8));
- }
- if (frameList != null) {
- sb.append(", list: [");
- boolean first = true;
- for (byte[] bs : frameList) {
- if (!first) {
- sb.append(", ");
- }
- sb.append(bs == null ? null : new String(bs, StandardCharsets.UTF_8));
- first = false;
- }
- sb.append("]");
- }
- return sb.append("}").toString();
- }
-
-}
diff --git a/src/main/java/org/redkalex/cache/redis/RedisCacheSource.java b/src/main/java/org/redkalex/cache/redis/RedisCacheSource.java
index 68ffeb5..b799e08 100644
--- a/src/main/java/org/redkalex/cache/redis/RedisCacheSource.java
+++ b/src/main/java/org/redkalex/cache/redis/RedisCacheSource.java
@@ -5,139 +5,149 @@
*/
package org.redkalex.cache.redis;
-import org.redkale.annotation.AutoLoad;
-import org.redkale.annotation.ResourceListener;
-import org.redkale.annotation.ResourceType;
-import org.redkale.annotation.*;
import org.redkale.convert.Convert;
+import org.redkale.convert.bson.BsonByteBufferWriter;
import org.redkale.convert.json.JsonConvert;
-import org.redkale.net.AsyncGroup;
-import org.redkale.net.WorkThread;
-import org.redkale.net.client.*;
+import org.redkale.convert.json.JsonFactory;
+import org.redkale.net.AsyncConnection;
+import org.redkale.net.Transport;
+import org.redkale.net.TransportFactory;
+import org.redkale.service.AbstractService;
import org.redkale.service.Local;
-import org.redkale.source.CacheEventListener;
-import org.redkale.source.CacheScoredValue;
+import org.redkale.service.Service;
import org.redkale.source.CacheSource;
+import org.redkale.source.Flipper;
import org.redkale.util.*;
+import org.redkale.util.AnyValue.DefaultAnyValue;
+import javax.annotation.Resource;
+import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Type;
import java.net.InetSocketAddress;
-import java.net.URI;
+import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.CompletionHandler;
import java.nio.charset.StandardCharsets;
import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Semaphore;
import java.util.logging.Level;
import java.util.logging.Logger;
-import static org.redkale.boot.Application.RESNAME_APP_CLIENT_ASYNCGROUP;
-import static org.redkale.boot.Application.RESNAME_APP_EXECUTOR;
-import static org.redkale.util.Utility.isEmpty;
-import static org.redkale.util.Utility.isNotEmpty;
-import static org.redkalex.cache.redis.RedisCacheRequest.BYTES_COUNT;
-import static org.redkalex.cache.redis.RedisCacheRequest.BYTES_MATCH;
-
/**
* 详情见: https://redkale.org
*
- *
+ * @param Value
* @author zhangjx
*/
@Local
@AutoLoad(false)
@ResourceType(CacheSource.class)
-public class RedisCacheSource extends AbstractRedisSource {
+public class RedisCacheSource extends AbstractService implements CacheSource, Service, AutoCloseable, Resourcable {
- static final boolean debug = false; //System.getProperty("os.name").contains("Window") || System.getProperty("os.name").contains("Mac");
+ protected static final byte DOLLAR_BYTE = '$';
- protected static final byte[] NX = "NX".getBytes();
+ protected static final byte ASTERISK_BYTE = '*';
- protected static final byte[] EX = "EX".getBytes();
+ protected static final byte PLUS_BYTE = '+';
- protected static final byte[] CHANNELS = "CHANNELS".getBytes();
+ protected static final byte MINUS_BYTE = '-';
- private final Logger logger = Logger.getLogger(getClass().getSimpleName());
+ protected static final byte COLON_BYTE = ':';
- @Resource(name = RESNAME_APP_CLIENT_ASYNCGROUP, required = false)
- private AsyncGroup clientAsyncGroup;
+ private final Logger logger = Logger.getLogger(this.getClass().getSimpleName());
- //配置 APP_EXECUTOR资源为null
- @Resource(name = RESNAME_APP_EXECUTOR, required = false)
- private ExecutorService workExecutor;
+ @Resource
+ protected JsonConvert defaultConvert;
- private RedisCacheClient client;
+ @Resource(name = "$_convert")
+ protected JsonConvert convert;
- private InetSocketAddress address;
+ protected Type objValueType = String.class;
- private RedisCacheConnection subConn;
+ protected Map passwords;
- private final ReentrantLock pubsubLock = new ReentrantLock();
+ protected List nodeAddrs;
- //key: topic
- private final Map>> pubsubListeners = new ConcurrentHashMap<>();
+ protected int db;
+
+ protected Transport transport;
@Override
public void init(AnyValue conf) {
- super.init(conf);
- if (conf == null) {
- conf = AnyValue.create();
+ if (this.convert == null) this.convert = this.defaultConvert;
+ if (conf == null) conf = new DefaultAnyValue();
+
+ AnyValue prop = conf.getAnyValue("properties");
+ if (prop != null) {
+ String storeValueStr = prop.getValue("value-type");
+ if (storeValueStr != null) {
+ try {
+ this.initValueType(Thread.currentThread().getContextClassLoader().loadClass(storeValueStr));
+ } catch (Throwable e) {
+ logger.log(Level.SEVERE, this.getClass().getSimpleName() + " load key & value store class (" + storeValueStr + ") error", e);
+ }
+ }
}
- initClient(conf);
+ final int bufferCapacity = conf.getIntValue("bufferCapacity", 8 * 1024);
+ final int bufferPoolSize = conf.getIntValue("bufferPoolSize", Runtime.getRuntime().availableProcessors() * 8);
+ final int threads = conf.getIntValue("threads", Runtime.getRuntime().availableProcessors() * 8);
+ final int readTimeoutSeconds = conf.getIntValue("readTimeoutSeconds", TransportFactory.DEFAULT_READTIMEOUTSECONDS);
+ final int writeTimeoutSeconds = conf.getIntValue("writeTimeoutSeconds", TransportFactory.DEFAULT_WRITETIMEOUTSECONDS);
+ final List addresses = new ArrayList<>();
+ Map passwords0 = new HashMap<>();
+ for (AnyValue node : conf.getAnyValues("node")) {
+ String addrstr = node.getValue("addr");
+ InetSocketAddress addr = null;
+ if (addrstr.startsWith("redis://")) {
+ addrstr = addrstr.substring("redis://".length());
+ int pos = addrstr.indexOf(':');
+ addr = new InetSocketAddress(addrstr.substring(0, pos), Integer.parseInt(addrstr.substring(pos + 1)));
+ addresses.add(addr);
+ } else {
+ addr = new InetSocketAddress(addrstr, node.getIntValue("port"));
+ addresses.add(addr);
+ }
+ String password = node.getValue("password", "").trim();
+ if (!password.isEmpty()) passwords0.put(addr, password.getBytes(StandardCharsets.UTF_8));
+ String db0 = node.getValue("db", "").trim();
+ if (!db0.isEmpty()) this.db = Integer.valueOf(db0);
+ }
+ if (!passwords0.isEmpty()) this.passwords = passwords0;
+ this.nodeAddrs = addresses;
+ TransportFactory transportFactory = TransportFactory.create(threads, bufferPoolSize, bufferCapacity, readTimeoutSeconds, writeTimeoutSeconds);
+ this.transport = transportFactory.createTransportTCP("Redis-Transport", null, addresses);
+ this.transport.setSemaphore(new Semaphore(conf.getIntValue("maxconns", threads)));
+ if (logger.isLoggable(Level.FINE)) logger.log(Level.FINE, RedisCacheSource.class.getSimpleName() + ": addrs=" + addresses + ", db=" + db);
+
}
- private void initClient(AnyValue conf) {
- RedisConfig config = RedisConfig.create(conf);
- if (config.getAddresses().size() != 1) {
- throw new RedkaleException("Only one address supported for " + getClass().getSimpleName());
+ @Override //ServiceLoader时判断配置是否符合当前实现类
+ public boolean match(AnyValue config) {
+ if (config == null) return false;
+ AnyValue[] nodes = config.getAnyValues("node");
+ if (nodes == null || nodes.length == 0) return false;
+ for (AnyValue node : nodes) {
+ if (node.getValue("addr") != null && node.getValue("port") != null) return true;
+ if (node.getValue("addr") != null && node.getValue("addr").startsWith("redis://")) return true;
}
- String oneAddr = config.getAddresses().get(0);
- if (oneAddr.contains("://")) {
- URI uri = URI.create(oneAddr);
- address = new InetSocketAddress(uri.getHost(), uri.getPort() > 0 ? uri.getPort() : 6379);
- } else {
- int pos = oneAddr.indexOf(':');
- address = new InetSocketAddress(pos < 0 ? oneAddr : oneAddr.substring(0, pos), pos < 0 ? 6379 : Integer.parseInt(oneAddr.substring(pos + 1)));
- }
- AsyncGroup ioGroup = clientAsyncGroup;
- if (clientAsyncGroup == null) {
- String f = "Redkalex-Redis-IOThread-" + resourceName() + "-%s";
- ioGroup = AsyncGroup.create(f, workExecutor, 16 * 1024, Utility.cpus() * 4).start();
- }
- RedisCacheClient old = this.client;
- this.client = new RedisCacheClient(appName, resourceName(), ioGroup, resourceName() + "." + config.getDb(),
- new ClientAddress(address), config.getMaxconns(Runtime.getRuntime().availableProcessors()), config.getPipelines(),
- isEmpty(config.getPassword()) ? null : new RedisCacheReqAuth(config.getPassword()),
- config.getDb() < 1 ? null : new RedisCacheReqDB(config.getDb()));
- if (this.subConn != null) {
- this.subConn.dispose(null);
- this.subConn = null;
- }
- if (old != null) {
- old.close();
- }
- if (!pubsubListeners.isEmpty()) {
- subConn().join();
- }
- //if (logger.isLoggable(Level.FINE)) logger.log(Level.FINE, RedisCacheSource.class.getSimpleName() + ": addr=" + address + ", db=" + db);
+ return false;
+ }
+
+ public void updateRemoteAddresses(final Collection addresses) {
+ this.transport.updateRemoteAddresses(addresses);
}
@Override
- @ResourceListener
- public void onResourceChange(ResourceEvent[] events) {
- if (events == null || events.length < 1) {
- return;
- }
- StringBuilder sb = new StringBuilder();
- for (ResourceEvent event : events) {
- sb.append("CacheSource(name=").append(resourceName()).append(") change '").append(event.name()).append("' to '").append(event.coverNewValue()).append("'\r\n");
- }
- initClient(this.conf);
- if (sb.length() > 0) {
- logger.log(Level.INFO, sb.toString());
- }
+ @Deprecated
+ public final void initValueType(Type valueType) {
+ this.objValueType = valueType == null ? String.class : valueType;
+ }
+
+ @Override
+ @Deprecated
+ public final void initTransient(boolean flag) {
}
@Override
@@ -145,1064 +155,839 @@ public class RedisCacheSource extends AbstractRedisSource {
return "redis";
}
+ public static void main(String[] args) throws Exception {
+ DefaultAnyValue conf = new DefaultAnyValue().addValue("maxconns", "1");
+ conf.addValue("node", new DefaultAnyValue().addValue("addr", "127.0.0.1").addValue("port", "6363"));
+
+ RedisCacheSource source = new RedisCacheSource();
+ source.defaultConvert = JsonFactory.root().getConvert();
+ source.init(conf);
+ InetSocketAddress addr = new InetSocketAddress("127.0.0.1", 7788);
+ try {
+ System.out.println("------------------------------------");
+ source.removeAsync("stritem1");
+ source.removeAsync("stritem2");
+ source.setStringAsync("stritem1", "value1");
+ source.setStringAsync("stritem2", "value2");
+ System.out.println("stritem开头的key有两个: " + source.queryKeysStartsWith("stritem"));
+ System.out.println("[有值] MGET : " + source.getStringMap("stritem1", "stritem2"));
+ System.out.println("[有值] MGET : " + Arrays.toString(source.getStringArray("stritem1", "stritem2")));
+
+ source.remove("intitem1");
+ source.remove("intitem2");
+ source.setLong("intitem1", 333);
+ source.setLong("intitem2", 444);
+ System.out.println("[有值] MGET : " + source.getStringMap("intitem1", "intitem22", "intitem2"));
+ System.out.println("[有值] MGET : " + Arrays.toString(source.getStringArray("intitem1", "intitem22", "intitem2")));
+ source.remove("objitem1");
+ source.remove("objitem2");
+ source.set("objitem1", Flipper.class, new Flipper(10));
+ source.set("objitem2", Flipper.class, new Flipper(20));
+ System.out.println("[有值] MGET : " + source.getMap(Flipper.class, "objitem1", "objitem2"));
+
+ source.remove("key1");
+ source.remove("key2");
+ source.remove("300");
+ source.set(1000, "key1", String.class, "value1");
+ source.set("key1", String.class, "value1");
+ source.setString("keystr1", "strvalue1");
+ source.setLong("keylong1", 333L);
+ source.set("300", String.class, "4000");
+ source.getAndRefresh("key1", 3500, String.class);
+ System.out.println("[有值] 300 GET : " + source.get("300", String.class));
+ System.out.println("[有值] key1 GET : " + source.get("key1", String.class));
+ System.out.println("[无值] key2 GET : " + source.get("key2", String.class));
+ System.out.println("[有值] keystr1 GET : " + source.getString("keystr1"));
+ System.out.println("[有值] keylong1 GET : " + source.getLong("keylong1", 0L));
+ System.out.println("[有值] key1 EXISTS : " + source.exists("key1"));
+ System.out.println("[无值] key2 EXISTS : " + source.exists("key2"));
+
+ source.remove("keys3");
+ source.appendListItem("keys3", String.class, "vals1");
+ source.appendListItem("keys3", String.class, "vals2");
+ System.out.println("-------- keys3 追加了两个值 --------");
+ System.out.println("[两值] keys3 VALUES : " + source.getCollection("keys3", String.class));
+ System.out.println("[有值] keys3 EXISTS : " + source.exists("keys3"));
+ source.removeListItem("keys3", String.class, "vals1");
+ System.out.println("[一值] keys3 VALUES : " + source.getCollection("keys3", String.class));
+ source.getCollectionAndRefresh("keys3", 3000, String.class);
+
+ source.remove("stringmap");
+ source.appendSetItem("stringmap", JsonConvert.TYPE_MAP_STRING_STRING, Utility.ofMap("a", "aa", "b", "bb"));
+ source.appendSetItem("stringmap", JsonConvert.TYPE_MAP_STRING_STRING, Utility.ofMap("c", "cc", "d", "dd"));
+ System.out.println("[两值] stringmap VALUES : " + source.getCollectionAsync("stringmap", JsonConvert.TYPE_MAP_STRING_STRING).join());
+
+ source.remove("sets3");
+ source.remove("sets4");
+ source.appendSetItem("sets3", String.class, "setvals1");
+ source.appendSetItem("sets3", String.class, "setvals2");
+ source.appendSetItem("sets3", String.class, "setvals1");
+ source.appendSetItem("sets4", String.class, "setvals2");
+ source.appendSetItem("sets4", String.class, "setvals1");
+ System.out.println("[两值] sets3 VALUES : " + source.getCollection("sets3", String.class));
+ System.out.println("[有值] sets3 EXISTS : " + source.exists("sets3"));
+ System.out.println("[有值] sets3-setvals2 EXISTSITEM : " + source.existsSetItem("sets3", String.class, "setvals2"));
+ System.out.println("[有值] sets3-setvals3 EXISTSITEM : " + source.existsSetItem("sets3", String.class, "setvals3"));
+ source.removeSetItem("sets3", String.class, "setvals1");
+ System.out.println("[一值] sets3 VALUES : " + source.getCollection("sets3", String.class));
+ System.out.println("sets3 大小 : " + source.getCollectionSize("sets3"));
+ System.out.println("all keys: " + source.queryKeys());
+ System.out.println("key startkeys: " + source.queryKeysStartsWith("key"));
+ System.out.println("newnum 值 : " + source.incr("newnum"));
+ System.out.println("newnum 值 : " + source.decr("newnum"));
+ System.out.println("sets3&sets4: " + source.getStringCollectionMap(true, "sets3", "sets4"));
+ System.out.println("------------------------------------");
+ source.set("myaddr", InetSocketAddress.class, addr);
+ System.out.println("myaddrstr: " + source.getString("myaddr"));
+ System.out.println("myaddr: " + source.get("myaddr", InetSocketAddress.class));
+ source.remove("myaddrs");
+ source.remove("myaddrs2");
+ source.appendSetItem("myaddrs", InetSocketAddress.class, new InetSocketAddress("127.0.0.1", 7788));
+ source.appendSetItem("myaddrs", InetSocketAddress.class, new InetSocketAddress("127.0.0.1", 7799));
+ System.out.println("myaddrs: " + source.getCollection("myaddrs", InetSocketAddress.class));
+ source.removeSetItem("myaddrs", InetSocketAddress.class, new InetSocketAddress("127.0.0.1", 7788));
+ System.out.println("myaddrs: " + source.getCollection("myaddrs", InetSocketAddress.class));
+ source.appendSetItem("myaddrs2", InetSocketAddress.class, new InetSocketAddress("127.0.0.1", 7788));
+ source.appendSetItem("myaddrs2", InetSocketAddress.class, new InetSocketAddress("127.0.0.1", 7799));
+ System.out.println("myaddrs&myaddrs2: " + source.getCollectionMap(true, InetSocketAddress.class, "myaddrs", "myaddrs2"));
+ System.out.println("------------------------------------");
+ source.remove("myaddrs");
+ Type mapType = new TypeToken>() {
+ }.getType();
+ Map map = new HashMap<>();
+ map.put("a", 1);
+ map.put("b", 2);
+ source.set("mapvals", mapType, map);
+ System.out.println("mapvals: " + source.get("mapvals", mapType));
+
+ source.remove("byteskey");
+ source.setBytes("byteskey", new byte[]{1, 2, 3});
+ System.out.println("byteskey 值 : " + Arrays.toString(source.getBytes("byteskey")));
+ //h
+ source.remove("hmap");
+ source.hincr("hmap", "key1");
+ System.out.println("hmap.key1 值 : " + source.hgetLong("hmap", "key1", -1));
+ source.hmset("hmap", "key2", "haha", "key3", 333);
+ source.hmset("hmap", "sm", (HashMap) Utility.ofMap("a", "aa", "b", "bb"));
+ System.out.println("hmap.sm 值 : " + source.hget("hmap", "sm", JsonConvert.TYPE_MAP_STRING_STRING));
+ System.out.println("hmap.[key1,key2,key3] 值 : " + source.hmget("hmap", String.class, "key1", "key2", "key3"));
+ System.out.println("hmap.keys 四值 : " + source.hkeys("hmap"));
+ source.hremove("hmap", "key1", "key3");
+ System.out.println("hmap.keys 两值 : " + source.hkeys("hmap"));
+ System.out.println("hmap.key2 值 : " + source.hgetString("hmap", "key2"));
+ System.out.println("hmap列表(2)大小 : " + source.hsize("hmap"));
+
+ source.remove("hmaplong");
+ source.hincr("hmaplong", "key1", 10);
+ source.hsetLong("hmaplong", "key2", 30);
+ System.out.println("hmaplong.所有两值 : " + source.hmap("hmaplong", long.class, 0, 10));
+
+ source.remove("hmapstr");
+ source.hsetString("hmapstr", "key1", "str10");
+ source.hsetString("hmapstr", "key2", null);
+ System.out.println("hmapstr.所有一值 : " + source.hmap("hmapstr", String.class, 0, 10));
+
+ source.remove("hmapstrmap");
+ source.hset("hmapstrmap", "key1", JsonConvert.TYPE_MAP_STRING_STRING, (HashMap) Utility.ofMap("ks11", "vv11"));
+ source.hset("hmapstrmap", "key2", JsonConvert.TYPE_MAP_STRING_STRING, null);
+ System.out.println("hmapstrmap.无值 : " + source.hmap("hmapstrmap", JsonConvert.TYPE_MAP_STRING_STRING, 0, 10, "key2*"));
+
+ source.remove("popset");
+ source.appendStringSetItem("popset", "111");
+ source.appendStringSetItem("popset", "222");
+ source.appendStringSetItem("popset", "333");
+ source.appendStringSetItem("popset", "444");
+ source.appendStringSetItem("popset", "555");
+ System.out.println("SPOP一个元素：" + source.spopStringSetItem("popset"));
+ System.out.println("SPOP两个元素：" + source.spopStringSetItem("popset", 2));
+ System.out.println("SPOP五个元素：" + source.spopStringSetItem("popset", 5));
+ source.appendLongSetItem("popset", 111);
+ source.appendLongSetItem("popset", 222);
+ source.appendLongSetItem("popset", 333);
+ source.appendLongSetItem("popset", 444);
+ source.appendLongSetItem("popset", 555);
+ System.out.println("SPOP一个元素：" + source.spopLongSetItem("popset"));
+ System.out.println("SPOP两个元素：" + source.spopLongSetItem("popset", 2));
+ System.out.println("SPOP五个元素：" + source.spopLongSetItem("popset", 5));
+ System.out.println("SPOP一个元素：" + source.spopLongSetItem("popset"));
+
+ //清除
+ int rs = source.remove("stritem1");
+ System.out.println("删除stritem1个数: " + rs);
+ source.remove("popset");
+ source.remove("stritem2");
+ source.remove("intitem1");
+ source.remove("intitem2");
+ source.remove("keylong1");
+ source.remove("keystr1");
+ source.remove("mapvals");
+ source.remove("myaddr");
+ source.remove("myaddrs2");
+ source.remove("newnum");
+ source.remove("objitem1");
+ source.remove("objitem2");
+ source.remove("key1");
+ source.remove("key2");
+ source.remove("keys3");
+ source.remove("sets3");
+ source.remove("sets4");
+ source.remove("myaddrs");
+ source.remove("300");
+ source.remove("stringmap");
+ source.remove("hmap");
+ source.remove("hmaplong");
+ source.remove("hmapstr");
+ source.remove("hmapstrmap");
+ source.remove("byteskey");
+ System.out.println("------------------------------------");
+// System.out.println("--------------测试大文本---------------");
+// HashMap bigmap = new HashMap<>();
+// StringBuilder sb = new StringBuilder();
+// sb.append("起始");
+// for (int i = 0; i < 1024 * 1024; i++) {
+// sb.append("abcde");
+// }
+// sb.append("结束");
+// bigmap.put("val", sb.toString());
+// System.out.println("文本长度: " + sb.length());
+// source.set("bigmap", JsonConvert.TYPE_MAP_STRING_STRING, bigmap);
+// System.out.println("写入完成");
+// for (int i = 0; i < 1; i++) {
+// HashMap fs = (HashMap) source.get("bigmap", JsonConvert.TYPE_MAP_STRING_STRING);
+// System.out.println("内容长度: " + fs.get("val").length());
+// }
+ source.remove("bigmap");
+
+ } finally {
+ source.close();
+ }
+ }
+
+ @Override
+ public void close() throws Exception { //在 Application 关闭时调用
+ destroy(null);
+ }
+
+ @Override
+ public String resourceName() {
+ Resource res = this.getClass().getAnnotation(Resource.class);
+ return res == null ? "" : res.name();
+ }
+
@Override
public String toString() {
- return getClass().getSimpleName() + "{name=" + resourceName() + ", addrs=" + this.address + ", db=" + this.db + "}";
+ return getClass().getSimpleName() + "{addrs = " + this.nodeAddrs + ", db=" + this.db + "}";
}
@Override
public void destroy(AnyValue conf) {
- super.destroy(conf);
- if (client != null) {
- client.close();
- }
- }
-
- protected CompletableFuture subConn() {
- RedisCacheConnection conn = this.subConn;
- if (conn != null) {
- return CompletableFuture.completedFuture(conn);
- }
- return client.newConnection().thenApply(r -> {
- pubsubLock.lock();
- try {
- if (subConn == null) {
- subConn = r;
- r.getCodec().withMessageListener(new ClientMessageListener() {
- @Override
- public void onMessage(ClientConnection conn, ClientResponse resp) {
- if (resp.getCause() == null) {
- RedisCacheResult result = (RedisCacheResult) resp.getMessage();
- if (result.getFrameValue() == null) {
- List events = result.getListValue(null, null, byte[].class);
- String type = new String(events.get(0), StandardCharsets.UTF_8);
- if (events.size() == 3 && "message".equals(type)) {
- String channel = new String(events.get(1), StandardCharsets.UTF_8);
- Set> set = pubsubListeners.get(channel);
- if (set != null) {
- byte[] msg = events.get(2);
- for (CacheEventListener item : set) {
- subExecutor().execute(() -> {
- try {
- item.onMessage(channel, msg);
- } catch (Throwable t) {
- logger.log(Level.SEVERE, "CacheSource subscribe message error, topic: " + channel, t);
- }
- });
- }
- }
- } else {
- RedisCacheRequest request = ((RedisCacheCodec) conn.getCodec()).nextRequest();
- ClientFuture respFuture = ((RedisCacheConnection) conn).pollRespFuture(request.getRequestid());
- respFuture.complete(result);
- }
- } else {
- RedisCacheRequest request = ((RedisCacheCodec) conn.getCodec()).nextRequest();
- ClientFuture respFuture = ((RedisCacheConnection) conn).pollRespFuture(request.getRequestid());
- respFuture.complete(result);
- }
- } else {
- RedisCacheRequest request = ((RedisCacheCodec) conn.getCodec()).nextRequest();
- ClientFuture respFuture = ((RedisCacheConnection) conn).pollRespFuture(request.getRequestid());
- respFuture.completeExceptionally(resp.getCause());
- }
- }
-
- public void onClose(ClientConnection conn) {
- subConn = null;
- }
- });
- //重连时重新订阅
- if (!pubsubListeners.isEmpty()) {
- final Map, HashSet> listeners = new HashMap<>();
- pubsubListeners.forEach((t, s) -> {
- s.forEach(l -> listeners.computeIfAbsent(l, x -> new HashSet<>()).add(t));
- });
- listeners.forEach((listener, topics) -> {
- subscribeAsync(listener, topics.toArray(Creator.funcStringArray()));
- });
- }
- }
- return subConn;
- } finally {
- pubsubLock.unlock();
- }
- });
- }
-
- @Override
- public CompletableFuture isOpenAsync() {
- return CompletableFuture.completedFuture(client != null);
- }
-
- //------------------------ 订阅发布 SUB/PUB ------------------------
- @Override
- public CompletableFuture> pubsubChannelsAsync(@Nullable String pattern) {
- CompletableFuture future = pattern == null ? sendAsync(RedisCommand.PUBSUB, "CHANNELS", CHANNELS)
- : sendAsync(RedisCommand.PUBSUB, "CHANNELS", CHANNELS, pattern.getBytes(StandardCharsets.UTF_8));
- return future.thenApply(v -> v.getListValue("CHANNELS", null, String.class));
- }
-
- @Override
- public CompletableFuture subscribeAsync(CacheEventListener listener, String... topics) {
- Objects.requireNonNull(listener);
- if (topics == null || topics.length < 1) {
- throw new RedkaleException("topics is empty");
- }
- WorkThread workThread = WorkThread.currentWorkThread();
- String traceid = Traces.currentTraceid();
- return subConn()
- .thenCompose(conn
- -> conn.writeRequest(conn.pollRequest(workThread, traceid).prepare(RedisCommand.SUBSCRIBE, null, keysArgs(topics)))
- .thenApply(v -> {
- for (String topic : topics) {
- pubsubListeners.computeIfAbsent(topic, y -> new CopyOnWriteArraySet<>()).add(listener);
- }
- return null;
- })
- );
- }
-
- @Override
- public CompletableFuture unsubscribeAsync(CacheEventListener listener, String... topics) {
- if (listener == null) { //清掉指定topic的所有订阅者
- Set delTopics = new HashSet<>();
- if (topics == null || topics.length < 1) {
- delTopics.addAll(pubsubListeners.keySet());
- } else {
- delTopics.addAll(Arrays.asList(topics));
- }
- List> futures = new ArrayList<>();
- delTopics.forEach(topic -> {
- futures.add(subConn().thenCompose(conn -> conn.writeRequest(RedisCacheRequest.create(RedisCommand.UNSUBSCRIBE, topic, topic.getBytes(StandardCharsets.UTF_8)))
- .thenApply(r -> {
- pubsubListeners.remove(topic);
- return null;
- })
- ));
- });
- return returnFutureSize(futures);
- } else { //清掉指定topic的指定订阅者
- List> futures = new ArrayList<>();
- for (String topic : topics) {
- CopyOnWriteArraySet> listens = pubsubListeners.get(topic);
- if (listens == null) {
- continue;
- }
- listens.remove(listener);
- if (listens.isEmpty()) {
- futures.add(subConn().thenCompose(conn -> conn.writeRequest(RedisCacheRequest.create(RedisCommand.UNSUBSCRIBE, topic, topic.getBytes(StandardCharsets.UTF_8)))
- .thenApply(r -> {
- pubsubListeners.remove(topic);
- return null;
- })
- ));
- }
- }
- return returnFutureSize(futures);
- }
- }
-
- @Override
- public CompletableFuture publishAsync(String topic, byte[] message) {
- Objects.requireNonNull(topic);
- Objects.requireNonNull(message);
- return sendAsync(RedisCommand.PUBLISH, topic, topic.getBytes(StandardCharsets.UTF_8), message).thenApply(v -> v.getIntValue(0));
+ if (transport != null) transport.close();
}
//--------------------- exists ------------------------------
@Override
public CompletableFuture existsAsync(String key) {
- return sendAsync(RedisCommand.EXISTS, key, keyArgs(key)).thenApply(v -> v.getIntValue(0) > 0);
+ return (CompletableFuture) send("EXISTS", null, (Type) null, key, key.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Override
+ public boolean exists(String key) {
+ return existsAsync(key).join();
}
//--------------------- get ------------------------------
+ @Override
+ @Deprecated
+ public CompletableFuture getAsync(String key) {
+ return (CompletableFuture) send("GET", CacheEntryType.OBJECT, (Type) null, key, key.getBytes(StandardCharsets.UTF_8));
+ }
+
@Override
public CompletableFuture getAsync(String key, Type type) {
- return sendAsync(RedisCommand.GET, key, keyArgs(key)).thenApply(v -> v.getObjectValue(key, cryptor, type));
- }
-
- //--------------------- getex ------------------------------
- @Override
- public CompletableFuture getexAsync(String key, int expireSeconds, final Type type) {
- return sendAsync(RedisCommand.GETEX, key, keyArgs(key, "EX", expireSeconds)).thenApply(v -> v.getObjectValue(key, cryptor, type));
+ return (CompletableFuture) send("GET", CacheEntryType.OBJECT, type, key, key.getBytes(StandardCharsets.UTF_8));
}
@Override
- public CompletableFuture msetAsync(final Serializable... keyVals) {
- if (keyVals.length % 2 != 0) {
- throw new RedkaleException("key value must be paired");
- }
- return sendAsync(RedisCommand.MSET, keyVals[0].toString(), keymArgs(keyVals)).thenApply(v -> v.getVoidValue());
+ public CompletableFuture getStringAsync(String key) {
+ return (CompletableFuture) send("GET", CacheEntryType.STRING, (Type) null, key, key.getBytes(StandardCharsets.UTF_8));
}
@Override
- public CompletableFuture msetAsync(final Map map) {
- if (isEmpty(map)) {
- return CompletableFuture.completedFuture(null);
- }
- return sendAsync(RedisCommand.MSET, map.keySet().stream().findFirst().orElse("").toString(), keymArgs(map)).thenApply(v -> v.getVoidValue());
+ public CompletableFuture getLongAsync(String key, long defValue) {
+ return ((CompletableFuture) send("GET", CacheEntryType.LONG, (Type) null, key, key.getBytes(StandardCharsets.UTF_8))).thenApplyAsync(v -> v == null ? defValue : v);
+ }
+
+ @Override
+ @Deprecated
+ public V get(String key) {
+ return getAsync(key).join();
+ }
+
+ @Override
+ public T get(String key, final Type type) {
+ return (T) getAsync(key, type).join();
+ }
+
+ @Override
+ public String getString(String key) {
+ return getStringAsync(key).join();
+ }
+
+ @Override
+ public long getLong(String key, long defValue) {
+ return getLongAsync(key, defValue).join();
+ }
+
+ //--------------------- getAndRefresh ------------------------------
+ @Override
+ @Deprecated
+ public CompletableFuture getAndRefreshAsync(String key, int expireSeconds) {
+ return (CompletableFuture) refreshAsync(key, expireSeconds).thenCompose(v -> getAsync(key));
+ }
+
+ @Override
+ public CompletableFuture getAndRefreshAsync(String key, int expireSeconds, final Type type) {
+ return (CompletableFuture) refreshAsync(key, expireSeconds).thenCompose(v -> getAsync(key, type));
+ }
+
+ @Override
+ @Deprecated
+ public V getAndRefresh(String key, final int expireSeconds) {
+ return getAndRefreshAsync(key, expireSeconds).join();
+ }
+
+ @Override
+ public T getAndRefresh(String key, final int expireSeconds, final Type type) {
+ return (T) getAndRefreshAsync(key, expireSeconds, type).join();
+ }
+
+ @Override
+ public CompletableFuture getStringAndRefreshAsync(String key, int expireSeconds) {
+ return (CompletableFuture) refreshAsync(key, expireSeconds).thenCompose(v -> getStringAsync(key));
+ }
+
+ @Override
+ public String getStringAndRefresh(String key, final int expireSeconds) {
+ return getStringAndRefreshAsync(key, expireSeconds).join();
+ }
+
+ @Override
+ public CompletableFuture getLongAndRefreshAsync(String key, int expireSeconds, long defValue) {
+ return (CompletableFuture) refreshAsync(key, expireSeconds).thenCompose(v -> getLongAsync(key, defValue));
+ }
+
+ @Override
+ public long getLongAndRefresh(String key, final int expireSeconds, long defValue) {
+ return getLongAndRefreshAsync(key, expireSeconds, defValue).join();
+ }
+
+ //--------------------- refresh ------------------------------
+ @Override
+ public CompletableFuture refreshAsync(String key, int expireSeconds) {
+ return setExpireSecondsAsync(key, expireSeconds);
+ }
+
+ @Override
+ public void refresh(String key, final int expireSeconds) {
+ setExpireSeconds(key, expireSeconds);
+ }
+
+ //--------------------- set ------------------------------
+ @Override
+ @Deprecated
+ public CompletableFuture setAsync(String key, V value) {
+ CacheEntryType cet = this.objValueType == String.class ? CacheEntryType.STRING : CacheEntryType.OBJECT;
+ return (CompletableFuture) send("SET", cet, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), formatValue(cet, (Convert) null, (Type) null, value));
+ }
+
+ @Override
+ public CompletableFuture setAsync(String key, Convert convert, T value) {
+ CacheEntryType cet = value instanceof CharSequence ? CacheEntryType.STRING : CacheEntryType.OBJECT;
+ return (CompletableFuture) send("SET", cet, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), formatValue(cet, convert, (Type) null, value));
+ }
+
+ @Override
+ public CompletableFuture setAsync(String key, final Type type, T value) {
+ CacheEntryType cet = type == String.class ? CacheEntryType.STRING : CacheEntryType.OBJECT;
+ return (CompletableFuture) send("SET", cet, type, key, key.getBytes(StandardCharsets.UTF_8), formatValue(cet, (Convert) null, type, value));
}
- //--------------------- setex ------------------------------
@Override
public CompletableFuture setAsync(String key, Convert convert, final Type type, T value) {
- return sendAsync(RedisCommand.SET, key, keyArgs(key, convert, type, value)).thenApply(v -> v.getVoidValue());
+ CacheEntryType cet = type == String.class ? CacheEntryType.STRING : CacheEntryType.OBJECT;
+ return (CompletableFuture) send("SET", cet, type, key, key.getBytes(StandardCharsets.UTF_8), formatValue(cet, convert, type, value));
}
@Override
- public CompletableFuture setnxAsync(String key, Convert convert, final Type type, T value) {
- return sendAsync(RedisCommand.SETNX, key, keyArgs(key, convert, type, value)).thenApply(v -> v.getBoolValue());
+ @Deprecated
+ public void set(final String key, V value) {
+ setAsync(key, value).join();
}
@Override
- public CompletableFuture getSetAsync(String key, Convert convert, final Type type, T value) {
- return sendAsync(RedisCommand.GETSET, key, keyArgs(key, convert, type, value)).thenApply(v -> v.getObjectValue(key, cryptor, type));
+ public void set(final String key, final Convert convert, T value) {
+ setAsync(key, convert, value).join();
}
@Override
- public CompletableFuture getDelAsync(String key, final Type type) {
- return sendAsync(RedisCommand.GETDEL, key, keyArgs(key)).thenApply(v -> v.getObjectValue(key, cryptor, type));
- }
-
- //--------------------- setex ------------------------------
- @Override
- public CompletableFuture setexAsync(String key, int expireSeconds, Convert convert, final Type type, T value) {
- return sendAsync(RedisCommand.SETEX, key, keyArgs(key, expireSeconds, convert, type, value)).thenApply(v -> v.getVoidValue());
+ public void set(final String key, final Type type, T value) {
+ setAsync(key, type, value).join();
}
@Override
- public CompletableFuture setnxexAsync(String key, int expireSeconds, Convert convert, final Type type, T value) {
- return sendAsync(RedisCommand.SET, key, keyArgs(key, expireSeconds, NX, EX, convert, type, value)).thenApply(v -> v.getBoolValue());
- }
-
- //--------------------- expire ------------------------------
- @Override
- public CompletableFuture expireAsync(String key, int expireSeconds) {
- return sendAsync(RedisCommand.EXPIRE, key, keyArgs(key, expireSeconds)).thenApply(v -> v.getVoidValue());
- }
-
- //--------------------- persist ------------------------------
- @Override
- public CompletableFuture persistAsync(String key) {
- return sendAsync(RedisCommand.PERSIST, key, keyArgs(key)).thenApply(v -> v.getBoolValue());
- }
-
- //--------------------- rename ------------------------------
- @Override
- public CompletableFuture renameAsync(String oldKey, String newKey) {
- return sendAsync(RedisCommand.RENAME, oldKey, keysArgs(oldKey, newKey)).thenApply(v -> v.getBoolValue());
+ public void set(String key, final Convert convert, final Type type, T value) {
+ setAsync(key, convert, type, value).join();
}
@Override
- public CompletableFuture renamenxAsync(String oldKey, String newKey) {
- return sendAsync(RedisCommand.RENAMENX, oldKey, keysArgs(oldKey, newKey)).thenApply(v -> v.getBoolValue());
+ public CompletableFuture setStringAsync(String key, String value) {
+ return (CompletableFuture) send("SET", CacheEntryType.STRING, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.STRING, (Convert) null, (Type) null, value));
}
- //--------------------- del ------------------------------
@Override
- public CompletableFuture delAsync(String... keys) {
- if (keys.length == 0) {
- return CompletableFuture.completedFuture(0L);
- }
- return sendAsync(RedisCommand.DEL, keys[0], keysArgs(keys)).thenApply(v -> v.getLongValue(0L));
+ public void setString(String key, String value) {
+ setStringAsync(key, value).join();
+ }
+
+ @Override
+ public CompletableFuture setLongAsync(String key, long value) {
+ return (CompletableFuture) send("SET", CacheEntryType.LONG, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.LONG, (Convert) null, (Type) null, value));
+ }
+
+ @Override
+ public void setLong(String key, long value) {
+ setLongAsync(key, value).join();
+ }
+
+ //--------------------- set ------------------------------
+ @Override
+ @Deprecated
+ public CompletableFuture setAsync(int expireSeconds, String key, V value) {
+ return (CompletableFuture) setAsync(key, value).thenCompose(v -> setExpireSecondsAsync(key, expireSeconds));
+ }
+
+ @Override
+ public CompletableFuture setAsync(int expireSeconds, String key, Convert convert, T value) {
+ return (CompletableFuture) setAsync(key, convert, value).thenCompose(v -> setExpireSecondsAsync(key, expireSeconds));
+ }
+
+ @Override
+ public CompletableFuture setAsync(int expireSeconds, String key, final Type type, T value) {
+ return (CompletableFuture) setAsync(key, type, value).thenCompose(v -> setExpireSecondsAsync(key, expireSeconds));
+ }
+
+ @Override
+ public CompletableFuture setAsync(int expireSeconds, String key, Convert convert, final Type type, T value) {
+ return (CompletableFuture) setAsync(key, convert, type, value).thenCompose(v -> setExpireSecondsAsync(key, expireSeconds));
+ }
+
+ @Override
+ @Deprecated
+ public void set(int expireSeconds, String key, V value) {
+ setAsync(expireSeconds, key, value).join();
+ }
+
+ @Override
+ public void set(int expireSeconds, String key, Convert convert, T value) {
+ setAsync(expireSeconds, key, convert, value).join();
+ }
+
+ @Override
+ public void set(int expireSeconds, String key, final Type type, T value) {
+ setAsync(expireSeconds, key, type, value).join();
+ }
+
+ @Override
+ public void set(int expireSeconds, String key, Convert convert, final Type type, T value) {
+ setAsync(expireSeconds, key, convert, type, value).join();
+ }
+
+ @Override
+ public CompletableFuture setStringAsync(int expireSeconds, String key, String value) {
+ return (CompletableFuture) setStringAsync(key, value).thenCompose(v -> setExpireSecondsAsync(key, expireSeconds));
+ }
+
+ @Override
+ public void setString(int expireSeconds, String key, String value) {
+ setStringAsync(expireSeconds, key, value).join();
+ }
+
+ @Override
+ public CompletableFuture setLongAsync(int expireSeconds, String key, long value) {
+ return (CompletableFuture) setLongAsync(key, value).thenCompose(v -> setExpireSecondsAsync(key, expireSeconds));
+ }
+
+ @Override
+ public void setLong(int expireSeconds, String key, long value) {
+ setLongAsync(expireSeconds, key, value).join();
+ }
+
+ //--------------------- setExpireSeconds ------------------------------
+ @Override
+ public CompletableFuture setExpireSecondsAsync(String key, int expireSeconds) {
+ return (CompletableFuture) send("EXPIRE", null, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(expireSeconds).getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Override
+ public void setExpireSeconds(String key, int expireSeconds) {
+ setExpireSecondsAsync(key, expireSeconds).join();
+ }
+
+ //--------------------- remove ------------------------------
+ @Override
+ public CompletableFuture removeAsync(String key) {
+ return (CompletableFuture) send("DEL", null, (Type) null, key, key.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Override
+ public int remove(String key) {
+ return removeAsync(key).join();
+ }
+
+ //--------------------- incr ------------------------------
+ @Override
+ public long incr(final String key) {
+ return incrAsync(key).join();
}
- //--------------------- incrby ------------------------------
@Override
public CompletableFuture incrAsync(final String key) {
- return sendAsync(RedisCommand.INCR, key, keyArgs(key)).thenApply(v -> v.getLongValue(0L));
+ return (CompletableFuture) send("INCR", CacheEntryType.ATOMIC, (Type) null, key, key.getBytes(StandardCharsets.UTF_8));
}
@Override
- public CompletableFuture incrbyAsync(final String key, long num) {
- return sendAsync(RedisCommand.INCRBY, key, keyArgs(key, num)).thenApply(v -> v.getLongValue(0L));
+ public long incr(final String key, long num) {
+ return incrAsync(key, num).join();
}
@Override
- public CompletableFuture incrbyFloatAsync(final String key, double num) {
- return sendAsync(RedisCommand.INCRBYFLOAT, key, keyArgs(key, num)).thenApply(v -> v.getDoubleValue(0.d));
+ public CompletableFuture incrAsync(final String key, long num) {
+ return (CompletableFuture) send("INCRBY", CacheEntryType.ATOMIC, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(num).getBytes(StandardCharsets.UTF_8));
+ }
+
+ //--------------------- decr ------------------------------
+ @Override
+ public long decr(final String key) {
+ return decrAsync(key).join();
}
- //--------------------- decrby ------------------------------
@Override
public CompletableFuture decrAsync(final String key) {
- return sendAsync(RedisCommand.DECR, key, keyArgs(key)).thenApply(v -> v.getLongValue(0L));
+ return (CompletableFuture) send("DECR", CacheEntryType.ATOMIC, (Type) null, key, key.getBytes(StandardCharsets.UTF_8));
}
@Override
- public CompletableFuture decrbyAsync(final String key, long num) {
- return sendAsync(RedisCommand.DECRBY, key, keyArgs(key, num)).thenApply(v -> v.getLongValue(0L));
+ public long decr(final String key, long num) {
+ return decrAsync(key, num).join();
}
@Override
- public CompletableFuture hdelAsync(final String key, String... fields) {
- return sendAsync(RedisCommand.HDEL, key, keysArgs(key, fields)).thenApply(v -> v.getLongValue(0L));
+ public CompletableFuture decrAsync(final String key, long num) {
+ return (CompletableFuture) send("DECRBY", CacheEntryType.ATOMIC, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), String.valueOf(num).getBytes(StandardCharsets.UTF_8));
}
@Override
- public CompletableFuture hlenAsync(final String key) {
- return sendAsync(RedisCommand.HLEN, key, keyArgs(key)).thenApply(v -> v.getLongValue(0L));
+ public int hremove(final String key, String... fields) {
+ return hremoveAsync(key, fields).join();
+ }
+
+ @Override
+ public int hsize(final String key) {
+ return hsizeAsync(key).join();
+ }
+
+ @Override
+ public List hkeys(final String key) {
+ return hkeysAsync(key).join();
+ }
+
+ @Override
+ public long hincr(final String key, String field) {
+ return hincrAsync(key, field).join();
+ }
+
+ @Override
+ public long hincr(final String key, String field, long num) {
+ return hincrAsync(key, field, num).join();
+ }
+
+ @Override
+ public long hdecr(final String key, String field) {
+ return hdecrAsync(key, field).join();
+ }
+
+ @Override
+ public long hdecr(final String key, String field, long num) {
+ return hdecrAsync(key, field, num).join();
+ }
+
+ @Override
+ public boolean hexists(final String key, String field) {
+ return hexistsAsync(key, field).join();
+ }
+
+ @Override
+ public void hset(final String key, final String field, final Convert convert, final T value) {
+ hsetAsync(key, field, convert, value).join();
+ }
+
+ @Override
+ public void hset(final String key, final String field, final Type type, final T value) {
+ hsetAsync(key, field, type, value).join();
+ }
+
+ @Override
+ public void hset(final String key, final String field, final Convert convert, final Type type, final T value) {
+ hsetAsync(key, field, convert, type, value).join();
+ }
+
+ @Override
+ public void hsetString(final String key, final String field, final String value) {
+ hsetStringAsync(key, field, value).join();
+ }
+
+ @Override
+ public void hsetLong(final String key, final String field, final long value) {
+ hsetLongAsync(key, field, value).join();
+ }
+
+ @Override
+ public void hmset(final String key, final Serializable... values) {
+ hmsetAsync(key, values).join();
+ }
+
+ @Override
+ public List hmget(final String key, final Type type, final String... fields) {
+ return hmgetAsync(key, type, fields).join();
+ }
+
+ @Override
+ public Map hmap(final String key, final Type type, int offset, int limit, String pattern) {
+ return (Map) hmapAsync(key, type, offset, limit, pattern).join();
+ }
+
+ @Override
+ public Map hmap(final String key, final Type type, int offset, int limit) {
+ return (Map) hmapAsync(key, type, offset, limit).join();
+ }
+
+ @Override
+ public T hget(final String key, final String field, final Type type) {
+ return (T) hgetAsync(key, field, type).join();
+ }
+
+ @Override
+ public String hgetString(final String key, final String field) {
+ return hgetStringAsync(key, field).join();
+ }
+
+ @Override
+ public long hgetLong(final String key, final String field, long defValue) {
+ return hgetLongAsync(key, field, defValue).join();
+ }
+
+ @Override
+ public CompletableFuture hremoveAsync(final String key, String... fields) {
+ byte[][] bs = new byte[fields.length + 1][];
+ bs[0] = key.getBytes(StandardCharsets.UTF_8);
+ for (int i = 0; i < fields.length; i++) {
+ bs[i + 1] = fields[i].getBytes(StandardCharsets.UTF_8);
+ }
+ return (CompletableFuture) send("HDEL", CacheEntryType.MAP, (Type) null, key, bs);
+ }
+
+ @Override
+ public CompletableFuture hsizeAsync(final String key) {
+ return (CompletableFuture) send("HLEN", CacheEntryType.LONG, (Type) null, key, key.getBytes(StandardCharsets.UTF_8));
}
@Override
public CompletableFuture> hkeysAsync(final String key) {
- return sendAsync(RedisCommand.HKEYS, key, keyArgs(key)).thenApply(v -> (List) v.getListValue(key, cryptor, String.class));
+ return (CompletableFuture) send("HKEYS", CacheEntryType.MAP, (Type) null, key, key.getBytes(StandardCharsets.UTF_8));
}
@Override
public CompletableFuture hincrAsync(final String key, String field) {
- return hincrbyAsync(key, field, 1);
+ return hincrAsync(key, field, 1);
}
@Override
- public CompletableFuture hincrbyAsync(final String key, String field, long num) {
- return sendAsync(RedisCommand.HINCRBY, key, keysArgs(key, field, String.valueOf(num))).thenApply(v -> v.getLongValue(0L));
- }
-
- @Override
- public CompletableFuture hincrbyFloatAsync(final String key, String field, double num) {
- return sendAsync(RedisCommand.HINCRBYFLOAT, key, keysArgs(key, field, String.valueOf(num))).thenApply(v -> v.getDoubleValue(0.d));
+ public CompletableFuture hincrAsync(final String key, String field, long num) {
+ return (CompletableFuture) send("HINCRBY", CacheEntryType.MAP, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), String.valueOf(num).getBytes(StandardCharsets.UTF_8));
}
@Override
public CompletableFuture hdecrAsync(final String key, String field) {
- return hincrbyAsync(key, field, -1);
+ return hincrAsync(key, field, -1);
}
@Override
- public CompletableFuture hdecrbyAsync(final String key, String field, long num) {
- return hincrbyAsync(key, field, -num);
+ public CompletableFuture hdecrAsync(final String key, String field, long num) {
+ return hincrAsync(key, field, -num);
}
@Override
public CompletableFuture hexistsAsync(final String key, String field) {
- return sendAsync(RedisCommand.HEXISTS, key, keysArgs(key, field)).thenApply(v -> v.getIntValue(0) > 0);
+ return (CompletableFuture) send("HEXISTS", CacheEntryType.MAP, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Override
+ public CompletableFuture hsetAsync(final String key, final String field, final Convert convert, final T value) {
+ return (CompletableFuture) send("HSET", CacheEntryType.MAP, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.MAP, convert, null, value));
+ }
+
+ @Override
+ public CompletableFuture hsetAsync(final String key, final String field, final Type type, final T value) {
+ return (CompletableFuture) send("HSET", CacheEntryType.MAP, type, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.MAP, null, type, value));
}
@Override
public CompletableFuture hsetAsync(final String key, final String field, final Convert convert, final Type type, final T value) {
- if (value == null) {
- return CompletableFuture.completedFuture(null);
- }
- return sendAsync(RedisCommand.HSET, key, keyArgs(key, field, convert, type, value)).thenApply(v -> v.getVoidValue());
+ return (CompletableFuture) send("HSET", CacheEntryType.MAP, type, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.MAP, convert, type, value));
}
@Override
- public CompletableFuture hsetnxAsync(final String key, final String field, final Convert convert, final Type type, final T value) {
- if (value == null) {
- return CompletableFuture.completedFuture(null);
- }
- return sendAsync(RedisCommand.HSETNX, key, keyArgs(key, field, convert, type, value)).thenApply(v -> v.getBoolValue());
+ public CompletableFuture hsetStringAsync(final String key, final String field, final String value) {
+ return (CompletableFuture) send("HSET", CacheEntryType.MAP, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.STRING, null, null, value));
+ }
+
+ @Override
+ public CompletableFuture hsetLongAsync(final String key, final String field, final long value) {
+ return (CompletableFuture) send("HSET", CacheEntryType.MAP, (Type) null, key, key.getBytes(StandardCharsets.UTF_8), field.getBytes(StandardCharsets.UTF_8), formatValue(CacheEntryType.LONG, null, null, value));
}
@Override
public CompletableFuture hmsetAsync(final String key, final Serializable... values) {
- return sendAsync(RedisCommand.HMSET, key, keyMapArgs(key, values)).thenApply(v -> v.getVoidValue());
- }
-
- @Override
- public CompletableFuture hmsetAsync(final String key, final Map map) {
- if (isEmpty(map)) {
- return CompletableFuture.completedFuture(null);
+ byte[][] bs = new byte[values.length + 1][];
+ bs[0] = key.getBytes(StandardCharsets.UTF_8);
+ for (int i = 0; i < values.length; i += 2) {
+ bs[i + 1] = String.valueOf(values[i]).getBytes(StandardCharsets.UTF_8);
+ bs[i + 2] = formatValue(CacheEntryType.MAP, null, null, values[i + 1]);
}
- return sendAsync(RedisCommand.HMSET, key, keyMapArgs(key, map)).thenApply(v -> v.getVoidValue());
+ return (CompletableFuture) send("HMSET", CacheEntryType.MAP, (Type) null, key, bs);
}
@Override
public CompletableFuture> hmgetAsync(final String key, final Type type, final String... fields) {
- return sendAsync(RedisCommand.HMGET, key, keysArgs(key, fields)).thenApply(v -> (List) v.getListValue(key, cryptor, type));
+ byte[][] bs = new byte[fields.length + 1][];
+ bs[0] = key.getBytes(StandardCharsets.UTF_8);
+ for (int i = 0; i < fields.length; i++) {
+ bs[i + 1] = fields[i].getBytes(StandardCharsets.UTF_8);
+ }
+ return (CompletableFuture) send("HMGET", CacheEntryType.MAP, type, key, bs);
}
@Override
- public CompletableFuture