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

fixed duplicated cf cache init in cf processor

pull/13941/head
IrynaMatveieva 9 months ago
parent
00381459c2
commit
61a39c2b33
9 changed files with 22 additions and 172 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. 3
      application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
  2. 12
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java
  3. 61
      application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
  4. 3
      application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
  5. 8
      application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java
  6. 3
      common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
  7. 34
      common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldInitMsg.java
  8. 36
      common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldInitProfileEntityMsg.java
  9. 34
      common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldLinkInitMsg.java

3
application/src/main/java/org/thingsboard/server/actors/app/AppActor.java
View File

@ -114,9 +114,6 @@ public class AppActor extends ContextAwareActor {
ctx.broadcastToChildrenByType(msg, EntityType.TENANT);
break;
case CF_CACHE_INIT_MSG:
case CF_INIT_PROFILE_ENTITY_MSG:
case CF_INIT_MSG:
case CF_LINK_INIT_MSG:
case CF_STATE_RESTORE_MSG:
//TODO: use priority from the message body. For example, messages about CF lifecycle are important and Device lifecycle are not.
// same for the Linked telemetry.

12
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerActor.java
View File

@ -24,9 +24,6 @@ import org.thingsboard.server.common.msg.TbActorStopReason;
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldInitProfileEntityMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldPartitionChangeMsg;
/**
@ -70,15 +67,6 @@ public class CalculatedFieldManagerActor extends AbstractCalculatedFieldActor {
case CF_CACHE_INIT_MSG:
processor.onCacheInitMsg((CalculatedFieldCacheInitMsg) msg);
break;
case CF_INIT_PROFILE_ENTITY_MSG:
processor.onProfileEntityMsg((CalculatedFieldInitProfileEntityMsg) msg);
break;
case CF_INIT_MSG:
processor.onFieldInitMsg((CalculatedFieldInitMsg) msg);
break;
case CF_LINK_INIT_MSG:
processor.onLinkInitMsg((CalculatedFieldLinkInitMsg) msg);
break;
case CF_STATE_RESTORE_MSG:
processor.onStateRestoreMsg((CalculatedFieldStateRestoreMsg) msg);
break;

61
application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java
View File

@ -35,9 +35,6 @@ import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageDataIterable;
import org.thingsboard.server.common.msg.cf.CalculatedFieldCacheInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldEntityLifecycleMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldInitProfileEntityMsg;
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;
@ -120,37 +117,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
msg.getCallback().onSuccess();
}
public void onProfileEntityMsg(CalculatedFieldInitProfileEntityMsg msg) {
log.debug("[{}] Processing profile entity message.", msg.getTenantId().getId());
entityProfileCache.add(msg.getProfileEntityId(), msg.getEntityId());
msg.getCallback().onSuccess();
}
public void onFieldInitMsg(CalculatedFieldInitMsg msg) throws CalculatedFieldException {
log.debug("[{}] Processing CF init message.", msg.getCf().getId());
var cf = msg.getCf();
var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService());
try {
cfCtx.init();
} catch (Exception e) {
throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build();
}
calculatedFields.put(cf.getId(), cfCtx);
// We use copy on write lists to safely pass the reference to another actor for the iteration.
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx);
msg.getCallback().onSuccess();
}
public void onLinkInitMsg(CalculatedFieldLinkInitMsg msg) {
log.debug("[{}] Processing CF link init message for entity [{}].", msg.getLink().getCalculatedFieldId(), msg.getLink().getEntityId());
var link = msg.getLink();
// We use copy on write lists to safely pass the reference to another actor for the iteration.
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link);
msg.getCallback().onSuccess();
}
public void onStateRestoreMsg(CalculatedFieldStateRestoreMsg msg) {
var cfId = msg.getId().cfId();
var calculatedField = calculatedFields.get(cfId);
@ -566,20 +532,37 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
cfs.forEach(cf -> {
log.trace("Processing calculated field record: {}", cf);
try {
onFieldInitMsg(new CalculatedFieldInitMsg(cf.getTenantId(), cf));
initCalculatedField(cf);
} catch (CalculatedFieldException e) {
log.error("Failed to process calculated field record: {}", cf, e);
}
});
calculatedFields.values().forEach(cf -> {
entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cf);
});
PageDataIterable<CalculatedFieldLink> cfls = new PageDataIterable<>(pageLink -> cfDaoService.findAllCalculatedFieldLinksByTenantId(tenantId, pageLink), cfSettings.getInitTenantFetchPackSize());
cfls.forEach(link -> {
onLinkInitMsg(new CalculatedFieldLinkInitMsg(link.getTenantId(), link));
log.trace("Processing calculated field link record: {}", link);
initCalculatedFieldLink(link);
});
}
private void initCalculatedField(CalculatedField cf) throws CalculatedFieldException {
var cfCtx = new CalculatedFieldCtx(cf, systemContext.getTbelInvokeService(), systemContext.getApiLimitService());
try {
cfCtx.init();
} catch (Exception e) {
throw CalculatedFieldException.builder().ctx(cfCtx).eventEntity(cf.getEntityId()).cause(e).errorMessage("Failed to initialize CF context").build();
}
calculatedFields.put(cf.getId(), cfCtx);
// We use copy on write lists to safely pass the reference to another actor for the iteration.
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
entityIdCalculatedFields.computeIfAbsent(cf.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(cfCtx);
}
private void initCalculatedFieldLink(CalculatedFieldLink link) {
// We use copy on write lists to safely pass the reference to another actor for the iteration.
// Alternative approach would be to use any list but avoid modifications to the list (change the complete map value instead)
entityIdCalculatedFieldLinks.computeIfAbsent(link.getEntityId(), id -> new CopyOnWriteArrayList<>()).add(link);
}
private void initEntityProfileCache() {
PageDataIterable<ProfileEntityIdInfo> deviceIdInfos = new PageDataIterable<>(pageLink -> deviceService.findProfileEntityIdInfosByTenantId(tenantId, pageLink), cfSettings.getInitTenantFetchPackSize());
for (ProfileEntityIdInfo idInfo : deviceIdInfos) {

3
application/src/main/java/org/thingsboard/server/actors/tenant/TenantActor.java
View File

@ -180,9 +180,6 @@ public class TenantActor extends RuleChainManagerActor {
onRuleChainMsg((RuleChainAwareMsg) msg);
break;
case CF_CACHE_INIT_MSG:
case CF_INIT_PROFILE_ENTITY_MSG:
case CF_INIT_MSG:
case CF_LINK_INIT_MSG:
case CF_STATE_RESTORE_MSG:
case CF_PARTITIONS_CHANGE_MSG:
onToCalculatedFieldSystemActorMsg((ToCalculatedFieldSystemMsg) msg, true);

8
application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldCache.java
View File

@ -19,11 +19,9 @@ import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.util.ConcurrentReferenceHashMap;
import org.thingsboard.script.api.tbel.TbelInvokeService;
import org.thingsboard.server.actors.ActorSystemContext;
import org.thingsboard.server.common.data.cf.CalculatedField;
import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
@ -31,8 +29,6 @@ import org.thingsboard.server.common.data.id.CalculatedFieldId;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.data.page.PageDataIterable;
import org.thingsboard.server.common.msg.cf.CalculatedFieldInitMsg;
import org.thingsboard.server.common.msg.cf.CalculatedFieldLinkInitMsg;
import org.thingsboard.server.dao.cf.CalculatedFieldService;
import org.thingsboard.server.dao.usagerecord.ApiLimitService;
import org.thingsboard.server.queue.util.AfterStartUp;
@ -56,8 +52,6 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
private final CalculatedFieldService calculatedFieldService;
private final TbelInvokeService tbelInvokeService;
private final ApiLimitService apiLimitService;
@Lazy
private final ActorSystemContext actorSystemContext;
private final ConcurrentMap<CalculatedFieldId, CalculatedField> calculatedFields = new ConcurrentHashMap<>();
private final ConcurrentMap<EntityId, List<CalculatedField>> entityIdCalculatedFields = new ConcurrentHashMap<>();
@ -75,7 +69,6 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
cfs.forEach(cf -> {
if (cf != null) {
calculatedFields.putIfAbsent(cf.getId(), cf);
actorSystemContext.tell(new CalculatedFieldInitMsg(cf.getTenantId(), cf));
}
});
calculatedFields.values().forEach(cf -> {
@ -84,7 +77,6 @@ public class DefaultCalculatedFieldCache implements CalculatedFieldCache {
PageDataIterable<CalculatedFieldLink> cfls = new PageDataIterable<>(calculatedFieldService::findAllCalculatedFieldLinks, initFetchPackSize);
cfls.forEach(link -> {
calculatedFieldLinks.computeIfAbsent(link.getCalculatedFieldId(), id -> new CopyOnWriteArrayList<>()).add(link);
actorSystemContext.tell(new CalculatedFieldLinkInitMsg(link.getTenantId(), link));
});
calculatedFieldLinks.values().stream()
.flatMap(List::stream)

3
common/message/src/main/java/org/thingsboard/server/common/msg/MsgType.java
View File

@ -137,9 +137,6 @@ public enum MsgType {
CF_CACHE_INIT_MSG, // Sent to init caches for CF actor;
CF_INIT_PROFILE_ENTITY_MSG, // Sent to init profile entities cache;
CF_INIT_MSG, // Sent to init particular calculated field;
CF_LINK_INIT_MSG, // Sent to init particular calculated field;
CF_STATE_RESTORE_MSG, // Sent to restore particular calculated field entity state;
CF_PARTITIONS_CHANGE_MSG, // Sent when cluster event occures;

34
common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldInitMsg.java
View File

@ -1,34 +0,0 @@
/**
* 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.common.msg.cf;
import lombok.Data;
import org.thingsboard.server.common.data.cf.CalculatedField;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
@Data
public class CalculatedFieldInitMsg implements ToCalculatedFieldSystemMsg {
private final TenantId tenantId;
private final CalculatedField cf;
@Override
public MsgType getMsgType() {
return MsgType.CF_INIT_MSG;
}
}

36
common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldInitProfileEntityMsg.java
View File

@ -1,36 +0,0 @@
/**
* 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.common.msg.cf;
import lombok.Data;
import org.thingsboard.server.common.data.id.EntityId;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
@Data
public class CalculatedFieldInitProfileEntityMsg implements ToCalculatedFieldSystemMsg {
private final TenantId tenantId;
private final EntityId profileEntityId;
private final EntityId entityId;
@Override
public MsgType getMsgType() {
return MsgType.CF_INIT_PROFILE_ENTITY_MSG;
}
}

34
common/message/src/main/java/org/thingsboard/server/common/msg/cf/CalculatedFieldLinkInitMsg.java
View File

@ -1,34 +0,0 @@
/**
* 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.common.msg.cf;
import lombok.Data;
import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
import org.thingsboard.server.common.data.id.TenantId;
import org.thingsboard.server.common.msg.MsgType;
import org.thingsboard.server.common.msg.ToCalculatedFieldSystemMsg;
@Data
public class CalculatedFieldLinkInitMsg implements ToCalculatedFieldSystemMsg {
private final TenantId tenantId;
private final CalculatedFieldLink link;
@Override
public MsgType getMsgType() {
return MsgType.CF_LINK_INIT_MSG;
}
}
Write Preview
Loading…
Cancel
Save