Browse Source

Merge branch 'master' of github.com:thingsboard/thingsboard

pull/12927/head
Igor Kulikov 2 years ago
parent
commit
a640ee11d9
  1. 5
      application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java
  2. 37
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  3. 10
      application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java
  4. 5
      application/src/main/java/org/thingsboard/server/actors/shared/AbstractContextAwareMsgProcessor.java
  5. 6
      application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java
  6. 2
      application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
  7. 8
      application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java
  8. 2
      application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java
  9. 74
      application/src/test/java/org/thingsboard/server/actors/service/ComponentActorTest.java
  10. 5
      application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java
  11. 1
      common/data/src/main/java/org/thingsboard/server/common/data/edqs/fields/EntityFields.java

5
application/src/main/java/org/thingsboard/server/actors/ActorSystemContext.java

@ -141,6 +141,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@ -856,9 +857,9 @@ public class ActorSystemContext {
appActor.tellWithHighPriority(tbActorMsg);
}
public void schedulePeriodicMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs, long periodInMs) {
public ScheduledFuture<?> schedulePeriodicMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs, long periodInMs) {
log.debug("Scheduling periodic msg {} every {} ms with delay {} ms", msg, periodInMs, delayInMs);
getScheduler().scheduleWithFixedDelay(() -> ctx.tell(msg), delayInMs, periodInMs, TimeUnit.MILLISECONDS);
return getScheduler().scheduleWithFixedDelay(() -> ctx.tell(msg), delayInMs, periodInMs, TimeUnit.MILLISECONDS);
}
public void scheduleMsgWithDelay(TbActorRef ctx, TbActorMsg msg, long delayInMs) {

37
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java

@ -22,6 +22,7 @@ import org.thingsboard.server.actors.TbActorRef;
import org.thingsboard.server.actors.TbCalculatedFieldEntityActorId;
import org.thingsboard.server.actors.service.DefaultActorService;
import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor;
import org.thingsboard.server.common.data.DataConstants;
import org.thingsboard.server.common.data.EntityType;
import org.thingsboard.server.common.data.cf.CalculatedField;
import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
@ -35,6 +36,7 @@ import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.dao.cf.CalculatedFieldService;
import org.thingsboard.server.service.cf.CalculatedFieldProcessingService;
@ -121,7 +123,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
if (calculatedField != null) {
msg.getState().setRequiredArguments(calculatedField.getArgNames());
log.info("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId());
log.debug("Pushing CF state restore msg to specific actor [{}]", msg.getId().entityId());
getOrCreateActor(msg.getId().entityId()).tell(msg);
} else {
cfStateService.removeState(msg.getId(), msg.getCallback());
@ -178,6 +180,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
EntityId entityId = msg.getEntityId();
EntityId profileId = getProfileId(tenantId, entityId);
cfEntityCache.add(tenantId, profileId, entityId);
if (!isMyPartition(entityId, callback)) {
return;
}
var entityIdFields = getCalculatedFieldsByEntityId(entityId);
var profileIdFields = getCalculatedFieldsByEntityId(profileId);
var fieldsCount = entityIdFields.size() + profileIdFields.size();
@ -193,6 +198,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
private void onEntityUpdated(ComponentLifecycleMsg msg, TbCallback callback) {
if (msg.getOldProfileId() != null && msg.getOldProfileId() != msg.getProfileId()) {
cfEntityCache.update(tenantId, msg.getOldProfileId(), msg.getProfileId(), msg.getEntityId());
if (!isMyPartition(msg.getEntityId(), callback)) {
return;
}
var oldProfileCfs = getCalculatedFieldsByEntityId(msg.getOldProfileId());
var newProfileCfs = getCalculatedFieldsByEntityId(msg.getProfileId());
var fieldsCount = oldProfileCfs.size() + newProfileCfs.size();
@ -209,8 +217,10 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
cfEntityCache.evict(tenantId, msg.getEntityId());
log.info("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId());
getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback));
if (isMyPartition(msg.getEntityId(), callback)) {
log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId());
getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback));
}
}
private void onCfCreated(ComponentLifecycleMsg msg, TbCallback callback) throws CalculatedFieldException {
@ -311,7 +321,9 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
callback.onSuccess();
}
} else {
deleteCfForEntity(entityId, cfId, callback);
if (isMyPartition(entityId, callback)) {
deleteCfForEntity(entityId, cfId, callback);
}
}
}
}
@ -418,20 +430,31 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
callback.onSuccess();
}
} else {
initCfForEntity(entityId, cfCtx, forceStateReinit, callback);
if (isMyPartition(entityId, callback)) {
initCfForEntity(entityId, cfCtx, forceStateReinit, callback);
}
}
}
private void deleteCfForEntity(EntityId entityId, CalculatedFieldId cfId, TbCallback callback) {
log.info("Pushing delete CF msg to specific actor [{}]", entityId);
log.debug("Pushing delete CF msg to specific actor [{}]", entityId);
getOrCreateActor(entityId).tell(new CalculatedFieldEntityDeleteMsg(tenantId, cfId, callback));
}
private void initCfForEntity(EntityId entityId, CalculatedFieldCtx cfCtx, boolean forceStateReinit, TbCallback callback) {
log.info("Pushing entity init CF msg to specific actor [{}]", entityId);
log.debug("Pushing entity init CF msg to specific actor [{}]", entityId);
getOrCreateActor(entityId).tell(new EntityInitCalculatedFieldMsg(tenantId, cfCtx, callback, forceStateReinit));
}
private boolean isMyPartition(EntityId entityId, TbCallback callback) {
if (!systemContext.getPartitionService().resolve(ServiceType.TB_RULE_ENGINE, DataConstants.CF_QUEUE_NAME, tenantId, entityId).isMyPartition()) {
log.debug("[{}] Entity belongs to external partition.", entityId);
callback.onSuccess();
return false;
}
return true;
}
private static boolean isProfileEntity(EntityType entityType) {
return EntityType.DEVICE_PROFILE.equals(entityType) || EntityType.ASSET_PROFILE.equals(entityType);
}

10
application/src/main/java/org/thingsboard/server/actors/service/ComponentActor.java

@ -29,6 +29,9 @@ import org.thingsboard.server.common.msg.TbActorStopReason;
import org.thingsboard.server.common.msg.plugin.ComponentLifecycleMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import java.util.Optional;
import java.util.concurrent.ScheduledFuture;
/**
* @author Andrew Shvayka
*/
@ -41,6 +44,7 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
protected P processor;
private long messagesProcessed;
private long errorsOccurred;
ScheduledFuture<?> statsScheduledFuture = null;
public ComponentActor(ActorSystemContext systemContext, TenantId tenantId, T id) {
super(systemContext);
@ -73,9 +77,9 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
}
}
private void scheduleStatsPersistTick() {
void scheduleStatsPersistTick() {
try {
processor.scheduleStatsPersistTick(ctx, systemContext.getStatisticsPersistFrequency());
this.statsScheduledFuture = processor.scheduleStatsPersistTick(ctx, systemContext.getStatisticsPersistFrequency());
} catch (Exception e) {
log.error("[{}][{}] Failed to schedule statistics store message. No statistics is going to be stored: {}", tenantId, id, e.getMessage());
logAndPersist("onScheduleStatsPersistMsg", e);
@ -90,6 +94,8 @@ public abstract class ComponentActor<T extends EntityId, P extends ComponentMsgP
processor.stop(ctx);
}
logLifecycleEvent(ComponentLifecycleEvent.STOPPED);
Optional.ofNullable(statsScheduledFuture).ifPresent(x -> x.cancel(false));
statsScheduledFuture = null;
} catch (Exception e) {
log.warn("[{}][{}] Failed to stop {} processor: {}", tenantId, id, id.getEntityType(), e.getMessage());
logAndPersist("OnStop", e, true);

5
application/src/main/java/org/thingsboard/server/actors/shared/AbstractContextAwareMsgProcessor.java

@ -21,6 +21,7 @@ import org.thingsboard.server.actors.TbActorCtx;
import org.thingsboard.server.common.msg.TbActorMsg;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@Slf4j
public abstract class AbstractContextAwareMsgProcessor {
@ -36,8 +37,8 @@ public abstract class AbstractContextAwareMsgProcessor {
return systemContext.getScheduler();
}
protected void schedulePeriodicMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs, long periodInMs) {
systemContext.schedulePeriodicMsgWithDelay(ctx, msg, delayInMs, periodInMs);
protected ScheduledFuture<?> schedulePeriodicMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs, long periodInMs) {
return systemContext.schedulePeriodicMsgWithDelay(ctx, msg, delayInMs, periodInMs);
}
protected void scheduleMsgWithDelay(TbActorCtx ctx, TbActorMsg msg, long delayInMs) {

6
application/src/main/java/org/thingsboard/server/actors/shared/ComponentMsgProcessor.java

@ -27,6 +27,8 @@ import org.thingsboard.server.common.msg.TbMsg;
import org.thingsboard.server.common.msg.queue.PartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.RuleNodeException;
import java.util.concurrent.ScheduledFuture;
@Slf4j
public abstract class ComponentMsgProcessor<T extends EntityId> extends AbstractContextAwareMsgProcessor {
@ -77,8 +79,8 @@ public abstract class ComponentMsgProcessor<T extends EntityId> extends Abstract
start(context);
}
public void scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) {
schedulePeriodicMsgWithDelay(context, StatsPersistTick.INSTANCE, statsPersistFrequency, statsPersistFrequency);
public ScheduledFuture<?> scheduleStatsPersistTick(TbActorCtx context, long statsPersistFrequency) {
return schedulePeriodicMsgWithDelay(context, StatsPersistTick.INSTANCE, statsPersistFrequency, statsPersistFrequency);
}
protected boolean checkMsgValid(TbMsg tbMsg) {

2
application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java

@ -112,7 +112,7 @@ public class TenantActor extends RuleChainManagerActor {
cantFindTenant = true;
}
} else {
log.info("Tenant {} is not managed by current service, skipping rule chains init", tenantId);
log.info("Tenant {} is not managed by current service, skipping rule chains and cf actor init", tenantId);
}
}
log.debug("[{}] Tenant actor started.", tenantId);

8
application/src/main/java/org/thingsboard/server/service/queue/DefaultTbCalculatedFieldConsumerService.java

@ -15,15 +15,12 @@
*/
package org.thingsboard.server.service.queue;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;
import org.thingsboard.common.util.ThingsBoardExecutors;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.calculatedField.CalculatedFieldLinkedTelemetryMsg;
import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg;
@ -35,7 +32,7 @@ import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
import org.thingsboard.server.common.msg.queue.ServiceType;
import org.thingsboard.server.common.msg.queue.TbCallback;
import org.thingsboard.server.common.msg.queue.TopicPartitionInfo;
import org.thingsboard.server.common.util.ProtoUtils;
import org.thingsboard.server.dao.tenant.TbTenantProfileCache;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto;
import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto;
@ -60,7 +57,6 @@ import org.thingsboard.server.service.queue.processing.IdMsgPair;
import org.thingsboard.server.service.security.auth.jwt.settings.JwtSettingsService;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@ -220,7 +216,7 @@ public class DefaultTbCalculatedFieldConsumerService extends AbstractConsumerSer
protected void handleNotification(UUID id, TbProtoQueueMsg<ToCalculatedFieldNotificationMsg> msg, TbCallback callback) {
ToCalculatedFieldNotificationMsg toCfNotification = msg.getValue();
if (toCfNotification.hasComponentLifecycleMsg()) {
// from upstream (maybe removed since we don't need to init state for each partition)
handleComponentLifecycleMsg(id, ProtoUtils.fromProto(toCfNotification.getComponentLifecycleMsg()));
log.trace("[{}] Forwarding component lifecycle message for processing {}", id, toCfNotification.getComponentLifecycleMsg());
forwardToActorSystem(toCfNotification.getComponentLifecycleMsg(), callback);
} else if (toCfNotification.hasLinkedTelemetryMsg()) {

2
application/src/main/java/org/thingsboard/server/service/transport/DefaultTransportApiService.java

@ -351,7 +351,7 @@ public class DefaultTransportApiService implements TransportApiService {
device.setAdditionalInfo(additionalInfo);
device = deviceService.saveDevice(device);
relationService.saveRelation(TenantId.SYS_TENANT_ID, new EntityRelation(gateway.getId(), device.getId(), "Created"));
relationService.saveRelation(tenantId, new EntityRelation(gateway.getId(), device.getId(), "Created"));
TbMsgMetaData metaData = new TbMsgMetaData();
CustomerId customerId = gateway.getCustomerId();

74
application/src/test/java/org/thingsboard/server/actors/service/ComponentActorTest.java

@ -0,0 +1,74 @@
/**
* Copyright © 2016-2025 The Thingsboard Authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.thingsboard.server.actors.service;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.BDDMockito;
import org.mockito.Mockito;
import org.springframework.test.util.ReflectionTestUtils;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.actors.shared.ComponentMsgProcessor;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.msg.TbActorStopReason;
import java.util.concurrent.ScheduledFuture;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
class ComponentActorTest {
ComponentActor componentActor;
@BeforeEach
void setUp() {
componentActor = Mockito.mock(ComponentActor.class);
}
@Test
void scheduleStatsPersistTickTest() {
Assertions.assertNull(componentActor.statsScheduledFuture);
ScheduledFuture<?> statsScheduledFuture = Mockito.mock(ScheduledFuture.class);
ActorSystemContext systemContext = Mockito.mock(ActorSystemContext.class);
ReflectionTestUtils.setField(componentActor, "systemContext", systemContext);
ComponentMsgProcessor<?> processor = Mockito.mock(ComponentMsgProcessor.class);
componentActor.processor = processor;
BDDMockito.willReturn(statsScheduledFuture).given(processor).scheduleStatsPersistTick(any(), anyLong());
BDDMockito.willCallRealMethod().given(componentActor).scheduleStatsPersistTick();
componentActor.scheduleStatsPersistTick();
Assertions.assertNotNull(componentActor.statsScheduledFuture);
}
@Test
void destroyTest() {
ScheduledFuture<?> statsScheduledFuture = Mockito.mock(ScheduledFuture.class);
componentActor.statsScheduledFuture = statsScheduledFuture;
Assertions.assertNotNull(componentActor.statsScheduledFuture);
Throwable cause = new Throwable();
EntityId id = Mockito.mock(EntityId.class);
ReflectionTestUtils.setField(componentActor, "id", id);
BDDMockito.willCallRealMethod().given(componentActor).destroy(any(), any());
componentActor.destroy(TbActorStopReason.STOPPED, cause);
Mockito.verify(statsScheduledFuture).cancel(false);
Assertions.assertNull(componentActor.statsScheduledFuture);
}
}

5
application/src/test/java/org/thingsboard/server/controller/AbstractWebTest.java

@ -38,6 +38,7 @@ import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.boot.test.mock.mockito.SpyBean;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
@ -143,6 +144,7 @@ import org.thingsboard.server.dao.device.ClaimDevicesService;
import org.thingsboard.server.dao.tenant.TenantProfileService;
import org.thingsboard.server.dao.timeseries.TimeseriesService;
import org.thingsboard.server.queue.memory.InMemoryStorage;
import org.thingsboard.server.service.cf.CfRocksDb;
import org.thingsboard.server.service.entitiy.tenant.profile.TbTenantProfileService;
import org.thingsboard.server.service.security.auth.jwt.RefreshTokenRequest;
import org.thingsboard.server.service.security.auth.rest.LoginRequest;
@ -276,6 +278,9 @@ public abstract class AbstractWebTest extends AbstractInMemoryStorageTest {
@Autowired
protected InMemoryStorage storage;
@MockBean
protected CfRocksDb cfRocksDb;
@Rule
public TestRule watcher = new TestWatcher() {
protected void starting(Description description) {

1
common/data/src/main/java/org/thingsboard/server/common/data/edqs/fields/EntityFields.java

@ -148,6 +148,7 @@ public interface EntityFields {
default String getAsString(String key) {
return switch (key) {
case "createdTime" -> Long.toString(getCreatedTime());
case "title" -> getName();
case "type" -> getType();
case "label" -> getLabel();
case "additionalInfo" -> getAdditionalInfo();

Loading…
Cancel
Save