feat(base): 新增了socket相关内容

kt-base/src/main/java/org/matrix/BaseBootApplication.java

@@ -3,6 +3,8 @@ package org.matrix;
 import lombok.extern.slf4j.Slf4j;
 import org.matrix.actuators.datasource.DataSourceDTO;
 import org.matrix.actuators.datasource.IDataSourceService;
+import org.matrix.socket.SocketServer;
+import org.matrix.util.SpringUtils;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -49,5 +51,7 @@ public class BaseBootApplication implements CommandLineRunner {
     public void run(String... args) throws Exception {
         log.info("[初始化] 初始化数据源" + driverClassName);
         dataSourceService.add(new DataSourceDTO("初始数据源",driverClassName,url,username,password));
+        //在spring容器启动后，取到已经初始化的SocketServer，启动Socket服务
+        new SocketServer(8009).start();
     }
 }

kt-base/src/main/java/org/matrix/actuators/checkpoint/CheckPointActuator.java

@@ -42,6 +42,7 @@ public class CheckPointActuator implements Actuator {
     CompleteExpressionUtil completeExpressionUtil ;
 
     public CheckPointActuator(@Value("${baseJsPath}") String baseJs, CompleteExpressionUtil completeExpressionUtil) {
+        System.out.println("-------------:"+Thread.currentThread().getId());
         this.completeExpressionUtil = completeExpressionUtil;
         ClassPathResource cpr = new ClassPathResource(baseJs);
         try {
@@ -53,6 +54,7 @@ public class CheckPointActuator implements Actuator {
 
 
     public CheckPointResult httpCheck(HttpResponseDetail httpResponseDetail, CheckPoint checkPoint,Long envId,Long projectId) {
+        System.out.println("-------------:"+Thread.currentThread().getId());
         CheckPointResult checkPointResult = new CheckPointResult();
         //根据checkPoint里的细节数据开始检测
         //异常检查点检测

kt-base/src/main/java/org/matrix/actuators/sql/SqlExpActuator.java

@@ -19,6 +19,7 @@ import org.matrix.database.service.IDynamicVariableService;
 import org.matrix.database.service.ITestCaseService;
 import org.matrix.enums.DynamicVarType;
 import org.matrix.exception.GlobalException;
+import org.matrix.socket.LogQueue;
 import org.springframework.http.ResponseEntity;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Component;

kt-base/src/main/java/org/matrix/actuators/usecase/CaseActuator.java

@@ -11,6 +11,8 @@ import org.matrix.actuators.checkpoint.CheckPoint;
 import org.matrix.actuators.checkpoint.CheckPointResult;
 import org.matrix.actuators.httpclient.HttpRequestDetail;
 import org.matrix.actuators.httpclient.HttpResponseDetail;
+import org.matrix.database.entity.TestData;
+import org.matrix.socket.LogQueue;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -37,7 +39,8 @@ public class CaseActuator implements Actuator {
     /**
      * 执行测试用例
      */
-    public TestCaseExecuteResult executeTestCase(TestCase testCase,Long envId,Long projectId) {
+    public TestCaseExecuteResult executeTestCase(TestCase testCase, TestData testData, Long envId, Long projectId) {
+        LogQueue.add(Thread.currentThread().getId(),String.format("[用例解析器] 当前线程ID为: %S", Thread.currentThread().getId()));
         //todo 李迪凡 执行前置动作
         //执行测试用例的本体内容
         HttpResponseDetail baseTestCaseResponseDetail = null;
@@ -49,27 +52,27 @@ public class CaseActuator implements Actuator {
         //进行检验
          CheckPointResult checkPointResult = null;
         if (testCase.getType().equals(TestCaseTypeEnum.HTTP.getValue())) {
-           checkPointResult = checkPointActuator.httpCheck(baseTestCaseResponseDetail, getCheckPointEntity(testCase),envId,projectId);
+           checkPointResult = checkPointActuator.httpCheck(baseTestCaseResponseDetail, getCheckPointEntity(testData),envId,projectId);
         }
         //todo 李迪凡 执行后置动作
         return new TestCaseExecuteResult(baseTestCaseResponseDetail, checkPointResult);
     }
 
-    private CheckPoint getCheckPointEntity(TestCase testCase) {
+    private CheckPoint getCheckPointEntity(TestData testData) {
         CheckPoint checkPoint = new CheckPoint();
-        if (testCase.getAbnormalCheckpoint()==0){
+        if (testData.getAbnormalCheckpoint()==0){
             checkPoint.setUseExceptionCheck(false);
         }else {
             checkPoint.setUseExceptionCheck(true);
         }
-        if (testCase.getNoEmptyCheckpoint()==0){
+        if (testData.getNoEmptyCheckpoint()==0){
             checkPoint.setUseNullCheck(false);
         }else {
             checkPoint.setUseNullCheck(true);
         }
-        checkPoint.setContainCheckPoint(testCase.getContainCheckpoint());
-        checkPoint.setNoContainCheckPoint(testCase.getNoContainCheckpoint());
-        checkPoint.setJsonPathCheckPoint(testCase.getJsonpathCheckpoint());
+        checkPoint.setContainCheckPoint(testData.getContainCheckpoint());
+        checkPoint.setNoContainCheckPoint(testData.getNoContainCheckpoint());
+        checkPoint.setJsonPathCheckPoint(testData.getJsonpathCheckpoint());
         return checkPoint;
     }
 

kt-base/src/main/java/org/matrix/database/controller/TestController.java

 package org.matrix.database.controller;
 
 import com.alibaba.fastjson.JSONObject;
+import org.matrix.actuators.usecase.CaseActuator;
+import org.matrix.actuators.usecase.TestCaseExecuteResult;
+import org.matrix.database.entity.TestCase;
+import org.matrix.database.entity.TestData;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
 
@@ -15,6 +20,8 @@ import org.springframework.web.bind.annotation.*;
 @RequestMapping("/test")
 public class TestController {
 
+    @Autowired
+    CaseActuator caseActuator;
 
     /**
      * 获得表名,用于测试http执行器的调用
@@ -71,4 +78,38 @@ public class TestController {
         jsonObject.put("message",String.format("你好: %s",tableName));
         return ResponseEntity.ok(jsonObject);
     }
+
+    @GetMapping("/testCase")
+    public void testCase() {
+        String json = "{\n" +
+                "    \"url\":\"http://127.0.0.1:8080/test/sendMessage\",\n" +
+                "    \"method\":\"GET\",\n" +
+                "    \"requestType\":\"QUERY\",\n" +
+                "    \"headers\":[\n" +
+                "        {\n" +
+                "            \"name\":\"cookie\",\n" +
+                "            \"value\":\"123456\"\n" +
+                "        }\n" +
+                "    ],\n" +
+                "    \"requestBodies\":[\n" +
+                "        {\n" +
+                "            \"key\":\"tableName\",\n" +
+                "            \"type\":\"TEXT\",\n" +
+                "            \"value\":\"张三\"\n" +
+                "        }\n" +
+                "    ]\n" +
+                "}";
+        System.out.println(json);
+        TestCase testCase = new TestCase();
+        testCase.setName("name");
+        testCase.setType(1);
+        testCase.setDetail(json);
+        TestData testData = new TestData();
+        testData.setAbnormalCheckpoint(1);
+        testData.setContainCheckpoint("张三,李四");
+        testData.setNoContainCheckpoint("张三,李四");
+        testData.setJsonpathCheckpoint("contains({$..category},'${componentName}[0]')");
+        TestCaseExecuteResult testCaseExecuteResult = caseActuator.executeTestCase(testCase,testData,1L,1L);
+        System.out.println(testCaseExecuteResult);
+    }
 }

kt-base/src/main/java/org/matrix/socket/ClientSocket.java

+package org.matrix.socket;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.matrix.actuators.usecase.CaseActuator;
+import org.matrix.actuators.usecase.TestCaseExecuteResult;
+import org.matrix.database.entity.TestCase;
+import org.matrix.util.SpringUtils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.TimeUnit;
+
+import static org.matrix.socket.SocketHandler.*;
+import static org.matrix.socket.SocketPool.add;
+
+/**
+ * @author zhoujian
+ * 自定义封装的连接的客户端
+ */
+@Slf4j
+@Data
+public class ClientSocket implements Runnable{
+
+    private Socket socket;
+    private DataInputStream inputStream;
+    private DataOutputStream outputStream;
+    private Long key;
+    private String message;
+    private CaseActuator caseActuator;
+
+    ClientSocket(){
+       this.caseActuator =  SpringUtils.getBean("caseActuator");
+    }
+
+    @Override
+    public void run() {
+        this.key = Thread.currentThread().getId();
+        add(this);
+        System.out.println("-----key:"+key);
+        while (true){
+            System.out.println("1111111");
+            byte[] bytes = new byte[1024];
+            try {
+                this.getInputStream().read(bytes);
+                System.out.println(new String(bytes, StandardCharsets.UTF_8));
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+//        System.out.println("意外停止了");
+
+
+
+
+        //接受socket消息，根据前端提交的消息来判断，是否应该执行下一步
+//        while (true){
+//            String message = onMessage(this);
+//            System.out.println("接受到了客户端传来的消息");
+//            System.out.println(message);
+//            String json = "{\n" +
+//                    "    \"url\":\"http://127.0.0.1:8080/test/sendMessage\",\n" +
+//                    "    \"method\":\"GET\",\n" +
+//                    "    \"requestType\":\"QUERY\",\n" +
+//                    "    \"headers\":[\n" +
+//                    "        {\n" +
+//                    "            \"name\":\"cookie\",\n" +
+//                    "            \"value\":\"123456\"\n" +
+//                    "        }\n" +
+//                    "    ],\n" +
+//                    "    \"requestBodies\":[\n" +
+//                    "        {\n" +
+//                    "            \"key\":\"tableName\",\n" +
+//                    "            \"type\":\"TEXT\",\n" +
+//                    "            \"value\":\"张三\"\n" +
+//                    "        }\n" +
+//                    "    ]\n" +
+//                    "}";
+//            System.out.println(json);
+//            TestCase testCase = new TestCase();
+//            testCase.setName("name");
+//            testCase.setType(1);
+//            testCase.setDetail(json);
+//            testCase.setAbnormalCheckpoint(1);
+//            testCase.setContainCheckpoint("张三,李四");
+//            testCase.setNoContainCheckpoint("张三,李四");
+//            testCase.setJsonpathCheckpoint("contains({$..category},'${componentName}[0]')");
+//            TestCaseExecuteResult testCaseExecuteResult = caseActuator.executeTestCase(testCase,1L,1L);
+//            System.out.println(testCaseExecuteResult);
+//            if (!StringUtils.isEmpty(message)){
+//                break;
+//            }
+//        }
+//        //每5秒进行一次客户端连接，判断是否需要释放资源
+//        while (true){
+//            try {
+//                TimeUnit.SECONDS.sleep(5);
+//            } catch (InterruptedException e) {
+//                e.printStackTrace();
+//            }
+//            if (isSocketClosed(this)){
+//                log.info("客户端已关闭,其Key值为：{}", this.getKey());
+//                //关闭对应的服务端资源
+//                close(this);
+//                break;
+//            }
+//        }
+    }
+}
\ No newline at end of file

kt-base/src/main/java/org/matrix/socket/LogQueue.java

+package org.matrix.socket;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+
+
+/**
+ * 日志消息队列
+ * @author huangxiahao
+ */
+public class LogQueue {
+
+    private static final ConcurrentHashMap<Long, List<String>> LOG_MAP = new ConcurrentHashMap<>();
+
+    public static void add(Long threadId,String log){
+        List<String> logList = LOG_MAP.get(threadId);
+        if (logList!=null){
+            logList.add(log);
+        }else {
+            logList = new ArrayList<>();
+            logList.add(log);
+            LOG_MAP.put(threadId,logList);
+        }
+        //将log发送出去
+        sendMessage(threadId,log);
+    }
+
+    public static void sendMessage(Long threadId,String log){
+        SocketHandler.sendMessage(SocketPool.get(threadId),log);
+    }
+
+}

kt-base/src/main/java/org/matrix/socket/SocketHandler.java

+package org.matrix.socket;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+
+import static org.matrix.socket.SocketPool.*;
+
+/**
+ * Socket操作处理类
+ * @author huangxiahao
+ */
+@Slf4j
+public class SocketHandler{
+
+
+    /**
+     * 将连接的Socket注册到Socket池中
+     * @param socket
+     * @return
+     */
+    public static ClientSocket register(Socket socket){
+        ClientSocket clientSocket = new ClientSocket();
+        clientSocket.setSocket(socket);
+        try {
+            clientSocket.setInputStream(new DataInputStream(socket.getInputStream()));
+            clientSocket.setOutputStream(new DataOutputStream(socket.getOutputStream()));
+            return clientSocket;
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+
+    /**
+     * 向指定客户端发送信息
+     * @param clientSocket
+     * @param message
+     */
+    public static void sendMessage(ClientSocket clientSocket, String message){
+        try {
+            clientSocket.getOutputStream().write(message.getBytes(StandardCharsets.UTF_8));
+        } catch (IOException e) {
+            log.error("发送信息异常：{}", e.getMessage());
+            close(clientSocket);
+        }
+    }
+
+    /**
+     * 获取指定客户端的上传信息
+     */
+    public static String onMessage(ClientSocket clientSocket){
+        byte[] bytes = new byte[1024];
+        try {
+            clientSocket.getInputStream().read(bytes);
+            return new String(bytes, StandardCharsets.UTF_8);
+        } catch (IOException e) {
+            e.printStackTrace();
+            close(clientSocket);
+        }
+        return null;
+    }
+
+    /**
+     * 指定Socket资源回收
+     */
+    public static void close(ClientSocket clientSocket){
+        log.info("进行资源回收");
+        if (clientSocket != null){
+            log.info("开始回收socket相关资源，其Key为{}", clientSocket.getKey());
+            remove(clientSocket.getKey());
+            Socket socket = clientSocket.getSocket();
+            try {
+                socket.shutdownInput();
+                socket.shutdownOutput();
+            } catch (IOException e) {
+                log.error("关闭输入输出流异常，{}", e.getMessage());
+            }finally {
+                try {
+                    socket.close();
+                } catch (IOException e) {
+                    log.error("关闭socket异常{}", e.getMessage());
+                }
+            }
+        }
+    }
+
+
+    /**
+     * 发送数据包，判断数据连接状态
+     * @param clientSocket
+     * @return
+     */
+    public static boolean isSocketClosed(ClientSocket clientSocket){
+        try {
+            clientSocket.getSocket().sendUrgentData(1);
+            return false;
+        } catch (IOException e) {
+            return true;
+        }
+    }
+
+}
\ No newline at end of file

kt-base/src/main/java/org/matrix/socket/SocketPool.java

+package org.matrix.socket;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author huangxiahao
+ */
+public class SocketPool {
+
+    private static final ConcurrentHashMap<Long, ClientSocket> ONLINE_SOCKET_MAP = new ConcurrentHashMap<>();
+
+
+    public static void add(ClientSocket clientSocket){
+        if (clientSocket != null && clientSocket.getKey()!=null) {
+            ONLINE_SOCKET_MAP.put(clientSocket.getKey(), clientSocket);
+        }
+    }
+
+    public static void remove(Long key){
+        if (key!=null) {
+            ONLINE_SOCKET_MAP.remove(key);
+        }
+    }
+
+    public static ClientSocket get(Long key){
+       return ONLINE_SOCKET_MAP.get(key);
+    }
+}
\ No newline at end of file

kt-base/src/main/java/org/matrix/socket/SocketServer.java

+package org.matrix.socket;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.PropertySource;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import static org.matrix.socket.SocketHandler.register;
+
+@EqualsAndHashCode(callSuper = true)
+@Slf4j
+@Data
+public class SocketServer extends Thread{
+
+    private Integer port;
+    private boolean started;
+    private ServerSocket serverSocket;
+    private ExecutorService executorService = Executors.newCachedThreadPool();
+
+    public SocketServer(Integer port) {
+        this.port = port;
+    }
+
+    @Override
+    public void run() {
+        try {
+            serverSocket =  new ServerSocket( port);
+            started = true;
+            log.info("Socket服务已启动，占用端口： {}", serverSocket.getLocalPort());
+        } catch (IOException e) {
+            log.error("端口冲突,异常信息：{}", e);
+            System.exit(0);
+        }
+        while (started){
+            try {
+                Socket socket = serverSocket.accept();
+                socket.setKeepAlive(true);
+                ClientSocket register = register(socket);
+                log.info("客户端已连接");
+                if (register != null){
+                    executorService.submit(register);
+                }
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+}
\ No newline at end of file

kt-base/src/main/java/org/matrix/testNg/TestPigeon.java

@@ -76,7 +76,8 @@ public class TestPigeon extends AbstractTestNGSpringContextTests {
         CaseActuator caseActuator = new CaseActuator(checkPointActuator, httpClientActuator);
         TestCase testCase = java.util.Optional.of(caseService.getById(caseId))
                 .orElseThrow(() -> new GlobalException(String.format("没有找到id = %d 的TestCase", caseId)));
-        TestCaseExecuteResult testCaseExecuteResult = caseActuator.executeTestCase(testCase, envId, projectId);
+        //todo 苗润雨 将这个NULL填一下，里面放testData
+        TestCaseExecuteResult testCaseExecuteResult = caseActuator.executeTestCase(testCase,null, envId, projectId);
         CheckPointResult checkPointResult = testCaseExecuteResult.getCheckPointResult();
         Reporter.log(checkPointResult.toString());
     }

kt-base/src/test/java/org/matrix/AppTest.java

@@ -2,6 +2,14 @@ package org.matrix;
 
 
 import org.junit.jupiter.api.Test;
+import org.junit.platform.commons.util.StringUtils;
+
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.File;
+import java.io.FileWriter;
+import java.net.Socket;
+import java.util.UUID;
 
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
@@ -18,4 +26,30 @@ public class AppTest
     {
         assertTrue( true );
     }
+
+    @Test
+    public void socketTest() throws Exception
+    {
+        String host = "127.0.0.1";
+        int port = 8009;
+
+        //与服务端建立连接
+        Socket socket = new Socket(host, port);
+        socket.setOOBInline(true);
+        //建立连接后获取输出流
+        DataOutputStream outputStream = new DataOutputStream(socket.getOutputStream());
+        DataInputStream inputStream = new DataInputStream(socket.getInputStream());
+        String uuid = UUID.randomUUID().toString();
+        outputStream.write(uuid.getBytes());
+        while (true){
+            byte[] buff = new byte[1024];
+            inputStream.read(buff);
+            String buffer = new String(buff, "utf-8");
+            if (StringUtils.isNotBlank(buffer)){
+                System.out.println(buffer);
+                outputStream.write(uuid.getBytes());
+            }
+
+        }
+    }
 }

kt-base/src/test/java/org/matrix/actuators/sql/CaseActuatorTest.java

@@ -7,6 +7,7 @@ import org.matrix.actuators.httpclient.HttpClientActuator;
 import org.matrix.actuators.usecase.CaseActuator;
 import org.matrix.actuators.usecase.TestCaseExecuteResult;
 import org.matrix.database.entity.TestCase;
+import org.matrix.database.entity.TestData;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.test.context.junit4.SpringRunner;
@@ -43,11 +44,12 @@ class CaseActuatorTest {
         testCase.setName("name");
         testCase.setType(1);
         testCase.setDetail(json);
-        testCase.setAbnormalCheckpoint(1);
-        testCase.setContainCheckpoint("张三,李四");
-        testCase.setNoContainCheckpoint("张三,李四");
-        testCase.setJsonpathCheckpoint("contains({$..category},'${componentName}[0]')");
-        TestCaseExecuteResult testCaseExecuteResult = caseActuator.executeTestCase(testCase,1L,1L);
+        TestData testData = new TestData();
+        testData.setAbnormalCheckpoint(1);
+        testData.setContainCheckpoint("张三,李四");
+        testData.setNoContainCheckpoint("张三,李四");
+        testData.setJsonpathCheckpoint("contains({$..category},'${componentName}[0]')");
+        TestCaseExecuteResult testCaseExecuteResult = caseActuator.executeTestCase(testCase,testData,1L,1L);
         System.out.println(testCaseExecuteResult);