!6 优化云快充ByteBuf

* 优化云快充ByteBuf

jcpp-app/src/main/java/sanbing/jcpp/app/service/DownlinkCallService.java

@@ -15,6 +15,8 @@ import sanbing.jcpp.infrastructure.cache.CacheValueWrapper;
 import sanbing.jcpp.infrastructure.cache.TransactionalCache;
 import sanbing.jcpp.infrastructure.queue.discovery.ServiceInfoProvider;
 import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
+import sanbing.jcpp.proto.gen.ProtocolProto.LoginRequest;
+import sanbing.jcpp.proto.gen.ProtocolProto.UplinkQueueMessage;
 import sanbing.jcpp.protocol.adapter.DownlinkController;
 
 import java.util.UUID;
@@ -55,22 +57,50 @@ public abstract class DownlinkCallService {
         if (downlinkMessageBuilder.getSessionIdLSB() == 0) {
             downlinkMessageBuilder.setSessionIdLSB(protocolSessionId.getLeastSignificantBits());
         }
-        if(downlinkMessageBuilder.getProtocolName() == null){
+        if (downlinkMessageBuilder.getProtocolName() == null) {
             downlinkMessageBuilder.setProtocolName(pileSession.getProtocolName());
         }
 
+        String nodeId = pileSession.getNodeId();
+        String nodeIp = pileSession.getNodeIp();
+        int nodeRestPort = pileSession.getNodeRestPort();
+        int nodeGrpcPort = pileSession.getNodeGrpcPort();
+
+        sendDownlinkMessage(downlinkMessageBuilder, nodeId, nodeIp, nodeRestPort, nodeGrpcPort);
+    }
+
+    public void sendDownlinkMessage(DownlinkRequestMessage.Builder downlinkMessageBuilder, UplinkQueueMessage uplinkQueueMessage, LoginRequest loginRequest) {
+
+        if (downlinkMessageBuilder.getSessionIdMSB() == 0) {
+            downlinkMessageBuilder.setSessionIdMSB(uplinkQueueMessage.getSessionIdMSB());
+        }
+        if (downlinkMessageBuilder.getSessionIdLSB() == 0) {
+            downlinkMessageBuilder.setSessionIdLSB(uplinkQueueMessage.getSessionIdLSB());
+        }
+        if (downlinkMessageBuilder.getProtocolName() == null) {
+            downlinkMessageBuilder.setProtocolName(uplinkQueueMessage.getProtocolName());
+        }
+
+        String nodeId = loginRequest.getNodeId();
+        String nodeIp = loginRequest.getNodeHostAddress();
+        int nodeRestPort = loginRequest.getNodeRestPort();
+        int nodeGrpcPort = loginRequest.getNodeGrpcPort();
+
+        sendDownlinkMessage(downlinkMessageBuilder, nodeId, nodeIp, nodeRestPort, nodeGrpcPort);
+    }
+
+    private void sendDownlinkMessage(DownlinkRequestMessage.Builder downlinkMessageBuilder, String nodeId, String nodeIp, int nodeRestPort, int nodeGrpcPort) {
         if (serviceInfoProvider.isMonolith() &&
-                ("caffeine".equalsIgnoreCase(cacheType) || serviceInfoProvider.getServiceId().equalsIgnoreCase(pileSession.getNodeId()))) {
+                ("caffeine".equalsIgnoreCase(cacheType) || serviceInfoProvider.getServiceId().equalsIgnoreCase(nodeId))) {
 
             downlinkController.onDownlink(downlinkMessageBuilder.build())
                     .setResultHandler(result -> log.debug("下行消息发送完成"));
 
         } else {
 
-            _sendDownlinkMessage(downlinkMessageBuilder.build(), pileSession);
+            _sendDownlinkMessage(downlinkMessageBuilder.build(), nodeIp, nodeRestPort, nodeGrpcPort);
         }
     }
 
-
-    protected abstract void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, PileSession pileSession);
+    protected abstract void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, String nodeIp, int nodeRestPort, int nodeGrpcPort);
 }

jcpp-app/src/main/java/sanbing/jcpp/app/service/impl/DefaultPileProtocolService.java

@@ -64,27 +64,36 @@ public class DefaultPileProtocolService implements PileProtocolService {
         DownlinkRequestMessage.Builder downlinkMessageBuilder = createDownlinkMessageBuilder(uplinkQueueMessage, loginRequest.getPileCode());
         downlinkMessageBuilder.setDownlinkCmd(DownlinkCmdEnum.LOGIN_ACK.name());
 
+
         if (pile != null) {
-            // 保存到缓存
-            cacheSession(uplinkQueueMessage, pile,
+
+            PileSession pileSession = createSession(uplinkQueueMessage, pile,
                     loginRequest.getRemoteAddress(),
                     loginRequest.getNodeId(),
                     loginRequest.getNodeHostAddress(),
                     loginRequest.getNodeRestPort(),
                     loginRequest.getNodeGrpcPort());
 
+            // 保存到缓存
+            pileSessionCache.put(new PileSessionCacheKey(pile.getPileCode()), pileSession);
+
             downlinkMessageBuilder.setLoginResponse(LoginResponse.newBuilder()
                     .setSuccess(true)
                     .setPileCode(loginRequest.getPileCode())
                     .build());
+
+            downlinkCallService.sendDownlinkMessage(downlinkMessageBuilder, pileCode);
         } else {
+
             downlinkMessageBuilder.setLoginResponse(LoginResponse.newBuilder()
                     .setSuccess(false)
                     .setPileCode(loginRequest.getPileCode())
                     .build());
+
+
+            downlinkCallService.sendDownlinkMessage(downlinkMessageBuilder, uplinkQueueMessage, loginRequest);
         }
 
-        downlinkCallService.sendDownlinkMessage(downlinkMessageBuilder, pileCode);
 
         callback.onSuccess();
     }
@@ -99,7 +108,7 @@ public class DefaultPileProtocolService implements PileProtocolService {
 
         if (pile != null) {
             // 重新保存到缓存
-            cacheSession(uplinkQueueMessage, pile,
+            createSession(uplinkQueueMessage, pile,
                     heartBeatRequest.getRemoteAddress(),
                     heartBeatRequest.getNodeId(),
                     heartBeatRequest.getNodeHostAddress(),
@@ -110,13 +119,13 @@ public class DefaultPileProtocolService implements PileProtocolService {
         callback.onSuccess();
     }
 
-    private void cacheSession(UplinkQueueMessage uplinkQueueMessage,
-                              Pile pile,
-                              String remoteAddress,
-                              String nodeId,
-                              String nodeIp,
-                              int restPort,
-                              int grpcPort) {
+    private PileSession createSession(UplinkQueueMessage uplinkQueueMessage,
+                                      Pile pile,
+                                      String remoteAddress,
+                                      String nodeId,
+                                      String nodeIp,
+                                      int restPort,
+                                      int grpcPort) {
         PileSession pileSession = new PileSession(pile.getId(), pile.getPileCode(), uplinkQueueMessage.getProtocolName());
         pileSession.setProtocolSessionId(new UUID(uplinkQueueMessage.getSessionIdMSB(), uplinkQueueMessage.getSessionIdLSB()));
         pileSession.setRemoteAddress(remoteAddress);
@@ -124,7 +133,8 @@ public class DefaultPileProtocolService implements PileProtocolService {
         pileSession.setNodeIp(nodeIp);
         pileSession.setNodeRestPort(restPort);
         pileSession.setNodeGrpcPort(grpcPort);
-        pileSessionCache.put(new PileSessionCacheKey(pile.getPileCode()), pileSession);
+
+        return pileSession;
     }
 
     @Override

jcpp-app/src/main/java/sanbing/jcpp/app/service/impl/GrpcDownlinkCallService.java

@@ -11,7 +11,6 @@ import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
-import sanbing.jcpp.app.data.PileSession;
 import sanbing.jcpp.app.service.DownlinkCallService;
 import sanbing.jcpp.app.service.grpc.DownlinkGrpcClient;
 import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
@@ -31,7 +30,7 @@ public class GrpcDownlinkCallService extends DownlinkCallService {
     DownlinkGrpcClient downlinkGrpcClient;
 
     @Override
-    protected void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, PileSession pileSession) {
+    protected void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, String nodeIp, int nodeRestPort, int nodeGrpcPort) {
         try {
 
             RequestMsg requestMsg = RequestMsg.newBuilder()
@@ -40,7 +39,7 @@ public class GrpcDownlinkCallService extends DownlinkCallService {
                     .setDownlinkRequestMessage(downlinkMessage)
                     .build();
 
-            downlinkGrpcClient.sendDownlinkRequest(HostAndPort.fromParts(pileSession.getNodeIp(), pileSession.getNodeGrpcPort()),
+            downlinkGrpcClient.sendDownlinkRequest(HostAndPort.fromParts(nodeIp, nodeGrpcPort),
                     requestMsg);
 
         } catch (Exception e) {

jcpp-app/src/main/java/sanbing/jcpp/app/service/impl/RestDownlinkCallService.java

@@ -16,7 +16,6 @@ import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestClientException;
 import org.springframework.web.client.RestTemplate;
-import sanbing.jcpp.app.data.PileSession;
 import sanbing.jcpp.app.service.DownlinkCallService;
 import sanbing.jcpp.infrastructure.util.trace.TracerContextUtil;
 import sanbing.jcpp.proto.gen.ProtocolProto.DownlinkRequestMessage;
@@ -35,10 +34,10 @@ public class RestDownlinkCallService extends DownlinkCallService {
     RestTemplate downlinkRestTemplate;
 
     @Override
-    protected void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, PileSession pileSession) {
+    protected void _sendDownlinkMessage(DownlinkRequestMessage downlinkMessage, String nodeIp, int nodeRestPort, int nodeGrpcPort) {
         try {
 
-            invokeDownlinkRestApi(downlinkMessage, pileSession.getNodeIp(), pileSession.getNodeRestPort());
+            invokeDownlinkRestApi(downlinkMessage, nodeIp, nodeGrpcPort);
 
         } catch (RestClientException e) {
             log.error("下行消息发送异常", e);