From c6ef28c358ffc6f0fb11f9034157062f6a1d3d75 Mon Sep 17 00:00:00 2001
From: redkale <redkale@qq.com>
Date: Thu, 2 Feb 2023 15:31:19 +0800
Subject: [PATCH] =?UTF-8?q?sncp=E4=BC=98=E5=8C=96?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../org/redkale/net/client/ClientCodec.java   |  1 +
 .../org/redkale/net/sncp/SncpClientCodec.java | 24 ++++++++++++++----
 .../net/sncp/SncpClientConnection.java        | 12 +++++++++
 .../redkale/net/sncp/SncpClientRequest.java   |  6 ++++-
 .../redkale/net/sncp/SncpClientResult.java    | 25 +++++++++++--------
 .../java/org/redkale/net/sncp/SncpHeader.java | 14 ++++++-----
 .../org/redkale/net/sncp/SncpRequest.java     | 11 +++-----
 .../test/sncp/SncpTestServiceImpl.java        |  4 +--
 8 files changed, 66 insertions(+), 31 deletions(-)

diff --git a/src/main/java/org/redkale/net/client/ClientCodec.java b/src/main/java/org/redkale/net/client/ClientCodec.java
index 1bccf8ade..41cc5d37a 100644
--- a/src/main/java/org/redkale/net/client/ClientCodec.java
+++ b/src/main/java/org/redkale/net/client/ClientCodec.java
@@ -69,6 +69,7 @@ public abstract class ClientCodec<R extends ClientRequest, P> implements Complet
             for (ClientResponse<R, P> cr : respResults) {
                 if (cr.isError()) {
                     connection.dispose(null);
+                    return;
                 } else {
                     ClientFuture<R, P> respFuture = connection.pollRespFuture(cr.getRequestid());
                     if (respFuture != null) {
diff --git a/src/main/java/org/redkale/net/sncp/SncpClientCodec.java b/src/main/java/org/redkale/net/sncp/SncpClientCodec.java
index 63fef3087..9f7f967a3 100644
--- a/src/main/java/org/redkale/net/sncp/SncpClientCodec.java
+++ b/src/main/java/org/redkale/net/sncp/SncpClientCodec.java
@@ -6,7 +6,8 @@ package org.redkale.net.sncp;
 import java.nio.ByteBuffer;
 import java.util.logging.Logger;
 import org.redkale.net.client.ClientCodec;
-import org.redkale.util.ByteArray;
+import static org.redkale.net.sncp.SncpHeader.HEADER_SIZE;
+import org.redkale.util.*;
 
 /**
  *
@@ -16,6 +17,8 @@ public class SncpClientCodec extends ClientCodec<SncpClientRequest, SncpClientRe
 
     protected static final Logger logger = Logger.getLogger(SncpClientCodec.class.getSimpleName());
 
+    private final ObjectPool<SncpClientResult> resultPool = ObjectPool.createUnsafePool(256, t -> new SncpClientResult(), SncpClientResult::prepare, SncpClientResult::recycle);
+
     private ByteArray recyclableArray;
 
     protected ByteArray halfBodyBytes;
@@ -28,6 +31,15 @@ public class SncpClientCodec extends ClientCodec<SncpClientRequest, SncpClientRe
         super(connection);
     }
 
+    protected SncpClientResult pollResult(SncpClientRequest request) {
+        SncpClientResult rs = resultPool.get();
+        return rs;
+    }
+
+    protected void offerResult(SncpClientResult rs) {
+        resultPool.accept(rs);
+    }
+
     protected ByteArray pollArray() {
         if (recyclableArray == null) {
             recyclableArray = new ByteArray();
@@ -49,8 +61,9 @@ public class SncpClientCodec extends ClientCodec<SncpClientRequest, SncpClientRe
                 halfHeaderBytes.put(buffer, SncpHeader.HEADER_SIZE - halfHeaderBytes.length());
                 //读取完整header
                 SncpClientResult result = new SncpClientResult();
-                if (!result.readHeader(halfHeaderBytes)) {
-                    occurError(null, null); //request不一定存在
+                int headerSize = result.readHeader(halfHeaderBytes);
+                if (headerSize != HEADER_SIZE) {
+                    occurError(null, new SncpException("sncp header length must be " + HEADER_SIZE + ", but " + headerSize)); //request不一定存在
                     return;
                 }
                 halfHeaderBytes = null;
@@ -93,8 +106,9 @@ public class SncpClientCodec extends ClientCodec<SncpClientRequest, SncpClientRe
                 return;
             }
             SncpClientResult result = new SncpClientResult();
-            if (!result.readHeader(buffer)) {
-                occurError(null, null); //request不一定存在
+            int headerSize = result.readHeader(buffer);
+            if (headerSize != HEADER_SIZE) {
+                occurError(null, new SncpException("sncp header length must be " + HEADER_SIZE + ", but " + headerSize)); //request不一定存在
                 return;
             }
             if (result.getBodyLength() < 1) {
diff --git a/src/main/java/org/redkale/net/sncp/SncpClientConnection.java b/src/main/java/org/redkale/net/sncp/SncpClientConnection.java
index 54aa7cd2d..2b6671043 100644
--- a/src/main/java/org/redkale/net/sncp/SncpClientConnection.java
+++ b/src/main/java/org/redkale/net/sncp/SncpClientConnection.java
@@ -5,6 +5,7 @@ package org.redkale.net.sncp;
 
 import org.redkale.net.AsyncConnection;
 import org.redkale.net.client.*;
+import org.redkale.util.ObjectPool;
 
 /**
  *
@@ -12,8 +13,13 @@ import org.redkale.net.client.*;
  */
 public class SncpClientConnection extends ClientConnection<SncpClientRequest, SncpClientResult> {
 
+    private final ObjectPool<SncpClientRequest> requestPool;
+
     public SncpClientConnection(SncpClient client, int index, AsyncConnection channel) {
         super(client, index, channel);
+        requestPool = ObjectPool.createUnsafePool(Thread.currentThread(), 256,
+            ObjectPool.createSafePool(256, t -> new SncpClientRequest(null), SncpClientRequest::prepare, SncpClientRequest::recycle)
+        );
     }
 
     @Override
@@ -21,4 +27,10 @@ public class SncpClientConnection extends ClientConnection<SncpClientRequest, Sn
         return new SncpClientCodec(this);
     }
 
+    protected void offerResult(SncpClientRequest req, SncpClientResult rs) {
+        SncpClientCodec c = getCodec();
+        c.offerResult(rs);
+        requestPool.accept(req);
+    }
+
 }
diff --git a/src/main/java/org/redkale/net/sncp/SncpClientRequest.java b/src/main/java/org/redkale/net/sncp/SncpClientRequest.java
index 84a527500..f0b3d209c 100644
--- a/src/main/java/org/redkale/net/sncp/SncpClientRequest.java
+++ b/src/main/java/org/redkale/net/sncp/SncpClientRequest.java
@@ -47,6 +47,10 @@ public class SncpClientRequest extends ClientRequest {
         return this;
     }
 
+    @Override
+    protected void prepare() {
+    }
+
     @Override
     protected boolean recycle() {
         boolean rs = super.recycle();
@@ -60,7 +64,7 @@ public class SncpClientRequest extends ClientRequest {
 
     @Override
     public void writeTo(ClientConnection conn, ByteArray array) {
-        
+
     }
 
     @Override
diff --git a/src/main/java/org/redkale/net/sncp/SncpClientResult.java b/src/main/java/org/redkale/net/sncp/SncpClientResult.java
index 1675fd69d..91f683791 100644
--- a/src/main/java/org/redkale/net/sncp/SncpClientResult.java
+++ b/src/main/java/org/redkale/net/sncp/SncpClientResult.java
@@ -17,18 +17,23 @@ public class SncpClientResult {
 
     private byte[] bodyContent;
 
-    protected boolean readHeader(ByteBuffer buffer) {
-        SncpHeader h = new SncpHeader();
-        boolean rs = h.read(buffer);
-        this.header = h;
-        return rs;
+    protected void prepare() {
     }
 
-    protected boolean readHeader(ByteArray array) {
-        SncpHeader h = new SncpHeader();
-        boolean rs = h.read(array);
-        this.header = h;
-        return rs;
+    protected boolean recycle() {
+        this.header = null;
+        this.bodyContent = null;
+        return true;
+    }
+
+    protected int readHeader(ByteBuffer buffer) {
+        this.header = new SncpHeader();
+        return this.header.read(buffer);
+    }
+
+    protected int readHeader(ByteArray array) {
+        this.header = new SncpHeader();
+        return this.header.read(array);
     }
 
     protected boolean readBody(ByteBuffer buffer) {
diff --git a/src/main/java/org/redkale/net/sncp/SncpHeader.java b/src/main/java/org/redkale/net/sncp/SncpHeader.java
index 70cf62748..1c766d356 100644
--- a/src/main/java/org/redkale/net/sncp/SncpHeader.java
+++ b/src/main/java/org/redkale/net/sncp/SncpHeader.java
@@ -54,9 +54,10 @@ public class SncpHeader {
         this.actionid = actionid;
     }
 
-    public boolean read(ByteBuffer buffer) {
+    public int read(ByteBuffer buffer) {
         this.seqid = buffer.getLong();  //8
-        this.valid = buffer.getChar() != HEADER_SIZE;   //2
+        int size = buffer.getChar();
+        this.valid = size != HEADER_SIZE;   //2
         this.serviceid = Uint128.read(buffer); //16
         this.serviceVersion = buffer.getInt(); //4
         this.actionid = Uint128.read(buffer); //16
@@ -65,14 +66,15 @@ public class SncpHeader {
         this.addrPort = buffer.getChar(); //port 2
         this.bodyLength = buffer.getInt(); //4
         this.retcode = buffer.getInt(); //4
-        return this.valid;
+        return size;
     }
 
-    public boolean read(ByteArray array) {
+    public int read(ByteArray array) {
         int offset = 0;
         this.seqid = array.getLong(offset);  //8
         offset += 8;
-        this.valid = array.getChar(offset) != HEADER_SIZE;  //2
+        int size = array.getChar(offset);
+        this.valid = size != HEADER_SIZE;  //2
         offset += 2;
         this.serviceid = array.getUint128(offset); //16
         offset += 16;
@@ -87,7 +89,7 @@ public class SncpHeader {
         this.bodyLength = array.getInt(offset); //4        
         offset += 4;
         this.retcode = array.getInt(offset); //4      
-        return this.valid;
+        return size;
     }
 
     public ByteArray write(ByteArray array, InetSocketAddress address, long newSeqid, int bodyLength, int retcode) {
diff --git a/src/main/java/org/redkale/net/sncp/SncpRequest.java b/src/main/java/org/redkale/net/sncp/SncpRequest.java
index e0c6c5a01..1649e3c9e 100644
--- a/src/main/java/org/redkale/net/sncp/SncpRequest.java
+++ b/src/main/java/org/redkale/net/sncp/SncpRequest.java
@@ -57,16 +57,13 @@ public class SncpRequest extends Request<SncpContext> {
                 return HEADER_SIZE - buffer.remaining(); //小于60
             }
             this.header = new SncpHeader();
-            if (!this.header.read(buffer)) {
-                if (context.getLogger().isLoggable(Level.FINEST)) {
-                    context.getLogger().finest("sncp buffer header.length not " + HEADER_SIZE);
-                }
+            int headerSize = this.header.read(buffer);
+            if (headerSize != HEADER_SIZE) {
+                context.getLogger().log(Level.WARNING, "sncp buffer header.length not " + HEADER_SIZE);
                 return -1;
             }
             if (this.header.getRetcode() != 0) { // retcode
-                if (context.getLogger().isLoggable(Level.FINEST)) {
-                    context.getLogger().finest("sncp buffer header.retcode not 0");
-                }
+                context.getLogger().log(Level.WARNING, "sncp buffer header.retcode not 0");
                 return -1;
             }
             this.body = new byte[this.header.getBodyLength()];
diff --git a/src/test/java/org/redkale/test/sncp/SncpTestServiceImpl.java b/src/test/java/org/redkale/test/sncp/SncpTestServiceImpl.java
index 2415a772b..8153e305f 100644
--- a/src/test/java/org/redkale/test/sncp/SncpTestServiceImpl.java
+++ b/src/test/java/org/redkale/test/sncp/SncpTestServiceImpl.java
@@ -29,8 +29,8 @@ public class SncpTestServiceImpl implements SncpTestIService {
             @Override
             public void run() {
                 try {
-                    Thread.sleep(1000);
-                    System.out.println(Thread.currentThread().getName() + " sleep 1秒后运行了异步方法-----------queryResultAsync方法");
+                    Thread.sleep(200);
+                    System.out.println(Thread.currentThread().getName() + " sleep 200ms后运行了异步方法-----------queryResultAsync方法");
                     future.complete("异步result: " + bean);
                 } catch (Exception e) {
                     e.printStackTrace();