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

Relation node fix (#1426) 

* Fix NPE in netty-mqtt client on puback

* Fix gateway mqtt extension configuration - filter is not required

* device type expression in mqtt extension config is not required

* Relation nodes: to and from Ids mad elocal variables
pull/1431/head
Valerii Sosliuk 7 years ago
committed by GitHub
parent
ef8604a21e
commit
06f5dfd1da
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 51 additions and 43 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. 38
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbAbstractRelationActionNode.java
  2. 45
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbCreateRelationNode.java
  3. 11
      rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/action/TbDeleteRelationNode.java

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

@ -61,10 +61,8 @@ import static org.thingsboard.rule.engine.api.util.DonAsynchron.withCallback;
public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationActionNodeConfiguration> implements TbNode {
protected C config;
protected EntityId fromId;
protected EntityId toId;
private LoadingCache<Entitykey, EntityContainer> entityIdCache;
private LoadingCache<EntityKey, EntityContainer> entityIdCache;
@Override
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException {
@ -99,29 +97,33 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
protected ListenableFuture<EntityContainer> getEntity(TbContext ctx, TbMsg msg) {
String entityName = TbNodeUtils.processPattern(this.config.getEntityNamePattern(), msg.getMetaData());
String type = null;
String type;
if (this.config.getEntityTypePattern() != null) {
type = TbNodeUtils.processPattern(this.config.getEntityTypePattern(), msg.getMetaData());
} else {
type = null;
}
EntityType entityType = EntityType.valueOf(this.config.getEntityType());
Entitykey key = new Entitykey(entityName, type, entityType);
EntityKey key = new EntityKey(entityName, type, entityType);
return ctx.getDbCallbackExecutor().executeAsync(() -> {
EntityContainer entityContainer = entityIdCache.get(key);
if (entityContainer.getEntityId() == null) {
throw new RuntimeException("No entity found with type '" + key.getEntityType() + " ' and name '" + key.getEntityName() + "'.");
throw new RuntimeException("No entity found with type '" + key.getEntityType() + "' and name '" + key.getEntityName() + "'.");
}
return entityContainer;
});
}
protected void processSingleSearchDirection(TbMsg msg, EntityContainer entityContainer) {
protected SearchDirectionIds processSingleSearchDirection(TbMsg msg, EntityContainer entityContainer) {
SearchDirectionIds searchDirectionIds = new SearchDirectionIds();
if (EntitySearchDirection.FROM.name().equals(config.getDirection())) {
fromId = EntityIdFactory.getByTypeAndId(entityContainer.getEntityType().name(), entityContainer.getEntityId().toString());
toId = msg.getOriginator();
searchDirectionIds.setFromId(EntityIdFactory.getByTypeAndId(entityContainer.getEntityType().name(), entityContainer.getEntityId().toString()));
searchDirectionIds.setToId(msg.getOriginator());
} else {
toId = EntityIdFactory.getByTypeAndId(entityContainer.getEntityType().name(), entityContainer.getEntityId().toString());
fromId = msg.getOriginator();
searchDirectionIds.setToId(EntityIdFactory.getByTypeAndId(entityContainer.getEntityType().name(), entityContainer.getEntityId().toString()));
searchDirectionIds.setFromId(msg.getOriginator());
}
return searchDirectionIds;
}
protected ListenableFuture<List<EntityRelation>> processListSearchDirection(TbContext ctx, TbMsg msg) {
@ -134,13 +136,19 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
@Data
@AllArgsConstructor
private static class Entitykey {
private static class EntityKey {
private String entityName;
private String type;
private EntityType entityType;
}
private static class EntityCacheLoader extends CacheLoader<Entitykey, EntityContainer> {
@Data
protected static class SearchDirectionIds {
private EntityId fromId;
private EntityId toId;
}
private static class EntityCacheLoader extends CacheLoader<EntityKey, EntityContainer> {
private final TbContext ctx;
private final boolean createIfNotExists;
@ -151,11 +159,11 @@ public abstract class TbAbstractRelationActionNode<C extends TbAbstractRelationA
}
@Override
public EntityContainer load(Entitykey key) {
public EntityContainer load(EntityKey key) {
return loadEntity(key);
}
private EntityContainer loadEntity(Entitykey entitykey) {
private EntityContainer loadEntity(EntityKey entitykey) {
EntityType type = entitykey.getEntityType();
EntityContainer targetEntity = new EntityContainer();
targetEntity.setEntityType(type);

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

@ -65,92 +65,91 @@ public class TbCreateRelationNode extends TbAbstractRelationActionNode<TbCreateR
}
private ListenableFuture<Boolean> createIfAbsent(TbContext ctx, TbMsg msg, EntityContainer entityContainer) {
processSingleSearchDirection(msg, entityContainer);
return Futures.transformAsync(ctx.getRelationService().checkRelation(ctx.getTenantId(), fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON),
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);
return processCreateRelation(ctx, entityContainer, sdId);
}
return Futures.immediateFuture(true);
});
}
private ListenableFuture<Boolean> processCreateRelation(TbContext ctx, EntityContainer entityContainer) {
private ListenableFuture<Boolean> processCreateRelation(TbContext ctx, EntityContainer entityContainer, SearchDirectionIds sdId) {
switch (entityContainer.getEntityType()) {
case ASSET:
return processAsset(ctx, entityContainer);
return processAsset(ctx, entityContainer, sdId);
case DEVICE:
return processDevice(ctx, entityContainer);
return processDevice(ctx, entityContainer, sdId);
case CUSTOMER:
return processCustomer(ctx, entityContainer);
return processCustomer(ctx, entityContainer, sdId);
case DASHBOARD:
return processDashboard(ctx, entityContainer);
return processDashboard(ctx, entityContainer, sdId);
case ENTITY_VIEW:
return processView(ctx, entityContainer);
return processView(ctx, entityContainer, sdId);
case TENANT:
return processTenant(ctx, entityContainer);
return processTenant(ctx, entityContainer, sdId);
}
return Futures.immediateFuture(true);
}
private ListenableFuture<Boolean> processView(TbContext ctx, EntityContainer entityContainer) {
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(fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON));
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
} else {
return Futures.immediateFuture(true);
}
});
}
private ListenableFuture<Boolean> processDevice(TbContext ctx, EntityContainer entityContainer) {
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(fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON));
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
} else {
return Futures.immediateFuture(true);
}
});
}
private ListenableFuture<Boolean> processAsset(TbContext ctx, EntityContainer entityContainer) {
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(fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON));
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
} else {
return Futures.immediateFuture(true);
}
});
}
private ListenableFuture<Boolean> processCustomer(TbContext ctx, EntityContainer entityContainer) {
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(fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON));
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
} else {
return Futures.immediateFuture(true);
}
});
}
private ListenableFuture<Boolean> processDashboard(TbContext ctx, EntityContainer entityContainer) {
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(fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON));
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
} else {
return Futures.immediateFuture(true);
}
});
}
private ListenableFuture<Boolean> processTenant(TbContext ctx, EntityContainer entityContainer) {
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(fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON));
return ctx.getRelationService().saveRelationAsync(ctx.getTenantId(), new EntityRelation(sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON));
} else {
return Futures.immediateFuture(true);
}
});
}
}

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

@ -93,18 +93,19 @@ public class TbDeleteRelationNode extends TbAbstractRelationActionNode<TbDeleteR
}
private ListenableFuture<Boolean> processSingle(TbContext ctx, TbMsg msg, EntityContainer entityContainer) {
processSingleSearchDirection(msg, entityContainer);
return Futures.transformAsync(ctx.getRelationService().checkRelation(ctx.getTenantId(), fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON),
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 processSingleDeleteRelation(ctx);
return processSingleDeleteRelation(ctx, sdId);
}
return Futures.immediateFuture(true);
});
}
private ListenableFuture<Boolean> processSingleDeleteRelation(TbContext ctx) {
return ctx.getRelationService().deleteRelationAsync(ctx.getTenantId(), fromId, toId, config.getRelationType(), RelationTypeGroup.COMMON);
private ListenableFuture<Boolean> processSingleDeleteRelation(TbContext ctx, SearchDirectionIds sdId) {
return ctx.getRelationService().deleteRelationAsync(ctx.getTenantId(), sdId.getFromId(), sdId.getToId(), config.getRelationType(), RelationTypeGroup.COMMON);
}
}

Write Preview
Loading…
Cancel
Save