|
|
@ -28,13 +28,25 @@ import org.thingsboard.rule.engine.util.EntitiesAlarmOriginatorIdAsyncLoader; |
|
|
import org.thingsboard.rule.engine.util.EntitiesCustomerIdAsyncLoader; |
|
|
import org.thingsboard.rule.engine.util.EntitiesCustomerIdAsyncLoader; |
|
|
import org.thingsboard.rule.engine.util.EntitiesRelatedEntityIdAsyncLoader; |
|
|
import org.thingsboard.rule.engine.util.EntitiesRelatedEntityIdAsyncLoader; |
|
|
import org.thingsboard.rule.engine.util.EntitiesTenantIdAsyncLoader; |
|
|
import org.thingsboard.rule.engine.util.EntitiesTenantIdAsyncLoader; |
|
|
|
|
|
import org.thingsboard.server.common.data.Customer; |
|
|
|
|
|
import org.thingsboard.server.common.data.DashboardInfo; |
|
|
|
|
|
import org.thingsboard.server.common.data.Device; |
|
|
|
|
|
import org.thingsboard.server.common.data.EntityType; |
|
|
|
|
|
import org.thingsboard.server.common.data.EntityView; |
|
|
|
|
|
import org.thingsboard.server.common.data.StringUtils; |
|
|
|
|
|
import org.thingsboard.server.common.data.User; |
|
|
|
|
|
import org.thingsboard.server.common.data.asset.Asset; |
|
|
|
|
|
import org.thingsboard.server.common.data.edge.Edge; |
|
|
import org.thingsboard.server.common.data.id.EntityId; |
|
|
import org.thingsboard.server.common.data.id.EntityId; |
|
|
|
|
|
import org.thingsboard.server.common.data.page.PageData; |
|
|
|
|
|
import org.thingsboard.server.common.data.page.PageLink; |
|
|
import org.thingsboard.server.common.data.plugin.ComponentType; |
|
|
import org.thingsboard.server.common.data.plugin.ComponentType; |
|
|
import org.thingsboard.server.common.msg.TbMsg; |
|
|
import org.thingsboard.server.common.msg.TbMsg; |
|
|
|
|
|
|
|
|
import java.util.Collections; |
|
|
import java.util.Collections; |
|
|
import java.util.HashSet; |
|
|
import java.util.HashSet; |
|
|
import java.util.List; |
|
|
import java.util.List; |
|
|
|
|
|
import java.util.Optional; |
|
|
|
|
|
|
|
|
@Slf4j |
|
|
@Slf4j |
|
|
@RuleNode( |
|
|
@RuleNode( |
|
|
@ -55,6 +67,7 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode { |
|
|
protected static final String TENANT_SOURCE = "TENANT"; |
|
|
protected static final String TENANT_SOURCE = "TENANT"; |
|
|
protected static final String RELATED_SOURCE = "RELATED"; |
|
|
protected static final String RELATED_SOURCE = "RELATED"; |
|
|
protected static final String ALARM_ORIGINATOR_SOURCE = "ALARM_ORIGINATOR"; |
|
|
protected static final String ALARM_ORIGINATOR_SOURCE = "ALARM_ORIGINATOR"; |
|
|
|
|
|
protected static final String ENTITY_SOURCE = "ENTITY_SOURCE"; |
|
|
|
|
|
|
|
|
private TbChangeOriginatorNodeConfiguration config; |
|
|
private TbChangeOriginatorNodeConfiguration config; |
|
|
|
|
|
|
|
|
@ -67,7 +80,7 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode { |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
protected ListenableFuture<List<TbMsg>> transform(TbContext ctx, TbMsg msg) { |
|
|
protected ListenableFuture<List<TbMsg>> transform(TbContext ctx, TbMsg msg) { |
|
|
ListenableFuture<? extends EntityId> newOriginator = getNewOriginator(ctx, msg.getOriginator()); |
|
|
ListenableFuture<? extends EntityId> newOriginator = getNewOriginator(ctx, msg.getOriginator(), msg); |
|
|
return Futures.transform(newOriginator, n -> { |
|
|
return Futures.transform(newOriginator, n -> { |
|
|
if (n == null || n.isNullUid()) { |
|
|
if (n == null || n.isNullUid()) { |
|
|
return null; |
|
|
return null; |
|
|
@ -76,7 +89,7 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode { |
|
|
}, ctx.getDbCallbackExecutor()); |
|
|
}, ctx.getDbCallbackExecutor()); |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
private ListenableFuture<? extends EntityId> getNewOriginator(TbContext ctx, EntityId original) { |
|
|
private ListenableFuture<? extends EntityId> getNewOriginator(TbContext ctx, EntityId original, TbMsg msg) { |
|
|
switch (config.getOriginatorSource()) { |
|
|
switch (config.getOriginatorSource()) { |
|
|
case CUSTOMER_SOURCE: |
|
|
case CUSTOMER_SOURCE: |
|
|
return EntitiesCustomerIdAsyncLoader.findEntityIdAsync(ctx, original); |
|
|
return EntitiesCustomerIdAsyncLoader.findEntityIdAsync(ctx, original); |
|
|
@ -86,13 +99,79 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode { |
|
|
return EntitiesRelatedEntityIdAsyncLoader.findEntityAsync(ctx, original, config.getRelationsQuery()); |
|
|
return EntitiesRelatedEntityIdAsyncLoader.findEntityAsync(ctx, original, config.getRelationsQuery()); |
|
|
case ALARM_ORIGINATOR_SOURCE: |
|
|
case ALARM_ORIGINATOR_SOURCE: |
|
|
return EntitiesAlarmOriginatorIdAsyncLoader.findEntityIdAsync(ctx, original); |
|
|
return EntitiesAlarmOriginatorIdAsyncLoader.findEntityIdAsync(ctx, original); |
|
|
|
|
|
case ENTITY_SOURCE: |
|
|
|
|
|
return getEntity(ctx, original, msg); |
|
|
default: |
|
|
default: |
|
|
return Futures.immediateFailedFuture(new IllegalStateException("Unexpected originator source " + config.getOriginatorSource())); |
|
|
return Futures.immediateFailedFuture(new IllegalStateException("Unexpected originator source " + config.getOriginatorSource())); |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
private ListenableFuture<? extends EntityId> getEntity(TbContext ctx, EntityId original, TbMsg msg) { |
|
|
|
|
|
EntityType entityType = EntityType.valueOf(config.getEntityType()); |
|
|
|
|
|
String entityName = TbNodeUtils.processPattern(config.getEntityNamePattern(), msg); |
|
|
|
|
|
EntityId targetEntity = null; |
|
|
|
|
|
switch (entityType) { |
|
|
|
|
|
case DEVICE: |
|
|
|
|
|
Device device = ctx.getDeviceService().findDeviceByTenantIdAndName(ctx.getTenantId(), entityName); |
|
|
|
|
|
if (device != null) { |
|
|
|
|
|
targetEntity = device.getId(); |
|
|
|
|
|
} |
|
|
|
|
|
break; |
|
|
|
|
|
case ASSET: |
|
|
|
|
|
Asset asset = ctx.getAssetService().findAssetByTenantIdAndName(ctx.getTenantId(), entityName); |
|
|
|
|
|
if (asset != null) { |
|
|
|
|
|
targetEntity = asset.getId(); |
|
|
|
|
|
} |
|
|
|
|
|
break; |
|
|
|
|
|
case CUSTOMER: |
|
|
|
|
|
Optional<Customer> customerOptional = ctx.getCustomerService().findCustomerByTenantIdAndTitle(ctx.getTenantId(), entityName); |
|
|
|
|
|
if (customerOptional.isPresent()) { |
|
|
|
|
|
targetEntity = customerOptional.get().getId(); |
|
|
|
|
|
} |
|
|
|
|
|
break; |
|
|
|
|
|
case TENANT: |
|
|
|
|
|
targetEntity = ctx.getTenantId(); |
|
|
|
|
|
break; |
|
|
|
|
|
case ENTITY_VIEW: |
|
|
|
|
|
EntityView entityView = ctx.getEntityViewService().findEntityViewByTenantIdAndName(ctx.getTenantId(), entityName); |
|
|
|
|
|
if (entityView != null) { |
|
|
|
|
|
targetEntity = entityView.getId(); |
|
|
|
|
|
} |
|
|
|
|
|
break; |
|
|
|
|
|
case EDGE: |
|
|
|
|
|
Edge edge = ctx.getEdgeService().findEdgeByTenantIdAndName(ctx.getTenantId(), entityName); |
|
|
|
|
|
if (edge != null) { |
|
|
|
|
|
targetEntity = edge.getId(); |
|
|
|
|
|
} |
|
|
|
|
|
break; |
|
|
|
|
|
case DASHBOARD: |
|
|
|
|
|
PageData<DashboardInfo> dashboardInfoTextPageData = ctx.getDashboardService().findDashboardsByTenantId(ctx.getTenantId(), new PageLink(200, 0, entityName)); |
|
|
|
|
|
Optional<DashboardInfo> currentDashboardInfo = dashboardInfoTextPageData.getData().stream() |
|
|
|
|
|
.filter(dashboardInfo -> dashboardInfo.getTitle().equals(entityName)) |
|
|
|
|
|
.findFirst(); |
|
|
|
|
|
if (currentDashboardInfo.isPresent()) { |
|
|
|
|
|
targetEntity = currentDashboardInfo.get().getId(); |
|
|
|
|
|
} |
|
|
|
|
|
break; |
|
|
|
|
|
case USER: |
|
|
|
|
|
User user = ctx.getUserService().findUserByEmail(ctx.getTenantId(), entityName); |
|
|
|
|
|
if (user != null) { |
|
|
|
|
|
targetEntity = user.getId(); |
|
|
|
|
|
} |
|
|
|
|
|
break; |
|
|
|
|
|
default: |
|
|
|
|
|
return Futures.immediateFailedFuture(new IllegalStateException("Unexpected entity type " + config.getEntityType())); |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
if (targetEntity != null) { |
|
|
|
|
|
return Futures.immediateFuture(targetEntity); |
|
|
|
|
|
} else { |
|
|
|
|
|
return Futures.immediateFailedFuture(new IllegalStateException("Entity '" + config.getEntityType() + "' not found by name '" + entityName + "'!")); |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
private void validateConfig(TbChangeOriginatorNodeConfiguration conf) { |
|
|
private void validateConfig(TbChangeOriginatorNodeConfiguration conf) { |
|
|
HashSet<String> knownSources = Sets.newHashSet(CUSTOMER_SOURCE, TENANT_SOURCE, RELATED_SOURCE, ALARM_ORIGINATOR_SOURCE); |
|
|
HashSet<String> knownSources = Sets.newHashSet(CUSTOMER_SOURCE, TENANT_SOURCE, RELATED_SOURCE, ALARM_ORIGINATOR_SOURCE, ENTITY_SOURCE); |
|
|
if (!knownSources.contains(conf.getOriginatorSource())) { |
|
|
if (!knownSources.contains(conf.getOriginatorSource())) { |
|
|
log.error("Unsupported source [{}] for TbChangeOriginatorNode", conf.getOriginatorSource()); |
|
|
log.error("Unsupported source [{}] for TbChangeOriginatorNode", conf.getOriginatorSource()); |
|
|
throw new IllegalArgumentException("Unsupported source TbChangeOriginatorNode" + conf.getOriginatorSource()); |
|
|
throw new IllegalArgumentException("Unsupported source TbChangeOriginatorNode" + conf.getOriginatorSource()); |
|
|
@ -106,6 +185,17 @@ public class TbChangeOriginatorNode extends TbAbstractTransformNode { |
|
|
} |
|
|
} |
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
if (conf.getOriginatorSource().equals(ENTITY_SOURCE)) { |
|
|
|
|
|
if (conf.getEntityType() == null) { |
|
|
|
|
|
log.error("Entity type not specified for [{}]", ENTITY_SOURCE); |
|
|
|
|
|
throw new IllegalArgumentException("Wrong config for [{}] in TbChangeOriginatorNode!" + ENTITY_SOURCE); |
|
|
|
|
|
} |
|
|
|
|
|
if (StringUtils.isEmpty(conf.getEntityNamePattern())) { |
|
|
|
|
|
log.error("EntityNamePattern not specified for type [{}]", conf.getEntityType()); |
|
|
|
|
|
throw new IllegalArgumentException("Wrong config for [{}] in TbChangeOriginatorNode!" + ENTITY_SOURCE); |
|
|
|
|
|
} |
|
|
|
|
|
} |
|
|
|
|
|
|
|
|
} |
|
|
} |
|
|
|
|
|
|
|
|
@Override |
|
|
@Override |
|
|
|