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

Merge pull request #12910 from thingsboard/rc

Browse Source 
rc
This commit is contained in:
Viacheslav Klimov 2025-03-13 15:35:55 +02:00 committed by GitHub
commit 40c865b176
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 92 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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) {

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

@ -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
View File

@ -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
View File

@ -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) {

74
application/src/test/java/org/thingsboard/server/actors/service/ComponentActorTest.java Normal file
View File

@ -0,0 +1,74 @@
/**
* Copyright © 2016-2024 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);
}
}