Skip to content

H
haikang-client
提交 9ec007c0 authored 作者: wangqiang's avatar wangqiang
浏览文件
操作

问题修改

上级 d31675c6
显示空白字符变更
内嵌 并排
正在显示 包含 523 行增加 和 126 行删除
apq-pc-client/src/main/java/com/priusis/client/extensions/http/DefaultHttpService.java
浏览文件 @ 9ec007c0
......@@ -4,11 +4,12 @@ import com.priusis.client.data.kv.KvEntry;
import com.priusis.client.extensions.ExtensionUpdate;
import com.priusis.client.extensions.http.conf.HttpConfiguration;
import com.priusis.client.service.MqttDeliveryFuture;
import com.priusis.client.service.MqttRpcDataMessage;
import com.priusis.client.service.conf.PcExtensionConfiguration;
import com.priusis.client.service.core.MqttService;
import com.priusis.client.service.data.DeviceData;
import com.priusis.client.service.data.RpcCommandResponse;
import com.priusis.client.util.ConfigurationTools;
import com.priusis.client.service.MqttRpcDataMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;
......@@ -87,6 +88,16 @@ public class DefaultHttpService extends ExtensionUpdate implements HttpService {
return mqttService.flushRpcDataToFile(mqttRpcDataMessage);
}
@Override
public void processRpcRes(String token, MqttRpcDataMessage mqttRpcDataMessage) throws Exception {
RpcCommandResponse rpcCommandResponse = new RpcCommandResponse();
rpcCommandResponse.setRequestId(mqttRpcDataMessage.getRequestId());
rpcCommandResponse.setBizRequestId(mqttRpcDataMessage.getBizRequestId());
rpcCommandResponse.setMethod(mqttRpcDataMessage.getMethod());
rpcCommandResponse.setData(mqttRpcDataMessage.getParams());
mqttService.onDeviceRpcResponse(rpcCommandResponse);
}
@Override
public MqttRpcDataMessage readFromFile(String method) throws IOException {
return mqttService.readFromFile(method);
......
apq-pc-client/src/main/java/com/priusis/client/extensions/http/HttpController.java
浏览文件 @ 9ec007c0
......@@ -3,8 +3,8 @@ package com.priusis.client.extensions.http;
import cn.hutool.json.JSONUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.priusis.client.extensions.http.conf.HttpRequestProcessingError;
import com.priusis.client.service.TenantManagerService;
import com.priusis.client.service.MqttRpcDataMessage;
import com.priusis.client.service.TenantManagerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
......@@ -14,8 +14,6 @@ import org.springframework.web.bind.annotation.*;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import static com.priusis.client.util.JsonTools.fromString;
@RestController
@Slf4j
public class HttpController {
......@@ -53,6 +51,16 @@ public class HttpController {
}
}
@RequestMapping(value = "/rpc_res/{token}", method = RequestMethod.POST)
public void handleRpcResDataRequest(@PathVariable String token, @RequestBody String body) {
MqttRpcDataMessage mqttRpcDataMessage = JSONUtil.toBean(body, MqttRpcDataMessage.class);
try {
service.processRpcRes(token, mqttRpcDataMessage);
} catch (Exception e) {
log.error("Failed to process rpc res persistent : {}", body);
}
}
@ExceptionHandler(Exception.class)
public void handlePriusisiotException(Exception exception, HttpServletResponse response) {
log.debug("Processing exception {}", exception.getMessage(), exception);
......
apq-pc-client/src/main/java/com/priusis/client/extensions/http/HttpService.java
浏览文件 @ 9ec007c0
......@@ -14,5 +14,7 @@ public interface HttpService extends ExtensionService {
File flushRpcDataToFile(MqttRpcDataMessage mqttRpcDataMessage) throws IOException;
void processRpcRes(String token, MqttRpcDataMessage mqttRpcDataMessage) throws Exception;
MqttRpcDataMessage readFromFile(String method) throws IOException;
}
apq-pc-client/src/main/java/com/priusis/client/service/DefaultTenantManagerService.java
浏览文件 @ 9ec007c0
......@@ -72,6 +72,11 @@ public abstract class DefaultTenantManagerService implements TenantManagerServic
httpService.processEventRequest(token, body);
}
@Override
public void processRpcRes(String token, MqttRpcDataMessage mqttRpcDataMessage) throws Exception {
httpService.processRpcRes(token, mqttRpcDataMessage);
}
@Override
public File flushRpcDataToFile(MqttRpcDataMessage mqttRpcDataMessage) throws IOException {
return httpService.flushRpcDataToFile(mqttRpcDataMessage);
......
apq-pc-client/src/main/java/com/priusis/client/service/TenantManagerService.java
浏览文件 @ 9ec007c0
......@@ -12,6 +12,8 @@ public interface TenantManagerService {
void processEventRequest(String token, String body) throws Exception;
void processRpcRes(String token, MqttRpcDataMessage mqttRpcDataMessage) throws Exception;
File flushRpcDataToFile(MqttRpcDataMessage mqttRpcDataMessage) throws IOException;
MqttRpcDataMessage readFromFile(String method) throws IOException;
......
apq-pc-client/src/main/java/com/priusis/client/service/core/MqttServiceImpl.java
浏览文件 @ 9ec007c0
......@@ -4,6 +4,7 @@ import cn.hutool.core.convert.Convert;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
......@@ -15,6 +16,7 @@ import com.priusis.client.service.conf.*;
import com.priusis.client.service.data.*;
import com.priusis.controller.ProgramController;
import com.priusis.controller.RemoteControlController;
import com.priusis.dto.HwPortControllerDto;
import com.priusis.lib.R;
import com.priusis.monitor.mqtt.*;
import com.priusis.service.common.MacAddrService;
......@@ -47,6 +49,8 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import static com.priusis.client.util.JsonTools.*;
......@@ -87,8 +91,9 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
private static final String DEVICE_TELEMETRY_TOPIC = "devices/telemetry";
private static final String DEVICE_ATTRIBUTES_TOPIC = "devices/attrs";
private static final String DEVICE_RPC_RES_TOPIC = "devices/rpc/req";
private static final String DEVICE_RPC_TOPIC = DEVICE_RPC_RES_TOPIC + "/+";
private static final String DEVICE_RPC_REQ_TOPIC = "devices/rpc/req";
private static final String DEVICE_RPC_RES_TOPIC = "devices/rpc/res";
private static final String DEVICE_RPC_TOPIC = DEVICE_RPC_REQ_TOPIC + "/+";
private static final String DEVICE_EVENT_TOPIC = "devices/event";
private static final String DEVICE_GET_ATTRIBUTES_REQUEST_TOPIC = "devices/attrs/req/1";
private static final String DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC = "devices/attrs/res/+";
......@@ -105,6 +110,10 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
private final Set<AttributesUpdateSubscription> attributeUpdateSubs = ConcurrentHashMap.newKeySet();
private final Map<AttributeRequestKey, AttributeRequestListener> pendingAttrRequestsMap = new ConcurrentHashMap<>();
private final String REQUEST_ID_PATTERN = "(?<requestId>\\d+)";
private final String DEVICE_RPC_REQUEST_TOPIC_PATTERN = DEVICE_RPC_REQ_TOPIC + "/" + REQUEST_ID_PATTERN;
private final Pattern DEVICE_RPC_REQUEST_PATTERN = Pattern.compile(DEVICE_RPC_REQUEST_TOPIC_PATTERN);
private String tenantLabel;
private PersistentFileService persistentFileService;
......@@ -271,20 +280,20 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
@Override
public void onDeviceRpcResponse(RpcCommandResponse response) {
final int msgId = msgIdSeq.incrementAndGet();
int requestId = response.getRequestId();
String bizRequestId = response.getBizRequestId();
String data = response.getData();
String method = response.getMethod();
/*ObjectNode paramsNode = newNode();
paramsNode.put("code", "1");
paramsNode.put("msg", "success");*/
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", data));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish RPC res", error));
ObjectNode node = newNode();
node.put("id", requestId);
node.put("data", data);
persistMessage(DEVICE_RPC_TOPIC, msgId, toBytes(node),
token -> {
log.debug("[{}] RPC response from device was delivered!", requestId);
},
error -> {
log.warn("[{}] Failed to report RPC response from device!", requestId, error);
});
}
@Override
......@@ -324,8 +333,8 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
onAttributesUpdate(message);
} else if (topic.equals(DEVICE_GET_ATTRIBUTES_RESPONSE_PLUS_TOPIC)) {
onDeviceAttributesResponse(message);
} else if (topic.startsWith(DEVICE_RPC_RES_TOPIC)) {
onRpcCommand(message);
} else if (topic.startsWith(DEVICE_RPC_REQ_TOPIC)) {
onRpcCommand(topic, message);
} else if (topic.startsWith(DEVICE_ATTRIBUTES_TOPIC)) {
onGatewayAttributesUpdate(message);
} else if (topic.equals(DEVICE_GET_ATTRIBUTES_RESPONSE_TOPIC)) {
......@@ -369,16 +378,44 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
}
}
private void onRpcCommand(String message) {
private void onRpcCommand(String topic, String message) {
JsonNode payload = fromString(message);
MqttRpcDataMessage mqttRpcDataMessage = MqttRpcDataMessage.builder()
String bizRequestId = payload.get("requestId").asText();
int requestId = 0;
Matcher fwMatcher = DEVICE_RPC_REQUEST_PATTERN.matcher(topic);
if (fwMatcher.find()) {
requestId = Integer.parseInt(fwMatcher.group("requestId"));
}
MqttRpcDataMessage mqttRpcDataMessage = MqttRpcDataMessage.builder().requestId(requestId).bizRequestId(bizRequestId)
.method(payload.get("method").asText()).sendTime(System.currentTimeMillis())
.params(Optional.ofNullable(payload.get("params")).map(JsonNode::toString).orElse(null)).build();
// 存储rpc下发的数据
powerHandler(message, requestId, bizRequestId, mqttRpcDataMessage);
remoteControlHandler(message, requestId, bizRequestId, mqttRpcDataMessage);
realDataMonitorScanHandler(message, requestId, bizRequestId, mqttRpcDataMessage);
remoteRtcHandler(message, requestId, bizRequestId, mqttRpcDataMessage);
resetControlHandler(message, requestId, bizRequestId, mqttRpcDataMessage);
controlHandler(message, requestId, bizRequestId, mqttRpcDataMessage);
try {
if ("power".equals(mqttRpcDataMessage.getMethod())) {
if (!Arrays.asList("power", "remote_control", "remote_control_exit", "remote_control_accept",
"real_data_monitor", "facility_scan", "remote_rtc", "control_allow", "controll").contains(mqttRpcDataMessage.getMethod())) {
persistentFileService.flushRpcDataToFile(mqttRpcDataMessage);
}
} catch (IOException e) {
log.error("Failed to process flushRpcDataToFile : {}", message, e);
}
}
private void powerHandler(String message, int requestId, String bizRequestId, MqttRpcDataMessage mqttRpcDataMessage) {
String method = mqttRpcDataMessage.getMethod();
if (!"power".equals(method)) {
return;
}
try {
if ("power".equals(method)) {
Integer type = JSONUtil.parseObj(mqttRpcDataMessage.getParams()).getInt("powerType");
if (type == 0) {
log.info("远程关机!!!");
......@@ -387,9 +424,309 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
log.info("远程重启!!!");
Runtime.getRuntime().exec("cmd /c shutdown -r");
}
// response
ObjectNode paramsNode = newNode();
paramsNode.put("code", "1");
paramsNode.put("msg", "success");
paramsNode.put("status", "success");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish power res", error));
}
} catch (Exception e) {
log.error("Failed to process rpc command persistent : {}", message, e);
ObjectNode paramsNode = newNode();
paramsNode.put("code", "0");
paramsNode.put("msg", "fail");
paramsNode.put("status", "fail");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish power res", error));
}
}
/**
* 远程硬件接口处理 control
*
* @param message
* @param requestId
* @param bizRequestId
* @param mqttRpcDataMessage
*/
private void controlHandler(String message, int requestId, String bizRequestId, MqttRpcDataMessage mqttRpcDataMessage) {
String method = mqttRpcDataMessage.getMethod();
if (!"controll".equals(method)) {
return;
}
if ("controll".equals(method)) {
String params = mqttRpcDataMessage.getParams();
if (StrUtil.isNotBlank(params)) {
int result;
try {
result = IHwPortController.instanceDll.APQ_HWPORT_Initialize();
log.info("APQ_HWPORT_Initialize ret:{}", result);
String status = "success";
Long sendTime = mqttRpcDataMessage.getSendTime();
long current = System.currentTimeMillis();
List<HwPortControllerDto> hwPortControllerDtos = JSONUtil.toList(params, HwPortControllerDto.class);
// 合并硬件请求检查
MqttRpcDataMessage persistentEntity = null;
try {
persistentEntity = persistentFileService.readFromFile("controll_4_merge");
} catch (Exception e) {
log.error("硬件监控上报异常merge", e.getMessage());
}
List<HwPortControllerDto> persistentHwPortControllerDtos = null;
if (null != persistentEntity) {
String persistentEntityParams = persistentEntity.getParams();
if (StrUtil.isNotBlank(persistentEntityParams)) {
persistentHwPortControllerDtos = JSONUtil.toList(persistentEntityParams, HwPortControllerDto.class);
persistentHwPortControllerDtos.removeAll(hwPortControllerDtos);
persistentHwPortControllerDtos.addAll(hwPortControllerDtos);
}
} else {
persistentHwPortControllerDtos = hwPortControllerDtos;
}
MqttRpcDataMessage mergeMqttRpcDataMessage = MqttRpcDataMessage.builder()
.sendTime(System.currentTimeMillis())
.method("controll_4_merge")
.params(JSONUtil.toJsonStr(persistentHwPortControllerDtos)).build();
persistentFileService.flushRpcDataToFile(mergeMqttRpcDataMessage);
// 收到硬件策略下发请求
log.info("收到硬件策略下发请求, sendTime:{}, params:{}", sendTime, params);
for (HwPortControllerDto hwPortControllerDto : hwPortControllerDtos) {
HwPortControllerDto.DeviceHardwareEnum deviceHardwareEnum = hwPortControllerDto.getKey();
int doType = hwPortControllerDto.getDoType();
int limitType = hwPortControllerDto.getLimitType();
int value = hwPortControllerDto.getValue();
List<Integer> hwDeviceIds = deviceHardwareEnum.getHwDeviceIds();
String hwDeviceName = deviceHardwareEnum.getName();
// 设置普通设备权限
if (Objects.equals(deviceHardwareEnum.getHwGroup(), "Normal")) {
// 1F/2T
for (Integer hwDeviceId : hwDeviceIds) {
int perm = (doType == 1 ? 2 : 1);
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_normal_device_permit(hwDeviceId, perm, null);
log.info("APQ_HWPORT_set_normal_device_permit doType:{}, limitType;{}，value:{}, hwDeviceId:{}, name;{}, perm:{}, ret:{}",
doType, limitType, value, hwDeviceId, hwDeviceName, perm, result32);
}
} else if (Objects.equals(deviceHardwareEnum.getHwGroup(), "Storage")) { // 设置存储设备权限
// 1F/2T/3T-ALL/4T-R/5T-CI/6T-CO 拷出目前无用！！ Storage移动硬盘，TYPEC 无效 ！！
for (Integer hwDeviceId : hwDeviceIds) {
int perm = (doType == 1 ? 2 : 1);
if (perm == 2) {
// limitType 1读写 2只读 3拷入 4拷出 5不限速 6限速
perm = (limitType == 1 ? 3 : limitType == 2 ? 4 : limitType == 3 ? 5 : limitType == 4 ? 6 : 3);
}
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_storage_device_permit(hwDeviceId, perm, null);
log.info("APQ_HWPORT_set_storage_device_permit doType:{}, limitType;{}，value:{}, hwDeviceId:{}, name;{}, perm:{}, ret:{}",
doType, limitType, value, hwDeviceId, hwDeviceName, perm, result32);
}
} else if (Objects.equals(deviceHardwareEnum.getHwGroup(), "NET")) { // 设置网络设备权限
// 1F/2T/3T-NS/4T-S-200 网口，4G/5G模块，无线WIFI区分不了 ！！
for (Integer hwDeviceId : hwDeviceIds) {
int perm = (doType == 1 ? 2 : 1);
if (perm == 2) {
// limitType 1读写 2只读 3拷入 4拷出 5不限速 6限速
perm = (limitType == 5 ? 3 : limitType == 6 ? 4 : 2);
}
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_net_device_permit(hwDeviceId, perm, value * 1024, null);
log.info("APQ_HWPORT_set_net_device_permit doType:{}, limitType;{}，value:{}, hwDeviceId:{}, name;{}, perm:{}, ret:{}",
doType, limitType, value, hwDeviceId, hwDeviceName, perm, result32);
}
}
// 应用硬件接口管理模块的配置内容
int result5 = IHwPortController.instanceDll.APQ_HWPORT_Apply();
log.info("APQ_HWPORT_Apply ret:{}", result5);
}
MqttRpcDataMessage controllMqttRpcDataMessage = MqttRpcDataMessage.builder()
.sendTime(System.currentTimeMillis())
.method("controll")
.params("").build();
persistentFileService.flushRpcDataToFile(controllMqttRpcDataMessage);
// response
ObjectNode paramsNode = newNode();
paramsNode.put("code", "1");
paramsNode.put("msg", "success");
paramsNode.put("status", "success");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish control reset res", error));
} catch (Exception e) {
log.error("APQ_HWPORT ERROR: {}", e.getMessage());
log.error("Failed to process rpc command persistent : {}", message, e);
ObjectNode paramsNode = newNode();
paramsNode.put("code", "0");
paramsNode.put("msg", "fail");
paramsNode.put("status", "fail");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish control reset res", error));
} finally {
result = IHwPortController.instanceDll.APQ_HWPORT_Uninitialize();
log.info("APQ_HWPORT_Uninitialize ret:{}", result);
}
} else {
log.info("未获取到获取硬件监控配置");
}
}
}
/**
* 远程硬件Reset接口处理 control_allow
*
* @param message
* @param requestId
* @param bizRequestId
* @param mqttRpcDataMessage
*/
private void resetControlHandler(String message, int requestId, String bizRequestId, MqttRpcDataMessage
mqttRpcDataMessage) {
String method = mqttRpcDataMessage.getMethod();
if (!"control_allow".equals(method)) {
return;
}
if ("control_allow".equals(method)) {
String params = mqttRpcDataMessage.getParams();
if (StrUtil.isNotBlank(params)) {
int result;
try {
result = IHwPortController.instanceDll.APQ_HWPORT_Initialize();
log.info("APQ_HWPORT_Initialize ret:{}", result);
Long sendTime = mqttRpcDataMessage.getSendTime();
List<HwPortControllerDto> hwPortControllerDtos = JSONUtil.toList(params, HwPortControllerDto.class);
// 收到硬件策略下发请求
log.info("收到硬件策略RESET下发请求, sendTime:{}, params:{}", sendTime, params);
for (HwPortControllerDto hwPortControllerDto : hwPortControllerDtos) {
HwPortControllerDto.DeviceHardwareEnum deviceHardwareEnum = hwPortControllerDto.getKey();
List<Integer> hwDeviceIds = deviceHardwareEnum.getHwDeviceIds();
String hwDeviceName = deviceHardwareEnum.getName();
// 设置普通设备权限
if (Objects.equals(deviceHardwareEnum.getHwGroup(), "Normal")) {
// 1F/2T
for (Integer hwDeviceId : hwDeviceIds) {
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_normal_device_permit(hwDeviceId, 2, null);
log.info("APQ_HWPORT_set_normal_device_permit hwDeviceId:{}, name;{}, ret:{}", hwDeviceId, hwDeviceName, result32);
}
} else if (Objects.equals(deviceHardwareEnum.getHwGroup(), "Storage")) { // 设置存储设备权限
for (Integer hwDeviceId : hwDeviceIds) {
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_storage_device_permit(hwDeviceId, 2, null);
log.info("APQ_HWPORT_set_storage_device_permit hwDeviceId:{}, name;{}, ret:{}", hwDeviceId, hwDeviceName, result32);
}
} else if (Objects.equals(deviceHardwareEnum.getHwGroup(), "NET")) { // 设置网络设备权限
for (Integer hwDeviceId : hwDeviceIds) {
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_net_device_permit(hwDeviceId, 3, 0, null);
log.info("APQ_HWPORT_set_net_device_permit hwDeviceId:{}, name;{}, ret:{}", hwDeviceId, hwDeviceName, result32);
}
}
// 应用硬件接口管理模块的配置内容
int result5 = IHwPortController.instanceDll.APQ_HWPORT_Apply();
log.info("APQ_HWPORT_Apply ret:{}", result5);
// 删除本地mergin数据
MqttRpcDataMessage persistentEntity = null;
try {
persistentEntity = persistentFileService.readFromFile("controll_4_merge");
} catch (Exception e) {
log.error("硬件监控上报异常merge", e.getMessage());
}
List<HwPortControllerDto> persistentHwPortControllerDtos = null;
if (null != persistentEntity) {
String persistentEntityParams = persistentEntity.getParams();
if (StrUtil.isNotBlank(persistentEntityParams)) {
persistentHwPortControllerDtos = JSONUtil.toList(persistentEntityParams, HwPortControllerDto.class);
persistentHwPortControllerDtos.removeAll(hwPortControllerDtos);
}
}
MqttRpcDataMessage resetMqttRpcDataMessage = MqttRpcDataMessage.builder()
.sendTime(System.currentTimeMillis())
.method("controll_4_merge")
.params(JSONUtil.toJsonStr(persistentHwPortControllerDtos)).build();
persistentFileService.flushRpcDataToFile(resetMqttRpcDataMessage);
}
// response
ObjectNode paramsNode = newNode();
paramsNode.put("code", "1");
paramsNode.put("msg", "success");
paramsNode.put("status", "success");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish control reset res", error));
} catch (Exception e) {
log.error("APQ_HWPORT ERROR: {}", e.getMessage());
log.error("Failed to process rpc command persistent : {}", message, e);
ObjectNode paramsNode = newNode();
paramsNode.put("code", "0");
paramsNode.put("msg", "fail");
paramsNode.put("status", "fail");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish control reset res", error));
} finally {
result = IHwPortController.instanceDll.APQ_HWPORT_Uninitialize();
log.info("APQ_HWPORT_Uninitialize ret:{}", result);
}
} else {
log.info("未获取到获取硬件RESET监控配置");
}
}
}
/**
* 远程RPC处理
*
* @param message
* @param requestId
* @param mqttRpcDataMessage
*/
private void remoteControlHandler(String message, int requestId, String bizRequestId, MqttRpcDataMessage
mqttRpcDataMessage) {
String method = mqttRpcDataMessage.getMethod();
if (!"remote_control".equals(method) && !"remote_control_exit".equals(method) && !"remote_control_accept".equals(method)) {
return;
}
if ("remote_control".equals(mqttRpcDataMessage.getMethod())) {// 远程控制请求
try {
if ("remote_control".equals(method)) {// 远程控制请求
log.info("收到远程控制请求!!!");
if (StrUtil.equals("RUNNING", remoteControlController.vncServerIsRunning())) {
log.warn("VNC Server 已在运行中，忽略此次请求");
......@@ -401,14 +738,12 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
SysConfigUtil.saveProperty("control.request", "-1");
remoteControlController.startVncServer(0);
}
return;
}
if ("remote_control_exit".equals(mqttRpcDataMessage.getMethod())) {
if ("remote_control_exit".equals(method)) {
log.info("退出远程控制!!!");
remoteControlController.stopVncServer();
return;
}
if ("remote_control_accept".equals(mqttRpcDataMessage.getMethod())) {
if ("remote_control_accept".equals(method)) {
log.info("远程控制请求接受!!! params: {}", mqttRpcDataMessage.getParams());
if (StrUtil.equals("RUNNING", remoteControlController.vncServerIsRunning())) {
log.warn("VNC Server 已在运行中，忽略此次请求");
......@@ -417,22 +752,95 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
Long minutes = Convert.toLong(mqttRpcDataMessage.getParams());
R<Void> startVncResult = remoteControlController.startVncServer(minutes);
log.info("远程控制请求接受!!! startVncResult: {}", startVncResult.getMsg());
return;
}
}
// response
ObjectNode paramsNode = newNode();
paramsNode.put("code", "1");
paramsNode.put("msg", "success");
paramsNode.put("status", "success");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish remoteControl res", error));
} catch (Exception e) {
log.error("Failed to process rpc command persistent : {}", message, e);
ObjectNode paramsNode = newNode();
paramsNode.put("code", "0");
paramsNode.put("msg", "fail");
paramsNode.put("status", "fail");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish remoteControl res", error));
}
}
/**
* 数据更新频率变为一秒刷新一次 / 扫描软件请求
*
* @param message
* @param requestId
* @param mqttRpcDataMessage
*/
private void realDataMonitorScanHandler(String message, int requestId, String bizRequestId, MqttRpcDataMessage
mqttRpcDataMessage) {
String method = mqttRpcDataMessage.getMethod();
if (!"real_data_monitor".equals(method) && !"facility_scan".equals(method)) {
return;
}
try {
// 数据更新频率变为一秒刷新一次
if ("real_data_monitor".equals(mqttRpcDataMessage.getMethod()) && !realDataMonitor) {
if ("real_data_monitor".equals(method) && !realDataMonitor) {
realDataMonitor = true;
log.info("数据上报频率请求接受!!! params: {}", mqttRpcDataMessage.getParams());
realDataMonitorScheduler(mqttRpcDataMessage);
}
if ("facility_scan".equals(mqttRpcDataMessage.getMethod())) {
if ("facility_scan".equals(method)) {
log.info("收到平台扫描软件请求");
programController.addProgram2Platform();
}
ObjectNode paramsNode = newNode();
paramsNode.put("code", "1");
paramsNode.put("msg", "success");
paramsNode.put("status", "success");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish realDataMonitorSca res", error));
} catch (Exception e) {
log.error("Failed to process rpc command persistent : {}", message, e);
ObjectNode paramsNode = newNode();
paramsNode.put("code", "0");
paramsNode.put("msg", "fail");
paramsNode.put("status", "fail");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish realDataMonitorSca res", error));
}
}
/**
* 收到语音通话请求
*
* @param message
* @param requestId
* @param mqttRpcDataMessage
*/
private void remoteRtcHandler(String message, int requestId, String bizRequestId, MqttRpcDataMessage
mqttRpcDataMessage) {
String method = mqttRpcDataMessage.getMethod();
if (!"remote_rtc".equals(method)) {
return;
}
if ("remote_rtc".equals(mqttRpcDataMessage.getMethod())) {
try {
if ("remote_rtc".equals(method)) {
log.info("收到语音通话请求");
try {
boolean hasApqExeProcess = WinExecuteUtils.findProcess("apq.exe");
......@@ -445,12 +853,27 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
log.error("语音通话请求异常", e);
SysConfigUtil.saveProperty("control.request", "0");
}
return;
ObjectNode paramsNode = newNode();
paramsNode.put("code", "1");
paramsNode.put("msg", "success");
paramsNode.put("status", "success");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish remoteRtc res", error));
}
persistentFileService.flushRpcDataToFile(mqttRpcDataMessage);
} catch (Exception e) {
log.error("Failed to process rpc command persistent : {}", message, e);
ObjectNode paramsNode = newNode();
paramsNode.put("code", "0");
paramsNode.put("msg", "fail");
paramsNode.put("status", "fail");
byte[] msgData = toBytes(newNode().put("requestId", bizRequestId).put("method", method)
.put("resTime", System.currentTimeMillis()).put("response", paramsNode.toString()));
persistMessage(DEVICE_RPC_RES_TOPIC + "/" + requestId, msgIdSeq.incrementAndGet(), msgData, null,
error ->
log.warn("Could not publish remoteRtc res", error));
}
}
......@@ -906,4 +1329,14 @@ public class MqttServiceImpl implements MqttService, MqttHandler, MqttClientCall
//return mqttClientSecurityConfiguration;
return null;
}
public static void main(String[] args) {
ObjectNode paramsNode = newNode();
paramsNode.put("code", "0");
paramsNode.put("msg", "fail");
paramsNode.put("status", "fail");
JSONObject jsonObject = JSONUtil.parseObj(paramsNode.toPrettyString());
System.out.println(jsonObject.getStr("msg"));
}
}
apq-pc-client/src/main/java/com/priusis/client/service/data/RpcCommandResponse.java
浏览文件 @ 9ec007c0
......@@ -9,7 +9,9 @@ import lombok.Data;
public class RpcCommandResponse {
private int requestId;
private String bizRequestId;
private String deviceName;
private String method;
private String data;
}
apq-pc-client/src/main/resources/application-adev.yml
浏览文件 @ 9ec007c0
apq:
iot-gateway: 192.168.124.19:7002
webrtc_path: https://ceshi.priusis.com:8443
client:
productId: 210
tenantId: 156
......
apq-pc-common/src/main/java/com/priusis/client/service/MqttRpcDataMessage.java
浏览文件 @ 9ec007c0
......@@ -16,6 +16,7 @@ public class MqttRpcDataMessage implements Serializable {
private static final long serialVersionUID = -3133461476074777891L;
private int requestId;
private String bizRequestId;
private String method;
private Long sendTime;
private String params;
......
apq-pc-common/src/main/java/com/priusis/job/ApqControlJob.java
浏览文件 @ 9ec007c0
package com.priusis.job;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.priusis.client.service.MqttRpcDataMessage;
......@@ -78,10 +77,11 @@ public class ApqControlJob {
// 获取硬件监控配置 params == data
ResponseEntity<MqttRpcDataMessage> forEntity = null;
try {
forEntity = restTemplate.getForEntity("http://localhost:8765/rpc_cmd/controll", MqttRpcDataMessage.class);
forEntity = restTemplate.getForEntity("http://localhost:8765/rpc_cmd/controll_4_merge", MqttRpcDataMessage.class);
} catch (Exception e) {
log.error("硬件监控上报异常", e.getMessage());
}
if (null != forEntity) {
MqttRpcDataMessage body = forEntity.getBody();
String params = body.getParams();
......@@ -90,99 +90,15 @@ public class ApqControlJob {
int result = IHwPortController.instanceDll.APQ_HWPORT_Initialize();
log.info("APQ_HWPORT_Initialize ret:{}", result);
try {
Long sendTime = body.getSendTime();
long current = System.currentTimeMillis();
List<HwPortControllerDto> hwPortControllerDtos = JSONUtil.toList(params, HwPortControllerDto.class);
// 合并硬件请求检查
// 先获取数据
// 获取硬件监控配置 params == data
ResponseEntity<MqttRpcDataMessage> persistentEntity = null;
try {
persistentEntity = restTemplate.getForEntity("http://localhost:8765/rpc_cmd/controll_4_merge", MqttRpcDataMessage.class);
} catch (Exception e) {
log.error("硬件监控上报异常merge", e.getMessage());
}
List<HwPortControllerDto> persistentHwPortControllerDtos = null;
if (null != persistentEntity) {
MqttRpcDataMessage persistentEntityBody = persistentEntity.getBody();
String persistentEntityParams = persistentEntityBody.getParams();
if (StrUtil.isNotBlank(persistentEntityParams)) {
persistentHwPortControllerDtos = JSONUtil.toList(persistentEntityParams, HwPortControllerDto.class);
persistentHwPortControllerDtos.removeAll(hwPortControllerDtos);
persistentHwPortControllerDtos.addAll(hwPortControllerDtos);
}
} else {
persistentHwPortControllerDtos = hwPortControllerDtos;
}
MqttRpcDataMessage mqttRpcDataMessage = MqttRpcDataMessage.builder()
.sendTime(System.currentTimeMillis())
.method("controll_4_merge")
.params(JSONUtil.toJsonStr(persistentHwPortControllerDtos)).build();
restTemplate.postForObject("http://localhost:8765/rpc_cmd", mqttRpcDataMessage, String.class);
if (current - sendTime <= FIXED_DELAY && CollUtil.isNotEmpty(hwPortControllerDtos)) {
// 收到硬件策略下发请求
log.info("收到硬件策略下发请求, sendTime:{}, params:{}", sendTime, params);
for (HwPortControllerDto hwPortControllerDto : hwPortControllerDtos) {
HwPortControllerDto.DeviceHardwareEnum deviceHardwareEnum = hwPortControllerDto.getKey();
int doType = hwPortControllerDto.getDoType();
int limitType = hwPortControllerDto.getLimitType();
int value = hwPortControllerDto.getValue();
List<Integer> hwDeviceIds = deviceHardwareEnum.getHwDeviceIds();
String hwDeviceName = deviceHardwareEnum.getName();
// 设置普通设备权限
if (Objects.equals(deviceHardwareEnum.getHwGroup(), "Normal")) {
// 1F/2T
for (Integer hwDeviceId : hwDeviceIds) {
int perm = (doType == 1 ? 2 : 1);
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_normal_device_permit(hwDeviceId, perm, null);
log.info("APQ_HWPORT_set_normal_device_permit doType:{}, limitType;{}，value:{}, hwDeviceId:{}, name;{}, perm:{}, ret:{}",
doType, limitType, value, hwDeviceId, hwDeviceName, perm, result32);
}
} else if (Objects.equals(deviceHardwareEnum.getHwGroup(), "Storage")) { // 设置存储设备权限
// 1F/2T/3T-ALL/4T-R/5T-CI/6T-CO 拷出目前无用！！ Storage移动硬盘，TYPEC 无效 ！！
for (Integer hwDeviceId : hwDeviceIds) {
int perm = (doType == 1 ? 2 : 1);
if (perm == 2) {
// limitType 1读写 2只读 3拷入 4拷出 5不限速 6限速
perm = (limitType == 1 ? 3 : limitType == 2 ? 4 : limitType == 3 ? 5 : limitType == 4 ? 6 : 3);
}
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_storage_device_permit(hwDeviceId, perm, null);
log.info("APQ_HWPORT_set_storage_device_permit doType:{}, limitType;{}，value:{}, hwDeviceId:{}, name;{}, perm:{}, ret:{}",
doType, limitType, value, hwDeviceId, hwDeviceName, perm, result32);
}
} else if (Objects.equals(deviceHardwareEnum.getHwGroup(), "NET")) { // 设置网络设备权限
// 1F/2T/3T-NS/4T-S-200 网口，4G/5G模块，无线WIFI区分不了 ！！
for (Integer hwDeviceId : hwDeviceIds) {
int perm = (doType == 1 ? 2 : 1);
if (perm == 2) {
// limitType 1读写 2只读 3拷入 4拷出 5不限速 6限速
perm = (limitType == 5 ? 3 : limitType == 6 ? 4 : 2);
}
int result32 = IHwPortController.instanceDll.APQ_HWPORT_set_net_device_permit(hwDeviceId, perm, value, null);
log.info("APQ_HWPORT_set_net_device_permit doType:{}, limitType;{}，value:{}, hwDeviceId:{}, name;{}, perm:{}, ret:{}",
doType, limitType, value, hwDeviceId, hwDeviceName, perm, result32);
}
}
// 应用硬件接口管理模块的配置内容
int result5 = IHwPortController.instanceDll.APQ_HWPORT_Apply();
log.info("APQ_HWPORT_Apply ret:{}", result5);
}
} else {
log.info("执行硬件监控, sendTime:{}, params:{}", sendTime, params);
// 上报硬件监控数据
Map mapRequest = new HashMap();
for (HwPortControllerDto hwPortControllerDto : persistentHwPortControllerDtos) {
for (HwPortControllerDto hwPortControllerDto : hwPortControllerDtos) {
HwPortControllerDto.DeviceHardwareEnum deviceHardwareEnum = hwPortControllerDto.getKey();
int doType = hwPortControllerDto.getDoType();
int limitType = hwPortControllerDto.getLimitType();
......@@ -247,7 +163,8 @@ public class ApqControlJob {
Map map = restTemplate.postForObject("http://localhost:8765/uplink_event/oc-client", eventMapRequest, Map.class);
log.info("存在硬件接口扫描告警，上报告警事件: mapData:{}, ret:{}", mapRequest, map);
}
}
} catch (Exception e) {
log.error("APQ_HWPORT ERROR: {}", e.getMessage());
} finally {
result = IHwPortController.instanceDll.APQ_HWPORT_Uninitialize();
log.info("APQ_HWPORT_Uninitialize ret:{}", result);
......
apq-pc-common/src/main/java/com/priusis/service/control/DeviceChangeCallbackImpl.java
浏览文件 @ 9ec007c0
......@@ -44,27 +44,41 @@ public class DeviceChangeCallbackImpl implements IDeviceChangeCallback {
if (changeDataSplits.length >= 5) {
String classData = changeDataSplits[4];
String changeDataDesc = changeDataSplits[0];
boolean needPush = true;
String deviceClass = classData.split("=")[1].toUpperCase();
if (StrUtil.equalsIgnoreCase("class=Keyboard", classData)) {
changeDataDesc = "键盘";
} else if (StrUtil.equalsIgnoreCase("class=Mouse", classData)) {
changeDataDesc = "鼠标";
} else if (StrUtil.equalsIgnoreCase("class=DiskDrive", classData)) {
changeDataDesc = "USB";
} else if (StrUtil.equalsIgnoreCase("class=Bluetooth", classData)) {
changeDataDesc = "无线蓝牙";
} else if (StrUtil.equalsIgnoreCase("class=CDROM", classData)) {
changeDataDesc = "光驱";
deviceClass = "CD_DRIVE";
} else if (StrUtil.equalsIgnoreCase("class=Ports", classData)) {
changeDataDesc = "串口";
deviceClass = "SERIAL_PORT";
} else if (StrUtil.equalsIgnoreCase("class=USB", classData)) {
changeDataDesc = "U盘/移动硬盘";
deviceClass = "USB_DRIVE";
List<String> addUSBs = USBCheck.getAddUSB();
if (null != addUSBs) {
for (String addUSB : addUSBs) {
antivirusDoService.doAntivirus(addUSB + ":");
}
}
} else {
needPush = false;
}
String changeContent = String.format("%s%s", changeType == 1 ? "插入" : "拔出", changeDataDesc);
mapRequest.put("deviceClass", classData.split("=")[1]);
mapRequest.put("deviceClass", deviceClass);
mapRequest.put("changeContent", changeContent);
mapRequest.put("changeRemark", changeData);
log.info("存在硬件接口扫描告警，上报告警事件！changeContent: {}, changeRemark: {}", changeContent, changeData);
if (needPush) {
Map eventMapRequest = new HashMap();
eventMapRequest.put("requestId", 4);
eventMapRequest.put("methodName", "control_dev_warning");
......@@ -73,4 +87,5 @@ public class DeviceChangeCallbackImpl implements IDeviceChangeCallback {
log.info("存在硬件接口扫描告警，上报告警事件: mapData:{}, ret:{}", mapRequest, map);
}
}
}
}
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论