|
|
@@ -16,6 +16,7 @@ import com.coffee.bus.registry.device.DeviceOperator;
|
|
|
import com.coffee.bus.service.dto.DeviceQuery;
|
|
|
import com.coffee.bus.service.dto.DeviceResult;
|
|
|
import com.coffee.common.crud.BaseService;
|
|
|
+import com.coffee.common.exception.CustomException;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
|
import org.springframework.beans.factory.annotation.Autowired;
|
|
|
import org.springframework.context.annotation.Lazy;
|
|
|
@@ -24,6 +25,8 @@ import org.springframework.transaction.annotation.Transactional;
|
|
|
|
|
|
import java.util.List;
|
|
|
import java.util.Objects;
|
|
|
+import java.util.concurrent.CompletableFuture;
|
|
|
+import java.util.concurrent.ExecutionException;
|
|
|
import java.util.concurrent.atomic.AtomicReference;
|
|
|
|
|
|
/**
|
|
|
@@ -109,33 +112,60 @@ public class LocalBusDeviceService extends BaseService<BusDeviceMapper, BusDevic
|
|
|
* 根据deviceId更新设备在线状态
|
|
|
* @param device
|
|
|
*/
|
|
|
- public boolean updateByDeviceId(BusDeviceEntity device){
|
|
|
+ public boolean updateDevice(BusDeviceEntity device){
|
|
|
return this.update(device,new QueryWrapper<BusDeviceEntity>().lambda()
|
|
|
.eq(BusDeviceEntity::getDeviceId,device.getDeviceId()));
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
* @author 龙三郎
|
|
|
- * 添加设备
|
|
|
+ * 保存一个设备,系统中存在时就更新,不存在时则插入。
|
|
|
* @param entity
|
|
|
*/
|
|
|
- public boolean saveByDeviceId(BusDeviceEntity entity) {
|
|
|
+ public boolean saveDevice(BusDeviceEntity entity) {
|
|
|
// 查询设备是否存在,无视逻辑删除
|
|
|
- BusDeviceEntity old = deviceMapper.selectOneByDeviceId(entity.getDeviceId());
|
|
|
+ BusDeviceEntity device = deviceMapper.selectOneByDeviceId(entity.getDeviceId());
|
|
|
+ // 判断设备是否存在
|
|
|
+ boolean isExists = Objects.nonNull(device);
|
|
|
// 设备存在,且处于删除状态,首先去掉逻辑删除标志 ?? 逻辑删除后设备数据是否不再接收
|
|
|
- if (Objects.nonNull(old) && old.getIsDelete() == 1){
|
|
|
+ if (isExists && device.getIsDelete() == 1){
|
|
|
deviceMapper.notDelete(entity.getDeviceId());
|
|
|
}
|
|
|
// 设备存在
|
|
|
- if (Objects.nonNull(old)){
|
|
|
+ if (isExists){
|
|
|
// 更新设备
|
|
|
- return this.updateByDeviceId(entity);
|
|
|
+ return this.updateDevice(entity);
|
|
|
}else {
|
|
|
// 添加设备
|
|
|
return this.save(entity);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * @author 龙三郎
|
|
|
+ * 更新一个系统中现有的设备
|
|
|
+ * @param deviceId
|
|
|
+ */
|
|
|
+ public boolean updateDeviceByDeviceId(String deviceId) {
|
|
|
+ // 查询设备是否存在
|
|
|
+ BusDeviceEntity device = getByDeviceId(deviceId);
|
|
|
+ // 设备不存在直接退出
|
|
|
+ if (Objects.isNull(device)){
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ // 从阿里云物联网查询设备
|
|
|
+ QueryDeviceDetailResponse response = aliyunIotSdk.queryDeviceDetail(deviceId);
|
|
|
+ // 设备存在
|
|
|
+ if (response.getSuccess()){
|
|
|
+ // 更新设备参数
|
|
|
+ device.updateFields(response.getData());
|
|
|
+ // 更新设备
|
|
|
+ return this.updateDevice(device);
|
|
|
+ }else {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* @author 龙三郎
|
|
|
* 根据deviceId获取设备
|
|
|
@@ -155,7 +185,7 @@ public class LocalBusDeviceService extends BaseService<BusDeviceMapper, BusDevic
|
|
|
* @param deviceId
|
|
|
* @return
|
|
|
*/
|
|
|
- public BusDeviceEntity add(String deviceId) {
|
|
|
+ public BusDeviceEntity addDevice(String deviceId) {
|
|
|
BusDeviceEntity device = new BusDeviceEntity();
|
|
|
device.setConfig(new AliIotConfig());
|
|
|
// 从阿里云物联网查询设备
|
|
|
@@ -164,7 +194,7 @@ public class LocalBusDeviceService extends BaseService<BusDeviceMapper, BusDevic
|
|
|
if (response.getSuccess()){
|
|
|
device.updateFields(response.getData());
|
|
|
// 存储设备
|
|
|
- this.saveByDeviceId(device);
|
|
|
+ this.saveDevice(device);
|
|
|
return device;
|
|
|
}
|
|
|
// 不存在返回null
|
|
|
@@ -180,14 +210,44 @@ public class LocalBusDeviceService extends BaseService<BusDeviceMapper, BusDevic
|
|
|
* 从阿里云平台获取全部的设备,同步到系统数据库
|
|
|
* @return
|
|
|
*/
|
|
|
- public int syncDevice() {
|
|
|
+ public int syncAllDevice(){
|
|
|
+ Integer m = 0;
|
|
|
+ // 创建异步执行任务,有返回值
|
|
|
+ CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(()->{
|
|
|
+ AtomicReference<Integer> n = new AtomicReference<>(0);
|
|
|
+ // 同步所有的设备
|
|
|
+ aliyunIotSdk.queryDevice().forEach(item ->{
|
|
|
+ BusDeviceEntity device = new BusDeviceEntity();
|
|
|
+ device.setConfig(new AliIotConfig());
|
|
|
+ // 更新设备属性
|
|
|
+ device.updateFields(item);
|
|
|
+ n.updateAndGet(v -> v + (this.saveDevice(device)?1:0));
|
|
|
+ });
|
|
|
+ return n.get();
|
|
|
+ });
|
|
|
+ //等待子任务执行完成
|
|
|
+ try {
|
|
|
+ m = cf.get();
|
|
|
+ }catch (Exception e){
|
|
|
+ throw new CustomException(e.getMessage());
|
|
|
+ }
|
|
|
+ return m;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * @author 龙三郎
|
|
|
+ * 根据ids获取指定的设备,更新本地设备
|
|
|
+ * @return
|
|
|
+ */
|
|
|
+ public int syncDevice(List<String> ids) {
|
|
|
AtomicReference<Integer> n = new AtomicReference<>(0);
|
|
|
- aliyunIotSdk.queryDevice().forEach(item ->{
|
|
|
- BusDeviceEntity device = new BusDeviceEntity();
|
|
|
- device.setConfig(new AliIotConfig());
|
|
|
- // 更新设备属性
|
|
|
- device.updateFields(item);
|
|
|
- n.updateAndGet(v -> v + (this.saveByDeviceId(device)?1:0));
|
|
|
+ // ids不能为空
|
|
|
+ if (Objects.isNull(ids) || ids.size() <= 0){
|
|
|
+ throw new CustomException("设备id不能为空");
|
|
|
+ }
|
|
|
+ // 同步指定的设备
|
|
|
+ ids.forEach(id->{
|
|
|
+ n.updateAndGet(v -> v + (this.updateDeviceByDeviceId(id)?1:0));
|
|
|
});
|
|
|
return n.get();
|
|
|
}
|