dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactoring: process correct log warning, remove extra locks to process msg from edge

Browse Source
This commit is contained in:
Andrii Landiak 2023-08-16 09:24:00 +03:00
parent 308674ddc7
commit 5ee110b47e
12 changed files with 80 additions and 101 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -674,9 +674,9 @@ public final class EdgeGrpcSession implements Closeable {
result.add(ctx.getDeviceProcessor().processDeviceCredentialsMsg(edge.getTenantId(), deviceCredentialsUpdateMsg));
}
}
if (uplinkMsg.getDeviceProfileUpdateMsgCount() > 0) {
for (DeviceProfileUpdateMsg deviceProfileUpdateMsg : uplinkMsg.getDeviceProfileUpdateMsgList()) {
result.add(ctx.getDeviceProfileProcessor().processDeviceProfileMsgFromEdge(edge.getTenantId(), edge, deviceProfileUpdateMsg));
if (uplinkMsg.getAssetProfileUpdateMsgCount() > 0) {
for (AssetProfileUpdateMsg assetProfileUpdateMsg : uplinkMsg.getAssetProfileUpdateMsgList()) {
result.add(ctx.getAssetProfileProcessor().processAssetProfileMsgFromEdge(edge.getTenantId(), edge, assetProfileUpdateMsg));
}
}
if (uplinkMsg.getAssetUpdateMsgCount() > 0) {
@ -684,16 +684,16 @@ public final class EdgeGrpcSession implements Closeable {
result.add(ctx.getAssetProcessor().processAssetMsgFromEdge(edge.getTenantId(), edge, assetUpdateMsg));
}
}
if (uplinkMsg.getAssetProfileUpdateMsgCount() > 0) {
for (AssetProfileUpdateMsg assetProfileUpdateMsg : uplinkMsg.getAssetProfileUpdateMsgList()) {
result.add(ctx.getAssetProfileProcessor().processAssetProfileMsgFromEdge(edge.getTenantId(), edge, assetProfileUpdateMsg));
}
}
if (uplinkMsg.getAlarmUpdateMsgCount() > 0) {
for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) {
result.add(ctx.getAlarmProcessor().processAlarmMsg(edge.getTenantId(), alarmUpdateMsg));
}
}
if (uplinkMsg.getDeviceProfileUpdateMsgCount() > 0) {
for (DeviceProfileUpdateMsg deviceProfileUpdateMsg : uplinkMsg.getDeviceProfileUpdateMsgList()) {
result.add(ctx.getDeviceProfileProcessor().processDeviceProfileMsgFromEdge(edge.getTenantId(), edge, deviceProfileUpdateMsg));
}
}
if (uplinkMsg.getRelationUpdateMsgCount() > 0) {
for (RelationUpdateMsg relationUpdateMsg : uplinkMsg.getRelationUpdateMsgList()) {
result.add(ctx.getRelationProcessor().processRelationMsg(edge.getTenantId(), relationUpdateMsg));

4
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/BaseEdgeProcessor.java
View File

@ -114,11 +114,7 @@ import java.util.concurrent.locks.ReentrantLock;
public abstract class BaseEdgeProcessor {
protected static final Lock deviceCreationLock = new ReentrantLock();
protected static final Lock deviceProfileCreationLock = new ReentrantLock();
protected static final Lock assetCreationLock = new ReentrantLock();
protected static final Lock assetProfileCreationLock = new ReentrantLock();
protected static final Lock dashboardCreationLock = new ReentrantLock();
protected static final Lock entityViewCreationLock = new ReentrantLock();
protected static final int DEFAULT_PAGE_SIZE = 100;

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/AssetEdgeProcessor.java
View File

@ -115,7 +115,7 @@ public class AssetEdgeProcessor extends BaseAssetProcessor {
@Override
public void onFailure(Throwable t) {
log.debug("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", asset, t);
log.warn("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", asset, t);
}
});
} catch (JsonProcessingException | IllegalArgumentException e) {

8
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/AssetProfileEdgeProcessor.java
View File

@ -64,12 +64,8 @@ public class AssetProfileEdgeProcessor extends BaseAssetProfileProcessor {
return handleUnsupportedMsgType(assetProfileUpdateMsg.getMsgType());
}
} catch (DataValidationException e) {
if (e.getMessage().contains("limit reached")) {
log.warn("[{}] Number of allowed asset profile violated {}", tenantId, assetProfileUpdateMsg, e);
return Futures.immediateFuture(null);
} else {
log.warn("Failed to process AssetProfileUpdateMsg from Edge [{}]", assetProfileUpdateMsg, e);
return Futures.immediateFailedFuture(e);
}
} finally {
edgeSynchronizationManager.getSync().remove();
}
@ -97,7 +93,7 @@ public class AssetProfileEdgeProcessor extends BaseAssetProfileProcessor {
@Override
public void onFailure(Throwable t) {
log.debug("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", assetProfile, t);
log.warn("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", assetProfile, t);
}
});
} catch (JsonProcessingException | IllegalArgumentException e) {

4
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/BaseAssetProfileProcessor.java
View File

@ -33,7 +33,7 @@ public class BaseAssetProfileProcessor extends BaseEdgeProcessor {
protected boolean saveOrUpdateAssetProfile(TenantId tenantId, AssetProfileId assetProfileId, AssetProfileUpdateMsg assetProfileUpdateMsg) {
boolean created = false;
assetProfileCreationLock.lock();
assetCreationLock.lock();
try {
AssetProfile assetProfile = assetProfileService.findAssetProfileById(tenantId, assetProfileId);
String assetProfileName = assetProfileUpdateMsg.getName();
@ -62,7 +62,7 @@ public class BaseAssetProfileProcessor extends BaseEdgeProcessor {
}
assetProfileService.saveAssetProfile(assetProfile, false);
} finally {
assetProfileCreationLock.unlock();
assetCreationLock.unlock();
}
return created;
}

8
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/BaseDashboardProcessor.java
View File

@ -34,8 +34,6 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor {
protected boolean saveOrUpdateDashboard(TenantId tenantId, DashboardId dashboardId, DashboardUpdateMsg dashboardUpdateMsg, CustomerId customerId) {
boolean created = false;
dashboardCreationLock.lock();
try {
Dashboard dashboard = dashboardService.findDashboardById(tenantId, dashboardId);
if (dashboard == null) {
created = true;
@ -47,7 +45,8 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor {
dashboard.setConfiguration(JacksonUtil.toJsonNode(dashboardUpdateMsg.getConfiguration()));
Set<ShortCustomerInfo> assignedCustomers = null;
if (dashboardUpdateMsg.hasAssignedCustomers()) {
assignedCustomers = JacksonUtil.fromString(dashboardUpdateMsg.getAssignedCustomers(), new TypeReference<>() {});
assignedCustomers = JacksonUtil.fromString(dashboardUpdateMsg.getAssignedCustomers(), new TypeReference<>() {
});
dashboard.setAssignedCustomers(assignedCustomers);
}
@ -65,9 +64,6 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor {
} else {
unassignCustomersFromDashboard(tenantId, savedDashboard);
}
} finally {
dashboardCreationLock.unlock();
}
return created;
}

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/dashboard/DashboardEdgeProcessor.java
View File

@ -105,7 +105,7 @@ public class DashboardEdgeProcessor extends BaseDashboardProcessor {
@Override
public void onFailure(Throwable t) {
log.debug("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", dashboard, t);
log.warn("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", dashboard, t);
}
});
} catch (JsonProcessingException | IllegalArgumentException e) {

4
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProfileProcessor.java
View File

@ -45,7 +45,7 @@ public class BaseDeviceProfileProcessor extends BaseEdgeProcessor {
protected boolean saveOrUpdateDeviceProfile(TenantId tenantId, DeviceProfileId deviceProfileId, DeviceProfileUpdateMsg deviceProfileUpdateMsg) {
boolean created = false;
deviceProfileCreationLock.lock();
deviceCreationLock.lock();
try {
DeviceProfile deviceProfile = deviceProfileService.findDeviceProfileById(tenantId, deviceProfileId);
if (deviceProfile == null) {
@ -95,7 +95,7 @@ public class BaseDeviceProfileProcessor extends BaseEdgeProcessor {
}
deviceProfileService.saveDeviceProfile(deviceProfile, false);
} finally {
deviceProfileCreationLock.unlock();
deviceCreationLock.unlock();
}
return created;
}

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceEdgeProcessor.java
View File

@ -124,7 +124,7 @@ public class DeviceEdgeProcessor extends BaseDeviceProcessor {
@Override
public void onFailure(Throwable t) {
log.debug("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", device, t);
log.warn("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", device, t);
}
});
} catch (JsonProcessingException | IllegalArgumentException e) {

8
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/DeviceProfileEdgeProcessor.java
View File

@ -65,12 +65,8 @@ public class DeviceProfileEdgeProcessor extends BaseDeviceProfileProcessor {
return handleUnsupportedMsgType(deviceProfileUpdateMsg.getMsgType());
}
} catch (DataValidationException e) {
if (e.getMessage().contains("limit reached")) {
log.warn("[{}] Number of allowed device profile violated {}", tenantId, deviceProfileUpdateMsg, e);
return Futures.immediateFuture(null);
} else {
log.warn("Failed to process DeviceProfileUpdateMsg from Edge [{}]", deviceProfileUpdateMsg, e);
return Futures.immediateFailedFuture(e);
}
} finally {
edgeSynchronizationManager.getSync().remove();
}
@ -98,7 +94,7 @@ public class DeviceProfileEdgeProcessor extends BaseDeviceProfileProcessor {
@Override
public void onFailure(Throwable t) {
log.debug("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", deviceProfile, t);
log.warn("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", deviceProfile, t);
}
});
} catch (JsonProcessingException | IllegalArgumentException e) {

5
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/BaseEntityViewProcessor.java
View File

@ -38,8 +38,6 @@ public abstract class BaseEntityViewProcessor extends BaseEdgeProcessor {
protected Pair<Boolean, Boolean> saveOrUpdateEntityView(TenantId tenantId, EntityViewId entityViewId, EntityViewUpdateMsg entityViewUpdateMsg, CustomerId customerId) {
boolean created = false;
boolean entityViewNameUpdated = false;
entityViewCreationLock.lock();
try {
EntityView entityView = entityViewService.findEntityViewById(tenantId, entityViewId);
String entityViewName = entityViewUpdateMsg.getName();
if (entityView == null) {
@ -73,9 +71,6 @@ public abstract class BaseEntityViewProcessor extends BaseEdgeProcessor {
entityView.setId(entityViewId);
}
entityViewService.saveEntityView(entityView, false);
} finally {
entityViewCreationLock.unlock();
}
return Pair.of(created, entityViewNameUpdated);
}
}

2
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/entityview/EntityViewEdgeProcessor.java
View File

@ -113,7 +113,7 @@ public class EntityViewEdgeProcessor extends BaseEntityViewProcessor {
@Override
public void onFailure(Throwable t) {
log.debug("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", entityView, t);
log.warn("Failed to send ENTITY_CREATED EVENT to rule engine [{}]", entityView, t);
}
});
} catch (JsonProcessingException | IllegalArgumentException e) {