微聊测试代码

src/main/java/com/tidecloud/dataacceptance/service/impl/JingWeiCardServerHandler.java

@@ -18,6 +18,7 @@ import io.netty.channel.socket.nio.NioServerSocketChannel;
 import io.netty.handler.codec.DelimiterBasedFrameDecoder;
 import io.netty.handler.logging.LogLevel;
 import io.netty.handler.logging.LoggingHandler;
+import io.netty.util.ByteProcessor;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -26,6 +27,8 @@ import org.springframework.stereotype.Component;
 
 import java.text.SimpleDateFormat;
 import java.util.Date;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
@@ -43,6 +46,9 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
     public static final String name = "JingWeiCardServerHandler";
     private static final Logger logger = LoggerFactory.getLogger(JingWeiCardServerHandler.class);
     private static ExecutorService executorService = Executors.newSingleThreadExecutor();
+    private static final ByteProcessor FIND_COMMA = new ByteProcessor.IndexOfProcessor((byte) ',');
+
+    private static Map<String, Channel> deviceChannelMap = new HashMap<>();
 
     @Autowired
     private DeviceMsgClient deviceMsgClient;
@@ -55,6 +61,7 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
         byte[] req = new byte[in.readableBytes()];
         in.readBytes(req);
         String msg = new String(req, "UTF-8");
+        System.out.println(req.length);
         logger.info("传入数据：》》》》》》》》》" + msg);
         Advice advice = setAdvice(msg);
         if (advice == null) {
@@ -67,6 +74,7 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
             case "LK": // 链路保持
                 sendMsg2Kafka((msg + DateUtil.formatDate2String(new Date())).getBytes(), advice.getDeviceId(), channel);
                 normalReply(advice, channel, "LK");
+                deviceChannelMap.put(advice.getDeviceId(), channel);
                 deviceMsgClient.acceptDeviceMsgParam("JingWei",
                         advice.getDeviceId(),
                         1, msg, System.currentTimeMillis());
@@ -82,6 +90,14 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
                 logger.info("盲点补传数据[UD2]：" + advice.toString());
                 sendMsg2Kafka(msg.getBytes(), advice.getDeviceId(), channel);
                 break;
+            case "TKQ": // 终端请求语音下发
+                logger.info("终端请求语音下发[TKQ]：" + advice.toString());
+                normalReply(advice, channel, "TKQ");
+                break;
+            case "TKQ2": // 终端请求好友语音下发
+                logger.info("终端请求好友语音下发[TKQ2]：" + advice.toString());
+                normalReply(advice, channel, "TKQ2");
+                break;
             case "TK": // 微聊对讲终端发送内容
                 if (advice.getAdviceSerial().equals("1")){
                     logger.info("发送对讲数据成功!");
@@ -91,9 +107,10 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
                     break;
                 } else {
                     logger.info("微聊对讲数据[TK]：" + advice.toString());
-                    VoiceMsgVo msgVo = splitVoiceMsg(msg, advice.getDeviceId());
+                    VoiceMsgVo msgVo = splitVoiceMsg(msg, advice.getDeviceId(), in);
                     voiceMsgClient.insVoiceMsg(msgVo);
                     normalReply(advice, channel, "TK,1");
+                    normalReply(channel, msgVo,msg);
                 }
                 break;
             case "UPLOAD": //数据上传间隔设置
@@ -109,9 +126,9 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
                 break;
         }
     }
-    //[CS*YYYYYYYYYY*LEN*TK,AMR 格式音频数据]
+    //[3G*YYYYYYYYYY*LEN*TK,AMR 格式音频数据]
 
-    protected VoiceMsgVo splitVoiceMsg(String msg, String deviceId) {
+    protected VoiceMsgVo splitVoiceMsg(String msg, String deviceId, ByteBuf byteBufIn) {
         VoiceMsgVo voiceMsg = new VoiceMsgVo();
         SimpleDateFormat df = new SimpleDateFormat("yyyyMMddHHmmss");//设置日期格式
         String time = df.format(new Date());
@@ -120,68 +137,17 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
         voiceMsg.setNu(1);
         voiceMsg.setVoiceTime(time);
         try {
-            String[] deviceMsg = msg.split("\\,|]");
-            StringBuilder replyCommand = new StringBuilder();
-            for (int i = 1; i <= (deviceMsg.length - 1) / 2; i++) {
-                replyCommand.append(amrToHex16(deviceMsg[(i*2)-1],deviceMsg[i*2])).append(",");
-            }
-            replyCommand.deleteCharAt(replyCommand.length()-1);
-            String replyCommandStr = replyCommand.toString();
-            byte[] bytes = replyCommandStr.getBytes();
+            int startIndex = msg.indexOf(",");
+            int endIndex = msg.indexOf("]");
+            String mic = msg.substring(startIndex + 1,endIndex);
+            byte[] bytes = mic.getBytes();
             voiceMsg.setMsg(bytes);
-            voiceMsg.setLength(Integer.valueOf(bytes.length));
         } catch (Exception e) {
             logger.warn("语音解析错误" + JSON.toJSONString(voiceMsg) + ":" + e.getStackTrace());
         }
         return voiceMsg;
     }
 
-    public static String amrToHex16(String a,String b) {
-        String exp = "0X7D";
-        if (a.equals(exp) && b.equals("0X01")){
-            return "0X7D";
-        } else if (a.equals(exp) && b.equals("0X02")){
-            return "0X5B";
-        } else if (a.equals(exp) && b.equals("0X03")){
-            return "0X5D";
-        } else if (a.equals(exp) && b.equals("0X04")){
-            return "0X2C";
-        } else if (a.equals(exp) && b.equals("0X05")){
-            return "0X2A";
-        }
-        return "";
-    }
-    public static String amrToReverse(String a) {
-        if (a.equals("0X7D")){
-            return "0X7D 0X01";
-        } else if (a.equals("0X5B")){
-            return "0X7D 0X02";
-        } else if (a.equals("0X5D")){
-            return "0X7D 0X03";
-        } else if (a.equals("0X2C")){
-            return "0X7D 0X04";
-        } else if (a.equals("0X2A")){
-            return "0X7D 0X05";
-        }
-        return "";
-    }
-
-    /**
-     * 回复内容
-     *
-     * @param msg      [CS*YYYYYYYYYY*LEN*TK,1]
-     * @param deviceId
-     * @return
-     */
-/*    protected VoiceMsgVo splitVoiceMsg(String msg, String deviceId) {
-        VoiceMsgVo voiceMsg = new VoiceMsgVo();
-        String[] msgArr = msg.split("\\*");
-        // 获取内容
-        String[] contents = msgArr[3].split(",");
-        voiceMsg.setDeviceId(deviceId);
-        voiceMsg.setLag(Integer.valueOf(contents[1]));
-        return voiceMsg;
-    }*/
 
     /**
      * 查询服务器传来的语音包并回复
@@ -192,9 +158,11 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
     protected void voiceReplyToClient(String deviceId, Channel channel) {
         try {
             VoiceMsgVo voiceF = voiceMsgClient.queryVoiceMsg(deviceId, 1);
-            normalReply(channel, voiceF);
-            //发送中的数据
-            voiceMsgClient.updateVoiceMsgSendFinish(deviceId, 1, 3);
+            if (voiceF.getLag() == 1) {
+                normalReply(channel, voiceF,"");
+                //发送中的数据
+                voiceMsgClient.updateVoiceMsgSendFinish(deviceId, 1, 3);
+            }
         } catch (Exception e) {
             logger.error("语音下行发送异常！！！！！ deviceId=" + deviceId);
         }
@@ -204,21 +172,43 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
      *  主动回复
      *
      * @param channel
-     * @content    获取下一条待发送终端语音  [CS*YYYYYYYYYY*LEN*TK,AMR 格式音频数据]
+     * @content    获取下一条待发送终端语音  [3G*YYYYYYYYYY*LEN*TK,AMR 格式音频数据]
      */
-    private void normalReply(Channel channel, VoiceMsgVo voiceMsg) {
+    private void normalReply(Channel channel, VoiceMsgVo voiceMsg, String msg) {
         StringBuilder replyCommand = new StringBuilder();
-        replyCommand.append("CS").append("*");
-        replyCommand.append(voiceMsg.getDeviceId()).append("*");
-        replyCommand.append(numToHex16(voiceMsg.getLength())).append("*");
-        String replyCommandStr = replyCommand.toString();
 
-        ByteBuf buffer = Unpooled.buffer(replyCommandStr.getBytes().length);
+        String message = String.valueOf(msg);
+        int startIndex = message.indexOf("[");
+        int endIndex = message.indexOf("]");
+        String data = message.substring(startIndex + 1, endIndex);
+        String[] bodys = data.split("\\*");
+        replyCommand.append("[");
+        replyCommand.append("3G").append("*");
+        String deviceId = voiceMsg.getDeviceId();
+        if (deviceId.equals("9513532727")) {
+            deviceId = "9513532780";
+        }
+        replyCommand.append(deviceId).append("*");
+        replyCommand.append(bodys[2]).append("*");
+        replyCommand.append("TK").append(",");
+        String replyCommandStr = replyCommand.toString();
+        StringBuilder replyCommand2 = new StringBuilder();
+        replyCommand2.append("]");
+        String replyCommandStr2 = replyCommand2.toString();
+        ByteBuf buffer = Unpooled.buffer(replyCommandStr.getBytes().length + voiceMsg.getMsg().length + replyCommandStr2.getBytes().length);
         buffer.writeBytes(replyCommandStr.getBytes());
         buffer.writeBytes(voiceMsg.getMsg());
+        buffer.writeBytes(replyCommandStr2.getBytes());
         buffer.readerIndex(0);
+
+        channel = deviceChannelMap.get(deviceId);
+        if (channel == null || !channel.isActive()) {
+            deviceChannelMap.remove(deviceId);
+            logger.warn("the channle of device[{}] has closed", deviceId);
+            return;
+        }
         ChannelFuture channelFuture = channel.writeAndFlush(buffer);
-        channelFuture.addListener(future -> logger.info("Normal reply :" + replyCommandStr + JSON.toJSON(voiceMsg.getMsg())));
+        channelFuture.addListener(future -> logger.info("Normal reply :" + replyCommandStr + JSON.toJSON(voiceMsg.getMsg())  + replyCommandStr2));
     }
 
 
@@ -262,7 +252,7 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
 
     /**
      * 回复
-     * [ZJ*YYYYYYYYYY*LEN*LK]
+     * [3G*YYYYYYYYYY*LEN*LK]
      *
      * @param advice
      * @param channel
@@ -333,7 +323,9 @@ public class JingWeiCardServerHandler extends HexBinaryAcceptanceHandlerAdapter
                 // 标识符
                 advice.setAdviceType(contents[0]);
                 //指令流水字段被用作判断音频是否接受成功
-                advice.setAdviceSerial(contents[1]);
+                if (contents.length == 2) {
+                    advice.setAdviceSerial(contents[1]);
+                }
                 return advice;
             } catch (Exception e) {
                 logger.error(e.getMessage(), e);