src/main/java/net/tccn/zhub/Rpc.java

@@ -2,8 +2,6 @@ package net.tccn.zhub;
 
 import org.redkale.convert.ConvertColumn;
 import org.redkale.service.RetResult;
-import org.redkale.util.TypeToken;
-import org.redkale.util.Utility;
 
 public class Rpc<T> {
     private String ruk;     // request unique key:
@@ -12,13 +10,11 @@ public class Rpc<T> {
 
     private RpcResult rpcResult;
 
-    private TypeToken typeToken;
-
     public Rpc() {
     }
 
-    protected Rpc(String appname, String topic, T value) {
-        this.ruk = appname + "::" + Utility.uuid();
+    protected Rpc(String appname, String ruk, String topic, T value) {
+        this.ruk = appname + "::" + ruk;
         this.topic = topic;
         this.value = value;
     }
@@ -56,16 +52,6 @@ public class Rpc<T> {
         this.rpcResult = rpcResult;
     }
 
-    @ConvertColumn(ignore = true)
-    public TypeToken getTypeToken() {
-        return typeToken;
-    }
-
-    @ConvertColumn(ignore = true)
-    public void setTypeToken(TypeToken typeToken) {
-        this.typeToken = typeToken;
-    }
-
     @ConvertColumn(ignore = true)
     public String getBackTopic() {
         return ruk.split("::")[0];

src/main/java/net/tccn/zhub/RpcResult.java

@@ -34,7 +34,7 @@ public class RpcResult<R> {
         return result;
     }
 
-    public void setResult(Object result) {
-        this.result = (R) result;
+    public void setResult(R result) {
+        this.result = result;
     }
 }

src/main/java/net/tccn/zhub/ZHubClient.java

@@ -9,6 +9,7 @@ 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 java.io.BufferedReader;
 import java.io.IOException;
@@ -99,11 +100,12 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         }
 
         CompletableFuture.runAsync(() -> {
-            if (!initSocket(0)) {
-                return;
-            }
             // 消息 事件接收
             new Thread(() -> {
+                if (!initSocket(0)) {
+                    return;
+                }
+
                 while (true) {
                     try {
                         String readLine = reader.readLine();
@@ -408,7 +410,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
                 send("auth", auth);
                 send("groupid " + groupid);
 
-                StringBuilder buf = new StringBuilder("subscribe lock trylock");
+                StringBuffer buf = new StringBuffer("subscribe lock trylock");
                 if (mainHub.containsValue(this)) {
                     buf.append(" " + APP_NAME);
                 }
@@ -422,9 +424,9 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
                     send("timer", name);
                 });
                 if (retry > 0) {
-                    logger.warning(String.format("ZHubClient[%s][%s] %s Succeed！", getGroupid(), i + 1, "reconnection"));
+                    logger.warning(String.format("ZHubClient[%s][%s] %s Succeed！", getGroupid(), i + 1, retry > 0 ? "reconnection" : "init"));
                 } else {
-                    logger.fine(String.format("ZHubClient[%s] %s Succeed！", getGroupid(), "init"));
+                    logger.fine(String.format("ZHubClient[%s] %s Succeed！", getGroupid(), retry > 0 ? "reconnection" : "init"));
                 }
                 return true;
             } catch (Exception e) {
@@ -592,6 +594,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
     // ================================================== rpc ==================================================
     // -- 调用端 --
     private static Map<String, Rpc> rpcMap = new ConcurrentHashMap<>();
+    private static Map<String, TypeToken> rpcRetType = new ConcurrentHashMap<>();
 
     @Comment("rpc call")
     public RpcResult<Void> rpc(String topic, Object v) {
@@ -605,11 +608,12 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
 
     @Comment("rpc call")
     public <T, R> RpcResult<R> rpc(String topic, T v, TypeToken<R> typeToken, long timeout) {
-        Rpc rpc = new Rpc<>(APP_NAME, topic, v);
-        rpc.setTypeToken(typeToken);
-
+        Rpc rpc = new Rpc<>(APP_NAME, Utility.uuid(), topic, v);
         String ruk = rpc.getRuk();
         rpcMap.put(ruk, rpc);
+        if (typeToken != null) {
+            rpcRetType.put(ruk, typeToken);
+        }
         try {
             if (eventMap.containsKey(topic)) { // 本地调用
                 rpcCallQueue.add(Event.of(topic, rpc));
@@ -667,21 +671,10 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
     // RpcResult: {ruk:xxx-xxxx, retcode:0}
     @Comment("rpc call back consumer")
     private <T> void rpcAccept(T value) {
-        // 接收到 本地调用返回的 RpcResult
+        // 接收到 本地调用返回的 RpcResult TODO, 本地模式下返回的数据对象类型需要和处理端一致，不然会出现类型转换异常 - 解决办法，当出现不一致的情况取数据做转换
         if (value instanceof RpcResult) {
             String ruk = ((RpcResult) value).getRuk();
             Rpc rpc = rpcMap.remove(ruk);
-            if (rpc == null) {
-                return;
-            }
-
-            // 本地模式下返回的数据对象类型需要和处理端一致，不然会出现类型转换异常 - 解决办法，当出现不一致的情况取数据做转换
-            TypeToken typeToken = rpc.getTypeToken();
-            if (typeToken.getType() != ((RpcResult<?>) value).getResult().getClass()) {
-                Object result = convert.convertFrom(typeToken.getType(), toStr(((RpcResult<?>) value).getResult()));
-                ((RpcResult<?>) value).setResult(result);
-            }
-
             rpc.setRpcResult((RpcResult) value);
             synchronized (rpc) {
                 rpc.notify();
@@ -697,7 +690,7 @@ public class ZHubClient extends AbstractConsumer implements IConsumer, IProducer
         if (rpc == null) {
             return;
         }
-        TypeToken typeToken = rpc.getTypeToken();
+        TypeToken typeToken = rpcRetType.get(ruk);
 
         Object result = resp.getResult();
         if (result != null && typeToken != null && !"java.lang.String".equals(typeToken.getType().getTypeName()) && !"java.lang.Void".equals(typeToken.getType().getTypeName())) {