tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

Merge remote-tracking branch 'upstream/master' into mqttConnectCommandFix

pull/9570/head
dashevchenko 3 years ago
parent
56390b53a4 0daca87226
commit
921a441b9a
7 changed files with 170 additions and 101 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 131
      application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
  2. 51
      application/src/test/java/org/thingsboard/server/controller/DeviceConnectivityControllerTest.java
  3. 50
      application/src/test/java/org/thingsboard/server/controller/DeviceConnectivityControllerWithDefaultPortTest.java
  4. 7
      application/src/test/java/org/thingsboard/server/controller/WidgetsBundleControllerTest.java
  5. 4
      application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java
  6. 11
      common/cache/src/main/java/org/thingsboard/server/cache/device/DeviceCacheKey.java
  7. 17
      dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java

131
application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
View File

@ -192,7 +192,7 @@ public class DefaultTransportApiService implements TransportApiService {
final String certChain = msg.getCertificateChain();
result = handlerExecutor.submit(() -> validateOrCreateDeviceX509Certificate(certChain));
} else if (transportApiRequestMsg.hasGetOrCreateDeviceRequestMsg()) {
result = handle(transportApiRequestMsg.getGetOrCreateDeviceRequestMsg());
result = handlerExecutor.submit(() -> handle(transportApiRequestMsg.getGetOrCreateDeviceRequestMsg()));
} else if (transportApiRequestMsg.hasEntityProfileRequestMsg()) {
result = handle(transportApiRequestMsg.getEntityProfileRequestMsg());
} else if (transportApiRequestMsg.hasLwM2MRequestMsg()) {
@ -223,7 +223,6 @@ public class DefaultTransportApiService implements TransportApiService {
}
private TransportApiResponseMsg validateCredentials(String credentialsId, DeviceCredentialsType credentialsType) {
//TODO: Make async and enable caching
DeviceCredentials credentials = deviceCredentialsService.findDeviceCredentialsByCredentialsId(credentialsId);
if (credentials != null && credentials.getCredentialsType() == credentialsType) {
return getDeviceInfo(credentials);
@ -336,76 +335,74 @@ public class DefaultTransportApiService implements TransportApiService {
return VALID;
}
private ListenableFuture<TransportApiResponseMsg> handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
private TransportApiResponseMsg handle(GetOrCreateDeviceFromGatewayRequestMsg requestMsg) {
DeviceId gatewayId = new DeviceId(new UUID(requestMsg.getGatewayIdMSB(), requestMsg.getGatewayIdLSB()));
ListenableFuture<Device> gatewayFuture = deviceService.findDeviceByIdAsync(TenantId.SYS_TENANT_ID, gatewayId);
return Futures.transform(gatewayFuture, gateway -> {
Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(requestMsg.getDeviceName(), id -> new ReentrantLock());
deviceCreationLock.lock();
try {
Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName());
if (device == null) {
TenantId tenantId = gateway.getTenantId();
device = new Device();
device.setTenantId(tenantId);
device.setName(requestMsg.getDeviceName());
device.setType(requestMsg.getDeviceType());
device.setCustomerId(gateway.getCustomerId());
DeviceProfile deviceProfile = deviceProfileCache.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType());
device.setDeviceProfileId(deviceProfile.getId());
ObjectNode additionalInfo = JacksonUtil.newObjectNode();
additionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
device.setAdditionalInfo(additionalInfo);
Device savedDevice = deviceService.saveDevice(device);
tbClusterService.onDeviceUpdated(savedDevice, null);
device = savedDevice;
relationService.saveRelation(TenantId.SYS_TENANT_ID, new EntityRelation(gateway.getId(), device.getId(), "Created"));
TbMsgMetaData metaData = new TbMsgMetaData();
CustomerId customerId = gateway.getCustomerId();
if (customerId != null && !customerId.isNullUid()) {
metaData.putValue("customerId", customerId.toString());
}
metaData.putValue("gatewayId", gatewayId.toString());
Device gateway = deviceService.findDeviceById(TenantId.SYS_TENANT_ID, gatewayId);
Lock deviceCreationLock = deviceCreationLocks.computeIfAbsent(requestMsg.getDeviceName(), id -> new ReentrantLock());
deviceCreationLock.lock();
try {
Device device = deviceService.findDeviceByTenantIdAndName(gateway.getTenantId(), requestMsg.getDeviceName());
if (device == null) {
TenantId tenantId = gateway.getTenantId();
device = new Device();
device.setTenantId(tenantId);
device.setName(requestMsg.getDeviceName());
device.setType(requestMsg.getDeviceType());
device.setCustomerId(gateway.getCustomerId());
DeviceProfile deviceProfile = deviceProfileCache.findOrCreateDeviceProfile(gateway.getTenantId(), requestMsg.getDeviceType());
device.setDeviceProfileId(deviceProfile.getId());
ObjectNode additionalInfo = JacksonUtil.newObjectNode();
additionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
device.setAdditionalInfo(additionalInfo);
Device savedDevice = deviceService.saveDevice(device);
tbClusterService.onDeviceUpdated(savedDevice, null);
device = savedDevice;
relationService.saveRelation(TenantId.SYS_TENANT_ID, new EntityRelation(gateway.getId(), device.getId(), "Created"));
TbMsgMetaData metaData = new TbMsgMetaData();
CustomerId customerId = gateway.getCustomerId();
if (customerId != null && !customerId.isNullUid()) {
metaData.putValue("customerId", customerId.toString());
}
metaData.putValue("gatewayId", gatewayId.toString());
DeviceId deviceId = device.getId();
JsonNode entityNode = JacksonUtil.valueToTree(device);
TbMsg tbMsg = TbMsg.newMsg(TbMsgType.ENTITY_CREATED, deviceId, customerId, metaData, TbMsgDataType.JSON, JacksonUtil.toString(entityNode));
tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, null);
} else {
JsonNode deviceAdditionalInfo = device.getAdditionalInfo();
if (deviceAdditionalInfo == null) {
deviceAdditionalInfo = JacksonUtil.newObjectNode();
}
if (deviceAdditionalInfo.isObject() &&
(!deviceAdditionalInfo.has(DataConstants.LAST_CONNECTED_GATEWAY)
|| !gatewayId.toString().equals(deviceAdditionalInfo.get(DataConstants.LAST_CONNECTED_GATEWAY).asText()))) {
ObjectNode newDeviceAdditionalInfo = (ObjectNode) deviceAdditionalInfo;
newDeviceAdditionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
Device savedDevice = deviceService.saveDevice(device);
tbClusterService.onDeviceUpdated(savedDevice, device);
}
DeviceId deviceId = device.getId();
JsonNode entityNode = JacksonUtil.valueToTree(device);
TbMsg tbMsg = TbMsg.newMsg(TbMsgType.ENTITY_CREATED, deviceId, customerId, metaData, TbMsgDataType.JSON, JacksonUtil.toString(entityNode));
tbClusterService.pushMsgToRuleEngine(tenantId, deviceId, tbMsg, null);
} else {
JsonNode deviceAdditionalInfo = device.getAdditionalInfo();
if (deviceAdditionalInfo == null) {
deviceAdditionalInfo = JacksonUtil.newObjectNode();
}
GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
.setDeviceInfo(getDeviceInfoProto(device));
DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
if (deviceProfile != null) {
builder.setProfileBody(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile)));
} else {
log.warn("[{}] Failed to find device profile [{}] for device. ", device.getId(), device.getDeviceProfileId());
if (deviceAdditionalInfo.isObject() &&
(!deviceAdditionalInfo.has(DataConstants.LAST_CONNECTED_GATEWAY)
|| !gatewayId.toString().equals(deviceAdditionalInfo.get(DataConstants.LAST_CONNECTED_GATEWAY).asText()))) {
ObjectNode newDeviceAdditionalInfo = (ObjectNode) deviceAdditionalInfo;
newDeviceAdditionalInfo.put(DataConstants.LAST_CONNECTED_GATEWAY, gatewayId.toString());
Device savedDevice = deviceService.saveDevice(device);
tbClusterService.onDeviceUpdated(savedDevice, device);
}
return TransportApiResponseMsg.newBuilder()
.setGetOrCreateDeviceResponseMsg(builder.build())
.build();
} catch (JsonProcessingException e) {
log.warn("[{}] Failed to lookup device by gateway id and name: [{}]", gatewayId, requestMsg.getDeviceName(), e);
throw new RuntimeException(e);
} finally {
deviceCreationLock.unlock();
}
}, dbCallbackExecutorService);
GetOrCreateDeviceFromGatewayResponseMsg.Builder builder = GetOrCreateDeviceFromGatewayResponseMsg.newBuilder()
.setDeviceInfo(getDeviceInfoProto(device));
DeviceProfile deviceProfile = deviceProfileCache.get(device.getTenantId(), device.getDeviceProfileId());
if (deviceProfile != null) {
builder.setProfileBody(ByteString.copyFrom(dataDecodingEncodingService.encode(deviceProfile)));
} else {
log.warn("[{}] Failed to find device profile [{}] for device. ", device.getId(), device.getDeviceProfileId());
}
return TransportApiResponseMsg.newBuilder()
.setGetOrCreateDeviceResponseMsg(builder.build())
.build();
} catch (JsonProcessingException e) {
log.warn("[{}] Failed to lookup device by gateway id and name: [{}]", gatewayId, requestMsg.getDeviceName(), e);
throw new RuntimeException(e);
} finally {
deviceCreationLock.unlock();
}
}
private ListenableFuture<TransportApiResponseMsg> handle(ProvisionDeviceRequestMsg requestMsg) {

51
application/src/test/java/org/thingsboard/server/controller/DeviceConnectivityControllerTest.java
View File

@ -121,18 +121,47 @@ public class DeviceConnectivityControllerTest extends AbstractControllerTest {
loginSysAdmin();
ObjectNode config = JacksonUtil.newObjectNode();
ObjectNode http = JacksonUtil.newObjectNode();
http.put("enabled", true);
http.put("host", "");
http.put("port", 8080);
config.set("http", http);
ObjectNode https = JacksonUtil.newObjectNode();
https.put("enabled", true);
https.put("host", "");
https.put("port", 444);
config.set("https", https);
ObjectNode mqtt = JacksonUtil.newObjectNode();
mqtt.put("enabled", true);
mqtt.put("host", "");
mqtt.put("port", 1883);
config.set("mqtt", mqtt);
ObjectNode mqtts = JacksonUtil.newObjectNode();
mqtts.put("enabled", true);
mqtts.put("host", "");
mqtts.put("port", 8883);
config.set("mqtts", mqtts);
ObjectNode coap = JacksonUtil.newObjectNode();
coap.put("enabled", true);
coap.put("host", "");
coap.put("port", 5683);
config.set("coap", coap);
ObjectNode coaps = JacksonUtil.newObjectNode();
coaps.put("enabled", true);
coaps.put("host", "");
coaps.put("port", 5684);
config.set("coaps", coaps);
AdminSettings adminSettings = doGet("/api/admin/settings/connectivity", AdminSettings.class);
JsonNode connectivity = adminSettings.getJsonValue();
((ObjectNode)connectivity.get("http")).put("port", 8080);
((ObjectNode)connectivity.get("http")).put("enabled", true);
((ObjectNode)connectivity.get("https")).put("enabled", true);
((ObjectNode)connectivity.get("https")).put("port", 444);
((ObjectNode)connectivity.get("mqtt")).put("enabled", true);
((ObjectNode)connectivity.get("mqtts")).put("enabled", true);
((ObjectNode)connectivity.get("coap")).put("enabled", true);
((ObjectNode)connectivity.get("coaps")).put("enabled", true);
doPost("/api/admin/settings", adminSettings);
adminSettings.setJsonValue(config);
doPost("/api/admin/settings", adminSettings).andExpect(status().isOk());
Tenant tenant = new Tenant();
tenant.setTitle("My tenant");

50
application/src/test/java/org/thingsboard/server/controller/DeviceConnectivityControllerWithDefaultPortTest.java
View File

@ -29,6 +29,7 @@ import org.mockito.Mockito;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.test.context.ContextConfiguration;
import org.thingsboard.common.util.JacksonUtil;
import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.server.common.data.AdminSettings;
import org.thingsboard.server.common.data.Device;
@ -66,16 +67,47 @@ public class DeviceConnectivityControllerWithDefaultPortTest extends AbstractCon
loginSysAdmin();
ObjectNode config = JacksonUtil.newObjectNode();
ObjectNode http = JacksonUtil.newObjectNode();
http.put("enabled", true);
http.put("host", "");
http.put("port", 80);
config.set("http", http);
ObjectNode https = JacksonUtil.newObjectNode();
https.put("enabled", true);
https.put("host", "");
https.put("port", 443);
config.set("https", https);
ObjectNode mqtt = JacksonUtil.newObjectNode();
mqtt.put("enabled", false);
mqtt.put("host", "");
mqtt.put("port", 1883);
config.set("mqtt", mqtt);
ObjectNode mqtts = JacksonUtil.newObjectNode();
mqtts.put("enabled", false);
mqtts.put("host", "");
mqtts.put("port", 8883);
config.set("mqtts", mqtts);
ObjectNode coap = JacksonUtil.newObjectNode();
coap.put("enabled", false);
coap.put("host", "");
coap.put("port", 5683);
config.set("coap", coap);
ObjectNode coaps = JacksonUtil.newObjectNode();
coaps.put("enabled", false);
coaps.put("host", "");
coaps.put("port", 5684);
config.set("coaps", coaps);
AdminSettings adminSettings = doGet("/api/admin/settings/connectivity", AdminSettings.class);
JsonNode connectivity = adminSettings.getJsonValue();
((ObjectNode) connectivity.get("http")).put("port", 80);
((ObjectNode) connectivity.get("https")).put("enabled", true);
((ObjectNode) connectivity.get("mqtt")).put("enabled", false);
((ObjectNode) connectivity.get("mqtts")).put("enabled", false);
((ObjectNode) connectivity.get("coaps")).put("enabled", false);
((ObjectNode) connectivity.get("coap")).put("enabled", false);
doPost("/api/admin/settings", adminSettings);
adminSettings.setJsonValue(config);
doPost("/api/admin/settings", adminSettings).andExpect(status().isOk());
Tenant tenant = new Tenant();
tenant.setTitle("My tenant");

7
application/src/test/java/org/thingsboard/server/controller/WidgetsBundleControllerTest.java
View File

@ -268,6 +268,13 @@ public class WidgetsBundleControllerTest extends AbstractControllerTest {
Collections.sort(loadedWidgetsBundles2, idComparator);
Assert.assertEquals(tenantWidgetsBundles, loadedWidgetsBundles2);
// cleanup
loginSysAdmin();
for (WidgetsBundle sysWidgetsBundle : sysWidgetsBundles) {
doDelete("/api/widgetsBundle/" + sysWidgetsBundle.getId().getId().toString())
.andExpect(status().isOk());
}
}
@Test

4
application/src/test/java/org/thingsboard/server/service/queue/ruleengine/TbRuleEngineQueueConsumerManagerTest.java
View File

@ -458,7 +458,7 @@ public class TbRuleEngineQueueConsumerManagerTest {
verify(consumer2, never()).unsubscribe();
int msgCount = totalConsumedMsgs.get();
await().atLeast(4, TimeUnit.SECONDS) // based on topicDeletionDelayInSec
await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call)
.atMost(7, TimeUnit.SECONDS)
.untilAsserted(() -> {
partitions.stream()
@ -498,7 +498,7 @@ public class TbRuleEngineQueueConsumerManagerTest {
verify(consumer, never()).unsubscribe();
int msgCount = totalConsumedMsgs.get();
await().atLeast(4, TimeUnit.SECONDS)
await().atLeast(2, TimeUnit.SECONDS) // based on topicDeletionDelayInSec(5) = 5 - ( 3 seconds the code may execute starting consumerManager.delete() call)
.atMost(7, TimeUnit.SECONDS)
.untilAsserted(() -> {
partitions.stream()

11
common/cache/src/main/java/org/thingsboard/server/cache/device/DeviceCacheKey.java
View File

@ -38,17 +38,22 @@ public class DeviceCacheKey implements Serializable {
this(null, deviceId, null);
}
public DeviceCacheKey(TenantId tenantId, DeviceId deviceId) {
this(tenantId, deviceId, null);
}
public DeviceCacheKey(TenantId tenantId, String deviceName) {
this(tenantId, null, deviceName);
}
@Override
public String toString() {
if (deviceId != null) {
if (deviceId == null) {
return tenantId + "_n_" + deviceName;
} else if (tenantId == null) {
return deviceId.toString();
} else {
return tenantId + "_n_" + deviceName;
return tenantId + "_" + deviceId;
}
}
}

17
dao/src/main/java/org/thingsboard/server/dao/device/DeviceServiceImpl.java
View File

@ -130,15 +130,13 @@ public class DeviceServiceImpl extends AbstractCachedEntityService<DeviceCacheKe
public Device findDeviceById(TenantId tenantId, DeviceId deviceId) {
log.trace("Executing findDeviceById [{}]", deviceId);
validateId(deviceId, INCORRECT_DEVICE_ID + deviceId);
return cache.getAndPutInTransaction(new DeviceCacheKey(deviceId),
() -> {
//TODO: possible bug source since sometimes we need to clear cache by tenant id and sometimes by sys tenant id?
if (TenantId.SYS_TENANT_ID.equals(tenantId)) {
return deviceDao.findById(tenantId, deviceId.getId());
} else {
return deviceDao.findDeviceByTenantIdAndId(tenantId, deviceId.getId());
}
}, true);
if (TenantId.SYS_TENANT_ID.equals(tenantId)) {
return cache.getAndPutInTransaction(new DeviceCacheKey(deviceId),
() -> deviceDao.findById(tenantId, deviceId.getId()), true);
} else {
return cache.getAndPutInTransaction(new DeviceCacheKey(tenantId, deviceId),
() -> deviceDao.findDeviceByTenantIdAndId(tenantId, deviceId.getId()), true);
}
}
@Override
@ -259,6 +257,7 @@ public class DeviceServiceImpl extends AbstractCachedEntityService<DeviceCacheKe
keys.add(new DeviceCacheKey(event.getTenantId(), event.getNewName()));
if (event.getDeviceId() != null) {
keys.add(new DeviceCacheKey(event.getDeviceId()));
keys.add(new DeviceCacheKey(event.getTenantId(), event.getDeviceId()));
}
if (StringUtils.isNotEmpty(event.getOldName()) && !event.getOldName().equals(event.getNewName())) {
keys.add(new DeviceCacheKey(event.getTenantId(), event.getOldName()));

Write Preview
Loading…
Cancel
Save