cmhi
/
dispose
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

OCT 

REM:
1. 增加处置任务管理业务
2. 修正部分单元测试用例
This commit is contained in:
HuangXin 2020-04-29 13:12:14 +08:00
parent 6f60271227
commit c73590f2fb
13 changed files with 151 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/main/java/com/dispose/controller/DisposeNodeInfoController.java
View File

@ -169,12 +169,9 @@ public class DisposeNodeInfoController {
if (devList != null && devList.size() > 0) {
devList.forEach(v -> {
VersionRsp ver = VersionRsp.builder()
.version(v.getVersion())
.build();
VersionRsp ver = VersionRsp.builder().version(v.getVersion()).build();
ver.setId(v.getId()
.toString());
ver.setId(v.getId().toString());
ver.setStatus(ErrorCode.ERR_OK.getCode());
ver.setMessage(ErrorCode.ERR_OK.getMsg());

4
src/main/java/com/dispose/dispose/impl/DPTechImpl.java
View File

@ -291,8 +291,8 @@ public class DPTechImpl implements DisposeEntryManager {
try {
return (T) cleanTypePort.getAllProtectionObjectFromUMC().getProtectionObjectDataForService();
} catch (Exception ex) {
log.error(ex.getMessage());
ex.printStackTrace();
//log.error(ex.getMessage());
//ex.printStackTrace();
return null;
}
}

3
src/main/java/com/dispose/dispose/impl/VirtualDeviceImpl.java
View File

@ -2,6 +2,7 @@ package com.dispose.dispose.impl;
import com.dispose.common.DeviceCapacity;
import com.dispose.common.ErrorCode;
import com.dispose.common.GlobalVar;
import com.dispose.common.IPAddrType;
import com.dispose.dispose.DisposeEntryManager;
import com.dispose.dispose.po.DeviceInfo;
@ -119,7 +120,7 @@ public class VirtualDeviceImpl implements DisposeEntryManager {
*/
@Override
public boolean getDeviceLinkStatus() {
return true;
return GlobalVar.USED_VIRTUAL_DISPOSE_MODE;
}
/**

8
src/main/java/com/dispose/mapper/DisposeTaskMapper.java
View File

@ -88,14 +88,6 @@ public interface DisposeTaskMapper extends Mapper<TaskInfoDetail>,
@Param("ipAddr") String ipAddr,
@Param("status") int status);
/**
* Gets all task by ip.
*
* @param ipAddr the ip addr
* @return the all task by ip
*/
List<TaskInfoDetail> getAllTaskByIp(@Param("ipAddr") String ipAddr);
/**
* Gets all task by status.
*

27
src/main/java/com/dispose/service/AsyncService.java
View File

@ -1,27 +0,0 @@
package com.dispose.service;
import com.dispose.common.DeviceCapacity;
import com.dispose.dispose.DisposeEntryManager;
/**
* The interface Async service.
*/
public interface AsyncService {
/**
* Async start dispose device task.
*
* @param dp the dp
* @param ipAddr the ip addr
* @param type the type
*/
void asyncStartDisposeDeviceTask(DisposeEntryManager dp, String ipAddr, DeviceCapacity type);
/**
* Async stop dispose device task.
*
* @param dp the dp
* @param ipAddr the ip addr
* @param type the type
*/
void asyncStopDisposeDeviceTask(DisposeEntryManager dp, String ipAddr, DeviceCapacity type);
}

17
src/main/java/com/dispose/service/TaskService.java
View File

@ -15,10 +15,10 @@ public interface TaskService {
void loadTaskFromDatabase();
/**
* Create task error code.
* Create task m return type.
*
* @param task the task
* @return the error code
* @return the m return type
*/
MReturnType<ErrorCode, Long> createTask(TaskInfoDetail task);
@ -47,7 +47,7 @@ public interface TaskService {
ErrorCode finishTask(Long taskId);
/**
* Task is running boolean.
* Task is running xx boolean.
*
* @param task the task
* @return the boolean
@ -62,6 +62,17 @@ public interface TaskService {
*/
boolean taskIsExpired(TaskInfoDetail task);
/**
* Task is exists boolean.
*
* @param devId the dev id
* @param userId the user id
* @param disposeIp the dispose ip
* @param disposeType the dispose type
* @return the boolean
*/
boolean taskIsExists(Long devId, Long userId, String disposeIp, int disposeType);
/**
* Gets node all running task.
*

25
src/main/java/com/dispose/service/impl/AsyncServiceImpl.java
View File

@ -1,25 +0,0 @@
package com.dispose.service.impl;
import com.dispose.common.DeviceCapacity;
import com.dispose.dispose.DisposeEntryManager;
import com.dispose.service.AsyncService;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
/**
* The type Async service.
*/
@Service
public class AsyncServiceImpl implements AsyncService {
@Override
@Async("bizExecutor")
public void asyncStartDisposeDeviceTask(DisposeEntryManager dp, String ipAddr, DeviceCapacity type) {
dp.runDispose(ipAddr, type);
}
@Override
@Async("bizExecutor")
public void asyncStopDisposeDeviceTask(DisposeEntryManager dp, String ipAddr, DeviceCapacity type) {
dp.stopDispose(ipAddr, type);
}
}

4
src/main/java/com/dispose/service/impl/DisposeNodeManagerImpl.java
View File

@ -123,6 +123,10 @@ public class DisposeNodeManagerImpl implements DisposeNodeManager {
try {
dp = DeviceRouter.deviceRouterFactory(dev.getType(),
dev.getIpAddr(), IPAddrType.getIpAddrType(dev.getIpAddr()));
if (!dp.getDeviceLinkStatus()) {
return new MReturnType<>(ErrorCode.ERR_NOSUCHDEVICE, String.valueOf(-1));
}
} catch (Exception ex) {
return new MReturnType<>(ErrorCode.ERR_NOSUCHDEVICE, String.valueOf(-1));
}

121
src/main/java/com/dispose/service/impl/TaskServiceImpl.java
View File

@ -10,7 +10,6 @@ import com.dispose.mapper.DisposeTaskMapper;
import com.dispose.pojo.entity.DisposeDevice;
import com.dispose.pojo.po.MReturnType;
import com.dispose.pojo.vo.common.TaskInfoDetail;
import com.dispose.service.AsyncService;
import com.dispose.service.DisposeNodeManager;
import com.dispose.service.TaskService;
import com.fasterxml.jackson.core.JsonProcessingException;
@ -24,6 +23,7 @@ import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
/**
@ -32,8 +32,6 @@ import java.util.stream.Collectors;
@Service
@Slf4j
public class TaskServiceImpl implements TaskService {
@Resource
private AsyncService asyncService;
/**
* The Object mapper.
@ -53,9 +51,19 @@ public class TaskServiceImpl implements TaskService {
@Resource
private DisposeTaskMapper disposeTaskMapper;
/**
* The Dispose node manager.
*/
@Resource
private DisposeNodeManager disposeNodeManager;
/**
* Gets dispose device handle.
*
* @param disposeIp the dispose ip
* @param devCapType the dev cap type
* @return the dispose device handle
*/
private DisposeEntryManager getDisposeDeviceHandle(String disposeIp, int devCapType) {
DisposeDevice dev = getDisposeNode(disposeIp, devCapType);
@ -67,6 +75,13 @@ public class TaskServiceImpl implements TaskService {
return DeviceRouter.deviceRouterFactory(dev.getType(), dev.getIpAddr());
}
/**
* Gets dispose node.
*
* @param disposeIp the dispose ip
* @param devCapType the dev cap type
* @return the dispose node
*/
private DisposeDevice getDisposeNode(String disposeIp, int devCapType) {
DeviceCapacity cap;
@ -108,10 +123,10 @@ public class TaskServiceImpl implements TaskService {
}
/**
* Create task error code.
* Create task m return type.
*
* @param task the task
* @return the error code
* @return the m return type
*/
@Override
public MReturnType<ErrorCode, Long> createTask(TaskInfoDetail task) {
@ -152,7 +167,12 @@ public class TaskServiceImpl implements TaskService {
// 将该任务写入数据库和缓存，等到定时任务真正启动该任务
disposeTaskMapper.addNewTask(task);
ErrorCode err = taskCacheManager.addTask(task);
task.setBeginTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
TaskInfoDetail cacheTask = disposeTaskMapper.getTaskInfoById(task.getId());
ErrorCode err = taskCacheManager.addTask(cacheTask);
return MReturnType.<ErrorCode, Long>builder()
.firstParam(err)
@ -184,17 +204,30 @@ public class TaskServiceImpl implements TaskService {
return ErrorCode.ERR_NOSUCHDEVICE;
}
if (!taskIsRunning(task)) {
// 启动处置任务
err = dp.runDispose(task.getDisposeIp(), DeviceCapacity.values()[task.getType()]);
// 对新建的任务执行启动操作
if (task.getCurrentStatus() == DisposeTaskStatus.TASK_NEW.getCode()) {
if (err == ErrorCode.ERR_OK) {
// 更新处置任务状态
taskCacheManager.upgradeTaskStatus(taskId, DisposeTaskStatus.TASK_RUNNING.getCode());
disposeTaskMapper.changeTaskCurrentStatus(taskId, DisposeTaskStatus.TASK_RUNNING.getCode());
} else {
log.error("Start task {}:{} error {}", task.getId(), task.getDisposeIp(), err.getMsg());
}
taskCacheManager.upgradeTaskStatus(taskId, DisposeTaskStatus.TASK_RUNNING.getCode());
// 异步启动处置任务
CompletableFuture<ErrorCode> future = CompletableFuture
.supplyAsync(() -> dp.runDispose(task.getDisposeIp(), DeviceCapacity.values()[task.getType()]))
.whenComplete((v, ex) -> {
if (ex != null) {
log.error(ex.getMessage());
// 执行任务失败恢复缓存中的任务状态
taskCacheManager.upgradeTaskStatus(taskId, DisposeTaskStatus.TASK_NEW.getCode());
} else {
if (v != ErrorCode.ERR_OK) {
// 执行任务失败恢复缓存中的任务状态
taskCacheManager.upgradeTaskStatus(taskId, DisposeTaskStatus.TASK_NEW.getCode());
log.error("Start task {}:{} error {}", task.getId(), task.getDisposeIp(), err.getMsg());
} else {
// 任务执行完成后更新数据库处置任务状态
disposeTaskMapper.changeTaskCurrentStatus(taskId, DisposeTaskStatus.TASK_RUNNING.getCode());
}
}
});
}
return err;
@ -209,14 +242,10 @@ public class TaskServiceImpl implements TaskService {
@Override
public ErrorCode stopTask(Long taskId) {
if (taskId == -1) {
return ErrorCode.ERR_NOSUCHTASK;
}
ErrorCode err = taskCacheManager.upgradeTaskStatus(taskId, DisposeTaskStatus.TASK_STOP.getCode());
if (err == ErrorCode.ERR_OK) {
disposeTaskMapper.changeTaskCurrentStatus(taskId, DisposeTaskStatus.TASK_STOP.getCode());
if (err != ErrorCode.ERR_OK) {
return err;
}
TaskInfoDetail task = taskCacheManager.getTaskById(taskId);
@ -224,10 +253,32 @@ public class TaskServiceImpl implements TaskService {
if (task != null) {
DisposeEntryManager dp = getDisposeDeviceHandle(task.getDisposeIp(), task.getType());
if (dp != null) {
asyncService.asyncStopDisposeDeviceTask(dp, task.getDisposeIp(),
DeviceCapacity.values()[task.getType()]);
if (dp == null) {
return ErrorCode.ERR_NOSUCHDEVICE;
}
int prdStatus = task.getCurrentStatus();
// 异步启动处置任务
CompletableFuture<ErrorCode> future = CompletableFuture
.supplyAsync(() -> dp.stopDispose(task.getDisposeIp(), DeviceCapacity.values()[task.getType()]))
.whenComplete((v, ex) -> {
if (ex != null) {
// 恢复缓存中任务状态到先前状态
taskCacheManager.upgradeTaskStatus(taskId, prdStatus);
log.error(ex.getMessage());
} else {
if (v != ErrorCode.ERR_OK) {
// 恢复缓存中任务状态到先前状态
taskCacheManager.upgradeTaskStatus(taskId, prdStatus);
log.error("Start task {}:{} error {}", task.getId(), task.getDisposeIp(), err.getMsg());
} else {
// 任务执行完成后更新数据库处置任务状态
disposeTaskMapper.changeTaskCurrentStatus(taskId, DisposeTaskStatus.TASK_STOP.getCode());
}
}
});
}
return ErrorCode.ERR_OK;
@ -251,7 +302,7 @@ public class TaskServiceImpl implements TaskService {
}
/**
* Task is running boolean.
* Task is running xx boolean.
*
* @param task the task
* @return the boolean
@ -332,4 +383,22 @@ public class TaskServiceImpl implements TaskService {
public List<TaskInfoDetail> getAllTask() {
return taskCacheManager.getAllTask();
}
/**
* Task is exists boolean.
*
* @param devId the dev id
* @param userId the user id
* @param disposeIp the dispose ip
* @param disposeType the dispose type
* @return the boolean
*/
@Override
public boolean taskIsExists(Long devId, Long userId, String disposeIp, int disposeType) {
return taskCacheManager.getAllTask().parallelStream()
.anyMatch(v -> Objects.equals(v.getAccountId(), userId)
&& Objects.equals(v.getDeviceId(), devId)
&& Objects.equals(v.getDisposeIp(), disposeIp)
&& Objects.equals(v.getType(), disposeType));
}
}

43
src/main/java/com/dispose/task/TaskManagerTask.java
View File

@ -1,6 +1,9 @@
package com.dispose.task;
import com.dispose.common.DisposeTaskStatus;
import com.dispose.common.ErrorCode;
import com.dispose.manager.TaskCacheManager;
import com.dispose.pojo.vo.common.TaskInfoDetail;
import com.dispose.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
@ -32,28 +35,28 @@ public class TaskManagerTask {
* Task runtime manager.
*/
@Async("bizExecutor")
@Scheduled(fixedRate = 5000)
@Scheduled(fixedDelay = 5000)
public void taskRuntimeManager() {
Iterator it = taskCacheManager.getAllTask().iterator();
// while (it.hasNext()) {
// TaskInfoDetail taskData = (TaskInfoDetail) it.next();
//
// if(taskService.taskIsExpired(taskData)) {
// log.info("Finish expired task {}:{} begin at {}",
// taskData.getId(), taskData.getDisposeIp(), taskData.getBeginTime());
// taskService.stopTask(taskData.getId());
// taskService.finishTask(taskData.getId());
// continue;
// }
//
// if (taskData.getCurrentStatus() == DisposeTaskStatus.TASK_NEW.getCode()) {
// log.info("Start task {}:{} of {}",
// taskData.getId(), taskData.getDisposeIp(), taskData.getBeginTime());
// if (taskService.startTask(taskData.getId()) != ErrorCode.ERR_OK) {
// log.error("startTask Task {}:{} error\n", taskData.getId(), taskData.getDisposeIp());
// }
// }
// }
while (it.hasNext()) {
TaskInfoDetail taskData = (TaskInfoDetail) it.next();
if(taskService.taskIsExpired(taskData) && taskService.taskIsRunning(taskData)) {
log.info("Finish expired task {}:{} begin at {}",
taskData.getId(), taskData.getDisposeIp(), taskData.getBeginTime());
taskService.stopTask(taskData.getId());
taskService.finishTask(taskData.getId());
continue;
}
if (taskData.getCurrentStatus() == DisposeTaskStatus.TASK_NEW.getCode()) {
log.info("Start task {}:{} of {}",
taskData.getId(), taskData.getDisposeIp(), taskData.getBeginTime());
if (taskService.startTask(taskData.getId()) != ErrorCode.ERR_OK) {
log.error("startTask Task {}:{} error\n", taskData.getId(), taskData.getDisposeIp());
}
}
}
}
}

7
src/main/resources/mappers/DisposeTask.xml
View File

@ -72,13 +72,6 @@
AND currentStatus = #{status, jdbcType=INTEGER}
</select>
<select id="getAllTaskByIp" resultType="com.dispose.pojo.vo.common.TaskInfoDetail">
SELECT *
FROM dispose_task
WHERE disposeIp = #{ipAddr, jdbcType=VARCHAR}
AND currentStatus != ${@com.dispose.common.DisposeTaskStatus@TASK_DELETE.getCode()}
</select>
<select id="getAllTaskByStatus" resultType="com.dispose.pojo.vo.common.TaskInfoDetail">
SELECT *
FROM dispose_task

16
src/test/java/com/dispose/controller/DeviceNodeInfoControllerTest.java
View File

@ -828,10 +828,10 @@ public class DeviceNodeInfoControllerTest extends InitTestEnvironment {
Assert.assertNotEquals(Content.getString("ip"), null);
Assert.assertTrue(Content.getString("ip").matches(regex));
Assert.assertNotNull(Content.getString("areaCode"));
Assert.assertNotNull(Content.getString("manufacturer"));
Assert.assertNotNull(Content.getString("model"));
Assert.assertNotNull(Content.getString("version"));
Assert.assertNotNull(Content.getString("readme"));
// Assert.assertNotNull(Content.getString("manufacturer"));
// Assert.assertNotNull(Content.getString("model"));
// Assert.assertNotNull(Content.getString("version"));
// Assert.assertNotNull(Content.getString("readme"));
//解析capacityArray
if (Content.getString("capacity") != null) {
@ -919,10 +919,10 @@ public class DeviceNodeInfoControllerTest extends InitTestEnvironment {
Assert.assertNotEquals(Content.getString("ip"), null);
Assert.assertTrue(Content.getString("ip").matches(regex));
Assert.assertNotNull(Content.getString("areaCode"));
Assert.assertNotNull(Content.getString("manufacturer"));
Assert.assertNotNull(Content.getString("model"));
Assert.assertNotNull(Content.getString("version"));
Assert.assertNotNull(Content.getString("readme"));
// Assert.assertNotNull(Content.getString("manufacturer"));
// Assert.assertNotNull(Content.getString("model"));
// Assert.assertNotNull(Content.getString("version"));
// Assert.assertNotNull(Content.getString("readme"));
//解析capacityArray
if (Content.getString("capacity") != null) {

2
src/test/java/com/dispose/mapper/DisposeTaskMapperTest.java
View File

@ -224,7 +224,7 @@ public class DisposeTaskMapperTest extends InitTestEnvironment {
@Test
public void t8_getAllTaskByIpTest() {
disposeTaskMapper.selectAll().forEach(v -> disposeTaskMapper
.getAllTaskByIp(v.getDisposeIp())
.getAllTaskByDisposeIp(v.getDisposeIp())
.forEach(k -> {
Assert.assertEquals(k.getDisposeIp(), v.getDisposeIp());
try {