feat: 增加104客户端服务和断线重连机制

src/main/java/com/xydl/cac/iec104/Iec104Client.java

@@ -1,5 +1,7 @@
 package com.xydl.cac.iec104;
 
+import com.xydl.cac.model.StaticVariable;
+import com.xydl.cac.socket.WebSocketServer;
 import lombok.extern.slf4j.Slf4j;
 import org.openmuc.j60870.*;
 import org.openmuc.j60870.ie.*;
@@ -12,12 +14,23 @@ public class Iec104Client implements ConnectionEventListener {
     public String ip;
     public Integer port;
     Connection connection = null;
+    ClientConnectionBuilder clientConnectionBuilder = null;
+    private WebSocketServer webSocketServer;
+    public boolean keep = false;
+    public boolean connected = false;
+    public int retry = 0;
+    public int seconds = 0;
+    private boolean inRetry = false;
+
+    public Iec104Client(WebSocketServer _webSocketServer) {
+        webSocketServer = _webSocketServer;
+    }
 
     public void connect(String _ip, Integer _port) throws Exception {
         ip = _ip;
         port = _port;
         InetAddress address = InetAddress.getByName(ip);
-        ClientConnectionBuilder clientConnectionBuilder = new ClientConnectionBuilder(address)
+        clientConnectionBuilder = new ClientConnectionBuilder(address)
                 .setMessageFragmentTimeout(5_000)
                 .setConnectionTimeout(20_000)
                 .setPort(port)
@@ -25,10 +38,12 @@ public class Iec104Client implements ConnectionEventListener {
         connection = clientConnectionBuilder.build();
         connection.startDataTransfer();
         log.info("IEC104客户端已连接上" + ip + ":" + port);
+        connected = true;
     }
 
     public void disconnect() {
         try {
+            connected = false;
             connection.stopDataTransfer();
             connection.close();
             connection = null;
@@ -37,6 +52,19 @@ public class Iec104Client implements ConnectionEventListener {
         }
     }
 
+    private void reconnect() throws Exception {
+        if (StaticVariable.shutdown == 1) {
+            return;
+        }
+        retry++;
+        seconds = 0;
+        connection = clientConnectionBuilder.build();
+        connection.startDataTransfer();
+        log.info("IEC104客户端断线重连成功, ip=" + ip + ", port=" + port);
+        connected = true;
+        retry = 0;
+    }
+
     public void interrogation() throws Exception {
         connection.interrogation(1, CauseOfTransmission.ACTIVATION,
                 new IeQualifierOfInterrogation(20));
@@ -44,21 +72,43 @@ public class Iec104Client implements ConnectionEventListener {
 
     @Override
     public void newASdu(Connection connection, ASdu aSdu) {
-        log.debug("Got new ASdu:" + aSdu.toString());
+        if (aSdu == null)
+            return;
         InformationObject[] objs = aSdu.getInformationObjects();
         if (objs != null && objs.length > 0) {
-            log.debug("InformationObject size=" + objs.length);
             for (InformationObject obj : objs) {
-                log.debug("ObjectAddress=" + obj.getInformationObjectAddress());
+                int addr = obj.getInformationObjectAddress();
                 InformationElement[][] elements = obj.getInformationElements();
-                log.debug("elements=" + elements);
             }
         }
     }
 
     @Override
     public void connectionClosed(Connection connection, IOException e) {
-
+        if (keep && !inRetry) {
+            inRetry = true;
+            retry = 0;
+            seconds = 0;
+            this.disconnect();
+            while (!connected && StaticVariable.shutdown == 0) {
+                try {
+                    Thread.sleep(1000);
+                    seconds++;
+                    if (retry < 10 && seconds > 60) {
+                        this.reconnect();
+                    } else if (seconds > 60 * 60) {
+                        this.reconnect();
+                    }
+                } catch (Exception ignore) {
+                    String err = "IEC104客户端断线重连失败" + retry + "次, ip=" + ip + ", port=" + port;
+                    log.warn(err);
+                    if (webSocketServer != null && retry >= 10) {
+                        webSocketServer.sendMessage(err, null);
+                    }
+                }
+            }
+            inRetry = false;
+        }
     }
 
     @Override
@@ -67,9 +117,9 @@ public class Iec104Client implements ConnectionEventListener {
     }
 
     public static void main(String[] args) {
-        Iec104Client client = new Iec104Client();
+        Iec104Client client = new Iec104Client(null);
         try {
-            client.connect("127.0.0.1", 2404);
+            client.connect("192.168.1.177", 2404);
             client.interrogation();
             while (true) {
                 Thread.sleep(1000);

src/main/java/com/xydl/cac/iec104/RealTime104DataService.java

@@ -0,0 +1,60 @@
+package com.xydl.cac.iec104;
+
+
+import com.xydl.cac.exception.BusinessException;
+import com.xydl.cac.model.StaticVariable;
+import com.xydl.cac.socket.WebSocketServer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import javax.annotation.Resource;
+
+@Service
+@Slf4j
+public class RealTime104DataService {
+    @Resource
+    WebSocketServer webSocketServer;
+
+    public static boolean inDoing = false;
+
+    @PostConstruct
+    private void init() {
+        try {
+//            this.startCollect("192.168.1.177");
+        } catch (Exception ignore) {
+        }
+    }
+
+    public void startCollect(String ip) throws BusinessException {
+        inDoing = true;
+        try {
+            Iec104Client client = new Iec104Client(webSocketServer);
+            client.connect(ip, 2404);
+            client.keep = true;
+        } catch (Exception ex) {
+            String err = "IEC104客户端连接异常, ip=" + ip;
+            log.error(err, ex);
+            throw new BusinessException(err);
+        } finally {
+            inDoing = false;
+        }
+    }
+
+    public void stopCollect() {
+        if (StaticVariable.realTime104Client != null) {
+            StaticVariable.realTime104Client.keep = false;
+            StaticVariable.realTime104Client.disconnect();
+            StaticVariable.realTime104Client = null;
+        }
+    }
+
+    @PreDestroy
+    private void stop() {
+        log.info("关闭IEC104客户端.");
+        StaticVariable.shutdown = 1;
+        this.stopCollect();
+    }
+
+}

src/main/java/com/xydl/cac/model/StaticVariable.java

@@ -4,6 +4,8 @@ import com.beanit.iec61850bean.BasicDataAttribute;
 import com.xydl.cac.entity.*;
 import com.xydl.cac.iec.IecClient;
 import com.xydl.cac.iec.IecServer;
+import com.xydl.cac.iec104.Iec104Client;
+import com.xydl.cac.iec104.Iec104Server;
 import com.xydl.cac.util.DateUtil;
 
 import java.util.*;
@@ -11,6 +13,9 @@ import java.util.concurrent.ConcurrentHashMap;
 
 public class StaticVariable {
 
+    public static Iec104Server iec104Server = null;
+    public static Iec104Client realTime104Client = null;
+
     public static HashMap<Integer, IecServer> iecServerMap = new HashMap<>();
     public static HashMap<Integer, HashMap<String, String>> sensorLastDataMap = new HashMap<>();
     public static HashMap<String, String> paramRelationMap = new HashMap<>();