添加日志

commons/src/main/java/org/yzh/commons/constant/BusinessConstant.java

@@ -1,13 +1,30 @@
 package org.yzh.commons.constant;
 
 public interface BusinessConstant {
+    /**
+     * 通用日志前缀
+     */
     String LOGGER_PREFIX = " --- custom log --- ";
 
+    /**
+     * 默认通道号
+     */
     Integer DEFAULT_CHANNEL_NO = 9;
 
+    /**
+     * 直播推流状态
+     */
     String LIVE_STATUS_ONLINE = "online";
     String LIVE_STATUS_WAITING = "waiting";
     String LIVE_STATUS_OFFLINE = "offline";
 
+    /**
+     * 慢直播服务更新推流车辆和直播推流状态接口地址
+     */
     String LIVE_SERVER_UPDATE_PUSH_CAR = "http://42.192.165.208:8002/blade-business/common-api/set-push-car-and-stat";
+
+    /**
+     * jtt1078服务IP地址
+     */
+    String JTT1078_SERVER_HOST = "112.6.27.89";
 }

commons/src/main/java/org/yzh/commons/util/BusinessCacheUtil.java

@@ -11,6 +11,7 @@ public class BusinessCacheUtil {
     private static String CACHE_KEY_CURRENT_PUSH_CHANNEL = "current-push-channel";
     private static String CACHE_KEY_1078_PUSH_DEVICE = "1078-push-device";
     private static String CACHE_KEY_PUSH_STATUS = "push-status";
+    private static String CACHE_KEY_PUSH_EMPTY_TIMES = "1078-push-empty-times";
 
     /**
      * 获取当前推流车辆的clientID
@@ -85,4 +86,22 @@ public class BusinessCacheUtil {
     public static void setPushStatus(String status) {
         businessCache.put(CACHE_KEY_PUSH_STATUS, status);
     }
+
+    /**
+     * 获取1078推送为空的次数缓存
+     *
+     * @return
+     */
+    public static int getPushEmptyTimes() {
+        String times = "0";
+        if (businessCache.containsKey(CACHE_KEY_PUSH_EMPTY_TIMES)) {
+            times = businessCache.get(CACHE_KEY_PUSH_EMPTY_TIMES);
+        }
+
+        return Integer.parseInt(times);
+    }
+
+    public static void setPushEmptyTimes(int times) {
+        businessCache.put(CACHE_KEY_PUSH_EMPTY_TIMES, String.valueOf(times));
+    }
 }

jtt808-server/src/main/java/io/github/yezhihao/netmc/websocket/WsChannelInboundHandler.java

@@ -39,12 +39,21 @@ public class WsChannelInboundHandler extends SimpleChannelInboundHandler<TextWeb
         // 获取客户端传输过来的消息
         String content = msg.text();
 
+        // 由于有多个客户端发送数据, 故此处添加判断, 非1078发送的数据直接返回
+        if (!content.contains("jtt1078:")) {
+            currentChannel.writeAndFlush(new TextWebSocketFrame("服务器时间" + LocalDateTime.now() + " : " + content));
+            return;
+        }
+
         // 每分钟的0s, 打印当前推流通道
         LocalTime localTime = LocalTime.now();
         if (localTime.getSecond() == 0) {
-            log.info(BusinessConstant.LOGGER_PREFIX + " : jt1078推送信息 : content = {}", content);
+            log.info(BusinessConstant.LOGGER_PREFIX + " : jt1078推送信息 : currentChannel = {}:{}:{},  content = {}",
+                    currentChannel.id(), currentChannel.localAddress(), currentChannel.remoteAddress(), content);
         }
 
+        content = content.replace("jtt1078:", "");
+
         if (content.contains("-")) {
             // 由于1078传过来的信息为(0+clientId+'-'+channelNo), 故此处需要解析
             // 接收jt1078通过websocket传输的数据
@@ -52,6 +61,7 @@ public class WsChannelInboundHandler extends SimpleChannelInboundHandler<TextWeb
             String channelNo = content.split("-")[1];
 
             BusinessCacheUtil.set1078PushDevice(clientId);
+            BusinessCacheUtil.setPushEmptyTimes(0);
             String curPush = BusinessCacheUtil.getCurrentPushDevice();
             int curPushChannel = BusinessCacheUtil.getCurrentPushChannel();
             log.debug(BusinessConstant.LOGGER_PREFIX + " : websocket接收 : 1078推流设备 = {} - {} ", clientId, channelNo);
@@ -72,6 +82,19 @@ public class WsChannelInboundHandler extends SimpleChannelInboundHandler<TextWeb
                     SendCommandUtil.getService().noticeCarStatus(null, BusinessConstant.DEFAULT_CHANNEL_NO, BusinessConstant.LIVE_STATUS_OFFLINE);
                 }
             }
+
+            int emptyTimes = BusinessCacheUtil.getPushEmptyTimes();
+            // 若持续2分钟收到的消息都为空, 调用推流方法
+            if (emptyTimes == 120) {
+                log.info(BusinessConstant.LOGGER_PREFIX + " : websocket接收 : 持续2分钟收到的消息都为空, 调用推流方法 ");
+                if (SendCommandUtil.getService() != null) {
+                    SendCommandUtil.getService().sendToDevice();
+                }
+                emptyTimes = 0;
+            } else {
+                emptyTimes++;
+            }
+            BusinessCacheUtil.setPushEmptyTimes(emptyTimes);
         }
 
         currentChannel.writeAndFlush(new TextWebSocketFrame("服务器时间" + LocalDateTime.now() + " : " + content));
@@ -92,7 +115,7 @@ public class WsChannelInboundHandler extends SimpleChannelInboundHandler<TextWeb
 
         BusinessCacheUtil.set1078PushDevice("");
         if (SendCommandUtil.getService() != null) {
-            // 1078上线后, 停止原推流线程, 重新启动线程发送推流命令, 避免
+            // 1078上线后, 停止原推流线程, 重新启动线程发送推流命令
             SendCommandUtil.getService().restartSendToDevice();
         }
     }

jtt808-server/src/main/java/org/yzh/web/controller/JT1078Controller.java

@@ -2,6 +2,7 @@ package org.yzh.web.controller;
 
 import io.github.yezhihao.netmc.session.Session;
 import io.github.yezhihao.netmc.session.SessionManager;
+import io.netty.util.internal.StringUtil;
 import io.swagger.v3.oas.annotations.Operation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -67,7 +68,7 @@ public class JT1078Controller {
 
         log.info(BusinessConstant.LOGGER_PREFIX + " : T9101 - step1 : 当前推流设备 = {} - {} ", BusinessCacheUtil.getCurrentPushDevice(), BusinessCacheUtil.getCurrentPushChannel());
         // 先调用停止推流方法, 然后指定设备推流
-        if (BusinessCacheUtil.getCurrentPushDevice() != null) {
+        if (StringUtil.isNullOrEmpty(BusinessCacheUtil.getCurrentPushDevice())) {
             T9102 t9102 = new T9102();
             t9102.setChannelNo(BusinessCacheUtil.getCurrentPushChannel());
             t9102.setClientId(BusinessCacheUtil.getCurrentPushDevice());

jtt808-server/src/main/java/org/yzh/web/endpoint/MessageManager.java

@@ -88,7 +88,7 @@ public class MessageManager {
                 .timeout(Duration.ofSeconds(10), TIMEOUT_RESULT)
                 .onErrorResume(e -> {
 //                    log.warn("消息发送失败", e);
-                    log.warn(BusinessConstant.LOGGER_PREFIX + " : 发送推流命令 : 结果 = 消息发送失败. msg : {}", e);
+                    log.info(BusinessConstant.LOGGER_PREFIX + " : 发送推流命令 : 结果 = 消息发送失败. msg : {}", e);
                     return SENDFAIL_RESULT;
                 });
         log.info(BusinessConstant.LOGGER_PREFIX + " : 发送推流命令 : 结果 = {}", resultMono.block().toString());

jtt808-server/src/main/java/org/yzh/web/service/SendCommandService.java

@@ -2,6 +2,7 @@ package org.yzh.web.service;
 
 import io.github.yezhihao.netmc.session.Session;
 import io.github.yezhihao.netmc.session.SessionManager;
+import io.netty.util.internal.ObjectUtil;
 import io.netty.util.internal.StringUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -50,7 +51,7 @@ public class SendCommandService {
                 T9101 request = new T9101();
                 request.setChannelNo(9);
                 request.setClientId(clientId);
-                request.setIp("112.6.27.89");
+                request.setIp(BusinessConstant.JTT1078_SERVER_HOST);
                 request.setMediaType(1);
                 request.setStreamType(0);
                 request.setTcpPort(4001);
@@ -77,6 +78,11 @@ public class SendCommandService {
      * 管理当前所有在线设备, 自动发送推流命令, 直到成功一个或全部失败, 通知慢直播平台当前推流车辆及推流状态
      */
     public void sendToDevice() {
+        if (sessionManager.all() == null || sessionManager.all().size() == 0) {
+            log.info(BusinessConstant.LOGGER_PREFIX + " : 自定义推流方法 sendToDevice : 当前无在线设备, 直接返回 ");
+            return;
+        }
+
         log.info(BusinessConstant.LOGGER_PREFIX + " : 自定义推流方法 sendToDevice : 开始执行 ");
 
         if (sendCommandToDeviceThread != null && sendCommandToDeviceThread.isAlive()) {
@@ -145,7 +151,7 @@ public class SendCommandService {
             T9101 request = new T9101();
             request.setChannelNo(channelNo);
             request.setClientId(clientId);
-            request.setIp("112.6.27.89");
+            request.setIp(BusinessConstant.JTT1078_SERVER_HOST);
             request.setMediaType(1);
             request.setStreamType(0);
             request.setTcpPort(4001);
@@ -214,6 +220,7 @@ public class SendCommandService {
         params.put("clientId", clientId);
         params.put("carStat", status);
         log.info(BusinessConstant.LOGGER_PREFIX + " : 调用慢直播服务接口 : 当前推流设备 = {} - {} - {}", clientId, channelNo, status);
-        HttpClientUtils.doGet(url, params);
+        String res = HttpClientUtils.doGet(url, params);
+        log.info(BusinessConstant.LOGGER_PREFIX + " : 调用慢直播服务接口 : 返回结果 = {}", res);
     }
 }