|
|
@@ -1,25 +1,9 @@
|
|
|
package com.tsi.comm.server.tcp.handler;
|
|
|
|
|
|
-import com.tsi.comm.server.config.TsiCvimServerConfig;
|
|
|
-import com.tsi.comm.server.process.dbms.TsiCvimDbmsProcess;
|
|
|
-import com.tsi.comm.server.process.packet.TsiChannelSession;
|
|
|
-import com.tsi.comm.server.process.packet.TsiCvimPacketProcess;
|
|
|
-import com.tsi.comm.server.protocol.AbstractTsiPacket;
|
|
|
-import com.tsi.comm.server.protocol.TsiCpuDisconnected;
|
|
|
-import com.tsi.comm.server.repository.TsiAlarmManager;
|
|
|
-import com.tsi.comm.server.repository.TsiNodeManager;
|
|
|
-import com.tsi.comm.server.repository.TsiSessionManager;
|
|
|
-import com.tsi.comm.server.service.TsiQueueDistributorService;
|
|
|
-import com.tsi.comm.server.vo.TsiAlarmConfigVo;
|
|
|
-import com.tsi.comm.server.vo.TsiNodeVo;
|
|
|
-import com.tsi.comm.server.vo.mariadb.AbstractDbmsVo;
|
|
|
-import com.tsi.comm.server.vo.mariadb.AlarmOccrVo;
|
|
|
-import com.tsi.comm.server.vo.mariadb.NodeStatusVo;
|
|
|
-import com.tsi.comm.server.xnet.NettyUtils;
|
|
|
+import com.tsi.comm.server.tcp.service.ConnectionLifecycleService;
|
|
|
import io.netty.channel.ChannelHandler;
|
|
|
import io.netty.channel.ChannelHandlerContext;
|
|
|
import io.netty.channel.ChannelInboundHandlerAdapter;
|
|
|
-import io.netty.handler.timeout.IdleState;
|
|
|
import io.netty.handler.timeout.IdleStateEvent;
|
|
|
import lombok.RequiredArgsConstructor;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
@@ -31,147 +15,31 @@ import org.springframework.stereotype.Component;
|
|
|
@ChannelHandler.Sharable
|
|
|
public class CvimServerInboundMessageHandler extends ChannelInboundHandlerAdapter {
|
|
|
|
|
|
- private final TsiCvimServerConfig config;
|
|
|
- private final TsiNodeManager nodeManager;
|
|
|
- private final TsiSessionManager sessionManager;
|
|
|
- private final TsiAlarmManager alarmManager;
|
|
|
- private final TsiCvimPacketProcess packetProcess;
|
|
|
- private final TsiCvimDbmsProcess dbmsProcess;
|
|
|
- private final TsiQueueDistributorService queueDistributorService;
|
|
|
-
|
|
|
- public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
|
|
- //log.error("Handler: channelRead: {}", NettyUtils.getRemoteIpAddress(ctx.channel()));
|
|
|
- if (msg instanceof AbstractTsiPacket) {
|
|
|
- //TsiCpuPacket packet = (TsiCpuPacket)msg;
|
|
|
- //packet.setPop(System.nanoTime());
|
|
|
- }
|
|
|
- else {
|
|
|
- log.error("Handler: channelRead: Unknown Instance Type: {}", msg.getClass().getSimpleName());
|
|
|
- }
|
|
|
- }
|
|
|
+ private final ConnectionLifecycleService connectionService;
|
|
|
|
|
|
@Override
|
|
|
public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
|
|
- // session count increment
|
|
|
- int sessions = this.sessionManager.add();
|
|
|
- // IP Address 로 제어기를 판달할 수 있으면 여기서 제어기 등록 여부를 판단해서
|
|
|
- // 접속 여부를 처리할 수 있다.
|
|
|
- final String remoteIpAddr = NettyUtils.getRemoteIpAddress(ctx.channel());
|
|
|
- log.info("--channelActive: {}, {} Sessions.", remoteIpAddr, sessions);
|
|
|
- TsiNodeVo ipNodeVo = this.nodeManager.getIpAddr(remoteIpAddr);
|
|
|
- if (ipNodeVo == null) {
|
|
|
- this.alarmManager.reportUnknownIp(remoteIpAddr);
|
|
|
- }
|
|
|
- else {
|
|
|
- TsiChannelSession.sessionActive(ipNodeVo, remoteIpAddr);
|
|
|
-// if (ipNodeVo.isConnect()) {
|
|
|
-// // 이미 연결된 상태라면 새 채널을 종료, 이전 채널이 통신을 수행중일수 있으므로 현재 채널을 종료시킴
|
|
|
-// log.warn("--channelActive: {}, Duplicate connection detected, closing new channel: {}",
|
|
|
-// ipNodeVo.getKey(), remoteIpAddr);
|
|
|
-// ctx.channel().close();
|
|
|
-// return;
|
|
|
-// }
|
|
|
- }
|
|
|
+ this.connectionService.handleChannelActive(ctx);
|
|
|
ctx.fireChannelActive();
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
|
|
- final String remoteIpAddr = NettyUtils.getRemoteIpAddress(ctx.channel());
|
|
|
- int sessions = this.sessionManager.remove();
|
|
|
- TsiNodeVo nodeVo = this.sessionManager.getNodeVo(ctx.channel());
|
|
|
- if (nodeVo != null) {
|
|
|
- NodeStatusVo status = new NodeStatusVo(AbstractDbmsVo.DBMS_NODE_STATUS);
|
|
|
- status.setServerId(this.config.getServerId());
|
|
|
- status.setNodeId(nodeVo.getNodeId());
|
|
|
- status.setStatus(0);
|
|
|
- status.setIpAddr(remoteIpAddr);
|
|
|
- this.dbmsProcess.add(status, (int)Thread.currentThread().getId());
|
|
|
-
|
|
|
- if (this.sessionManager.isServerRun()) {
|
|
|
- TsiCpuDisconnected packet = new TsiCpuDisconnected(nodeVo.getNodeId(), ctx.channel());
|
|
|
- packet.setBuf(null);
|
|
|
- packet.setObj(nodeVo);
|
|
|
- this.packetProcess.add(packet, nodeVo.getPktQIdx()); // 패킷인덱스큐로 데이터 전송
|
|
|
- }
|
|
|
-// log.info("channelInactive: Node {}, {}, sessions: {}", nodeVo.getNodeId(), remoteIpAddr, sessions);
|
|
|
-
|
|
|
- TsiChannelSession.sessionInactive(nodeVo, remoteIpAddr);
|
|
|
-
|
|
|
- this.sessionManager.removeChannel(ctx.channel());
|
|
|
- ctx.channel().attr(TsiSessionManager.TSI_NODE_ATTRIBUTE_KEY).set(null);
|
|
|
- this.queueDistributorService.releaseQueue(nodeVo); // 작업큐를 할당을 해제한다.
|
|
|
-
|
|
|
- synchronized (nodeVo) {
|
|
|
- if (nodeVo.getChannel() == ctx.channel()) {
|
|
|
- // 현재 비활성화되는 채널이 nodeVo에 등록된 활성 채널과 동일한 경우에만
|
|
|
- nodeVo.setConnect(null);
|
|
|
- log.info("channelInactive: {}, {}, sessions: {}", nodeVo.getNodeId(), remoteIpAddr, sessions);
|
|
|
- } else {
|
|
|
- // 새로운 연결에 의해 강제로 종료되었음
|
|
|
- log.warn("channelInactive: {}, {}, sessions: {}, Duplicate Connected.", nodeVo.getNodeId(), remoteIpAddr, sessions);
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- else {
|
|
|
- log.info("channelInactive: {}, sessions: {}, Unknown node.", remoteIpAddr, sessions);
|
|
|
- }
|
|
|
-
|
|
|
+ this.connectionService.handleChannelInactive(ctx);
|
|
|
ctx.fireChannelInactive();
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public void userEventTriggered(ChannelHandlerContext ctx, Object e) throws Exception {
|
|
|
if (e instanceof IdleStateEvent) {
|
|
|
- IdleStateEvent evt = (IdleStateEvent) e;
|
|
|
- TsiNodeVo nodeVo = this.sessionManager.getNodeVo(ctx.channel());
|
|
|
- // 연결이 완료된 후 송수신 데이터가 일정시간 동안 없을 경우 이곳에서 처리
|
|
|
- if (evt.state() == IdleState.READER_IDLE) {
|
|
|
- String remoteIpAddr = NettyUtils.getRemoteIpAddress(ctx.channel());
|
|
|
- if (nodeVo == null) {
|
|
|
- log.warn("userEventTriggered: Recv Timeout: {}", remoteIpAddr);
|
|
|
- // 통신 접속 후 수신 데이터가 없이 READ 타임아웃이 발생한 경우임
|
|
|
- if (this.alarmManager.checkAlarm(TsiAlarmConfigVo.COMM_02)) {
|
|
|
- AlarmOccrVo alarm = new AlarmOccrVo(AbstractDbmsVo.DBMS_ALARM_OCCR_HS);
|
|
|
- alarm.setAlarmCode(TsiAlarmConfigVo.COMM_02);
|
|
|
- alarm.setAlarmTarget(remoteIpAddr);
|
|
|
- alarm.setAlarmValue(remoteIpAddr);
|
|
|
- this.dbmsProcess.add(alarm, (int) Thread.currentThread().getId());
|
|
|
- }
|
|
|
-// ctx.channel().disconnect();
|
|
|
- ctx.channel().close();
|
|
|
- }
|
|
|
- else {
|
|
|
- log.warn("userEventTriggered: Recv Timeout: {}, {}", remoteIpAddr, nodeVo.getNodeId());
|
|
|
- // 통신 접속 후 데이터를 한번이라도 수신한 경우에는 스케쥴러에서 처리한다.
|
|
|
- if (this.alarmManager.checkAlarm(TsiAlarmConfigVo.COMM_02)) {
|
|
|
- AlarmOccrVo alarm = new AlarmOccrVo(AbstractDbmsVo.DBMS_ALARM_OCCR_HS);
|
|
|
- alarm.setAlarmCode(TsiAlarmConfigVo.COMM_02);
|
|
|
- alarm.setAlarmTarget(nodeVo.getKey());
|
|
|
- alarm.setAlarmValue(remoteIpAddr);
|
|
|
- this.dbmsProcess.add(alarm, (int) Thread.currentThread().getId());
|
|
|
- }
|
|
|
-
|
|
|
- TsiChannelSession.sessionTimeout(nodeVo, NettyUtils.getRemoteIpAddress(ctx.channel()));
|
|
|
-
|
|
|
- ctx.channel().close();
|
|
|
- }
|
|
|
- } else if (evt.state() == IdleState.WRITER_IDLE) {
|
|
|
- log.error("{}.userEventTriggered: WRITER_IDLE: {}", this.getClass().getSimpleName(), NettyUtils.getAddress(ctx.channel()));
|
|
|
- } else if (evt.state() == IdleState.ALL_IDLE) {
|
|
|
- log.error("{}.userEventTriggered: ALL_IDLE: {}", this.getClass().getSimpleName(), NettyUtils.getAddress(ctx.channel()));
|
|
|
- }
|
|
|
+ this.connectionService.handleIdleStateEvent(ctx, (IdleStateEvent) e);
|
|
|
}
|
|
|
ctx.fireUserEventTriggered(e);
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
|
|
- TsiNodeVo nodeVo = this.sessionManager.getNodeVo(ctx.channel());
|
|
|
- if (nodeVo != null) {
|
|
|
- log.error("{}.exceptionCaught: {}, {}", this.getClass().getSimpleName(), nodeVo.getNodeId(), ctx.channel());
|
|
|
- }
|
|
|
- log.error("{}.exceptionCaught: {}", this.getClass().getSimpleName(), NettyUtils.getAddress(ctx.channel()));
|
|
|
+ this.connectionService.handleExceptionCaught(ctx, cause);
|
|
|
ctx.fireExceptionCaught(cause);
|
|
|
ctx.channel().close();
|
|
|
}
|