|
|
@@ -0,0 +1,73 @@
|
|
|
+package com.inspur.netty.handler;
|
|
|
+
|
|
|
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
|
|
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
|
|
|
+import com.inspur.device.domain.SmsbDeviceLogPush;
|
|
|
+import com.inspur.device.domain.SmsbDeviceTask;
|
|
|
+import com.inspur.device.domain.SmsbDeviceTaskDetail;
|
|
|
+import com.inspur.device.domain.constants.DeviceTaskConstants;
|
|
|
+import com.inspur.device.domain.vo.SmsbDeviceTaskVo;
|
|
|
+import com.inspur.device.domain.vo.SmsbDeviceVo;
|
|
|
+import com.inspur.device.mapper.SmsbDeviceLogPushMapper;
|
|
|
+import com.inspur.device.mapper.SmsbDeviceTaskDetailMapper;
|
|
|
+import com.inspur.device.mapper.SmsbDeviceTaskMapper;
|
|
|
+import com.inspur.device.service.ISmsbDeviceService;
|
|
|
+import com.inspur.device.service.impl.SmsbDeviceServiceImpl;
|
|
|
+import com.inspur.netty.message.receive.ReceiveMessageType;
|
|
|
+import io.netty.channel.ChannelHandlerContext;
|
|
|
+import io.netty.channel.ChannelInboundHandlerAdapter;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+import org.dromara.common.core.utils.SpringUtils;
|
|
|
+
|
|
|
+/**
|
|
|
+ * 日志开始抓取回复
|
|
|
+ *
|
|
|
+ * @author lihao16
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+public class TaskLogPushStartHandler extends ChannelInboundHandlerAdapter {
|
|
|
+
|
|
|
+ private static final ISmsbDeviceService smsbDeviceService = SpringUtils.getBean(SmsbDeviceServiceImpl.class);
|
|
|
+ private static final SmsbDeviceTaskMapper smsbDeviceTaskMapper = SpringUtils.getBean(SmsbDeviceTaskMapper.class);
|
|
|
+ private static final SmsbDeviceTaskDetailMapper smsbDeviceTaskDetailMapper = SpringUtils.getBean(SmsbDeviceTaskDetailMapper.class);
|
|
|
+ private static final SmsbDeviceLogPushMapper smsbDeviceLogPushMapper = SpringUtils.getBean(SmsbDeviceLogPushMapper.class);
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
|
|
|
+ String message = (String) msg;
|
|
|
+ // 处理结束推流回复reply消息
|
|
|
+ if (message.contains(ReceiveMessageType.CONTROL_START_LOG_PUSH_REPLAY.getValue())) {
|
|
|
+ log.info("收到设备日志开始抓取回复消息:{}", message);
|
|
|
+ String identifier = message.split("/")[0];
|
|
|
+ // 1、 查询当前设备最近一条日志抓取任务
|
|
|
+ SmsbDeviceVo smsbDeviceVo = smsbDeviceService.getDeviceByIdentifier(identifier);
|
|
|
+ if (null == smsbDeviceVo) {
|
|
|
+ log.error("未查询到设备信息:{}", identifier);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ LambdaQueryWrapper<SmsbDeviceTask> lqw = Wrappers.lambdaQuery();
|
|
|
+ lqw.eq(SmsbDeviceTask::getIdentifier, identifier);
|
|
|
+ lqw.eq(SmsbDeviceTask::getTaskType, DeviceTaskConstants.DEVICE_LOG_PUSH_START);
|
|
|
+ lqw.orderByDesc(SmsbDeviceTask::getCreateTime);
|
|
|
+ lqw.last("limit 1");
|
|
|
+ SmsbDeviceTaskVo smsbDeviceTask = smsbDeviceTaskMapper.selectVoOne(lqw);
|
|
|
+ if (null == smsbDeviceTask) {
|
|
|
+ log.error("未查询到设备日志开始抓取任务:{}", identifier);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ // 2、收到回复 认定接收任务
|
|
|
+ SmsbDeviceTaskDetail taskDetail = new SmsbDeviceTaskDetail();
|
|
|
+ taskDetail.setTaskId(smsbDeviceTask.getId());
|
|
|
+ taskDetail.setTaskStatus(DeviceTaskConstants.DEVICE_TASK_STATUS_INIT);
|
|
|
+ taskDetail.setTenantId(smsbDeviceVo.getTenantId());
|
|
|
+ smsbDeviceTaskDetailMapper.insert(taskDetail);
|
|
|
+ // 3、更新抓取状态
|
|
|
+ SmsbDeviceLogPush lastLogPush = smsbDeviceLogPushMapper.selectOne(new LambdaQueryWrapper<SmsbDeviceLogPush>()
|
|
|
+ .eq(SmsbDeviceLogPush::getDeviceId, smsbDeviceVo.getId()).orderByDesc(SmsbDeviceLogPush::getId).last("limit 1"));
|
|
|
+ lastLogPush.setLogStatus(1);
|
|
|
+ smsbDeviceLogPushMapper.updateById(lastLogPush);
|
|
|
+ } else {
|
|
|
+ ctx.fireChannelRead(message);
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|