IOT/MiddlePlatform
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
rqian
2026-06-10 16:31:39 +08:00
parent 1ece1b2336
commit 9657fb42f6
11 changed files with 463 additions and 194 deletions
Download Patch File Download Diff File Expand all files Collapse all files

124
maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java
View File

@@ -1,12 +1,17 @@
package com.maibu.core.business.device;
import java.time.LocalDateTime;
import java.util.LinkedList;
import java.util.Queue;
import org.springframework.util.CollectionUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.maibu.core.business.DevicePlanTask;
import com.maibu.core.business.DeviceRunStatistics;
import com.maibu.core.business.PlanPath;
import com.maibu.core.business.WorkRecord;
import com.maibu.core.business.path.LatAndLngEntity;
import com.maibu.core.business.path.RoutePlanSendEntity;
import com.maibu.core.enums.ConnectorType;
import com.maibu.core.enums.DeviceTaskStaus;
@@ -15,35 +20,28 @@ import com.maibu.mapper.WorkRecordMapper;
import com.maibu.memory.GlobalMemory;
import com.maibu.memory.SiteMemory;
import com.maibu.utils.spring.SpringUtils;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;
import lombok.Data;
import lombok.EqualsAndHashCode;
import com.maibu.core.business.path.LatAndLngEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.CollectionUtils;
import java.time.LocalDateTime;
import java.util.LinkedList;
import java.util.Queue;
@Slf4j
@EqualsAndHashCode(callSuper = true)
@Data
public class NettyDevice extends Connector {
private ConnectorType deviceType; // 设备角色类型：MASTER、SLAVE、GENERAL
private ConnectorType deviceType; // 设备角色类型：MASTER、SLAVE、GENERAL
private String currentSlaveId;// 当前对应的下位机 master
private Long currentTaskId; //当前执行的任务;
private Long currentTaskId; // 当前执行的任务;
private DevicePlanTask task; //当前执行的任务;
private DevicePlanTask task; // 当前执行的任务;
private Integer onlineStatus; //在线状态 0 离线 1在线
private Integer onlineStatus; // 在线状态 0 离线 1在线
private LatAndLngEntity currentPoint; //执行任务的当前点
private LatAndLngEntity currentPoint; // 执行任务的当前点
private Queue<LatAndLngEntity> locationQueue = new LinkedList<>();
@@ -69,41 +67,15 @@ public class NettyDevice extends Connector {
this.deviceType = deviceType;
}
// abaa 00 18 06 18 06 00 00 00 00 00 00 aaab
//
// ab aa 01 01 00 3c 88
// 2d e0 48 05 40 40
// 22 07 d0 42 c6 33
// 5e 40 e8 03 00 00 aa ab
private static final byte[] PACKET = {(byte) 0xAB, (byte) 0xAA,
(byte) 0x00, (byte) 0x18,(byte) 0x06,(byte) 0x18,(byte) 0x06,
(byte) 0x00,(byte) 0x00,(byte) 0x00,(byte) 0x00,(byte) 0x00,(byte) 0x00,
(byte) 0xAA, (byte) 0xAB};
private static final byte[] route = {(byte) 0xab, (byte) 0xaa,
(byte) 0x01, (byte) 0x01,(byte) 0x00,(byte) 0x3c,(byte) 0x88,
(byte) 0x2d, (byte) 0xe0,(byte) 0x48,(byte) 0x05,(byte) 0x40,(byte) 0x40,
(byte) 0x22, (byte) 0x07,(byte) 0xd0,(byte) 0x42,(byte) 0xc6,(byte) 0x33,
(byte) 0x5e,(byte) 0x40,(byte) 0xe8,(byte) 0x03,(byte) 0x00,(byte) 0x00,
(byte) 0xaa, (byte) 0xab};
//todo 重发
// todo 重发
public void sendNextPoint() {
if (task.getTaskStaus().equals(DeviceTaskStaus.PAUSE)) return;
if (task.getTaskStaus().equals(DeviceTaskStaus.PAUSE))
return;
if (!locationQueue.isEmpty()) {
LatAndLngEntity entity = locationQueue.poll();
if (entity != null) {
RoutePlanSendEntity routePlanSendEntity = new RoutePlanSendEntity();
routePlanSendEntity.setCommandType((byte) 0x01);
routePlanSendEntity.setPointCounts((short) 1);
routePlanSendEntity.setTargetLatitude(entity.getLat());
routePlanSendEntity.setTargetLongitude(entity.getLng());
routePlanSendEntity.setSpeed((short) 1000);
byte[] bytes = routePlanSendEntity.toBytes();
log.debug("下发路径的报文：{}", toHexString(bytes));
sendPathCommand((byte) 0x01, (short) 1, entity.getLat(), entity.getLng(), (short) 1000);
currentPoint = entity;
if (this.getChannel() != null && this.getChannel().isActive()) {
this.getChannel().writeAndFlush(Unpooled.wrappedBuffer(bytes));
}
}
}
}
@@ -116,21 +88,22 @@ public class NettyDevice extends Connector {
return sb.toString().trim();
}
public synchronized void finishTask() {
if (executingTask) {
task.setTaskStaus(DeviceTaskStaus.FINISH);
task.setFinishTime(LocalDateTime.now());
devicePlanTaskMapper.saveOrUpdate(task);
SiteMemory siteMemory = GlobalMemory.getSiteMemory(task.getOrgId(), task.getSiteId());
siteMemory.removeDevicePlanExecute(task.getDeviceId());
task = null;
currentTaskId = null;
executingTask = false;
}
public synchronized Long finishTask() {
long finishId = task.getId();
task.setTaskStaus(DeviceTaskStaus.FINISH);
task.setFinishTime(LocalDateTime.now());
devicePlanTaskMapper.saveOrUpdate(task);
SiteMemory siteMemory = GlobalMemory.getSiteMemory(task.getOrgId(), task.getSiteId());
siteMemory.removeDevicePlanExecute(task.getDeviceId());
task = null;
currentTaskId = null;
executingTask = false;
return finishId;
}
public synchronized boolean cancelTask() {
if (executingTask) {
if (task != null) {
sendPathCommand((byte) 0x01, (short) 0, 0.00d, 0.00d, (short) 0);
task.setTaskStaus(DeviceTaskStaus.CANCELED);
task.setUpdateTime(LocalDateTime.now());
devicePlanTaskMapper.saveOrUpdate(task);
@@ -141,17 +114,30 @@ public class NettyDevice extends Connector {
currentTaskId = null;
locationQueue.clear();
executingTask = false;
return true;
} else {
return false;
}
return true;
}
public void sendPathCommand(byte coommadnType, short pointCounts, double lat, double lng, short speed) {
RoutePlanSendEntity routePlanSendEntity = new RoutePlanSendEntity();
routePlanSendEntity.setCommandType(coommadnType);
routePlanSendEntity.setPointCounts(pointCounts);
routePlanSendEntity.setTargetLatitude(lat);
routePlanSendEntity.setTargetLongitude(lng);
routePlanSendEntity.setSpeed(speed);
byte[] bytes = routePlanSendEntity.toBytes();
log.debug("下发路径的报文：{}", toHexString(bytes));
if (this.getChannel() != null && this.getChannel().isActive()) {
this.getChannel().writeAndFlush(Unpooled.wrappedBuffer(bytes));
}
}
//todo 优化
// todo 优化
public synchronized boolean pauseTask() {
if (executingTask) {
task.setTaskStaus(DeviceTaskStaus.PAUSE);
executingTask = false;
sendPathCommand((byte) 0x01, (short) 2, 0.00d, 0.00d, (short) 0);
// executingTask = false;
return true;
} else {
return false;
@@ -159,22 +145,25 @@ public class NettyDevice extends Connector {
}
//todo 优化
public synchronized boolean recoveryTask() {
if (!executingTask) {
// todo 优化
public synchronized boolean recoveryTask() throws InterruptedException {
if (task.getTaskStaus().equals(DeviceTaskStaus.PAUSE)) {
task.setTaskStaus(DeviceTaskStaus.EXECUTING);
executingTask = true;
sendNextPoint();
sendPathCommand((byte) 0x01, (short) 3, 0.00d, 0.00d, (short) 0);
Thread.sleep(1000);
sendPathCommand((byte) 0x01, (short) 1, currentPoint.getLat(), currentPoint.getLng(), (short) 0);
// sendNextPoint();
return true;
} else {
return false;
}
}
public void startTask() {
try {
if (executingTask) return;
if (executingTask)
return;
if (task != null) {
Long routeId = task.getRouteId();
WorkRecord workRecord = workRecordMapper.selectById(routeId);
@@ -196,4 +185,3 @@ public class NettyDevice extends Connector {
}
}

6
maibu-common/src/main/java/com/maibu/core/enums/RespondCode.java
View File

@@ -13,5 +13,9 @@ public enum RespondCode {
/**
* 设备错误推送
*/
device_error_push
device_error_push,
/**
* 任务状态变化推送
*/
device_task_change
}

2
maibu-netty-server/src/main/java/com/maibu/controller/DeviceTaskController.java
View File

@@ -137,7 +137,7 @@ public class DeviceTaskController extends BaseController {
}
@PostMapping("/recoveryTask")
public AjaxResult recoveryTask(@RequestBody DeviceTaskCommandDTO dto) {
public AjaxResult recoveryTask(@RequestBody DeviceTaskCommandDTO dto) throws InterruptedException {
String deviceId = dto.getDeviceId();
if (StringUtils.isEmpty(deviceId) && dto.getTaskId() == null) {
return AjaxResult.error("deviceId或taskId为空!");

18
maibu-netty-server/src/main/java/com/maibu/dto/DeviceTaskReportDTO.java Normal file
View File

@@ -0,0 +1,18 @@
package com.maibu.dto;
import com.maibu.core.enums.DeviceTaskStaus;
import com.maibu.core.enums.RespondCode;
import lombok.Data;
@Data
public class DeviceTaskReportDTO {
private RespondCode event;
private String deviceId;
private Long taskId;
private DeviceTaskStaus status;
}

112
maibu-netty-server/src/main/java/com/maibu/netty/handler/DeviceConnectHandler.java
View File

@@ -8,7 +8,6 @@ import com.maibu.core.business.path.LatAndLngEntity;
import com.maibu.core.domain.entity.SysUser;
import com.maibu.core.domain.entity.UserClient;
import com.maibu.core.enums.*;
import com.maibu.core.enums.DeviceStatus;
import com.maibu.dto.*;
import com.maibu.manager.DeviceSessionManager;
import com.maibu.mapper.DeviceRunStatisticsMapper;
@@ -46,7 +45,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
private static final Logger logger = LoggerFactory.getLogger(DeviceConnectHandler.class);
//记录切换控制权请求来源 key 是请求的masterId,value是控制当前设备的主机
// 记录切换控制权请求来源 key 是请求的masterId,value是控制当前设备的主机
private final ConcurrentHashMap<String, String> controlRequestMap = new ConcurrentHashMap<>();
@Autowired
@@ -73,7 +72,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
logger.info("Channel active: {}", ctx.channel().remoteAddress());
ctx.fireChannelActive();
//todo 更新单次统计消息
// todo 更新单次统计消息
}
@@ -95,12 +94,12 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
nettyDevice.setCurrentSlaveId(null);
sysUserClientService.updateAliveStatus(deviceId, "", 0);
logger.info("{}: 设备断开 status: {}", deviceId, ConnectorStatus.DISABLED);
// updateLogoutSingle(deviceId);
//todo 自动切换到其他？
// updateLogoutSingle(deviceId);
// todo 自动切换到其他？
}
if (nettyDevice != null && nettyDevice.getDeviceType() == ConnectorType.SLAVE) {
nettyDevice.setOnlineStatus(0);
//推送json 找到设备绑定的主机进行推送
// 推送json 找到设备绑定的主机进行推送
List<NettyDevice> controlMasters = sessionManager.getAllSlaveControl(deviceId);
if (!CollectionUtils.isEmpty(controlMasters)) {
controlMasters.forEach(x -> {
@@ -122,7 +121,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
device.setOnlineStatus(DeviceStatus.offline.getType());
device.setUpdateTime(LocalDateTime.now());
deviceService.updateDeviceBySelf(device);
//todo 更新历史作业消息
// todo 更新历史作业消息
}
if (nettyDevice.getLatestStatus() != null) {
nettyDevice.getLatestStatus().setEndTime(LocalDateTime.now());
@@ -153,7 +152,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
if (!StringUtils.isEmpty(data)) {
if (data.contains(":")) {
logger.info("上位机连接 data:{}", data);
//判断 是否更新连接了 更新连接就返回失败
// 判断 是否更新连接了 更新连接就返回失败
String[] parts = data.split(":");
if (parts.length == 3) {
String deviceId = parts[0] + ":" + parts[1];
@@ -167,37 +166,41 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
if (sysUser != null) {
sysUserClientService.insertUserClient(sysUser.getUserId(), userName, deviceId, "", 0);
}
//如果 当前角色存在sessionId同时和当前的穿过来不同则下发剔除指令
// 如果 当前角色存在sessionId同时和当前的穿过来不同则下发剔除指令
boolean t = false;
if (StringUtils.isEmpty(sessionId) || sysUser == null) {
t = true;
}
if (sysUser != null && !StringUtils.isEmpty(sessionId) && !StringUtils.isEmpty(sysUser.getSessionId())) {
if (sysUser != null && !StringUtils.isEmpty(sessionId)
&& !StringUtils.isEmpty(sysUser.getSessionId())) {
if (!sysUser.getSessionId().equals(sessionId)) {
t = true;
}
}
if (t) {
//发送已登录状态到客户端
// 发送已登录状态到客户端
if ("web".equals(parts[1])) {
NettyDevice device = sessionManager.getDevice(deviceId);
logger.info("web 上位机连接重连: {}", JsonUtils.toJsonString(device));
if (device != null && !StringUtils.isEmpty(device.getCurrentSlaveId())) {
logger.info("上位机连接重连: {}", deviceId);
} else {
//判断是否账号在控制空 如果在控制设置账号为非active
List<UserClient> userClients = sysUserClientService.selectUserClientByUserName(userName);
// 判断是否账号在控制空 如果在控制设置账号为非active
List<UserClient> userClients = sysUserClientService
.selectUserClientByUserName(userName);
ConnectorStatus status = ConnectorStatus.ACTIVE;
logger.info("web 上位机连接重连2: {}", JsonUtils.toJsonString(userClients));
UserClient nowClient = null;
if (!CollectionUtils.isEmpty(userClients)) {
long count = userClients.stream().filter(x ->
x.getIsAlive() == 1 && !deviceId.equals(x.getClientName())).count();
long count = userClients.stream().filter(
x -> x.getIsAlive() == 1 && !deviceId.equals(x.getClientName()))
.count();
if (count > 0) {
status = ConnectorStatus.RECEIVER;
}
nowClient = userClients.stream().filter(x ->
x.getIsAlive() == 1 && deviceId.equals(x.getClientName())).findFirst().orElse(null);
nowClient = userClients.stream().filter(
x -> x.getIsAlive() == 1 && deviceId.equals(x.getClientName()))
.findFirst().orElse(null);
}
logger.info("web 上位机连接重连3: {}", JsonUtils.toJsonString(status));
if (!ConnectorStatus.ACTIVE.equals(status)) {
@@ -220,22 +223,24 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
ctx.channel().writeAndFlush(buf);
}
} else {
//判断是否账号在控制空 如果在控制设置账号为非active
List<UserClient> userClients = sysUserClientService.selectUserClientByUserName(userName);
// 判断是否账号在控制空 如果在控制设置账号为非active
List<UserClient> userClients = sysUserClientService
.selectUserClientByUserName(userName);
ConnectorStatus status = ConnectorStatus.ACTIVE;
if (!CollectionUtils.isEmpty(userClients)) {
long count = userClients.stream().filter(x ->
x.getIsAlive() == 1 && !deviceId.equals(x.getClientName())).count();
long count = userClients.stream()
.filter(x -> x.getIsAlive() == 1 && !deviceId.equals(x.getClientName()))
.count();
if (count > 0) {
status = ConnectorStatus.RECEIVER;
}
}
logger.info("上位机上线:{}", status);
// if (ConnectorStatus.ACTIVE.equals(status)) {
// //todo 设置当前控制的token
// sysUser.setSessionId(sessionId);
// userService.updateUserProfile(sysUser);
// }
// if (ConnectorStatus.ACTIVE.equals(status)) {
// //todo 设置当前控制的token
// sysUser.setSessionId(sessionId);
// userService.updateUserProfile(sysUser);
// }
handleMasterConnect(ctx, deviceId, null, status);
}
}
@@ -266,7 +271,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
if (slaveDevice != null) {
slaveDevice.setOnlineStatus(1);
}
//推送json 找到设备绑定的主机进行推送
// 推送json 找到设备绑定的主机进行推送
List<NettyDevice> controlMasters = sessionManager.getAllSlaveControl(device.getSerialNumber());
if (!CollectionUtils.isEmpty(controlMasters)) {
controlMasters.forEach(x -> {
@@ -304,17 +309,37 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
}
}
} else if (cmdType == CommandConstant.path) {
//路径下发完成
if(bytes.length < 11){
//01 表示完成
// 路径下发完成
if (bytes.length < 11) {
// 01 表示完成
if (bytes[5] == (byte) 0x01) {
Channel channel = ctx.channel();
String deviceId = channel.attr(Constant.ATT_DEVICE_ID).get();
NettyDevice device = sessionManager.getDevice(deviceId);
if(CollectionUtils.isEmpty(device.getLocationQueue())) {
device.finishTask();
}else{
if (CollectionUtils.isEmpty(device.getLocationQueue())) {
Long taskId = device.finishTask();
// todo 推送信息给前端
DeviceTaskReportDTO changeDTO = new DeviceTaskReportDTO();
changeDTO.setDeviceId(deviceId);
changeDTO.setStatus(DeviceTaskStaus.FINISH);
changeDTO.setEvent(RespondCode.device_task_change);
changeDTO.setTaskId(taskId);
// ByteBuf buf = ctx.alloc().buffer();
ByteBuf buf = Unpooled.buffer();
CommandUtils.buildCommand(buf, changeDTO, CommandConstant.interaction);
List<NettyDevice> controlMasters = sessionManager
.getAllSlaveControl(deviceId);
if (!CollectionUtils.isEmpty(controlMasters)) {
controlMasters.forEach(x -> {
if (x.getChannel() != null && x.getChannel().isActive()) {
x.getChannel().writeAndFlush(buf);
logger.info("任务完成状态推送 task:{},device:{}", taskId, deviceId);
}
});
}
} else {
device.sendNextPoint();
//todo 推送到点给前端
}
}
}
@@ -326,23 +351,23 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
}
/**
* 处理反馈 待扩展
* 处理反馈 待扩展
*/
private void responseDispatch(DeviceRespondDTO respond, byte[] bytes) {
Boolean switchResult = respond.getSwitchResult();
String slaveId = respond.getDeviceId();
//找到请求的 设备 他这个是先记录切换请求做个绑定关系,在同意或者反对时能直接找到对应 请求的设备 直接发送报文
// 找到请求的 设备 他这个是先记录切换请求做个绑定关系,在同意或者反对时能直接找到对应 请求的设备 直接发送报文
String masterId = getRequestMaster(slaveId);
if (switchResult) {
List<NettyDevice> controlList = sessionManager.getSlaveControl(slaveId);
NettyDevice masterDevice = sessionManager.getDevice(masterId);
NettyDevice slaveDevice = sessionManager.getDevice(slaveId);
if (!CollectionUtils.isEmpty(controlList)) {
//如果当前设备有控制
// 如果当前设备有控制
controlList.forEach(c -> {
logger.info("12指令切换权限，c:{}", c.getConnectorId());
if (!c.getConnectorId().equals(masterId)) {
// c.setCurrentSlaveId(null);
// c.setCurrentSlaveId(null);
c.removeConnectedDevice(slaveDevice);
c.setConnectorStatus(ConnectorStatus.RECEIVER);
sysUserClientService.updateAliveStatus(c.getConnectorId(), slaveId, 2);
@@ -367,7 +392,8 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
* 获取当前请求目标设备的主机
*/
private String getRequestMaster(String slaveId) {
return controlRequestMap.keySet().stream().filter(x -> controlRequestMap.get(x).equals(slaveId)).findFirst().orElse(null);
return controlRequestMap.keySet().stream().filter(x -> controlRequestMap.get(x).equals(slaveId)).findFirst()
.orElse(null);
}
/**
@@ -393,7 +419,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
String masterId = requestDTO.getUserId() + ":" + requestDTO.getPlatform();
logger.info("转发控制请求 controlList :{}", JsonUtils.toJsonString(controlList));
if (!CollectionUtils.isEmpty(controlList)) {
//如果当前设备有控制
// 如果当前设备有控制
logger.info("转发控制请求 发送报文");
NettyDevice controlDevice = controlList.get(0);
if (controlDevice != null && controlDevice.getChannel() != null && controlDevice.getChannel().isActive()) {
@@ -404,7 +430,7 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
}
} else {
//判断直接给权限 还是 判断是否有该账号其他端的的控制有权限
// 判断直接给权限 还是 判断是否有该账号其他端的的控制有权限
NettyDevice activeDevice = sessionManager.getUserControl(requestDTO.getUserId());
boolean sendMes = true;
if (activeDevice == null) {
@@ -456,13 +482,13 @@ public class DeviceConnectHandler extends SimpleChannelInboundHandler<String> {
sessionManager.registerDevice(slaveId, ctx.channel(), ConnectorType.SLAVE, ConnectorStatus.ACTIVE, null);
}
private void handleMasterConnect(ChannelHandlerContext ctx, String masterId, String slaveId, ConnectorStatus status) {
private void handleMasterConnect(ChannelHandlerContext ctx, String masterId, String slaveId,
ConnectorStatus status) {
ctx.channel().attr(Constant.ATT_DEVICE_ID).set(masterId);
sessionManager.registerDevice(masterId, ctx.channel(), ConnectorType.MASTER, status, slaveId);
// sessionManager.bindConnector(masterId, slaveId);
// sessionManager.bindConnector(masterId, slaveId);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
String deviceId = ctx.channel().attr(Constant.ATT_DEVICE_ID).get();

1
maibu-netty-server/src/main/java/com/maibu/service/DevicePlanTaskMonitorService.java
View File

@@ -230,6 +230,7 @@ public class DevicePlanTaskMonitorService {
DevicePlanTask devicePlanTask = tasks.get(0);
String deviceId = devicePlanTask.getDeviceId();
if (!StringUtils.isEmpty(deviceId) && deviceId.equals(key)) {
devicePlanTask.setTaskStaus(DeviceTaskStaus.EXECUTING);
siteMemory.addDevicePlanExecute(devicePlanTask);
device.setCurrentTaskId(devicePlanTask.getId());
device.setTask(devicePlanTask);

93
maibu-netty-server/src/main/java/com/maibu/service/DeviceTaskService.java
View File

@@ -21,6 +21,7 @@ import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.Random;
@@ -42,14 +43,12 @@ public class DeviceTaskService {
@Autowired
private WorkRecordMapper workRecordMapper;
@Autowired
private DeviceSessionManager deviceSessionManager;
@Autowired
private SysSiteMapper sysSiteMapper;
public int insertOrUpdate(DevicePlan devicePlan, LoginUser loginUser) {
Long id = devicePlan.getId();
SiteMemory siteMemory = GlobalMemory.getSiteMemory(devicePlan.getOrgId(), devicePlan.getSiteId());
@@ -116,7 +115,8 @@ public class DeviceTaskService {
}
}
public List<DevicePlan> getSiteDevicePlan(Long siteId, LocalDate startTime, LocalDate endTime, String deviceId, DeviceTaskStaus taskStaus) {
public List<DevicePlan> getSiteDevicePlan(Long siteId, LocalDate startTime, LocalDate endTime, String deviceId,
DeviceTaskStaus taskStaus) {
if (siteId != null) {
LambdaQueryWrapper<DevicePlan> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(DevicePlan::getSiteId, siteId);
@@ -206,10 +206,10 @@ public class DeviceTaskService {
List<DevicePlanStatisticsDTO> list = new ArrayList<>();
LocalDate today = LocalDate.now();
LocalDateTime monthStart = LocalDateTime.of(
today.getYear(), // 当前年
today.getMonth(), // 当前月
1, // 当月第一天
0, 0, 0, 0 // 时:分:秒:纳秒（00:00:00.000）
today.getYear(), // 当前年
today.getMonth(), // 当前月
1, // 当月第一天
0, 0, 0, 0 // 时:分:秒:纳秒（00:00:00.000）
);
LocalDate lastDayOfMonth = today.withDayOfMonth(today.lengthOfMonth()); // 获取当月最后一天
LocalDateTime monthEnd = LocalDateTime.of(
@@ -226,7 +226,8 @@ public class DeviceTaskService {
if (!CollectionUtils.isEmpty(devicePlans)) {
List<LocalDate> dates = getMonthDays();
for (LocalDate date : dates) {
List<DevicePlan> plans = devicePlans.stream().filter(x -> x.getCreateTime().toLocalDate().equals(date)).collect(Collectors.toList());
List<DevicePlan> plans = devicePlans.stream().filter(x -> x.getCreateTime().toLocalDate().equals(date))
.collect(Collectors.toList());
if (!CollectionUtils.isEmpty(plans)) {
DevicePlanStatisticsDTO dto = new DevicePlanStatisticsDTO();
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("M.d");
@@ -260,10 +261,10 @@ public class DeviceTaskService {
LocalDate today = LocalDate.now();
// 2. 构造当前月份的起始时间：当月第一天 00:00:00
LocalDateTime monthStart = LocalDateTime.of(
today.getYear(), // 当前年
today.getMonth(), // 当前月
1, // 当月第一天
0, 0, 0, 0 // 时:分:秒:纳秒（00:00:00.000）
today.getYear(), // 当前年
today.getMonth(), // 当前月
1, // 当月第一天
0, 0, 0, 0 // 时:分:秒:纳秒（00:00:00.000）
);
// 3. 构造当前月份的结束时间：当月最后一天 23:59:59.999
LocalDate lastDayOfMonth = today.withDayOfMonth(today.lengthOfMonth()); // 获取当月最后一天
@@ -277,9 +278,11 @@ public class DeviceTaskService {
LambdaQueryWrapper<DevicePlan> query = new LambdaQueryWrapper<>();
query.eq(DevicePlan::getUserId, userId);
List<DevicePlan> devicePlans = transferDevicePlanData(devicePlanMapper.selectList(query));
if (CollectionUtils.isEmpty(devicePlans)) return new ArrayList<>();
if (CollectionUtils.isEmpty(devicePlans))
return new ArrayList<>();
List<Long> planIds = devicePlans.stream().map(DevicePlan::getId).collect(Collectors.toList());
if (CollectionUtils.isEmpty(planIds)) return new ArrayList<>();
if (CollectionUtils.isEmpty(planIds))
return new ArrayList<>();
LambdaQueryWrapper<DevicePlanTask> queryWrapper = new LambdaQueryWrapper<>();
// 核心：筛选 createTime 大于等于起始时间，且小于等于结束时间
@@ -290,16 +293,20 @@ public class DeviceTaskService {
if (!CollectionUtils.isEmpty(devicePlanTasks)) {
List<LocalDate> dates = getMonthDays();
for (LocalDate date : dates) {
List<DevicePlanTask> planTasks = devicePlanTasks.stream().filter(x -> x.getCreateTime().toLocalDate().equals(date)).collect(Collectors.toList());
List<DevicePlanTask> planTasks = devicePlanTasks.stream()
.filter(x -> x.getCreateTime().toLocalDate().equals(date)).collect(Collectors.toList());
if (!CollectionUtils.isEmpty(planTasks)) {
DevicePlanTaskStatisticsDTO dto = new DevicePlanTaskStatisticsDTO();
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("M.d");
dto.setDate(formatter.format(date));
long finishCount = planTasks.stream().filter(x -> DeviceTaskStaus.FINISH.equals(x.getTaskStaus())).count();
long finishCount = planTasks.stream().filter(x -> DeviceTaskStaus.FINISH.equals(x.getTaskStaus()))
.count();
dto.setSuccess((int) finishCount);
long failedCount = planTasks.stream().filter(x -> DeviceTaskStaus.FAILED.equals(x.getTaskStaus())).count();
long failedCount = planTasks.stream().filter(x -> DeviceTaskStaus.FAILED.equals(x.getTaskStaus()))
.count();
dto.setFailed((int) failedCount);
long pauseCount = planTasks.stream().filter(x -> DeviceTaskStaus.PAUSE.equals(x.getTaskStaus())).count();
long pauseCount = planTasks.stream().filter(x -> DeviceTaskStaus.PAUSE.equals(x.getTaskStaus()))
.count();
dto.setPaused((int) pauseCount);
list.add(dto);
}
@@ -308,7 +315,6 @@ public class DeviceTaskService {
return list;
}
public WorkRecord concurrentDevicePlanTask(String deviceId) {
LambdaQueryWrapper<DevicePlanTask> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(DevicePlanTask::getDeviceId, deviceId)
@@ -318,9 +324,8 @@ public class DeviceTaskService {
return routId == null ? null : workRecordMapper.selectById(routId);
}
public List<DeviceWorkStatisticsDTO> workStatistics(Long userId) {
//todo update
// todo update
List<DeviceWorkStatisticsDTO> result = new ArrayList<>();
Random random = new Random();
@@ -341,7 +346,6 @@ public class DeviceTaskService {
return result;
}
public List<DevicePlanTask> deviceTaskPool(DeviceTaskQueryDTO dto) {
List<DevicePlanTask> list = new ArrayList<>();
if (dto.getOrgId() == null) {
@@ -394,9 +398,9 @@ public class DeviceTaskService {
return transferDeviceTaskData(devicePlanTaskMapper.selectList(queryWrapper));
}
public List<DevicePlan> transferDevicePlanData(List<DevicePlan> list) {
if (CollectionUtils.isEmpty(list)) return list;
if (CollectionUtils.isEmpty(list))
return list;
list.forEach(x -> {
DeviceTaskStaus taskStaus = x.getTaskStaus();
x.setTaskStausTranslate(taskStaus.getDescription());
@@ -405,11 +409,17 @@ public class DeviceTaskService {
}
public List<DevicePlanTask> transferDeviceTaskData(List<DevicePlanTask> list) {
if (CollectionUtils.isEmpty(list)) return list;
if (CollectionUtils.isEmpty(list))
return list;
list.forEach(x -> {
DeviceTaskStaus taskStaus = x.getTaskStaus();
x.setTaskStausTranslate(taskStaus.getDescription());
});
list.sort(
Comparator.comparing(DevicePlanTask::getFinishTime,
Comparator.nullsLast(Comparator.reverseOrder()))
.thenComparing(DevicePlanTask::getCreateTime,
Comparator.nullsLast(Comparator.reverseOrder())));
return list;
}
@@ -443,8 +453,8 @@ public class DeviceTaskService {
}
} else {
Long taskId = dto.getTaskId();
// Long siteId = dto.getSiteId();
// Long orgId = dto.getOrgId();
// Long siteId = dto.getSiteId();
// Long orgId = dto.getOrgId();
DevicePlanTask devicePlanTask = devicePlanTaskMapper.selectById(taskId);
if (devicePlanTask != null) {
NettyDevice device = deviceSessionManager.getDevice(devicePlanTask.getDeviceId());
@@ -456,7 +466,7 @@ public class DeviceTaskService {
return false;
}
//todo 完善
// todo 完善
public boolean pauseTask(DeviceTaskCommandDTO dto, String username) {
if (!StringUtils.isEmpty(dto.getDeviceId())) {
NettyDevice device = deviceSessionManager.getDevice(dto.getDeviceId());
@@ -467,18 +477,17 @@ public class DeviceTaskService {
Long taskId = dto.getTaskId();
DevicePlanTask devicePlanTask = devicePlanTaskMapper.selectById(taskId);
if (devicePlanTask != null) {
devicePlanTask.setUpdateTime(LocalDateTime.now());
devicePlanTask.setTaskStaus(DeviceTaskStaus.PAUSE);
devicePlanTask.setUpdateBy(username);
devicePlanTaskMapper.saveOrUpdate(devicePlanTask);
NettyDevice device = deviceSessionManager.getDevice(devicePlanTask.getDeviceId());
if (device != null) {
return device.pauseTask();
}
}
}
return false;
}
//todo 完善
public boolean recoveryTask(DeviceTaskCommandDTO dto, String username) {
// todo 完善
public boolean recoveryTask(DeviceTaskCommandDTO dto, String username) throws InterruptedException {
if (!StringUtils.isEmpty(dto.getDeviceId())) {
NettyDevice device = deviceSessionManager.getDevice(dto.getDeviceId());
if (device != null) {
@@ -486,16 +495,12 @@ public class DeviceTaskService {
}
} else {
Long taskId = dto.getTaskId();
Long siteId = dto.getSiteId();
Long orgId = dto.getOrgId();
DevicePlanTask devicePlanTask = devicePlanTaskMapper.selectById(taskId);
SiteMemory siteMemory = GlobalMemory.getSiteMemory(orgId, siteId);
DevicePlanTask executeTask = siteMemory.getDeviceExecuteTask(devicePlanTask.getDeviceId());
if (Objects.equals(devicePlanTask.getId(), executeTask.getId())) {
devicePlanTask.setUpdateTime(LocalDateTime.now());
devicePlanTask.setTaskStaus(DeviceTaskStaus.EXECUTING);
devicePlanTask.setUpdateBy(username);
devicePlanTaskMapper.saveOrUpdate(devicePlanTask);
if (devicePlanTask != null) {
NettyDevice device = deviceSessionManager.getDevice(devicePlanTask.getDeviceId());
if (device != null) {
return device.recoveryTask();
}
}
}
return false;

202
maibu-service/maibu-system-service/src/main/java/com/maibu/service/impl/DeviceProductServiceImpl.java
View File

File diff suppressed because one or more lines are too long

1
maibu-web-middleware/src/main/java/com/maibu/dto/MesType.java
View File

@@ -7,6 +7,7 @@ public enum MesType {
have_logged_in,//消息
device_status_changed,//上下线状态变化
device_error_push, //错误推送
device_task_change, //任务状态变化推送
heartbeat,
remoteControl,
detectionReport,//障碍物消息

15
maibu-web-middleware/src/main/java/com/maibu/dto/WebDeviceTaskStatusMessageDTO.java Normal file
View File

@@ -0,0 +1,15 @@
package com.maibu.dto;
import lombok.Data;
@Data
public class WebDeviceTaskStatusMessageDTO {
private MesType event;
private String deviceId;
private Long taskId;
private String status;
}

83
maibu-web-middleware/src/main/java/com/maibu/netty/handler/ClientHandler.java
View File

@@ -3,6 +3,7 @@ package com.maibu.netty.handler;
import cn.hutool.json.JSONObject;
import com.maibu.common.MiddleCommandConstant;
import com.maibu.common.MiddleConstant;
import com.maibu.core.enums.RespondCode;
import com.maibu.dto.*;
import com.maibu.memory.MiddleGlobalMemory;
import com.maibu.netty.NettyClient;
@@ -39,17 +40,18 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
try {
byte[] data = (byte[]) msg;
logger.debug("channelRead2: {}", Arrays.toString(data));
//假如是心跳包
// 假如是心跳包
String key = ctx.channel().attr(MiddleConstant.ATT_MASTER_KEY).get();
WebSocket webSocket = MiddleGlobalMemory.onlineSockets.get(key);
if (isHeartbeat(data)) {
//websocket 不在了停止发 心跳
// websocket 不在了停止发 心跳
if (webSocket != null && webSocket.isOpen()) {
ctx.writeAndFlush(data);
}
NettyClient nettyClient = MiddleGlobalMemory.nettyClientMap.get(key);
if (nettyClient != null && StringUtils.isEmpty(nettyClient.currentDeviceId)) {
if (nettyClient.lastSwitchTime == -1 || System.currentTimeMillis() - nettyClient.lastSwitchTime > nettyClient.Interval) {
if (nettyClient.lastSwitchTime == -1
|| System.currentTimeMillis() - nettyClient.lastSwitchTime > nettyClient.Interval) {
if (!StringUtils.isEmpty(nettyClient.requestDeviceId) && !StringUtils.isEmpty(key)) {
switchDevice(nettyClient, key);
nettyClient.lastSwitchTime = System.currentTimeMillis();
@@ -76,7 +78,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
if (StringUtils.isNotBlank(json)) {
JSONObject obj = new JSONObject(json);
if (obj.get("event") != null) {
//状态变化 上下线
// 状态变化 上下线
String eventType = obj.get("event").toString();
String deviceId = obj.get("deviceId").toString();
if ("device_status_changed".equals(eventType)) {
@@ -91,8 +93,8 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
}
WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto));
} else if ("device_error_push".equals(eventType)) {
//todo 更新推送的错误
String details = obj.get("details").toString();
// todo 更新推送的错误
// String details = obj.get("details").toString();
WebOlineStatusMessageDTO dto = new WebOlineStatusMessageDTO();
dto.setDeviceId(deviceId);
dto.setType(MesType.device_error_push);
@@ -100,6 +102,15 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
dto.setDescription("");
dto.setErrorCode("");
WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto));
} else if ("device_task_change".equals(eventType)) {
String taskId = obj.get("taskId").toString();
String status = obj.get("status").toString();
WebDeviceTaskStatusMessageDTO dto = new WebDeviceTaskStatusMessageDTO();
dto.setDeviceId(deviceId);
dto.setEvent(MesType.device_task_change);
dto.setTaskId(Long.valueOf(taskId));
dto.setStatus(status);
WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto));
}
} else {
/* 判断 respond 字段 */
@@ -107,7 +118,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
String deviceId = obj.get("deviceId").toString();
String userName = obj.get("userId").toString();
String platform = obj.get("platform").toString();
//直接转发 websocket 申请
// 直接转发 websocket 申请
WebAuthResponseDTO request = new WebAuthResponseDTO();
request.setType(MesType.switchPermission);
request.setUserName(userName);
@@ -119,44 +130,47 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
if (obj.get("respond") != null) {
if ("have_logged_in".equals(obj.get("respond").toString())) {
/* 如果回复 have_logged_in 则调用logout */
//System.out.println("检测到重复登录，执行 logout");
//NettyClient nettyClient = GlobalMemory.nettyClientMap.get(key);
//WebHasLoginDTO dto = new WebHasLoginDTO();
//dto.setToken(nettyClient.token);
//dto.setUserName(nettyClient.userName);
//dto.setType(MesType.have_logged_in);
//WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto));
// System.out.println("检测到重复登录，执行 logout");
// NettyClient nettyClient = GlobalMemory.nettyClientMap.get(key);
// WebHasLoginDTO dto = new WebHasLoginDTO();
// dto.setToken(nettyClient.token);
// dto.setUserName(nettyClient.userName);
// dto.setType(MesType.have_logged_in);
// WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto));
} else {
//todo respond netty_server返回结果处理
DeviceRespondDTO respond = JsonUtils.parseObject(obj.get("respond").toString(), DeviceRespondDTO.class);
// todo respond netty_server返回结果处理
DeviceRespondDTO respond = JsonUtils.parseObject(obj.get("respond").toString(),
DeviceRespondDTO.class);
if (respond != null) {
//直接转发 websocket 申请
// 直接转发 websocket 申请
WebSwitchControlResponseDTO responseDTO = new WebSwitchControlResponseDTO();
responseDTO.setType(MesType.switchResult);
responseDTO.setRespond(respond);
WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(responseDTO));
WebsocketHandler.sendMessageToClient(webSocket,
JsonUtils.toJsonString(responseDTO));
}
}
} else {
if (obj.get("type").toString() != null && "detectionReport".equals(obj.get("type").toString())) {
if (obj.get("type").toString() != null
&& "detectionReport".equals(obj.get("type").toString())) {
WebsocketHandler.sendMessageToClient(webSocket, json);
}
}
}
}
//error 推送
// error 推送
}
//TODO web 无切换控制请求
// TODO web 无切换控制请求
} catch (Exception e) {
logger.error("事件推送解析失败:{}", e.getMessage());
}
} else if (isStatusDate(data)) {
//状态消息推送给前端
// 状态消息推送给前端
WebStatusMessageDTO dto = createWebDeviceStatusMessage(data);
WebsocketHandler.sendMessageToClient(webSocket, JsonUtils.toJsonString(dto));
} else if (isPath(data)) {
//已到达
// 已到达
if (data[5] == (byte) 0x01) {
}
@@ -289,7 +303,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
detail17.setUnit("");
transferData.add(detail17);
//18 时间暂未给值
// 18 时间暂未给值
DeviceStatusDetail detail19 = new DeviceStatusDetail();
detail19.setName("cuttingSpeed");
@@ -324,13 +338,12 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
return transferData;
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
logger.debug("连接开始channelActive:ChannelId:{} ", ctx.channel().id().asLongText());
//websocket连上后连接netty成功后发送登录认证包
// websocket连上后连接netty成功后发送登录认证包
// key username + : web + token
String key = ctx.channel().attr(MiddleConstant.ATT_MASTER_KEY).get();
System.out.println(key);
@@ -350,11 +363,10 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
packet[tailStart + 3] = (byte) 0xAB;
ctx.writeAndFlush(packet);
// startHeartbeat(nettyClient, ctx, key);
// startHeartbeat(nettyClient, ctx, key);
}
}
public static void switchDevice(NettyClient nettyClient, String key) {
try {
// 2. 调用 POST 接口：传入 Token B（与 Token A 不同）
@@ -368,9 +380,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
if (resultDTO != null && resultDTO.getData() != null) {
if ((boolean) resultDTO.getData()) {
logger.debug("用户:{}切换设备:{}成功", key, nettyClient.requestDeviceId);
//切换成功推送 websocket
// 切换成功推送 websocket
nettyClient.currentDeviceId = nettyClient.requestDeviceId;
// nettyClient.requestDeviceId = null;
// nettyClient.requestDeviceId = null;
WebSocket webSocket = MiddleGlobalMemory.onlineSockets.get(key);
if (webSocket != null && webSocket.isOpen()) {
WebAuthResponseDTO responseDTO = new WebAuthResponseDTO();
@@ -389,7 +401,6 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
}
}
/**
* 启动心跳检测
*/
@@ -422,7 +433,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
public void channelInactive(ChannelHandlerContext ctx) {
String key = ctx.channel().attr(MiddleConstant.ATT_MASTER_KEY).get();
NettyClient nettyClient = MiddleGlobalMemory.nettyClientMap.get(key);
// stopHeartbeat(nettyClient);
// stopHeartbeat(nettyClient);
nettyClient.currentDeviceId = null;
logger.debug("连接断开channelInactive,ChannelId:{} ", ctx.channel().id().asLongText());
WebSocket webSocket = MiddleGlobalMemory.onlineSockets.get(key);
@@ -435,7 +446,7 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
String key = ctx.channel().attr(MiddleConstant.ATT_MASTER_KEY).get();
NettyClient nettyClient = MiddleGlobalMemory.nettyClientMap.get(key);
// stopHeartbeat(nettyClient);
// stopHeartbeat(nettyClient);
cause.printStackTrace();
logger.error("异常断开exceptionCaught:{}", cause.getMessage());
ctx.close();
@@ -471,9 +482,9 @@ public class ClientHandler extends ChannelInboundHandlerAdapter {
data[data.length - 2] == (byte) 0xAA &&
data[data.length - 1] == (byte) 0xAB;
// // 只要是以 '{' 开头、以 '}' 结尾，就认为是事件 JSON
// if (data == null || data.length < 2) return false;
// return data[0] == '{' && data[data.length - 1] == '}';
// // 只要是以 '{' 开头、以 '}' 结尾，就认为是事件 JSON
// if (data == null || data.length < 2) return false;
// return data[0] == '{' && data[data.length - 1] == '}';
}
private boolean isStatusDate(byte[] data) {