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

Merge branch 'rc' into error-processing-device-profile

Browse Source
This commit is contained in:
yevhenii_zahrebelnyi 2025-02-11 16:59:51 +02:00 committed by GitHub
commit ca1c8f71c0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
27 changed files with 187 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
application/src/main/data/json/edge/instructions/upgrade/docker/upgrade_db.md
View File

@ -43,19 +43,6 @@ volumes:
Execute the following command to start upgrade process:
```bash
docker compose -f docker-compose-upgrade.yml up
{:copy-code}
```
Once upgrade process successfully completed, exit from the docker-compose shell by this combination:
```text
Ctrl + C
```
Execute the following command to stop TB Edge upgrade container:
```bash
docker compose -f docker-compose-upgrade.yml stop
docker compose -f docker-compose-upgrade.yml up --abort-on-container-exit
{:copy-code}
```

2
application/src/main/java/org/thingsboard/server/actors/device/DeviceActorMessageProcessor.java
View File

@ -865,6 +865,8 @@ public class DeviceActorMessageProcessor extends AbstractContextAwareMsgProcesso
}
private void notifyTransportAboutClosedSessionMaxSessionsLimit(UUID sessionId, SessionInfoMetaData sessionMd) {
attributeSubscriptions.remove(sessionId);
rpcSubscriptions.remove(sessionId);
notifyTransportAboutClosedSession(sessionId, sessionMd, TransportSessionCloseReason.MAX_CONCURRENT_SESSIONS_LIMIT_REACHED);
}

3
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcService.java
View File

@ -404,6 +404,9 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
private void scheduleEdgeEventsCheck(EdgeGrpcSession session) {
EdgeId edgeId = session.getEdge().getId();
TenantId tenantId = session.getEdge().getTenantId();
cancelScheduleEdgeEventsCheck(edgeId);
if (sessions.containsKey(edgeId)) {
ScheduledFuture<?> edgeEventCheckTask = edgeEventProcessingExecutorService.schedule(() -> {
try {

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

@ -448,7 +448,11 @@ public abstract class EdgeGrpcSession implements Closeable {
private void scheduleDownlinkMsgsPackSend(int attempt) {
Runnable sendDownlinkMsgsTask = () -> {
try {
if (isConnected() && !sessionState.getPendingMsgsMap().values().isEmpty()) {
if (!isConnected()) {
stopCurrentSendDownlinkMsgsTask(true);
return;
}
if (!sessionState.getPendingMsgsMap().values().isEmpty()) {
List<DownlinkMsg> copy = new ArrayList<>(sessionState.getPendingMsgsMap().values());
if (attempt > 1) {
String error = "Failed to deliver the batch";
@ -529,6 +533,11 @@ public abstract class EdgeGrpcSession implements Closeable {
log.debug("[{}][{}][{}] Msg has been processed successfully! Msg Id: [{}], Msg: {}", tenantId, edge.getId(), sessionId, msg.getDownlinkMsgId(), msg);
} else {
log.error("[{}][{}][{}] Msg processing failed! Msg Id: [{}], Error msg: {}", tenantId, edge.getId(), sessionId, msg.getDownlinkMsgId(), msg.getErrorMsg());
DownlinkMsg downlinkMsg = sessionState.getPendingMsgsMap().get(msg.getDownlinkMsgId());
// if NOT timeseries or attributes failures - ack failed downlink
if (downlinkMsg.getEntityDataCount() == 0) {
sessionState.getPendingMsgsMap().remove(msg.getDownlinkMsgId());
}
}
if (sessionState.getPendingMsgsMap().isEmpty()) {
log.debug("[{}][{}][{}] Pending msgs map is empty. Stopping current iteration", tenantId, edge.getId(), sessionId);

18
application/src/main/java/org/thingsboard/server/service/edge/rpc/fetch/GeneralEdgeEventFetcher.java
View File

@ -54,23 +54,25 @@ public class GeneralEdgeEventFetcher implements EdgeEventFetcher {
log.trace("[{}] Finding general edge events [{}], seqIdStart = {}, pageLink = {}",
tenantId, edge.getId(), seqIdStart, pageLink);
PageData<EdgeEvent> edgeEvents = edgeEventService.findEdgeEvents(tenantId, edge.getId(), seqIdStart, null, (TimePageLink) pageLink);
if (edgeEvents.getData().isEmpty()) {
if (!edgeEvents.getData().isEmpty()) {
return edgeEvents;
}
if (seqIdStart > this.maxReadRecordsCount) {
edgeEvents = edgeEventService.findEdgeEvents(tenantId, edge.getId(), 0L, Math.max(this.maxReadRecordsCount, seqIdStart - this.maxReadRecordsCount), (TimePageLink) pageLink);
if (edgeEvents.getData().stream().anyMatch(ee -> ee.getSeqId() < seqIdStart)) {
log.info("[{}] seqId column of edge_event table started new cycle [{}]", tenantId, edge.getId());
this.seqIdNewCycleStarted = true;
this.seqIdStart = 0L;
} else {
edgeEvents = new PageData<>();
log.warn("[{}] unexpected edge notification message received. " +
"no new events found and seqId column of edge_event table doesn't started new cycle [{}]", tenantId, edge.getId());
return edgeEvents;
}
}
return edgeEvents;
log.info("[{}] Unexpected edge notification message received. " +
"No new events found, and the seqId column of the edge_event table has not started a new cycle [{}].", tenantId, edge.getId());
return new PageData<>();
} catch (Exception e) {
log.error("[{}] failed to find edge events [{}]", tenantId, edge.getId());
log.error("[{}] Failed to find edge events [{}]", tenantId, edge.getId(), e);
return new PageData<>();
}
return new PageData<>();
}
}

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

@ -26,6 +26,7 @@ import org.thingsboard.server.dao.service.DataValidator;
import org.thingsboard.server.gen.edge.v1.DashboardUpdateMsg;
import org.thingsboard.server.service.edge.rpc.processor.BaseEdgeProcessor;
import java.util.HashSet;
import java.util.Set;
@Slf4j
@ -40,54 +41,66 @@ public abstract class BaseDashboardProcessor extends BaseEdgeProcessor {
if (dashboard == null) {
throw new RuntimeException("[{" + tenantId + "}] dashboardUpdateMsg {" + dashboardUpdateMsg + "} cannot be converted to dashboard");
}
Set<ShortCustomerInfo> assignedCustomers = null;
Set<ShortCustomerInfo> newAssignedCustomers = new HashSet<>();
if (dashboard.getAssignedCustomers() != null && !dashboard.getAssignedCustomers().isEmpty()) {
newAssignedCustomers.addAll(dashboard.getAssignedCustomers());
}
Dashboard dashboardById = edgeCtx.getDashboardService().findDashboardById(tenantId, dashboardId);
if (dashboardById == null) {
created = true;
dashboard.setId(null);
dashboard.setAssignedCustomers(null);
} else {
dashboard.setId(dashboardId);
assignedCustomers = filterNonExistingCustomers(tenantId, dashboardById.getAssignedCustomers());
dashboard.setAssignedCustomers(dashboardById.getAssignedCustomers());
}
dashboardValidator.validate(dashboard, Dashboard::getTenantId);
if (created) {
dashboard.setId(dashboardId);
}
Set<ShortCustomerInfo> msgAssignedCustomers = filterNonExistingCustomers(tenantId, dashboard.getAssignedCustomers());
if (msgAssignedCustomers != null) {
if (assignedCustomers == null) {
assignedCustomers = msgAssignedCustomers;
} else {
assignedCustomers.addAll(msgAssignedCustomers);
}
}
dashboard.setAssignedCustomers(assignedCustomers);
Dashboard savedDashboard = edgeCtx.getDashboardService().saveDashboard(dashboard, false);
if (msgAssignedCustomers != null && !msgAssignedCustomers.isEmpty()) {
for (ShortCustomerInfo assignedCustomer : msgAssignedCustomers) {
if (assignedCustomer.getCustomerId().equals(customerId)) {
edgeCtx.getDashboardService().assignDashboardToCustomer(tenantId, savedDashboard.getId(), assignedCustomer.getCustomerId());
}
}
} else {
unassignCustomersFromDashboard(tenantId, savedDashboard, customerId);
}
updateDashboardAssignments(tenantId, dashboardById, savedDashboard, newAssignedCustomers);
return created;
}
private void unassignCustomersFromDashboard(TenantId tenantId, Dashboard dashboard, CustomerId customerId) {
if (dashboard.getAssignedCustomers() != null && !dashboard.getAssignedCustomers().isEmpty()) {
for (ShortCustomerInfo assignedCustomer : dashboard.getAssignedCustomers()) {
if (assignedCustomer.getCustomerId().equals(customerId)) {
edgeCtx.getDashboardService().unassignDashboardFromCustomer(tenantId, dashboard.getId(), assignedCustomer.getCustomerId());
}
private void updateDashboardAssignments(TenantId tenantId, Dashboard dashboardById, Dashboard savedDashboard, Set<ShortCustomerInfo> newAssignedCustomers) {
Set<ShortCustomerInfo> currentAssignedCustomers = new HashSet<>();
if (dashboardById != null) {
if (dashboardById.getAssignedCustomers() != null) {
currentAssignedCustomers.addAll(dashboardById.getAssignedCustomers());
}
}
newAssignedCustomers = filterNonExistingCustomers(tenantId, currentAssignedCustomers, newAssignedCustomers);
Set<CustomerId> addedCustomerIds = new HashSet<>();
Set<CustomerId> removedCustomerIds = new HashSet<>();
for (ShortCustomerInfo newAssignedCustomer : newAssignedCustomers) {
if (!savedDashboard.isAssignedToCustomer(newAssignedCustomer.getCustomerId())) {
addedCustomerIds.add(newAssignedCustomer.getCustomerId());
}
}
for (ShortCustomerInfo currentAssignedCustomer : currentAssignedCustomers) {
if (!newAssignedCustomers.contains(currentAssignedCustomer)) {
removedCustomerIds.add(currentAssignedCustomer.getCustomerId());
}
}
for (CustomerId customerIdToAdd : addedCustomerIds) {
edgeCtx.getDashboardService().assignDashboardToCustomer(tenantId, savedDashboard.getId(), customerIdToAdd);
}
for (CustomerId customerIdToRemove : removedCustomerIds) {
edgeCtx.getDashboardService().unassignDashboardFromCustomer(tenantId, savedDashboard.getId(), customerIdToRemove);
}
}
protected abstract Dashboard constructDashboardFromUpdateMsg(TenantId tenantId, DashboardId dashboardId, DashboardUpdateMsg dashboardUpdateMsg);
protected abstract Set<ShortCustomerInfo> filterNonExistingCustomers(TenantId tenantId, Set<ShortCustomerInfo> assignedCustomers);
protected abstract Set<ShortCustomerInfo> filterNonExistingCustomers(TenantId tenantId, Set<ShortCustomerInfo> currentAssignedCustomers, Set<ShortCustomerInfo> newAssignedCustomers);
}

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

@ -128,9 +128,9 @@ public abstract class DashboardEdgeProcessor extends BaseDashboardProcessor impl
}
@Override
protected Set<ShortCustomerInfo> filterNonExistingCustomers(TenantId tenantId, Set<ShortCustomerInfo> assignedCustomers) {
// do nothing on cloud
return assignedCustomers;
protected Set<ShortCustomerInfo> filterNonExistingCustomers(TenantId tenantId, Set<ShortCustomerInfo> currentAssignedCustomers, Set<ShortCustomerInfo> newAssignedCustomers) {
newAssignedCustomers.addAll(currentAssignedCustomers);
return newAssignedCustomers;
}
}

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

@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.gen.edge.v1.DashboardUpdateMsg;
import org.thingsboard.server.queue.util.TbCoreComponent;
import java.util.HashSet;
import java.util.Set;
@Component
@ -44,7 +45,7 @@ public class DashboardEdgeProcessorV1 extends DashboardEdgeProcessor {
Set<ShortCustomerInfo> assignedCustomers;
if (dashboardUpdateMsg.hasAssignedCustomers()) {
assignedCustomers = JacksonUtil.fromString(dashboardUpdateMsg.getAssignedCustomers(), new TypeReference<>() {});
assignedCustomers = filterNonExistingCustomers(tenantId, assignedCustomers);
assignedCustomers = filterNonExistingCustomers(tenantId, new HashSet<>(), assignedCustomers);
dashboard.setAssignedCustomers(assignedCustomers);
}
dashboard.setMobileOrder(dashboardUpdateMsg.hasMobileOrder() ? dashboardUpdateMsg.getMobileOrder() : null);

3
application/src/main/java/org/thingsboard/server/service/edge/rpc/processor/resource/BaseResourceProcessor.java
View File

@ -53,6 +53,9 @@ public abstract class BaseResourceProcessor extends BaseEdgeProcessor {
}
String resourceKey = resource.getResourceKey();
ResourceType resourceType = resource.getResourceType();
if (!created && !resourceType.isUpdatable()) {
resource.setData(null);
}
PageDataIterable<TbResource> resourcesIterable = new PageDataIterable<>(
link -> edgeCtx.getResourceService().findTenantResourcesByResourceTypeAndPageLink(tenantId, resourceType, link), 1024);
for (TbResource tbResource : resourcesIterable) {

5
application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java
View File

@ -28,17 +28,22 @@ import org.thingsboard.server.gen.edge.v1.WidgetBundleTypesRequestMsg;
public interface EdgeRequestsService {
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processRuleChainMetadataRequestMsg(TenantId tenantId, Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg);
ListenableFuture<Void> processAttributesRequestMsg(TenantId tenantId, Edge edge, AttributesRequestMsg attributesRequestMsg);
ListenableFuture<Void> processRelationRequestMsg(TenantId tenantId, Edge edge, RelationRequestMsg relationRequestMsg);
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processDeviceCredentialsRequestMsg(TenantId tenantId, Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg);
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processUserCredentialsRequestMsg(TenantId tenantId, Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg);
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processWidgetBundleTypesRequestMsg(TenantId tenantId, Edge edge, WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg);
@Deprecated(since = "3.9.1", forRemoval = true)
ListenableFuture<Void> processEntityViewsRequestMsg(TenantId tenantId, Edge edge, EntityViewsRequestMsg entityViewsRequestMsg);
}

5
application/src/main/java/org/thingsboard/server/service/rpc/DefaultTbRuleEngineRpcService.java
View File

@ -115,8 +115,9 @@ public class DefaultTbRuleEngineRpcService implements TbRuleEngineDeviceRpcServi
ToDeviceRpcRequest request = new ToDeviceRpcRequest(src.getRequestUUID(), src.getTenantId(), src.getDeviceId(),
src.isOneway(), src.getExpirationTime(), new ToDeviceRpcRequestBody(src.getMethod(), src.getBody()), src.isPersisted(), src.getRetries(), src.getAdditionalInfo());
forwardRpcRequestToDeviceActor(request, response -> {
if (src.isRestApiCall()) {
sendRpcResponseToTbCore(src.getOriginServiceId(), response);
String originServiceId = src.getOriginServiceId();
if (src.isRestApiCall() && originServiceId != null) {
sendRpcResponseToTbCore(originServiceId, response);
}
consumer.accept(RuleEngineDeviceRpcResponse.builder()
.deviceId(src.getDeviceId())

4
application/src/main/resources/thingsboard.yml
View File

@ -212,9 +212,9 @@ ui:
database:
ts_max_intervals: "${DATABASE_TS_MAX_INTERVALS:700}" # Max number of DB queries generated by a single API call to fetch telemetry records
ts:
type: "${DATABASE_TS_TYPE:sql}" # cassandra or sql. timescale option is deprecated and will no longer be supported in ThingsBoard 4.0
type: "${DATABASE_TS_TYPE:sql}" # cassandra, sql, or timescale (for hybrid mode, DATABASE_TS_TYPE value should be cassandra, or timescale)
ts_latest:
type: "${DATABASE_TS_LATEST_TYPE:sql}" # cassandra or sql. timescale option is deprecated and will no longer be supported in ThingsBoard 4.0
type: "${DATABASE_TS_LATEST_TYPE:sql}" # cassandra, sql, or timescale (for hybrid mode, DATABASE_TS_TYPE value should be cassandra, or timescale)
# Cassandra driver configuration parameters
cassandra:

38
application/src/test/java/org/thingsboard/server/edge/DashboardEdgeTest.java
View File

@ -16,6 +16,7 @@
package org.thingsboard.server.edge;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.collect.Sets;
import com.google.protobuf.AbstractMessage;
import org.junit.Assert;
import org.junit.Test;
@ -47,12 +48,14 @@ public class DashboardEdgeTest extends AbstractEdgeTest {
private static final int MOBILE_ORDER = 5;
private static final String IMAGE = "data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHZpZXdCb3g9IjAgMCAyNCAyNCIgd2lkdGg9IjQ4IiBoZWlnaHQ9IjQ4Ij48cGF0aCBkPSJNMTMuMjMgMTAuNTZWMTBjLTEuOTQgMC0zLjk5LjM5LTMuOTkgMi42NyAwIDEuMTYuNjEgMS45NSAxLjYzIDEuOTUuNzYgMCAxLjQzLS40NyAxLjg2LTEuMjIuNTItLjkzLjUtMS44LjUtMi44NG0yLjcgNi41M2MtLjE4LjE2LS40My4xNy0uNjMuMDYtLjg5LS43NC0xLjA1LTEuMDgtMS41NC0xLjc5LTEuNDcgMS41LTIuNTEgMS45NS00LjQyIDEuOTUtMi4yNSAwLTQuMDEtMS4zOS00LjAxLTQuMTcgMC0yLjE4IDEuMTctMy42NCAyLjg2LTQuMzggMS40Ni0uNjQgMy40OS0uNzYgNS4wNC0uOTNWNy41YzAtLjY2LjA1LTEuNDEtLjMzLTEuOTYtLjMyLS40OS0uOTUtLjctMS41LS43LTEuMDIgMC0xLjkzLjUzLTIuMTUgMS42MS0uMDUuMjQtLjI1LjQ4LS40Ny40OWwtMi42LS4yOGMtLjIyLS4wNS0uNDYtLjIyLS40LS41Ni42LTMuMTUgMy40NS00LjEgNi00LjEgMS4zIDAgMyAuMzUgNC4wMyAxLjMzQzE3LjExIDQuNTUgMTcgNi4xOCAxNyA3Ljk1djQuMTdjMCAxLjI1LjUgMS44MSAxIDIuNDguMTcuMjUuMjEuNTQgMCAuNzFsLTIuMDYgMS43OGgtLjAxIj48L3BhdGg+PHBhdGggZD0iTTIwLjE2IDE5LjU0QzE4IDIxLjE0IDE0LjgyIDIyIDEyLjEgMjJjLTMuODEgMC03LjI1LTEuNDEtOS44NS0zLjc2LS4yLS4xOC0uMDItLjQzLjI1LS4yOSAyLjc4IDEuNjMgNi4yNSAyLjYxIDkuODMgMi42MSAyLjQxIDAgNS4wNy0uNSA3LjUxLTEuNTMuMzctLjE2LjY2LjI0LjMyLjUxIj48L3BhdGg+PHBhdGggZD0iTTIxLjA3IDE4LjVjLS4yOC0uMzYtMS44NS0uMTctMi41Ny0uMDgtLjE5LjAyLS4yMi0uMTYtLjAzLS4zIDEuMjQtLjg4IDMuMjktLjYyIDMuNTMtLjMzLjI0LjMtLjA3IDIuMzUtMS4yNCAzLjMyLS4xOC4xNi0uMzUuMDctLjI2LS4xMS4yNi0uNjcuODUtMi4xNC41Ny0yLjV6Ij48L3BhdGg+PC9zdmc+";
private static final String DASHBOARD_TITLE = "Edge Test Dashboard";
@Test
public void testDashboards() throws Exception {
// create dashboard and assign to edge
edgeImitator.expectMessageAmount(2);
Dashboard dashboard = new Dashboard();
dashboard.setTitle("Edge Test Dashboard");
dashboard.setTitle(DASHBOARD_TITLE);
dashboard.setMobileHide(true);
dashboard.setImage(IMAGE);
dashboard.setMobileOrder(MOBILE_ORDER);
@ -175,7 +178,11 @@ public class DashboardEdgeTest extends AbstractEdgeTest {
@Test
public void testSendDashboardToCloud() throws Exception {
Dashboard dashboard = buildDashboardForUplinkMsg();
Customer customer = new Customer();
customer.setTitle("Edge Customer");
Customer savedCustomer = doPost("/api/customer", customer, Customer.class);
Dashboard dashboard = buildDashboardForUplinkMsg(savedCustomer);
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
DashboardUpdateMsg.Builder dashboardUpdateMsgBuilder = DashboardUpdateMsg.newBuilder();
@ -195,7 +202,27 @@ public class DashboardEdgeTest extends AbstractEdgeTest {
Dashboard foundDashboard = doGet("/api/dashboard/" + dashboard.getUuidId(), Dashboard.class);
Assert.assertNotNull(foundDashboard);
Assert.assertEquals("Edge Test Dashboard", foundDashboard.getName());
Assert.assertEquals(DASHBOARD_TITLE, foundDashboard.getName());
PageData<DashboardInfo> pageData = doGetTypedWithPageLink("/api/customer/" + savedCustomer.getId().toString() + "/dashboards?",
new TypeReference<>() {}, new PageLink(100));
Assert.assertEquals(1, pageData.getData().size());
Assert.assertEquals(DASHBOARD_TITLE, pageData.getData().get(0).getTitle());
dashboard.setTitle(DASHBOARD_TITLE + " Updated");
dashboard.setAssignedCustomers(null);
dashboardUpdateMsgBuilder.setEntity(JacksonUtil.toString(dashboard));
dashboardUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE);
uplinkMsgBuilder = UplinkMsg.newBuilder();
uplinkMsgBuilder.addDashboardUpdateMsg(dashboardUpdateMsgBuilder.build());
edgeImitator.expectResponsesAmount(1);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
Assert.assertTrue(edgeImitator.waitForResponses());
foundDashboard = doGet("/api/dashboard/" + dashboard.getUuidId(), Dashboard.class);
Assert.assertEquals(DASHBOARD_TITLE + " Updated", foundDashboard.getName());
}
@Test
@ -242,11 +269,12 @@ public class DashboardEdgeTest extends AbstractEdgeTest {
return savedDashboard;
}
private Dashboard buildDashboardForUplinkMsg() {
private Dashboard buildDashboardForUplinkMsg(Customer savedCustomer) {
Dashboard dashboard = new Dashboard();
dashboard.setId(new DashboardId(UUID.randomUUID()));
dashboard.setTenantId(tenantId);
dashboard.setTitle("Edge Test Dashboard");
dashboard.setTitle(DASHBOARD_TITLE);
dashboard.setAssignedCustomers(Sets.newHashSet(new ShortCustomerInfo(savedCustomer.getId(), savedCustomer.getTitle(), savedCustomer.isPublic())));
return dashboard;
}

76
application/src/test/java/org/thingsboard/server/edge/ResourceEdgeTest.java
View File

@ -17,6 +17,7 @@ package org.thingsboard.server.edge;
import com.datastax.oss.driver.api.core.uuid.Uuids;
import com.google.protobuf.AbstractMessage;
import com.google.protobuf.InvalidProtocolBufferException;
import org.junit.Assert;
import org.junit.Test;
import org.thingsboard.common.util.JacksonUtil;
@ -98,30 +99,23 @@ public class ResourceEdgeTest extends AbstractEdgeTest {
public void testSendResourceToCloud() throws Exception {
TbResource tbResource = createTbResource();
UUID uuid = Uuids.timeBased();
UplinkMsg uplinkMsg = getUplinkMsg(uuid, tbResource, UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE);
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
ResourceUpdateMsg.Builder resourceUpdateMsgBuilder = ResourceUpdateMsg.newBuilder();
resourceUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits());
resourceUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits());
resourceUpdateMsgBuilder.setEntity(JacksonUtil.toString(tbResource));
resourceUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE);
testAutoGeneratedCodeByProtobuf(resourceUpdateMsgBuilder);
uplinkMsgBuilder.addResourceUpdateMsg(resourceUpdateMsgBuilder.build());
checkResourceOnCloud(uplinkMsg, uuid, tbResource.getTitle());
}
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
@Test
public void testUpdateResourceTitleOnCloud() throws Exception {
TbResource tbResource = createTbResource();
UUID uuid = Uuids.timeBased();
UplinkMsg uplinkMsg = getUplinkMsg(uuid, tbResource, UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE);
edgeImitator.expectResponsesAmount(1);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
checkResourceOnCloud(uplinkMsg, uuid, tbResource.getTitle());
Assert.assertTrue(edgeImitator.waitForResponses());
tbResource.setTitle("Updated Edge Test Resource");
UplinkMsg updatedUplinkMsg = getUplinkMsg(uuid, tbResource, UpdateMsgType.ENTITY_UPDATED_RPC_MESSAGE);
UplinkResponseMsg latestResponseMsg = edgeImitator.getLatestResponseMsg();
Assert.assertTrue(latestResponseMsg.getSuccess());
TbResource tb = doGet("/api/resource/" + uuid, TbResource.class);
Assert.assertNotNull(tb);
Assert.assertEquals("Edge Test Resource", tb.getName());
Assert.assertEquals(TEST_DATA, tb.getEncodedData());
checkResourceOnCloud(updatedUplinkMsg, uuid, tbResource.getTitle());
}
@Test
@ -134,21 +128,12 @@ public class ResourceEdgeTest extends AbstractEdgeTest {
UUID uuid = Uuids.timeBased();
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
ResourceUpdateMsg.Builder resourceUpdateMsgBuilder = ResourceUpdateMsg.newBuilder();
resourceUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits());
resourceUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits());
resourceUpdateMsgBuilder.setEntity(JacksonUtil.toString(resource));
resourceUpdateMsgBuilder.setMsgType(UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE);
testAutoGeneratedCodeByProtobuf(resourceUpdateMsgBuilder);
uplinkMsgBuilder.addResourceUpdateMsg(resourceUpdateMsgBuilder.build());
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
UplinkMsg uplinkMsg = getUplinkMsg(uuid, resource, UpdateMsgType.ENTITY_CREATED_RPC_MESSAGE);
edgeImitator.expectResponsesAmount(1);
edgeImitator.expectMessageAmount(1);
edgeImitator.sendUplinkMsg(uplinkMsgBuilder.build());
edgeImitator.sendUplinkMsg(uplinkMsg);
Assert.assertTrue(edgeImitator.waitForResponses());
Assert.assertTrue(edgeImitator.waitForMessages());
@ -177,4 +162,35 @@ public class ResourceEdgeTest extends AbstractEdgeTest {
tbResource.setEncodedData(TEST_DATA);
return tbResource;
}
private UplinkMsg getUplinkMsg(UUID uuid, TbResource tbResource, UpdateMsgType updateMsgType) throws InvalidProtocolBufferException {
UplinkMsg.Builder uplinkMsgBuilder = UplinkMsg.newBuilder();
ResourceUpdateMsg.Builder resourceUpdateMsgBuilder = ResourceUpdateMsg.newBuilder();
resourceUpdateMsgBuilder.setIdMSB(uuid.getMostSignificantBits());
resourceUpdateMsgBuilder.setIdLSB(uuid.getLeastSignificantBits());
resourceUpdateMsgBuilder.setEntity(JacksonUtil.toString(tbResource));
resourceUpdateMsgBuilder.setMsgType(updateMsgType);
testAutoGeneratedCodeByProtobuf(resourceUpdateMsgBuilder);
uplinkMsgBuilder.addResourceUpdateMsg(resourceUpdateMsgBuilder.build());
testAutoGeneratedCodeByProtobuf(uplinkMsgBuilder);
return uplinkMsgBuilder.build();
}
private void checkResourceOnCloud(UplinkMsg uplinkMsg, UUID uuid, String resourceTitle) throws Exception {
edgeImitator.expectResponsesAmount(1);
edgeImitator.sendUplinkMsg(uplinkMsg);
Assert.assertTrue(edgeImitator.waitForResponses());
UplinkResponseMsg latestResponseMsg = edgeImitator.getLatestResponseMsg();
Assert.assertTrue(latestResponseMsg.getSuccess());
TbResource tb = doGet("/api/resource/" + uuid, TbResource.class);
Assert.assertNotNull(tb);
Assert.assertEquals(resourceTitle, tb.getName());
Assert.assertEquals(TEST_DATA, tb.getEncodedData());
}
}

3
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java
View File

@ -96,6 +96,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
private final TbQueueAdmin edgeEventAdmin;
private final AtomicLong consumerCount = new AtomicLong();
private final AtomicLong edgeConsumerCount = new AtomicLong();
public KafkaMonolithQueueFactory(TopicService topicService, TbKafkaSettings kafkaSettings,
TbServiceInfoProvider serviceInfoProvider,
@ -472,7 +473,7 @@ public class KafkaMonolithQueueFactory implements TbCoreQueueFactory, TbRuleEngi
TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> consumerBuilder = TbKafkaConsumerTemplate.builder();
consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(topicService.buildTopicName("tb_edge_event.notifications." + tenantId + "." + edgeId));
consumerBuilder.clientId("monolith-to-edge-event-consumer" + serviceInfoProvider.getServiceId());
consumerBuilder.clientId("monolith-to-edge-event-consumer-" + serviceInfoProvider.getServiceId() + "-" + edgeConsumerCount.incrementAndGet());
consumerBuilder.groupId(topicService.buildTopicName("monolith-edge-event-consumer"));
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToEdgeEventNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(edgeEventAdmin);

3
common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java
View File

@ -95,6 +95,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
private final TbQueueAdmin edgeEventAdmin;
private final AtomicLong consumerCount = new AtomicLong();
private final AtomicLong edgeConsumerCount = new AtomicLong();
public KafkaTbCoreQueueFactory(TopicService topicService,
TbKafkaSettings kafkaSettings,
@ -421,7 +422,7 @@ public class KafkaTbCoreQueueFactory implements TbCoreQueueFactory {
TbKafkaConsumerTemplate.TbKafkaConsumerTemplateBuilder<TbProtoQueueMsg<ToEdgeEventNotificationMsg>> consumerBuilder = TbKafkaConsumerTemplate.builder();
consumerBuilder.settings(kafkaSettings);
consumerBuilder.topic(topicService.buildTopicName("tb_edge_event.notifications." + tenantId + "." + edgeId));
consumerBuilder.clientId("tb-core-edge-event-consumer" + serviceInfoProvider.getServiceId());
consumerBuilder.clientId("tb-core-edge-event-consumer-" + serviceInfoProvider.getServiceId() + "-" + edgeConsumerCount.incrementAndGet());
consumerBuilder.groupId(topicService.buildTopicName("tb-core-edge-event-consumer"));
consumerBuilder.decoder(msg -> new TbProtoQueueMsg<>(msg.getKey(), ToEdgeEventNotificationMsg.parseFrom(msg.getData()), msg.getHeaders()));
consumerBuilder.admin(edgeEventAdmin);

2
common/transport/mqtt/src/main/java/org/thingsboard/server/transport/mqtt/MqttTransportHandler.java
View File

@ -963,7 +963,7 @@ public class MqttTransportHandler extends ChannelInboundHandlerAdapter implement
unSubResults.add((short) MqttReasonCodes.UnsubAck.NO_SUBSCRIPTION_EXISTED.byteValue());
}
}
if (!activityReported) {
if (!activityReported && !deviceSessionCtx.isProvisionOnly()) {
transportService.recordActivity(deviceSessionCtx.getSessionInfo());
}
ctx.writeAndFlush(createUnSubAckMessage(mqttMsg.variableHeader().messageId(), unSubResults));

6
common/transport/transport-api/src/main/java/org/thingsboard/server/common/transport/service/DefaultTransportService.java
View File

@ -736,7 +736,11 @@ public class DefaultTransportService extends TransportActivityManager implements
}
private void recordActivityInternal(TransportProtos.SessionInfoProto sessionInfo) {
onActivity(toSessionId(sessionInfo), sessionInfo, getCurrentTimeMillis());
if (sessionInfo != null) {
onActivity(toSessionId(sessionInfo), sessionInfo, getCurrentTimeMillis());
} else {
log.warn("Session info is missing, unable to record activity");
}
}
@Override

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/rest/TbHttpClient.java
View File

@ -343,7 +343,7 @@ public class TbHttpClient {
if (CredentialsType.BASIC == credentials.getType()) {
BasicCredentials basicCredentials = (BasicCredentials) credentials;
String authString = basicCredentials.getUsername() + ":" + basicCredentials.getPassword();
String encodedAuthString = new String(Base64.getDecoder().decode(authString.getBytes(StandardCharsets.UTF_8)));
String encodedAuthString = new String(Base64.getEncoder().encode(authString.getBytes(StandardCharsets.UTF_8)));
headers.add("Authorization", "Basic " + encodedAuthString);
}
}

2
ui-ngx/src/app/modules/home/components/widget/config/basic/indicator/liquid-level-card-basic-config.component.html
View File

@ -63,7 +63,7 @@
</tb-color-input>
</div>
</div>
<div class="tb-form-row space-between flex-1 flex-col">
<div class="tb-form-row space-between flex-1 !flex-col">
<div class="tb-flex row space-between align-center no-gap fill-width">
<div class="fixed-title-width" translate>widgets.liquid-level-card.shape</div>
<tb-toggle-select formControlName="tankSelectionType">

2
ui-ngx/src/app/modules/home/components/widget/lib/settings/indicator/liquid-level-card-widget-settings.component.html
View File

@ -18,7 +18,7 @@
<ng-container *ngIf="levelCardWidgetSettingsForm" [formGroup]="levelCardWidgetSettingsForm">
<div class="tb-form-panel no-padding no-border">
<div class="tb-form-panel">
<div class="tb-form-row space-between flex flex-1 flex-col">
<div class="tb-form-row space-between flex flex-1 !flex-col">
<div class="tb-flex row space-between align-center no-gap fill-width">
<div class="fixed-title-width" translate>widgets.liquid-level-card.shape</div>
<tb-toggle-select formControlName="tankSelectionType">

2
ui-ngx/src/app/modules/home/pages/admin/queue/queue.component.html
View File

@ -22,7 +22,7 @@
[class.!hidden]="isEdit || isDetailsPage">
{{'common.open-details-page' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'delete')"
[class.!hidden]="hideDelete() || isEdit">

4
ui-ngx/src/app/modules/home/pages/admin/resource/js-resource.component.html
View File

@ -22,13 +22,13 @@
[class.!hidden]="isEdit || isDetailsPage">
{{'common.open-details-page' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'downloadResource')"
[class.!hidden]="isEdit">
{{ 'javascript.download' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'delete')"
[class.!hidden]="hideDelete() || isEdit">

4
ui-ngx/src/app/modules/home/pages/admin/resource/resources-library.component.html
View File

@ -22,13 +22,13 @@
[class.!hidden]="isEdit || isDetailsPage">
{{'common.open-details-page' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'downloadResource')"
[class.!hidden]="isEdit">
{{ 'resource.download' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'delete')"
[class.!hidden]="hideDelete() || isEdit">

2
ui-ngx/src/app/modules/home/pages/device/device.component.html
View File

@ -58,7 +58,7 @@
[class.!hidden]="isEdit || deviceScope !== 'edge'">
{{ 'edge.unassign-from-edge' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'delete')"
[class.!hidden]="hideDelete() || isEdit">

2
ui-ngx/src/app/modules/home/pages/edge/edge.component.html
View File

@ -70,7 +70,7 @@
[class.!hidden]="isEdit || edgeScope !== 'tenant'">
{{'edge.manage-rulechains' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'delete')"
[class.!hidden]="hideDelete() || isEdit">

6
ui-ngx/src/app/modules/home/pages/ota-update/ota-update.component.html
View File

@ -16,19 +16,19 @@
-->
<div class="tb-details-buttons xs:flex xs:flex-col">
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'open')"
[class.!hidden]="isEdit || isDetailsPage">
{{'common.open-details-page' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async) || !(entity?.hasData && !entity?.url)"
(click)="onEntityAction($event, 'uploadPackage')"
[class.!hidden]="isEdit">
{{ 'ota-update.download' | translate }}
</button>
<button mat-raised-button color="primary" class="xs:flex-1"
<button mat-raised-button color="primary"
[disabled]="(isLoading$ | async)"
(click)="onEntityAction($event, 'delete')"
[class.!hidden]="hideDelete() || isEdit">