作者 crossoverJie

:white_check_mark: Adding tests.

... ... @@ -186,6 +186,10 @@ public class CIMClient {
if (channel != null && channel.isActive()) {
return;
}
//首先清除路由信息,下线
routeRequest.offLine();
LOGGER.info("开始重连。。");
start();
}
... ...
... ... @@ -69,7 +69,8 @@ public class CIMClientHandle extends SimpleChannelInboundHandler<CIMResponseProt
@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
LOGGER.info("客户端断开了!");
LOGGER.info("客户端断开了,重新连接!");
if (scheduledExecutorService == null){
scheduledExecutorService = SpringBeanFactory.getBean("scheduledTask",ScheduledExecutorService.class) ;
}
... ...
package com.crossoverjie.cim.client.service.impl;
import com.crossoverjie.cim.client.client.CIMClient;
import com.crossoverjie.cim.client.config.AppConfiguration;
import com.crossoverjie.cim.client.service.RouteRequest;
import com.crossoverjie.cim.common.kit.HeartBeatHandler;
import io.netty.channel.ChannelHandlerContext;
import okhttp3.MediaType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
... ... @@ -22,10 +20,6 @@ import org.springframework.stereotype.Service;
public class ClientHeartBeatHandlerImpl implements HeartBeatHandler {
private final static Logger LOGGER = LoggerFactory.getLogger(ClientHeartBeatHandlerImpl.class);
private final MediaType mediaType = MediaType.parse("application/json");
@Autowired
private AppConfiguration appConfiguration;
@Autowired
private CIMClient cimClient;
... ... @@ -36,9 +30,6 @@ public class ClientHeartBeatHandlerImpl implements HeartBeatHandler {
@Override
public void process(ChannelHandlerContext ctx) throws Exception {
//首先清除路由信息,下线
routeRequest.offLine();
//重连
cimClient.reconnect();
... ...
... ... @@ -174,6 +174,9 @@ public class AccountServiceRedisImpl implements AccountService {
@Override
public void offLine(Long userId) throws Exception {
// TODO: 2019-01-21 改为一个原子命令,以防数据一致性
//删除路由
redisTemplate.delete(ROUTE_PREFIX + userId) ;
... ...
... ... @@ -51,7 +51,7 @@ public class BeanConfig {
public CIMRequestProto.CIMReqProtocol heartBeat() {
CIMRequestProto.CIMReqProtocol heart = CIMRequestProto.CIMReqProtocol.newBuilder()
.setRequestId(0L)
.setReqMsg("ping")
.setReqMsg("pong")
.setType(Constants.CommandType.PING)
.build();
return heart;
... ...
... ... @@ -61,18 +61,6 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
IdleStateEvent idleStateEvent = (IdleStateEvent) evt;
if (idleStateEvent.state() == IdleState.READER_IDLE) {
LOGGER.info("服务端没有收到消息,向客户端发送心跳!");
//向客户端发送消息
CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",
CIMRequestProto.CIMReqProtocol.class);
ctx.writeAndFlush(heartBeat).addListeners((ChannelFutureListener) future -> {
if (!future.isSuccess()) {
LOGGER.error("IO error,close Channel");
future.channel().close();
}
}) ;
HeartBeatHandler heartBeatHandler = SpringBeanFactory.getBean(ServerHeartBeatHandlerImpl.class) ;
heartBeatHandler.process(ctx) ;
}
... ... @@ -140,6 +128,15 @@ public class CIMServerHandle extends SimpleChannelInboundHandler<CIMRequestProto
//心跳更新时间
if (msg.getType() == Constants.CommandType.PING){
NettyAttrUtil.updateReaderTime(ctx.channel(),System.currentTimeMillis());
//向客户端响应 pong 消息
CIMRequestProto.CIMReqProtocol heartBeat = SpringBeanFactory.getBean("heartBeat",
CIMRequestProto.CIMReqProtocol.class);
ctx.writeAndFlush(heartBeat).addListeners((ChannelFutureListener) future -> {
if (!future.isSuccess()) {
LOGGER.error("IO error,close Channel");
future.channel().close();
}
}) ;
}
}
... ...