Merge branch 'Jw' into 'master'

新增心跳处理器,修复已知bug

See merge request hewei/Jumeirah!73

customer-service/src/main/java/com/ym/im/core/MessageDecoder.java

@@ -8,6 +8,7 @@ package com.ym.im.core;
  */
 
 import com.ym.im.entity.ChatRecord;
+import com.ym.im.entity.HeartBeat;
 import com.ym.im.entity.MsgBody;
 import com.ym.im.util.JsonUtils;
 import io.netty.channel.ChannelHandler;
@@ -30,10 +31,16 @@ import java.util.List;
 @ChannelHandler.Sharable
 public class MessageDecoder extends MessageToMessageDecoder<TextWebSocketFrame> {
 
+    private final static String PING = "ping";
+
     @Override
     protected void decode(ChannelHandlerContext ctx, TextWebSocketFrame textWebSocketFrame, List out) throws Exception {
 
         final String jsonStr = textWebSocketFrame.text();
+        if (PING.equals(jsonStr)) {
+            out.add(new HeartBeat());
+            return;
+        }
         MsgBody<ChatRecord> msgBody = new MsgBody<ChatRecord>().setCode(MsgBody.ERROR);
         try {
             msgBody = JsonUtils.json2Obj(jsonStr, MsgBody.class, ChatRecord.class);

customer-service/src/main/java/com/ym/im/core/WebSocketChannelInitializer.java

@@ -2,6 +2,7 @@ package com.ym.im.core;
 
 
 import com.ym.im.entity.base.NettyConstant;
+import com.ym.im.handler.HeartBeatServerHandler;
 import com.ym.im.handler.SingleChatHandler;
 import com.ym.im.handler.WebSocketHandshakerHandler;
 import io.netty.channel.ChannelInitializer;
@@ -9,6 +10,7 @@ import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
 import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
+import io.netty.handler.timeout.IdleStateHandler;
 import io.netty.util.concurrent.EventExecutorGroup;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
@@ -35,6 +37,9 @@ public class WebSocketChannelInitializer extends ChannelInitializer<NioSocketCha
     private SingleChatHandler singleChatHandler;
 
     @Autowired
+    private HeartBeatServerHandler heartBeatServerHandler;
+
+    @Autowired
     private WebSocketHandshakerHandler webSocketHandshakerHandler;
 
 
@@ -45,8 +50,10 @@ public class WebSocketChannelInitializer extends ChannelInitializer<NioSocketCha
                         new HttpObjectAggregator(NettyConstant.MAX_FRAME_LENGTH),
                         webSocketHandshakerHandler,
                         new WebSocketServerProtocolHandler(NettyConstant.CS),
+                        new IdleStateHandler(60, 0, 0),
                         messageDecoder,
                         messageEncoder)
-                .addLast(businessGroup, singleChatHandler);//复杂业务绑定businessGroup
+                .addLast(businessGroup, singleChatHandler)//复杂业务绑定businessGroup
+                .addLast(heartBeatServerHandler);
     }
 }

customer-service/src/main/java/com/ym/im/entity/HeartBeat.java

+package com.ym.im.entity;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+/**
+ * @author: JJww
+ * @Date:2020/11/6
+ */
+@Data
+public class HeartBeat implements Serializable {
+}

customer-service/src/main/java/com/ym/im/handler/HeartBeatServerHandler.java

+package com.ym.im.handler;
+
+import com.ym.im.entity.HeartBeat;
+import com.ym.im.entity.base.ChannelAttributeKey;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
+import io.netty.handler.timeout.IdleState;
+import io.netty.handler.timeout.IdleStateEvent;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author: JJww
+ * @Date:2020/11/6
+ */
+@Slf4j
+@Component
+@ChannelHandler.Sharable
+public class HeartBeatServerHandler extends BaseHandler<HeartBeat> {
+
+    private final static String PONG = "pong";
+
+    @Override
+    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
+        if (evt instanceof IdleStateEvent) {
+            IdleStateEvent event = (IdleStateEvent) evt;
+            if (event.state() == IdleState.READER_IDLE) {
+                ctx.channel().close();
+                log.info("用户: " + ctx.channel().attr(ChannelAttributeKey.ROLE_ID).get() + "  无心跳 ");
+            }
+        } else {
+            super.userEventTriggered(ctx, evt);
+        }
+    }
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext ctx, HeartBeat heartBeat) throws Exception {
+        ctx.channel().writeAndFlush(new TextWebSocketFrame(PONG));
+    }
+}

customer-service/src/main/java/com/ym/im/service/impl/StaffServiceImpl.java

@@ -10,7 +10,10 @@ import com.ym.im.service.StaffService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.LinkedHashMap;
+import java.util.List;
 
 import static java.util.Map.Entry.comparingByValue;
 import static java.util.stream.Collectors.toMap;
@@ -72,7 +75,10 @@ public class StaffServiceImpl implements StaffService {
             return new MsgBody<>().setCode(MsgBody.BINDINGFAILURE).setMessage(ResultStatus.FORWARD_FAILURE.getMessage());
         }
         //移除原客服绑定
-        channelGroup.getMerchantStaff(userSocketInfo.getStaffId(merchantId)).getUserIds().remove(userId);
+        final StaffSocketInfo merchantStaff = channelGroup.getMerchantStaff(userSocketInfo.getStaffId(merchantId));
+        if (merchantStaff != null) {
+            merchantStaff.getUserIds().remove(userId);
+        }
         //设置新的客服
         staffSocketInfo.getUserIds().add(userId);
         userSocketInfo.setStaff(merchantId, staffId);

customer-service/src/main/resources/application-dev.yml

@@ -16,9 +16,9 @@ spring:
       max-wait: 60000
       # 配置间隔多久才进行一次检测，检测需要关闭的空闲连接，单位毫秒
       time-between-eviction-runs-millis: 60000
-      url: jdbc:mysql://127.0.0.1/customer_service?useUnicode=true&useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf8
+      url: jdbc:mysql://47.99.47.225/customer_service?useUnicode=true&useSSL=false&serverTimezone=Asia/Shanghai&characterEncoding=utf8
       username: root
-      password: 101020
+      password: temple123456
       driver-class-name: com.mysql.cj.jdbc.Driver
       connectionInitSqls: set names utf8mb4
   jackson:
@@ -28,9 +28,9 @@ spring:
     port: 5672
     username: root
     password: root
-    staff-offline-Queue-Name: staff.offline.dev
-    order-queue-name: push.order.dev
-    exchange-name: delay.exchange.dev
+    staff-offline-Queue-Name: staff.offline
+    order-queue-name: push.order
+    exchange-name: delay.exchange
     listener:
       simple:
         default-requeue-rejected: false