Browse Source

Remove transactional from saveDevice (fromEdge). Fix order of uplink messages. Fix asset profile

pull/10264/head
Andrii Landiak 3 years ago
parent
commit
59a8858ef1
  1. 25
      application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
  2. 1
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/profile/BaseAssetProfileProcessor.java
  3. 13
      application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java
  4. 1
      dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java
  5. 20
      dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java

25
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java

@ -696,11 +696,6 @@ public final class EdgeGrpcSession implements Closeable {
private ListenableFuture<List<Void>> processUplinkMsg(UplinkMsg uplinkMsg) {
List<ListenableFuture<Void>> result = new ArrayList<>();
try {
if (uplinkMsg.getEntityDataCount() > 0) {
for (EntityDataProto entityData : uplinkMsg.getEntityDataList()) {
result.addAll(ctx.getTelemetryProcessor().processTelemetryMsg(edge.getTenantId(), entityData));
}
}
if (uplinkMsg.getDeviceProfileUpdateMsgCount() > 0) {
for (DeviceProfileUpdateMsg deviceProfileUpdateMsg : uplinkMsg.getDeviceProfileUpdateMsgList()) {
result.add(((DeviceProfileProcessor) ctx.getDeviceProfileEdgeProcessorFactory().getProcessorByEdgeVersion(this.edgeVersion))
@ -731,6 +726,17 @@ public final class EdgeGrpcSession implements Closeable {
.processAssetMsgFromEdge(edge.getTenantId(), edge, assetUpdateMsg));
}
}
if (uplinkMsg.getEntityViewUpdateMsgCount() > 0) {
for (EntityViewUpdateMsg entityViewUpdateMsg : uplinkMsg.getEntityViewUpdateMsgList()) {
result.add(((EntityViewProcessor) ctx.getEntityViewProcessorFactory().getProcessorByEdgeVersion(this.edgeVersion))
.processEntityViewMsgFromEdge(edge.getTenantId(), edge, entityViewUpdateMsg));
}
}
if (uplinkMsg.getEntityDataCount() > 0) {
for (EntityDataProto entityData : uplinkMsg.getEntityDataList()) {
result.addAll(ctx.getTelemetryProcessor().processTelemetryMsg(edge.getTenantId(), entityData));
}
}
if (uplinkMsg.getAlarmUpdateMsgCount() > 0) {
for (AlarmUpdateMsg alarmUpdateMsg : uplinkMsg.getAlarmUpdateMsgList()) {
result.add(((AlarmProcessor) ctx.getAlarmEdgeProcessorFactory().getProcessorByEdgeVersion(this.edgeVersion))
@ -743,12 +749,6 @@ public final class EdgeGrpcSession implements Closeable {
.processAlarmCommentMsgFromEdge(edge.getTenantId(), edge.getId(), alarmCommentUpdateMsg));
}
}
if (uplinkMsg.getEntityViewUpdateMsgCount() > 0) {
for (EntityViewUpdateMsg entityViewUpdateMsg : uplinkMsg.getEntityViewUpdateMsgList()) {
result.add(((EntityViewProcessor) ctx.getEntityViewProcessorFactory().getProcessorByEdgeVersion(this.edgeVersion))
.processEntityViewMsgFromEdge(edge.getTenantId(), edge, entityViewUpdateMsg));
}
}
if (uplinkMsg.getRelationUpdateMsgCount() > 0) {
for (RelationUpdateMsg relationUpdateMsg : uplinkMsg.getRelationUpdateMsgList()) {
result.add(((RelationProcessor) ctx.getRelationEdgeProcessorFactory().getProcessorByEdgeVersion(this.edgeVersion))
@ -763,7 +763,8 @@ public final class EdgeGrpcSession implements Closeable {
}
if (uplinkMsg.getResourceUpdateMsgCount() > 0) {
for (ResourceUpdateMsg resourceUpdateMsg : uplinkMsg.getResourceUpdateMsgList()) {
result.add(((ResourceProcessor) ctx.getResourceEdgeProcessorFactory().getProcessorByEdgeVersion(this.edgeVersion)).processResourceMsgFromEdge(edge.getTenantId(), edge, resourceUpdateMsg));
result.add(((ResourceProcessor) ctx.getResourceEdgeProcessorFactory().getProcessorByEdgeVersion(this.edgeVersion))
.processResourceMsgFromEdge(edge.getTenantId(), edge, resourceUpdateMsg));
}
}
if (uplinkMsg.getRuleChainMetadataRequestMsgCount() > 0) {

1
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/asset/profile/BaseAssetProfileProcessor.java

@ -47,7 +47,6 @@ public abstract class BaseAssetProfileProcessor extends BaseEdgeProcessor {
assetProfile.setId(assetProfileId);
assetProfile.setDefault(assetProfileById.isDefault());
}
assetProfile.setDefault(false);
String assetProfileName = assetProfile.getName();
AssetProfile assetProfileByName = assetProfileService.findAssetProfileByName(tenantId, assetProfileName);
if (assetProfileByName != null && !assetProfileByName.getId().equals(assetProfileId)) {

13
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/device/BaseDeviceProcessor.java

@ -23,7 +23,6 @@ import org.thingsboard.server.common.data.id.CustomerId;
import org.thingsboard.server.common.data.id.DeviceId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.security.DeviceCredentials;
import org.thingsboard.server.common.data.security.DeviceCredentialsType;
import org.thingsboard.server.gen.edge.v1.DeviceCredentialsUpdateMsg;
import org.thingsboard.server.gen.edge.v1.DeviceUpdateMsg;
import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor;
@ -63,13 +62,6 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
device.setId(deviceId);
}
Device savedDevice = deviceService.saveDevice(device, false);
if (created) {
DeviceCredentials deviceCredentials = new DeviceCredentials();
deviceCredentials.setDeviceId(new DeviceId(savedDevice.getUuidId()));
deviceCredentials.setCredentialsType(DeviceCredentialsType.ACCESS_TOKEN);
deviceCredentials.setCredentialsId(StringUtils.randomAlphanumeric(20));
deviceCredentialsService.createDeviceCredentials(device.getTenantId(), deviceCredentials);
}
tbClusterService.onDeviceUpdated(savedDevice, created ? null : device);
} catch (Exception e) {
log.error("[{}] Failed to process device update msg [{}]", tenantId, deviceUpdateMsg, e);
@ -91,6 +83,10 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
tenantId, device.getName(), deviceCredentials.getCredentialsId(), deviceCredentials.getCredentialsValue());
try {
DeviceCredentials deviceCredentialsByDeviceId = deviceCredentialsService.findDeviceCredentialsByDeviceId(tenantId, device.getId());
if (deviceCredentialsByDeviceId == null) {
deviceCredentialsByDeviceId = new DeviceCredentials();
deviceCredentialsByDeviceId.setDeviceId(device.getId());
}
deviceCredentialsByDeviceId.setCredentialsType(deviceCredentials.getCredentialsType());
deviceCredentialsByDeviceId.setCredentialsId(deviceCredentials.getCredentialsId());
deviceCredentialsByDeviceId.setCredentialsValue(deviceCredentials.getCredentialsValue());
@ -111,4 +107,5 @@ public abstract class BaseDeviceProcessor extends BaseEdgeProcessor {
protected abstract void setCustomerId(TenantId tenantId, CustomerId customerId, Device device, DeviceUpdateMsg deviceUpdateMsg);
protected abstract DeviceCredentials constructDeviceCredentialsFromUpdateMsg(TenantId tenantId, DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg);
}

1
dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java

@ -163,7 +163,6 @@ public class DeviceServiceImpl extends AbstractCachedEntityService<DeviceCacheKe
return doSaveDevice(device, accessToken, true);
}
@Transactional
@Override
public Device saveDevice(Device device, boolean doValidate) {
return doSaveDevice(device, null, doValidate);

20
dao/src/test/java/org/thingsboard/server/dao/service/DeviceServiceTest.java

@ -16,7 +16,6 @@
package org.thingsboard.server.dao.service;
import com.datastax.oss.driver.api.core.uuid.Uuids;
import org.hibernate.exception.ConstraintViolationException;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@ -57,7 +56,6 @@ import org.thingsboard.server.dao.service.validator.DeviceCredentialsDataValidat
import org.thingsboard.server.dao.tenant.TenantProfileService;
import java.nio.ByteBuffer;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@ -160,24 +158,6 @@ public class DeviceServiceTest extends AbstractServiceTest {
Assertions.assertNull(deviceByName);
}
@Test
public void testShouldRollbackNotValidatedDeviceIfDeviceCredentialsValidationFailed() {
Mockito.reset(validator);
Mockito.doThrow(new DataValidationException("mock message"))
.when(validator).validate(any(), any());
Device device = new Device();
device.setTenantId(tenantId);
device.setName(StringUtils.randomAlphabetic(10));
device.setType("default");
assertThatThrownBy(() -> deviceService.saveDevice(device, false))
.isInstanceOf(DataValidationException.class)
.hasMessageContaining("mock message");
Device deviceByName = deviceService.findDeviceByTenantIdAndName(tenantId, device.getName());
Assertions.assertNull(deviceByName);
}
@Test
public void testShouldRollbackValidatedDeviceIfDeviceCredentialsValidationFailed() {
Mockito.reset(validator);

Loading…
Cancel
Save