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

Feature/relation nodes (#1452)

Browse Source 
* modified relation-action-nodes

* add rule node description

* fix rule node description

* fix typo

* updated relation-action-nodes
This commit is contained in:
ShvaykaD 2019-02-06 12:54:55 +02:00 committed by Andrew Shvayka
parent 9eb1cd911b
commit 75b1f84805
8 changed files with 138 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java
View File

@ -148,7 +148,7 @@ public class BaseTimeseriesService implements TimeseriesService {
private void saveAndRegisterFutures(TenantId tenantId, List<ListenableFuture<Void>> futures, EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
if (entityId.getEntityType().equals(EntityType.ENTITY_VIEW)) {
throw new IncorrectParameterException("Telemetry data can't be stored for entity view. Only read only");
throw new IncorrectParameterException("Telemetry data can't be stored for entity view. Read only");
}
futures.add(timeseriesDao.savePartition(tenantId, entityId, tsKvEntry.getTs(), tsKvEntry.getKey(), ttl));
futures.add(timeseriesDao.saveLatest(tenantId, entityId, tsKvEntry));

34
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractRelationActionNode.java
View File

@ -22,6 +22,7 @@ import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.thingsboard.rule.engine.api.TbContext;
import org.thingsboard.rule.engine.api.TbNode;
@ -78,28 +79,28 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
@Override
public void onMsg(TbContext ctx, TbMsg msg) {
withCallback(processEntityRelationAction(ctx, msg),
filterResult -> ctx.tellNext(msg, filterResult ? SUCCESS : FAILURE), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
filterResult -> ctx.tellNext(filterResult.getMsg(), filterResult.isResult() ? SUCCESS : FAILURE), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor());
}
@Override
public void destroy() {
}
protected ListenableFuture<Boolean> processEntityRelationAction(TbContext ctx, TbMsg msg) {
protected ListenableFuture<RelationContainer> processEntityRelationAction(TbContext ctx, TbMsg msg) {
return Futures.transformAsync(getEntity(ctx, msg), entityContainer -> doProcessEntityRelationAction(ctx, msg, entityContainer));
}
protected abstract boolean createEntityIfNotExists();
protected abstract ListenableFuture<Boolean> doProcessEntityRelationAction(TbContext ctx, TbMsg msg, EntityContainer entityContainer);
protected abstract ListenableFuture<RelationContainer> doProcessEntityRelationAction(TbContext ctx, TbMsg msg, EntityContainer entityContainer);
protected abstract C loadEntityNodeActionConfig(TbNodeConfiguration configuration) throws TbNodeException;
protected ListenableFuture<EntityContainer> getEntity(TbContext ctx, TbMsg msg) {
String entityName = TbNodeUtils.processPattern(this.config.getEntityNamePattern(), msg.getMetaData());
String entityName = processPattern(msg, this.config.getEntityNamePattern());
String type;
if (this.config.getEntityTypePattern() != null) {
type = TbNodeUtils.processPattern(this.config.getEntityTypePattern(), msg.getMetaData());
type = processPattern(msg, this.config.getEntityTypePattern());
} else {
type = null;
}
@ -116,24 +117,30 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
protected SearchDirectionIds processSingleSearchDirection(TbMsg msg, EntityContainer entityContainer) {
SearchDirectionIds searchDirectionIds = new SearchDirectionIds();
if (EntitySearchDirection.FROM.name().equals(config.getDirection())) {
if (EntitySearchDirection.FROM.name().equals(this.config.getDirection())) {
searchDirectionIds.setFromId(EntityIdFactory.getByTypeAndId(entityContainer.getEntityType().name(), entityContainer.getEntityId().toString()));
searchDirectionIds.setToId(msg.getOriginator());
searchDirectionIds.setOrignatorDirectionFrom(false);
} else {
searchDirectionIds.setToId(EntityIdFactory.getByTypeAndId(entityContainer.getEntityType().name(), entityContainer.getEntityId().toString()));
searchDirectionIds.setFromId(msg.getOriginator());
searchDirectionIds.setOrignatorDirectionFrom(true);
}
return searchDirectionIds;
}
protected ListenableFuture<List<EntityRelation>> processListSearchDirection(TbContext ctx, TbMsg msg) {
if (EntitySearchDirection.FROM.name().equals(config.getDirection())) {
return ctx.getRelationService().findByToAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), config.getRelationType(), RelationTypeGroup.COMMON);
if (EntitySearchDirection.FROM.name().equals(this.config.getDirection())) {
return ctx.getRelationService().findByToAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), processPattern(msg, this.config.getRelationTypePattern()), RelationTypeGroup.COMMON);
} else {
return ctx.getRelationService().findByFromAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), config.getRelationType(), RelationTypeGroup.COMMON);
return ctx.getRelationService().findByFromAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), processPattern(msg, this.config.getRelationTypePattern()), RelationTypeGroup.COMMON);
}
}
protected String processPattern(TbMsg msg, String pattern){
return TbNodeUtils.processPattern(pattern, msg.getMetaData());
}
@Data
@AllArgsConstructor
private static class EntityKey {
@ -146,6 +153,7 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
protected static class SearchDirectionIds {
private EntityId fromId;
private EntityId toId;
private boolean orignatorDirectionFrom;
}
private static class EntityCacheLoader extends CacheLoader<EntityKey, EntityContainer> {
@ -233,7 +241,15 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
}
return targetEntity;
}
}
@Data
@NoArgsConstructor
@AllArgsConstructor
protected static class RelationContainer {
private TbMsg msg;
private boolean result;
}

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractRelationActionNodeConfiguration.java
View File

@ -21,7 +21,7 @@ import lombok.Data;
public abstract class TbAbstractRelationActionNodeConfiguration {
private String direction;
private String relationType;
private String relationTypePattern;
private String entityType;
private String entityNamePattern;

97
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateRelationNode.java
View File

@ -35,13 +35,20 @@ import org.thingsboard.server.common.data.relation.EntityRelation;
import org.thingsboard.server.common.data.relation.RelationTypeGroup;
import org.thingsboard.server.common.msg.TbMsg;
import java.util.ArrayList;
import java.util.List;
@Slf4j
@RuleNode(
type = ComponentType.ACTION,
name = "create relation",
configClazz = TbCreateRelationNodeConfiguration.class,
nodeDescription = "Finds target Entity by entity name pattern and (entity type pattern for Asset, Device) and then create a relation to Originator Entity by type and direction." +
" If Selected entity type: Asset, Device or Customer will create new Entity if it doesn't exist and 'Create new entity if not exists' is set to true.",
" If Selected entity type: Asset, Device or Customer will create new Entity if it doesn't exist and selected checkbox 'Create new entity if not exists'.<br>" +
" In case that relation from the message originator to the selected entity not exist and If selected checkbox 'Remove current relations'," +
" before creating the new relation all existed relations to message originator by type and direction will be removed.<br>" +
" If relation from the message originator to the selected entity created and If selected checkbox 'Change originator to related entity'," +
" outbound message will be processed as a message from this entity.",
nodeDetails = "If the relation already exists or successfully created - Message send via <b>Success</b> chain, otherwise <b>Failure</b> chain will be used.",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective = "tbActionNodeCreateRelationConfig",
@ -49,6 +56,8 @@ import org.thingsboard.server.common.msg.TbMsg;
)
public class TbCreateRelationNode extends TbAbstractRelationActionNode<TbCreateRelationNodeConfiguration> {
private String relationType;
@Override
protected TbCreateRelationNodeConfiguration loadEntityNodeActionConfig(TbNodeConfiguration configuration) throws TbNodeException {
return TbNodeUtils.convert(configuration, TbCreateRelationNodeConfiguration.class);
@ -60,19 +69,60 @@ public class TbCreateRelationNode extends TbAbstractRelationActionNode<TbCreateR
}
@Override
protected ListenableFuture<Boolean> doProcessEntityRelationAction(TbContext ctx, TbMsg msg, EntityContainer entity) {
return createIfAbsent(ctx, msg, entity);
protected ListenableFuture<RelationContainer> doProcessEntityRelationAction(TbContext ctx, TbMsg msg, EntityContainer entity) {
ListenableFuture<Boolean> future = createIfAbsent(ctx, msg, entity);
return Futures.transform(future, result -> {
RelationContainer container = new RelationContainer();
if (result && config.isChangeOriginatorToRelatedEntity()) {
TbMsg tbMsg = ctx.transformMsg(msg, msg.getType(), entity.getEntityId(), msg.getMetaData(), msg.getData());
container.setMsg(tbMsg);
} else {
container.setMsg(msg);
}
container.setResult(result);
return container;
});
}
private ListenableFuture<Boolean> createIfAbsent(TbContext ctx, TbMsg msg, EntityContainer entityContainer) {
relationType = processPattern(msg, config.getRelationTypePattern());
SearchDirectionIds sdId = processSingleSearchDirection(msg, entityContainer);
return Futures.transformAsync(ctx.getRelationService().checkRelation(ctx.getTenantId(), sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON),
result -> {
if (!result) {
return processCreateRelation(ctx, entityContainer, sdId);
}
return Futures.immediateFuture(true);
});
ListenableFuture<Boolean> checkRelationFuture = Futures.transformAsync(ctx.getRelationService().checkRelation(ctx.getTenantId(), sdId.getFromId(), sdId.getToId(), relationType, RelationTypeGroup.COMMON), result -> {
if (!result) {
if (config.isRemoveCurrentRelations()) {
return processDeleteRelations(ctx, processFindRelations(ctx, msg, sdId));
}
}
return Futures.immediateFuture(true);
}, ctx.getDbCallbackExecutor());
return Futures.transformAsync(checkRelationFuture, result -> {
if (!result) {
return processCreateRelation(ctx, entityContainer, sdId);
}
return Futures.immediateFuture(true);
}, ctx.getDbCallbackExecutor());
}
private ListenableFuture<List<EntityRelation>> processFindRelations(TbContext ctx, TbMsg msg, SearchDirectionIds sdId) {
if (sdId.isOrignatorDirectionFrom()) {
return ctx.getRelationService().findByFromAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), relationType, RelationTypeGroup.COMMON);
} else {
return ctx.getRelationService().findByToAndTypeAsync(ctx.getTenantId(), msg.getOriginator(), relationType, RelationTypeGroup.COMMON);
}
}
private ListenableFuture<Boolean> processDeleteRelations(TbContext ctx, ListenableFuture<List<EntityRelation>> listListenableFuture) {
return Futures.transformAsync(listListenableFuture, entityRelations -> {
if (!entityRelations.isEmpty()) {
List<ListenableFuture<Boolean>> list = new ArrayList<>();
for (EntityRelation relation : entityRelations) {
list.add(ctx.getRelationService().deleteRelationAsync(ctx.getTenantId(), relation));
}
return Futures.transform(Futures.allAsList(list), result -> false);
}
return Futures.immediateFuture(false);
}, ctx.getDbCallbackExecutor());
}
private ListenableFuture<Boolean> processCreateRelation(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId) {
@ -96,17 +146,17 @@ public class TbCreateRelationNode extends TbAbstractRelationActionNode<TbCreateR
private ListenableFuture<Boolean> processView(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId) {
return Futures.transformAsync(ctx.getEntityViewService().findEntityViewByIdAsync(ctx.getTenantId(), new EntityViewId(entityContainer.getEntityId().getId())), entityView -> {
if (entityView != null) {
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
return processSave(ctx, sdId);
} else {
return Futures.immediateFuture(true);
}
});
}, ctx.getDbCallbackExecutor());
}
private ListenableFuture<Boolean> processDevice(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId) {
return Futures.transformAsync(ctx.getDeviceService().findDeviceByIdAsync(ctx.getTenantId(), new DeviceId(entityContainer.getEntityId().getId())), device -> {
if (device != null) {
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
return processSave(ctx, sdId);
} else {
return Futures.immediateFuture(true);
}
@ -116,40 +166,45 @@ public class TbCreateRelationNode extends TbAbstractRelationActionNode<TbCreateR
private ListenableFuture<Boolean> processAsset(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId) {
return Futures.transformAsync(ctx.getAssetService().findAssetByIdAsync(ctx.getTenantId(), new AssetId(entityContainer.getEntityId().getId())), asset -> {
if (asset != null) {
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
return processSave(ctx, sdId);
} else {
return Futures.immediateFuture(true);
}
});
}, ctx.getDbCallbackExecutor());
}
private ListenableFuture<Boolean> processCustomer(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId) {
return Futures.transformAsync(ctx.getCustomerService().findCustomerByIdAsync(ctx.getTenantId(), new CustomerId(entityContainer.getEntityId().getId())), customer -> {
if (customer != null) {
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
return processSave(ctx, sdId);
} else {
return Futures.immediateFuture(true);
}
});
}, ctx.getDbCallbackExecutor());
}
private ListenableFuture<Boolean> processDashboard(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId) {
return Futures.transformAsync(ctx.getDashboardService().findDashboardByIdAsync(ctx.getTenantId(), new DashboardId(entityContainer.getEntityId().getId())), dashboard -> {
if (dashboard != null) {
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
return processSave(ctx, sdId);
} else {
return Futures.immediateFuture(true);
}
});
}, ctx.getDbCallbackExecutor());
}
private ListenableFuture<Boolean> processTenant(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId) {
return Futures.transformAsync(ctx.getTenantService().findTenantByIdAsync(ctx.getTenantId(), new TenantId(entityContainer.getEntityId().getId())), tenant -> {
if (tenant != null) {
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
return processSave(ctx, sdId);
} else {
return Futures.immediateFuture(true);
}
});
}, ctx.getDbCallbackExecutor());
}
private ListenableFuture<Boolean> processSave(TbContext ctx, SearchDirectionIds sdId) {
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), relationType, RelationTypeGroup.COMMON));
}
}

6
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateRelationNodeConfiguration.java
View File

@ -23,15 +23,19 @@ import org.thingsboard.server.common.data.relation.EntitySearchDirection;
public class TbCreateRelationNodeConfiguration extends TbAbstractRelationActionNodeConfiguration implements NodeConfiguration<TbCreateRelationNodeConfiguration> {
private boolean createEntityIfNotExists;
private boolean changeOriginatorToRelatedEntity;
private boolean removeCurrentRelations;
@Override
public TbCreateRelationNodeConfiguration defaultConfiguration() {
TbCreateRelationNodeConfiguration configuration = new TbCreateRelationNodeConfiguration();
configuration.setDirection(EntitySearchDirection.FROM.name());
configuration.setRelationType("Contains");
configuration.setRelationTypePattern("Contains");
configuration.setEntityNamePattern("");
configuration.setEntityCacheExpiration(300);
configuration.setCreateEntityIfNotExists(false);
configuration.setRemoveCurrentRelations(false);
configuration.setChangeOriginatorToRelatedEntity(false);
return configuration;
}
}

45
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeleteRelationNode.java
View File

@ -42,11 +42,13 @@ import java.util.List;
" if 'Delete single entity' is set to true, otherwise rule node will delete all relations to the originator of the message by type and direction.",
nodeDetails = "If the relation(s) successfully deleted - Message send via <b>Success</b> chain, otherwise <b>Failure</b> chain will be used.",
uiResources = {"static/rulenode/rulenode-core-config.js"},
configDirective ="tbActionNodeDeleteRelationConfig",
configDirective = "tbActionNodeDeleteRelationConfig",
icon = "remove_circle"
)
public class TbDeleteRelationNode extends TbAbstractRelationActionNode<TbDeleteRelationNodeConfiguration> {
private String relationType;
@Override
protected TbDeleteRelationNodeConfiguration loadEntityNodeActionConfig(TbNodeConfiguration configuration) throws TbNodeException {
return TbNodeUtils.convert(configuration, TbDeleteRelationNodeConfiguration.class);
@ -58,35 +60,39 @@ public class TbDeleteRelationNode extends TbAbstractRelationActionNode<TbDeleteR
}
@Override
protected ListenableFuture<Boolean> processEntityRelationAction(TbContext ctx, TbMsg msg) {
if(config.isDeleteForSingleEntity()){
return Futures.transformAsync(getEntity(ctx, msg), entityContainer -> doProcessEntityRelationAction(ctx, msg, entityContainer));
} else {
return processList(ctx, msg);
}
protected ListenableFuture<RelationContainer> processEntityRelationAction(TbContext ctx, TbMsg msg) {
return getRelationContainerListenableFuture(ctx, msg);
}
@Override
protected ListenableFuture<Boolean> doProcessEntityRelationAction(TbContext ctx, TbMsg msg, EntityContainer entityContainer) {
return processSingle(ctx, msg, entityContainer);
protected ListenableFuture<RelationContainer> doProcessEntityRelationAction(TbContext ctx, TbMsg msg, EntityContainer entityContainer) {
return Futures.transform(processSingle(ctx, msg, entityContainer), result -> new RelationContainer(msg, result));
}
private ListenableFuture<RelationContainer> getRelationContainerListenableFuture(TbContext ctx, TbMsg msg) {
relationType = processPattern(msg, config.getRelationTypePattern());
if (config.isDeleteForSingleEntity()) {
return Futures.transformAsync(getEntity(ctx, msg), entityContainer -> doProcessEntityRelationAction(ctx, msg, entityContainer));
} else {
return Futures.transform(processList(ctx, msg), result -> new RelationContainer(msg, result));
}
}
private ListenableFuture<Boolean> processList(TbContext ctx, TbMsg msg) {
return Futures.transformAsync(processListSearchDirection(ctx, msg), entityRelations -> {
if(entityRelations.isEmpty()){
if (entityRelations.isEmpty()) {
return Futures.immediateFuture(true);
} else {
List<ListenableFuture<Boolean>> listenableFutureList = new ArrayList<>();
for (EntityRelation entityRelation: entityRelations) {
for (EntityRelation entityRelation : entityRelations) {
listenableFutureList.add(ctx.getRelationService().deleteRelationAsync(ctx.getTenantId(), entityRelation));
}
return Futures.transformAsync(Futures.allAsList(listenableFutureList), booleans -> {
for (Boolean bool : booleans) {
if (!bool) {
return Futures.immediateFuture(false);
}
}
return Futures.immediateFuture(true);
for (Boolean bool : booleans) {
if (!bool) {
return Futures.immediateFuture(false);
}
}
return Futures.immediateFuture(true);
});
}
});
@ -94,8 +100,7 @@ public class TbDeleteRelationNode extends TbAbstractRelationActionNode<TbDeleteR
private ListenableFuture<Boolean> processSingle(TbContext ctx, TbMsg msg, EntityContainer entityContainer) {
SearchDirectionIds sdId = processSingleSearchDirection(msg, entityContainer);
return Futures.transformAsync(ctx.getRelationService().checkRelation(ctx.getTenantId(), sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON),
return Futures.transformAsync(ctx.getRelationService().checkRelation(ctx.getTenantId(), sdId.getFromId(), sdId.getToId(), relationType, RelationTypeGroup.COMMON),
result -> {
if (result) {
return processSingleDeleteRelation(ctx, sdId);
@ -105,7 +110,7 @@ public class TbDeleteRelationNode extends TbAbstractRelationActionNode<TbDeleteR
}
private ListenableFuture<Boolean> processSingleDeleteRelation(TbContext ctx, SearchDirectionIds sdId) {
return ctx.getRelationService().deleteRelationAsync(ctx.getTenantId(), sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON);
return ctx.getRelationService().deleteRelationAsync(ctx.getTenantId(), sdId.getFromId(), sdId.getToId(), relationType, RelationTypeGroup.COMMON);
}
}

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeleteRelationNodeConfiguration.java
View File

@ -29,7 +29,7 @@ public class TbDeleteRelationNodeConfiguration extends TbAbstractRelationActionN
TbDeleteRelationNodeConfiguration configuration = new TbDeleteRelationNodeConfiguration();
configuration.setDeleteForSingleEntity(true);
configuration.setDirection(EntitySearchDirection.FROM.name());
configuration.setRelationType("Contains");
configuration.setRelationTypePattern("Contains");
configuration.setEntityNamePattern("");
configuration.setEntityCacheExpiration(300);
return configuration;

8
rule-engine/rule-engine-components/src/main/resources/public/static/rulenode/rulenode-core-config.js
View File

File diff suppressed because one or more lines are too long