reverted refactoring with "return" use in the if statement & added @Override where required in DefaultTbContext & reverted changes to TbAbstractExternalNode
This commit is contained in:
parent
fabd26dc2c
commit
30bc082e0a
@ -371,10 +371,12 @@ class DefaultTbContext implements TbContext {
|
|||||||
return TbMsg.transformMsgOriginator(origMsg, originator);
|
return TbMsg.transformMsgOriginator(origMsg, originator);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
public TbMsg customerCreatedMsg(Customer customer, RuleNodeId ruleNodeId) {
|
public TbMsg customerCreatedMsg(Customer customer, RuleNodeId ruleNodeId) {
|
||||||
return entityActionMsg(customer, customer.getId(), ruleNodeId, ENTITY_CREATED);
|
return entityActionMsg(customer, customer.getId(), ruleNodeId, ENTITY_CREATED);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
public TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId) {
|
public TbMsg deviceCreatedMsg(Device device, RuleNodeId ruleNodeId) {
|
||||||
DeviceProfile deviceProfile = null;
|
DeviceProfile deviceProfile = null;
|
||||||
if (device.getDeviceProfileId() != null) {
|
if (device.getDeviceProfileId() != null) {
|
||||||
@ -383,6 +385,7 @@ class DefaultTbContext implements TbContext {
|
|||||||
return entityActionMsg(device, device.getId(), ruleNodeId, ENTITY_CREATED, deviceProfile);
|
return entityActionMsg(device, device.getId(), ruleNodeId, ENTITY_CREATED, deviceProfile);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) {
|
public TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId) {
|
||||||
AssetProfile assetProfile = null;
|
AssetProfile assetProfile = null;
|
||||||
if (asset.getAssetProfileId() != null) {
|
if (asset.getAssetProfileId() != null) {
|
||||||
@ -391,18 +394,33 @@ class DefaultTbContext implements TbContext {
|
|||||||
return entityActionMsg(asset, asset.getId(), ruleNodeId, ENTITY_CREATED, assetProfile);
|
return entityActionMsg(asset, asset.getId(), ruleNodeId, ENTITY_CREATED, assetProfile);
|
||||||
}
|
}
|
||||||
|
|
||||||
public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, TbMsgType actionMsgType) {
|
@Override
|
||||||
HasRuleEngineProfile profile = null;
|
public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action) {
|
||||||
if (EntityType.DEVICE.equals(alarm.getOriginator().getEntityType())) {
|
EntityId originator = alarm.getOriginator();
|
||||||
DeviceId deviceId = new DeviceId(alarm.getOriginator().getId());
|
HasRuleEngineProfile profile = getRuleEngineProfile(originator);
|
||||||
profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
|
return entityActionMsg(alarm, originator, ruleNodeId, action, profile);
|
||||||
} else if (EntityType.ASSET.equals(alarm.getOriginator().getEntityType())) {
|
|
||||||
AssetId assetId = new AssetId(alarm.getOriginator().getId());
|
|
||||||
profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId);
|
|
||||||
}
|
|
||||||
return entityActionMsg(alarm, alarm.getOriginator(), ruleNodeId, actionMsgType, profile);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, TbMsgType actionMsgType) {
|
||||||
|
EntityId originator = alarm.getOriginator();
|
||||||
|
HasRuleEngineProfile profile = getRuleEngineProfile(originator);
|
||||||
|
return entityActionMsg(alarm, originator, ruleNodeId, actionMsgType, profile);
|
||||||
|
}
|
||||||
|
|
||||||
|
private HasRuleEngineProfile getRuleEngineProfile(EntityId originator) {
|
||||||
|
HasRuleEngineProfile profile = null;
|
||||||
|
if (EntityType.DEVICE.equals(originator.getEntityType())) {
|
||||||
|
DeviceId deviceId = new DeviceId(originator.getId());
|
||||||
|
profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
|
||||||
|
} else if (EntityType.ASSET.equals(originator.getEntityType())) {
|
||||||
|
AssetId assetId = new AssetId(originator.getId());
|
||||||
|
profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId);
|
||||||
|
}
|
||||||
|
return profile;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List<AttributeKvEntry> attributes) {
|
public TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List<AttributeKvEntry> attributes) {
|
||||||
ObjectNode entityNode = JacksonUtil.newObjectNode();
|
ObjectNode entityNode = JacksonUtil.newObjectNode();
|
||||||
if (attributes != null) {
|
if (attributes != null) {
|
||||||
@ -411,6 +429,7 @@ class DefaultTbContext implements TbContext {
|
|||||||
return attributesActionMsg(originator, ruleNodeId, scope, ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode));
|
return attributesActionMsg(originator, ruleNodeId, scope, ATTRIBUTES_UPDATED, JacksonUtil.toString(entityNode));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
public TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List<String> keys) {
|
public TbMsg attributesDeletedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List<String> keys) {
|
||||||
ObjectNode entityNode = JacksonUtil.newObjectNode();
|
ObjectNode entityNode = JacksonUtil.newObjectNode();
|
||||||
ArrayNode attrsArrayNode = entityNode.putArray("attributes");
|
ArrayNode attrsArrayNode = entityNode.putArray("attributes");
|
||||||
@ -423,14 +442,7 @@ class DefaultTbContext implements TbContext {
|
|||||||
private TbMsg attributesActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, TbMsgType actionMsgType, String msgData) {
|
private TbMsg attributesActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, TbMsgType actionMsgType, String msgData) {
|
||||||
TbMsgMetaData tbMsgMetaData = getActionMetaData(ruleNodeId);
|
TbMsgMetaData tbMsgMetaData = getActionMetaData(ruleNodeId);
|
||||||
tbMsgMetaData.putValue("scope", scope);
|
tbMsgMetaData.putValue("scope", scope);
|
||||||
HasRuleEngineProfile profile = null;
|
HasRuleEngineProfile profile = getRuleEngineProfile(originator);
|
||||||
if (EntityType.DEVICE.equals(originator.getEntityType())) {
|
|
||||||
DeviceId deviceId = new DeviceId(originator.getId());
|
|
||||||
profile = mainCtx.getDeviceProfileCache().get(getTenantId(), deviceId);
|
|
||||||
} else if (EntityType.ASSET.equals(originator.getEntityType())) {
|
|
||||||
AssetId assetId = new AssetId(originator.getId());
|
|
||||||
profile = mainCtx.getAssetProfileCache().get(getTenantId(), assetId);
|
|
||||||
}
|
|
||||||
return entityActionMsg(originator, tbMsgMetaData, msgData, actionMsgType, profile);
|
return entityActionMsg(originator, tbMsgMetaData, msgData, actionMsgType, profile);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -443,6 +455,26 @@ class DefaultTbContext implements TbContext {
|
|||||||
return entityActionMsg(entity, id, ruleNodeId, actionMsgType, null);
|
return entityActionMsg(entity, id, ruleNodeId, actionMsgType, null);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Deprecated(since = "3.5.2", forRemoval = true)
|
||||||
|
public <E, I extends EntityId, K extends HasRuleEngineProfile> TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, String action, K profile) {
|
||||||
|
try {
|
||||||
|
return entityActionMsg(id, getActionMetaData(ruleNodeId), JacksonUtil.toString(JacksonUtil.valueToTree(entity)), action, profile);
|
||||||
|
} catch (IllegalArgumentException e) {
|
||||||
|
throw new RuntimeException("Failed to process " + id.getEntityType().name().toLowerCase() + " " + action + " msg: " + e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Deprecated(since = "3.5.2", forRemoval = true)
|
||||||
|
private <I extends EntityId, K extends HasRuleEngineProfile> TbMsg entityActionMsg(I id, TbMsgMetaData msgMetaData, String msgData, String action, K profile) {
|
||||||
|
String defaultQueueName = null;
|
||||||
|
RuleChainId defaultRuleChainId = null;
|
||||||
|
if (profile != null) {
|
||||||
|
defaultQueueName = profile.getDefaultQueueName();
|
||||||
|
defaultRuleChainId = profile.getDefaultRuleChainId();
|
||||||
|
}
|
||||||
|
return TbMsg.newMsg(defaultQueueName, action, id, msgMetaData, msgData, defaultRuleChainId, null);
|
||||||
|
}
|
||||||
|
|
||||||
public <E, I extends EntityId, K extends HasRuleEngineProfile> TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, TbMsgType actionMsgType, K profile) {
|
public <E, I extends EntityId, K extends HasRuleEngineProfile> TbMsg entityActionMsg(E entity, I id, RuleNodeId ruleNodeId, TbMsgType actionMsgType, K profile) {
|
||||||
try {
|
try {
|
||||||
return entityActionMsg(id, getActionMetaData(ruleNodeId), JacksonUtil.toString(JacksonUtil.valueToTree(entity)), actionMsgType, profile);
|
return entityActionMsg(id, getActionMetaData(ruleNodeId), JacksonUtil.toString(JacksonUtil.valueToTree(entity)), actionMsgType, profile);
|
||||||
@ -878,10 +910,17 @@ class DefaultTbContext implements TbContext {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private static String getFailureMessage(Throwable th) {
|
private static String getFailureMessage(Throwable th) {
|
||||||
if (th == null) {
|
String failureMessage;
|
||||||
return null;
|
if (th != null) {
|
||||||
|
if (!StringUtils.isEmpty(th.getMessage())) {
|
||||||
|
failureMessage = th.getMessage();
|
||||||
|
} else {
|
||||||
|
failureMessage = th.getClass().getSimpleName();
|
||||||
}
|
}
|
||||||
return StringUtils.isNotEmpty(th.getMessage()) ? th.getMessage() : th.getClass().getSimpleName();
|
} else {
|
||||||
|
failureMessage = null;
|
||||||
|
}
|
||||||
|
return failureMessage;
|
||||||
}
|
}
|
||||||
|
|
||||||
private class SimpleTbQueueCallback implements TbQueueCallback {
|
private class SimpleTbQueueCallback implements TbQueueCallback {
|
||||||
|
|||||||
@ -225,7 +225,9 @@ public interface TbContext {
|
|||||||
|
|
||||||
TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId);
|
TbMsg assetCreatedMsg(Asset asset, RuleNodeId ruleNodeId);
|
||||||
|
|
||||||
// TODO: Does this changes the message?
|
@Deprecated(since = "3.5.2", forRemoval = true)
|
||||||
|
TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, String action);
|
||||||
|
|
||||||
TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, TbMsgType actionMsgType);
|
TbMsg alarmActionMsg(Alarm alarm, RuleNodeId ruleNodeId, TbMsgType actionMsgType);
|
||||||
|
|
||||||
TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List<AttributeKvEntry> attributes);
|
TbMsg attributesUpdatedActionMsg(EntityId originator, RuleNodeId ruleNodeId, String scope, List<AttributeKvEntry> attributes);
|
||||||
|
|||||||
@ -38,11 +38,19 @@ public abstract class TbAbstractExternalNode implements TbNode {
|
|||||||
|
|
||||||
protected void tellFailure(TbContext ctx, TbMsg tbMsg, Throwable t) {
|
protected void tellFailure(TbContext ctx, TbMsg tbMsg, Throwable t) {
|
||||||
if (forceAck) {
|
if (forceAck) {
|
||||||
|
if (t == null) {
|
||||||
|
ctx.enqueueForTellNext(tbMsg.copyWithNewCtx(), TbNodeConnectionType.FAILURE);
|
||||||
|
} else {
|
||||||
ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx(), t);
|
ctx.enqueueForTellFailure(tbMsg.copyWithNewCtx(), t);
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
if (t == null) {
|
||||||
|
ctx.tellNext(tbMsg, TbNodeConnectionType.FAILURE);
|
||||||
} else {
|
} else {
|
||||||
ctx.tellFailure(tbMsg, t);
|
ctx.tellFailure(tbMsg, t);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
protected TbMsg ackIfNeeded(TbContext ctx, TbMsg msg) {
|
protected TbMsg ackIfNeeded(TbContext ctx, TbMsg msg) {
|
||||||
if (forceAck) {
|
if (forceAck) {
|
||||||
|
|||||||
@ -111,13 +111,9 @@ public class TbDeviceProfileNode implements TbNode {
|
|||||||
if (msg.checkType(TbMsgType.DEVICE_PROFILE_PERIODIC_SELF_MSG)) {
|
if (msg.checkType(TbMsgType.DEVICE_PROFILE_PERIODIC_SELF_MSG)) {
|
||||||
scheduleAlarmHarvesting(ctx, msg);
|
scheduleAlarmHarvesting(ctx, msg);
|
||||||
harvestAlarms(ctx, System.currentTimeMillis());
|
harvestAlarms(ctx, System.currentTimeMillis());
|
||||||
return;
|
} else if (msg.checkType(TbMsgType.DEVICE_PROFILE_UPDATE_SELF_MSG)) {
|
||||||
}
|
|
||||||
if (msg.checkType(TbMsgType.DEVICE_PROFILE_UPDATE_SELF_MSG)) {
|
|
||||||
updateProfile(ctx, new DeviceProfileId(UUID.fromString(msg.getData())));
|
updateProfile(ctx, new DeviceProfileId(UUID.fromString(msg.getData())));
|
||||||
return;
|
} else if (msg.checkType(TbMsgType.DEVICE_UPDATE_SELF_MSG)) {
|
||||||
}
|
|
||||||
if (msg.checkType(TbMsgType.DEVICE_UPDATE_SELF_MSG)) {
|
|
||||||
JsonNode data = JacksonUtil.toJsonNode(msg.getData());
|
JsonNode data = JacksonUtil.toJsonNode(msg.getData());
|
||||||
DeviceId deviceId = new DeviceId(UUID.fromString(data.get("deviceId").asText()));
|
DeviceId deviceId = new DeviceId(UUID.fromString(data.get("deviceId").asText()));
|
||||||
if (data.has("profileId")) {
|
if (data.has("profileId")) {
|
||||||
@ -125,31 +121,29 @@ public class TbDeviceProfileNode implements TbNode {
|
|||||||
} else {
|
} else {
|
||||||
removeDeviceState(deviceId);
|
removeDeviceState(deviceId);
|
||||||
}
|
}
|
||||||
return;
|
} else {
|
||||||
}
|
|
||||||
if (EntityType.DEVICE.equals(originatorType)) {
|
if (EntityType.DEVICE.equals(originatorType)) {
|
||||||
DeviceId deviceId = new DeviceId(msg.getOriginator().getId());
|
DeviceId deviceId = new DeviceId(msg.getOriginator().getId());
|
||||||
if (msg.checkType(TbMsgType.ENTITY_UPDATED)) {
|
if (msg.checkType(TbMsgType.ENTITY_UPDATED)) {
|
||||||
invalidateDeviceProfileCache(deviceId, msg.getData());
|
invalidateDeviceProfileCache(deviceId, msg.getData());
|
||||||
ctx.tellSuccess(msg);
|
ctx.tellSuccess(msg);
|
||||||
return;
|
} else if (msg.checkType(TbMsgType.ENTITY_DELETED)) {
|
||||||
}
|
|
||||||
if (msg.checkType(TbMsgType.ENTITY_DELETED)) {
|
|
||||||
removeDeviceState(deviceId);
|
removeDeviceState(deviceId);
|
||||||
ctx.tellSuccess(msg);
|
ctx.tellSuccess(msg);
|
||||||
return;
|
} else {
|
||||||
}
|
|
||||||
DeviceState deviceState = getOrCreateDeviceState(ctx, deviceId, null, false);
|
DeviceState deviceState = getOrCreateDeviceState(ctx, deviceId, null, false);
|
||||||
if (deviceState == null) {
|
if (deviceState != null) {
|
||||||
|
deviceState.process(ctx, msg);
|
||||||
|
} else {
|
||||||
log.info("Device was not found! Most probably device [" + deviceId + "] has been removed from the database. Acknowledging msg.");
|
log.info("Device was not found! Most probably device [" + deviceId + "] has been removed from the database. Acknowledging msg.");
|
||||||
ctx.ack(msg);
|
ctx.ack(msg);
|
||||||
|
}
|
||||||
|
}
|
||||||
} else {
|
} else {
|
||||||
deviceState.process(ctx, msg);
|
|
||||||
}
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
ctx.tellSuccess(msg);
|
ctx.tellSuccess(msg);
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@Override
|
@Override
|
||||||
public void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {
|
public void onPartitionChangeMsg(TbContext ctx, PartitionChangeMsg msg) {
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user