|
@@ -1,174 +0,0 @@
|
|
|
-package com.persagy.commons.netty.client;
|
|
|
-
|
|
|
-import cn.hutool.core.date.DateField;
|
|
|
-import cn.hutool.core.date.DateTime;
|
|
|
-import cn.hutool.core.date.DateUtil;
|
|
|
-import com.persagy.entity.CommandResult;
|
|
|
-import com.persagy.entity.NettyMessage;
|
|
|
-import com.persagy.service.CommandService;
|
|
|
-import com.persagy.utils.DateUtils;
|
|
|
-import com.persagy.utils.StringUtil;
|
|
|
-import io.netty.channel.ChannelHandlerContext;
|
|
|
-import io.netty.channel.ChannelInboundHandlerAdapter;
|
|
|
-import io.netty.handler.timeout.IdleState;
|
|
|
-import io.netty.handler.timeout.IdleStateEvent;
|
|
|
-import lombok.extern.slf4j.Slf4j;
|
|
|
-import org.apache.commons.lang3.StringUtils;
|
|
|
-import org.quartz.JobDataMap;
|
|
|
-import org.quartz.SchedulerException;
|
|
|
-import org.springframework.util.CollectionUtils;
|
|
|
-
|
|
|
-import java.time.LocalDateTime;
|
|
|
-import java.util.ArrayList;
|
|
|
-import java.util.Date;
|
|
|
-import java.util.List;
|
|
|
-import java.util.concurrent.TimeUnit;
|
|
|
-
|
|
|
-/**
|
|
|
- * @ClassName: EchoClientHandler
|
|
|
- * @Description: 客户端处理类
|
|
|
- * @author lqshi
|
|
|
- * @date 2020年8月29日 下午8:31:59
|
|
|
- */
|
|
|
-// 注意:SimpleChannelInboundHandler<ByteBuf>的<>中是什么,channelRead0第二参数是什么
|
|
|
-@Slf4j
|
|
|
-public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
|
|
|
- // Sleep 5 seconds before a reconnection attempt.
|
|
|
- static final int RECONNECT_DELAY = Integer.parseInt(System.getProperty("reconnectDelay", "5"));
|
|
|
- // Reconnect when the server sends nothing for 10 seconds.
|
|
|
- private static final int READ_TIMEOUT = Integer.parseInt(System.getProperty("readTimeout", "10"));
|
|
|
-
|
|
|
- private CommandService commandService;
|
|
|
-
|
|
|
- public GroupNettyClientHandler(CommandService commandService) {
|
|
|
- this.commandService = commandService;
|
|
|
- }
|
|
|
-
|
|
|
- public GroupNettyClientHandler() {
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
|
|
|
-
|
|
|
- if (!(evt instanceof IdleStateEvent)) {
|
|
|
- return;
|
|
|
- }
|
|
|
- IdleStateEvent e = (IdleStateEvent) evt;
|
|
|
- if (e.state() == IdleState.READER_IDLE) {
|
|
|
- System.out.println("no inbound traffic");
|
|
|
- // The connection was OK but there was no traffic for last period.
|
|
|
- // 长时间不操作的时候自动关闭连接; ctx.close();
|
|
|
- }
|
|
|
- }
|
|
|
- /**
|
|
|
- * 在到服务器的连接已经建立之后将被调用
|
|
|
- * @param ctx
|
|
|
- * @throws Exception
|
|
|
- */
|
|
|
- @Override
|
|
|
- public void channelActive(ChannelHandlerContext ctx) throws Exception {
|
|
|
- System.out.println("Connected to: " + ctx.channel().remoteAddress());
|
|
|
- //启动的时候发送消息
|
|
|
- NettyMessage msg = new NettyMessage();
|
|
|
- msg.setOpCode(3);
|
|
|
- ctx.channel().writeAndFlush(msg.toString());
|
|
|
- }
|
|
|
- /**
|
|
|
- * 当从服务器接收到一个消息时被调用
|
|
|
- * @param ctx
|
|
|
- * @param msg
|
|
|
- * @throws Exception
|
|
|
- */
|
|
|
- @Override
|
|
|
- public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
|
|
|
- System.out.println("Client received: "+ msg);
|
|
|
- try {
|
|
|
- handlerMsg(ctx, msg);
|
|
|
- } catch (Exception e) {
|
|
|
- log.error("channelRead",e);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- private void handlerMsg(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
|
|
|
- if(StringUtil.isJSONObject((String) msg)) {
|
|
|
- NettyMessage message = StringUtil.tranferItemToDTO((String) msg, NettyMessage.class);
|
|
|
- NettyMessage response = new NettyMessage();
|
|
|
- response.setStreamId(message.getStreamId());
|
|
|
- response.setSuccess(true);
|
|
|
- response.setOpCode(2);
|
|
|
- List<CommandResult> responseContent = new ArrayList<>();
|
|
|
- List<CommandResult> content = message.getContent();
|
|
|
- //删除这个时间点之后的所有工作内容
|
|
|
- String timeFlag = message.getClearBeforeTimeFlag();
|
|
|
- if(! CollectionUtils.isEmpty(content)){
|
|
|
- if(StringUtils.isNotBlank(timeFlag)) {
|
|
|
- LocalDateTime timeFlagDateTime = DateUtils.parse(timeFlag);
|
|
|
- Date endDate = DateUtils.localDateTime2Date(LocalDateTime.now().minusHours(48));
|
|
|
- List<DateTime> dateTimes = DateUtil.rangeToList(new Date(), endDate, DateField.HOUR);
|
|
|
- for (DateTime dateTime : dateTimes) {
|
|
|
- System.out.println(dateTime);
|
|
|
- }
|
|
|
- }
|
|
|
- for (CommandResult command:content){
|
|
|
- CommandResult tmpNewcommand = new CommandResult();
|
|
|
- tmpNewcommand.setId(command.getId());
|
|
|
- tmpNewcommand.setCommandResult(0);
|
|
|
- responseContent.add(tmpNewcommand);
|
|
|
- LocalDateTime commandTime = DateUtils.parse(command.getCommandTime(), DateUtils.date_format_show_minute);
|
|
|
- Date startTime = DateUtils.localDateTime2Date(commandTime);
|
|
|
- String hour = DateUtils.format(commandTime, DateUtils.sdfHour);
|
|
|
- String jobName = command.getFuncId()+"_"+command.getMeterId()+DateUtils.format(commandTime);
|
|
|
- JobDataMap jobDataMap = new JobDataMap();
|
|
|
- jobDataMap.put("commandResult",command.toString());
|
|
|
- try {
|
|
|
- commandService.addCommand(startTime,jobName,hour,jobDataMap);
|
|
|
- } catch (SchedulerException e) {
|
|
|
- log.error("addCommand error: ",e);
|
|
|
- }
|
|
|
-
|
|
|
- }
|
|
|
- }
|
|
|
- response.setContent(responseContent);
|
|
|
- System.out.println(message.toString());
|
|
|
- channelHandlerContext.write(response.toString());
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- @Override
|
|
|
- public void channelReadComplete(ChannelHandlerContext ctx) {
|
|
|
- ctx.flush();
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 在处理过程中引发异常时被调用
|
|
|
- * @param ctx
|
|
|
- * @param cause
|
|
|
- * @throws Exception
|
|
|
- */
|
|
|
- @Override
|
|
|
- public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
|
|
|
- cause.printStackTrace();
|
|
|
- ctx.close();
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 通道处于非活跃状态动作,该方法只会在失效时调用一次
|
|
|
- */
|
|
|
- @Override
|
|
|
- public void channelInactive(ChannelHandlerContext ctx) throws Exception {
|
|
|
- //客户端自己不正常情况下自己在重连一次
|
|
|
- System.out.println("Disconnected from: " + ctx.channel().remoteAddress());
|
|
|
-
|
|
|
- }
|
|
|
- @Override
|
|
|
- public void channelUnregistered(final ChannelHandlerContext ctx) throws Exception {
|
|
|
- System.out.println("Sleeping for: " + RECONNECT_DELAY + "s,Reconnecting to: " + GroupNettyClient.HOST + ':' + GroupNettyClient.PORT);
|
|
|
- ctx.channel().eventLoop().schedule(new Runnable() {
|
|
|
- @Override
|
|
|
- public void run() {
|
|
|
- System.out.println("Reconnecting to: " + GroupNettyClient.HOST + ':' + GroupNettyClient.PORT);
|
|
|
- GroupNettyClient.connect(ctx.channel());
|
|
|
- }
|
|
|
- }, RECONNECT_DELAY, TimeUnit.SECONDS);
|
|
|
- }
|
|
|
-}
|