Dev log stream

README.md

@@ -121,3 +121,5 @@ Long user_count = Convert.toLong(sysUserRole.get("user_count"));
 - 使用 JSON 作为储存值，例如：{ dictionary_key: "Gender", dictionary_value: "[{\"label\":\"男\",\"value\":1},{\"label\":\"女\",\"value\":2}]" }
 - 接口1：查询接口，首次调用Service时，将该值保存到Redis长效缓存
 - 接口2：手动更新缓存的接口
+
+2.日志log加.gz压缩

src/main/java/com/backendsys/modules/log/controller/LogStreamController.java

@@ -1,43 +1,86 @@
 package com.backendsys.modules.log.controller;
 
 import cn.hutool.core.convert.Convert;
+import com.backendsys.exception.CustomException;
 import com.backendsys.modules.common.config.security.annotations.Anonymous;
 import com.backendsys.modules.sse.emitter.SseEmitterManager;
 import com.backendsys.modules.sse.utils.SseEmitterUTF8;
 import com.backendsys.modules.sse.utils.SseUtil;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.http.MediaType;
+import org.springframework.http.codec.ServerSentEvent;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+import reactor.core.publisher.Flux;
 
+import java.io.BufferedReader;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.time.LocalTime;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
+@Slf4j
 @RestController
 public class LogStreamController {
 
     @Autowired
     private SseUtil sseUtil;
 
+    @Value("${log-stream.enable}")
+    private Boolean isEnable;
+    @Value("${log-stream.charset}")
+    private String charset;
+    @Value("${log-stream.exec}")
+    private String exec;
+
+
     /**
      * [SSE] 消息监听
      */
     @Anonymous
     @GetMapping(value = "/api/log/stream/watch", produces = "text/event-stream")
     public SseEmitter stream() {
+
+        if (!isEnable) return null;
+
         String userId = Convert.toStr(1L);
         SseEmitterUTF8 emitter = new SseEmitterUTF8(Long.MAX_VALUE);
         SseEmitterManager manager = SseEmitterManager.getInstance();
-
         try {
-
             // 如果存在，则关闭
             sseUtil.closeIfExist(userId);
-
             // 创建新连接
             manager.addEmitter(userId, emitter);
             emitter.send(SseEmitter.event().data("Connected successfully! (连接成功)"));
+
+            new Thread(() -> {
+                try {
+
+                    // Window 使用 GBK 字符
+                    Process process = Runtime.getRuntime().exec(exec);  // "ping 127.0.0.1" | "docker logs -f backendsys";
+                    BufferedReader reader = new BufferedReader(new InputStreamReader(process.getInputStream(), charset));
+                    String line;
+                    while ((line = reader.readLine()) != null) {
+                        // 将日志发送到客户端
+                        // System.out.println(line);
+                        emitter.send(SseEmitter.event().data(line));
+                    }
+
+//                    // 完成后关闭连接
+//                    emitter.complete();
+
+                } catch (IOException e) {
+                    emitter.completeWithError(e);
+                }
+            }).start();
+
         } catch (IOException e) {
             // 关闭连接
             // emitter.complete();

src/main/java/com/backendsys/modules/sse/emitter/SseEmitterManager.java

@@ -22,8 +22,14 @@ public class SseEmitterManager {
     // 公共方法，供外部添加SseEmitter
     public void addEmitter(String userId, SseEmitterUTF8 emitter) {
         this.emitters.put(userId, emitter);
-        emitter.onTimeout(() -> this.emitters.remove(emitter));
-        emitter.onCompletion(() -> this.emitters.remove(emitter));
+        emitter.onTimeout(() -> {
+            System.out.println("emitter (onTimeout) 超时");
+            this.emitters.remove(emitter);
+        });
+        emitter.onCompletion(() -> {
+            System.out.println("emitter (onCompletion) 中断");
+            this.emitters.remove(emitter);
+        });
     }
     // 公共方法，供外部移除SseEmitter
     public SseEmitterUTF8 getEmitter(String userId) {

src/main/java/com/backendsys/modules/sse/utils/SseUtil.java

@@ -18,7 +18,8 @@ public class SseUtil {
         SseEmitterUTF8 emitter = manager.getEmitter(Convert.toStr(userId));
         if (emitter != null) {
             try {
-                emitter.send(SseEmitter.event().data(data, MediaType.TEXT_PLAIN));
+//                emitter.send(SseEmitter.event().data(data));
+                emitter.send(SseEmitter.event().data(data));
             } catch (IOException e) {
                 System.out.println(e.getMessage());
                 manager.removeEmitter(emitter);
@@ -33,7 +34,7 @@ public class SseUtil {
         SseEmitterUTF8 oldEmitter = manager.getEmitter(userId);
         if (oldEmitter != null) {
             try {
-                oldEmitter.send(SseEmitter.event().data("Disconnected! (连接中断)", MediaType.TEXT_PLAIN));
+                oldEmitter.send(SseEmitter.event().data("Disconnected! (连接中断)"));
                 oldEmitter.complete();              // 关闭旧的连接
                 manager.removeEmitter(oldEmitter);  // 从管理器中移除旧的连接
             } catch (IOException e) {

src/main/resources/application-local.yml

@@ -11,6 +11,11 @@ HTTP_BASE_URI: http://127.0.0.1:48080
 HTTP_BASE_STATIC: http://127.0.0.1:48080
 HTTP_ACTUATOR_URI: https://jsonplaceholder.typicode.com
 
+log-stream:
+  enable: true
+  charset: GBK
+  exec: 'ping 127.0.0.1'
+
 spring:
   config:
     name: application-local

src/main/resources/application-prod.yml

@@ -11,6 +11,11 @@ HTTP_BASE_URI: http://43.128.1.201:48080
 HTTP_BASE_STATIC: http://ai.api.daoguyujia.com
 HTTP_ACTUATOR_URI: http://43.128.1.201:48080
 
+log-stream:
+  enable: true
+  charset: UTF-8
+  exec: 'docker logs -f backendsys'
+
 spring:
   config:
     name: application-prod

src/main/resources/logback.xml

@@ -54,7 +54,7 @@
              <level>:设置过滤级别
              <onMatch>:用于配置符合过滤条件的操作
              <onMismatch>:用于配置不符合过滤条件的操作
-                                       此处配置为只接收ERROR日志级别信息
+             此处配置为只接收ERROR日志级别信息
          -->
         <filter class="ch.qos.logback.classic.filter.LevelFilter">
             <level>ERROR</level>