Andrew Shvayka
5 years ago
committed by
GitHub
GitHub
69 changed files with 1577 additions and 436 deletions
@ -0,0 +1,41 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.actors.ruleChain; |
|||
|
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.Getter; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.03.18. |
|||
*/ |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@ToString |
|||
public final class RuleChainInputMsg extends TbToRuleChainActorMsg { |
|||
|
|||
public RuleChainInputMsg(RuleChainId target, TbMsg tbMsg) { |
|||
super(tbMsg, target); |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.RULE_CHAIN_INPUT_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,49 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.actors.ruleChain; |
|||
|
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.Getter; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
import org.thingsboard.server.common.msg.MsgType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.03.18. |
|||
*/ |
|||
@EqualsAndHashCode(callSuper = true) |
|||
@ToString |
|||
public final class RuleChainOutputMsg extends TbToRuleChainActorMsg { |
|||
|
|||
@Getter |
|||
private final RuleNodeId targetRuleNodeId; |
|||
|
|||
@Getter |
|||
private final String relationType; |
|||
|
|||
public RuleChainOutputMsg(RuleChainId target, RuleNodeId targetRuleNodeId, String relationType, TbMsg tbMsg) { |
|||
super(tbMsg, target); |
|||
this.targetRuleNodeId = targetRuleNodeId; |
|||
this.relationType = relationType; |
|||
} |
|||
|
|||
@Override |
|||
public MsgType getMsgType() { |
|||
return MsgType.RULE_CHAIN_OUTPUT_MSG; |
|||
} |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.actors.ruleChain; |
|||
|
|||
import lombok.EqualsAndHashCode; |
|||
import lombok.Getter; |
|||
import lombok.ToString; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.msg.TbActorStopReason; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbRuleEngineActorMsg; |
|||
import org.thingsboard.server.common.msg.aware.RuleChainAwareMsg; |
|||
import org.thingsboard.server.common.msg.queue.RuleEngineException; |
|||
|
|||
@EqualsAndHashCode(callSuper = true) |
|||
@ToString |
|||
public abstract class TbToRuleChainActorMsg extends TbRuleEngineActorMsg implements RuleChainAwareMsg { |
|||
|
|||
@Getter |
|||
private final RuleChainId target; |
|||
|
|||
public TbToRuleChainActorMsg(TbMsg msg, RuleChainId target) { |
|||
super(msg); |
|||
this.target = target; |
|||
} |
|||
|
|||
@Override |
|||
public RuleChainId getRuleChainId() { |
|||
return target; |
|||
} |
|||
|
|||
@Override |
|||
public void onTbActorStopped(TbActorStopReason reason) { |
|||
String message = reason == TbActorStopReason.STOPPED ? String.format("Rule chain [%s] stopped", target.getId()) : String.format("Failed to initialize rule chain [%s]!", target.getId()); |
|||
msg.getCallback().onFailure(new RuleEngineException(message)); |
|||
} |
|||
} |
|||
@ -0,0 +1,190 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.service.rule; |
|||
|
|||
import lombok.RequiredArgsConstructor; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.rule.engine.flow.TbRuleChainInputNode; |
|||
import org.thingsboard.rule.engine.flow.TbRuleChainInputNodeConfiguration; |
|||
import org.thingsboard.rule.engine.flow.TbRuleChainOutputNode; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.relation.EntityRelation; |
|||
import org.thingsboard.server.common.data.rule.RuleChain; |
|||
import org.thingsboard.server.common.data.rule.RuleChainMetaData; |
|||
import org.thingsboard.server.common.data.rule.RuleChainOutputLabelsUsage; |
|||
import org.thingsboard.server.common.data.rule.RuleChainUpdateResult; |
|||
import org.thingsboard.server.common.data.rule.RuleNode; |
|||
import org.thingsboard.server.common.data.rule.RuleNodeUpdateResult; |
|||
import org.thingsboard.server.dao.relation.RelationService; |
|||
import org.thingsboard.server.dao.rule.RuleChainService; |
|||
import org.thingsboard.server.queue.util.TbCoreComponent; |
|||
|
|||
import java.util.Comparator; |
|||
import java.util.HashMap; |
|||
import java.util.HashSet; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.Set; |
|||
import java.util.TreeSet; |
|||
import java.util.stream.Collectors; |
|||
|
|||
@RequiredArgsConstructor |
|||
@Service |
|||
@TbCoreComponent |
|||
@Slf4j |
|||
public class DefaultTbRuleChainService implements TbRuleChainService { |
|||
|
|||
private final RuleChainService ruleChainService; |
|||
private final RelationService relationService; |
|||
|
|||
@Override |
|||
public Set<String> getRuleChainOutputLabels(TenantId tenantId, RuleChainId ruleChainId) { |
|||
RuleChainMetaData metaData = ruleChainService.loadRuleChainMetaData(tenantId, ruleChainId); |
|||
Set<String> outputLabels = new TreeSet<>(); |
|||
for (RuleNode ruleNode : metaData.getNodes()) { |
|||
if (isOutputRuleNode(ruleNode)) { |
|||
outputLabels.add(ruleNode.getName()); |
|||
} |
|||
} |
|||
return outputLabels; |
|||
} |
|||
|
|||
@Override |
|||
public List<RuleChainOutputLabelsUsage> getOutputLabelUsage(TenantId tenantId, RuleChainId ruleChainId) { |
|||
List<RuleNode> ruleNodes = ruleChainService.findRuleNodesByTenantIdAndType(tenantId, TbRuleChainInputNode.class.getName(), ruleChainId.getId().toString()); |
|||
Map<RuleChainId, String> ruleChainNamesCache = new HashMap<>(); |
|||
// Additional filter, "just in case" the structure of the JSON configuration will change.
|
|||
var filteredRuleNodes = ruleNodes.stream().filter(node -> { |
|||
try { |
|||
TbRuleChainInputNodeConfiguration configuration = JacksonUtil.treeToValue(node.getConfiguration(), TbRuleChainInputNodeConfiguration.class); |
|||
return ruleChainId.getId().toString().equals(configuration.getRuleChainId()); |
|||
} catch (Exception e) { |
|||
log.warn("[{}][{}] Failed to decode rule node configuration", tenantId, ruleChainId, e); |
|||
return false; |
|||
} |
|||
}).collect(Collectors.toList()); |
|||
|
|||
|
|||
return filteredRuleNodes.stream() |
|||
.map(ruleNode -> { |
|||
RuleChainOutputLabelsUsage usage = new RuleChainOutputLabelsUsage(); |
|||
usage.setRuleNodeId(ruleNode.getId()); |
|||
usage.setRuleNodeName(ruleNode.getName()); |
|||
usage.setRuleChainId(ruleNode.getRuleChainId()); |
|||
List<EntityRelation> relations = ruleChainService.getRuleNodeRelations(tenantId, ruleNode.getId()); |
|||
if (relations != null && !relations.isEmpty()) { |
|||
usage.setLabels(relations.stream().map(EntityRelation::getType).collect(Collectors.toSet())); |
|||
} |
|||
return usage; |
|||
}) |
|||
.filter(usage -> usage.getLabels() != null) |
|||
.peek(usage -> { |
|||
String ruleChainName = ruleChainNamesCache.computeIfAbsent(usage.getRuleChainId(), |
|||
id -> ruleChainService.findRuleChainById(tenantId, id).getName()); |
|||
usage.setRuleChainName(ruleChainName); |
|||
}) |
|||
.sorted(Comparator |
|||
.comparing(RuleChainOutputLabelsUsage::getRuleChainName) |
|||
.thenComparing(RuleChainOutputLabelsUsage::getRuleNodeName)) |
|||
.collect(Collectors.toList()); |
|||
} |
|||
|
|||
@Override |
|||
public List<RuleChain> updateRelatedRuleChains(TenantId tenantId, RuleChainId ruleChainId, RuleChainUpdateResult result) { |
|||
Set<RuleChainId> ruleChainIds = new HashSet<>(); |
|||
log.debug("[{}][{}] Going to update links in related rule chains", tenantId, ruleChainId); |
|||
if (result.getUpdatedRuleNodes() == null || result.getUpdatedRuleNodes().isEmpty()) { |
|||
return null; |
|||
} |
|||
|
|||
Set<String> oldLabels = new HashSet<>(); |
|||
Set<String> newLabels = new HashSet<>(); |
|||
Set<String> confusedLabels = new HashSet<>(); |
|||
Map<String, String> updatedLabels = new HashMap<>(); |
|||
for (RuleNodeUpdateResult update : result.getUpdatedRuleNodes()) { |
|||
var oldNode = update.getOldRuleNode(); |
|||
var newNode = update.getNewRuleNode(); |
|||
if (isOutputRuleNode(newNode)) { |
|||
try { |
|||
oldLabels.add(oldNode.getName()); |
|||
newLabels.add(newNode.getName()); |
|||
if (!oldNode.getName().equals(newNode.getName())) { |
|||
String oldLabel = oldNode.getName(); |
|||
String newLabel = newNode.getName(); |
|||
if (updatedLabels.containsKey(oldLabel) && !updatedLabels.get(oldLabel).equals(newLabel)) { |
|||
confusedLabels.add(oldLabel); |
|||
log.warn("[{}][{}] Can't automatically rename the label from [{}] to [{}] due to conflict [{}]", tenantId, ruleChainId, oldLabel, newLabel, updatedLabels.get(oldLabel)); |
|||
} else { |
|||
updatedLabels.put(oldLabel, newLabel); |
|||
} |
|||
|
|||
} |
|||
} catch (Exception e) { |
|||
log.warn("[{}][{}][{}] Failed to decode rule node configuration", tenantId, ruleChainId, newNode.getId(), e); |
|||
} |
|||
} |
|||
} |
|||
// Remove all output labels that are renamed to two or more different labels, since we don't which new label to use;
|
|||
confusedLabels.forEach(updatedLabels::remove); |
|||
// Remove all output labels that are renamed but still present in the rule chain;
|
|||
newLabels.forEach(updatedLabels::remove); |
|||
if (!oldLabels.equals(newLabels)) { |
|||
ruleChainIds.addAll(updateRelatedRuleChains(tenantId, ruleChainId, updatedLabels)); |
|||
} |
|||
return ruleChainIds.stream().map(id -> ruleChainService.findRuleChainById(tenantId, id)).collect(Collectors.toList()); |
|||
} |
|||
|
|||
public Set<RuleChainId> updateRelatedRuleChains(TenantId tenantId, RuleChainId ruleChainId, Map<String, String> labelsMap) { |
|||
Set<RuleChainId> updatedRuleChains = new HashSet<>(); |
|||
List<RuleChainOutputLabelsUsage> usageList = getOutputLabelUsage(tenantId, ruleChainId); |
|||
for (RuleChainOutputLabelsUsage usage : usageList) { |
|||
labelsMap.forEach((oldLabel, newLabel) -> { |
|||
if (usage.getLabels().contains(oldLabel)) { |
|||
updatedRuleChains.add(usage.getRuleChainId()); |
|||
renameOutgoingLinks(tenantId, usage.getRuleNodeId(), oldLabel, newLabel); |
|||
} |
|||
}); |
|||
} |
|||
return updatedRuleChains; |
|||
} |
|||
|
|||
private void renameOutgoingLinks(TenantId tenantId, RuleNodeId ruleNodeId, String oldLabel, String newLabel) { |
|||
List<EntityRelation> relations = ruleChainService.getRuleNodeRelations(tenantId, ruleNodeId); |
|||
for (EntityRelation relation : relations) { |
|||
if (relation.getType().equals(oldLabel)) { |
|||
relationService.deleteRelation(tenantId, relation); |
|||
relation.setType(newLabel); |
|||
relationService.saveRelation(tenantId, relation); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private boolean isOutputRuleNode(RuleNode ruleNode) { |
|||
return isRuleNode(ruleNode, TbRuleChainOutputNode.class); |
|||
} |
|||
|
|||
private boolean isInputRuleNode(RuleNode ruleNode) { |
|||
return isRuleNode(ruleNode, TbRuleChainInputNode.class); |
|||
} |
|||
|
|||
private boolean isRuleNode(RuleNode ruleNode, Class<?> clazz) { |
|||
return ruleNode != null && ruleNode.getType().equals(clazz.getName()); |
|||
} |
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.service.rule; |
|||
|
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.rule.RuleChain; |
|||
import org.thingsboard.server.common.data.rule.RuleChainOutputLabelsUsage; |
|||
import org.thingsboard.server.common.data.rule.RuleChainUpdateResult; |
|||
|
|||
import java.util.List; |
|||
import java.util.Set; |
|||
|
|||
public interface TbRuleChainService { |
|||
|
|||
Set<String> getRuleChainOutputLabels(TenantId tenantId, RuleChainId ruleChainId); |
|||
|
|||
List<RuleChainOutputLabelsUsage> getOutputLabelUsage(TenantId tenantId, RuleChainId ruleChainId); |
|||
|
|||
List<RuleChain> updateRelatedRuleChains(TenantId tenantId, RuleChainId ruleChainId, RuleChainUpdateResult result); |
|||
} |
|||
@ -0,0 +1,45 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.data.rule; |
|||
|
|||
import io.swagger.annotations.ApiModel; |
|||
import io.swagger.annotations.ApiModelProperty; |
|||
import lombok.Data; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.jetbrains.annotations.NotNull; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
|
|||
import java.util.Set; |
|||
|
|||
@ApiModel |
|||
@Data |
|||
@Slf4j |
|||
public class RuleChainOutputLabelsUsage { |
|||
|
|||
@ApiModelProperty(position = 1, required = true, value = "Rule Chain Id", accessMode = ApiModelProperty.AccessMode.READ_ONLY) |
|||
private RuleChainId ruleChainId; |
|||
@ApiModelProperty(position = 2, required = true, value = "Rule Node Id", accessMode = ApiModelProperty.AccessMode.READ_ONLY) |
|||
private RuleNodeId ruleNodeId; |
|||
|
|||
@ApiModelProperty(position = 3, required = true, value = "Rule Chain Name", accessMode = ApiModelProperty.AccessMode.READ_ONLY) |
|||
private String ruleChainName; |
|||
@ApiModelProperty(position = 4, required = true, value = "Rule Node Name", accessMode = ApiModelProperty.AccessMode.READ_ONLY) |
|||
private String ruleNodeName; |
|||
@ApiModelProperty(position = 5, required = true, value = "Output labels", accessMode = ApiModelProperty.AccessMode.READ_ONLY) |
|||
private Set<String> labels; |
|||
|
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.data.rule; |
|||
|
|||
import lombok.AccessLevel; |
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
|
|||
import java.util.List; |
|||
import java.util.Map; |
|||
|
|||
/** |
|||
* Created by igor on 3/13/18. |
|||
*/ |
|||
@Data |
|||
@AllArgsConstructor(access = AccessLevel.PRIVATE) |
|||
public class RuleChainUpdateResult { |
|||
|
|||
private final boolean success; |
|||
private final List<RuleNodeUpdateResult> updatedRuleNodes; |
|||
|
|||
public static RuleChainUpdateResult failed(){ |
|||
return new RuleChainUpdateResult(false, null); |
|||
} |
|||
|
|||
public static RuleChainUpdateResult successful(List<RuleNodeUpdateResult> updatedRuleNodes){ |
|||
return new RuleChainUpdateResult(true, updatedRuleNodes); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,33 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.data.rule; |
|||
|
|||
import com.fasterxml.jackson.databind.JsonNode; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
|
|||
import java.util.Map; |
|||
|
|||
/** |
|||
* Created by igor on 3/13/18. |
|||
*/ |
|||
@Data |
|||
public class RuleNodeUpdateResult { |
|||
|
|||
private final RuleNode oldRuleNode; |
|||
private final RuleNode newRuleNode; |
|||
|
|||
} |
|||
@ -0,0 +1,95 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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; |
|||
|
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
import org.thingsboard.server.common.msg.gen.MsgProtos; |
|||
|
|||
import java.io.Serializable; |
|||
import java.util.LinkedList; |
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.atomic.AtomicInteger; |
|||
|
|||
/** |
|||
* Created by ashvayka on 13.01.18. |
|||
*/ |
|||
public final class TbMsgProcessingCtx implements Serializable { |
|||
|
|||
private final AtomicInteger ruleNodeExecCounter; |
|||
private volatile LinkedList<TbMsgProcessingStackItem> stack; |
|||
|
|||
public TbMsgProcessingCtx() { |
|||
this(0); |
|||
} |
|||
|
|||
public TbMsgProcessingCtx(int ruleNodeExecCounter) { |
|||
this(ruleNodeExecCounter, null); |
|||
} |
|||
|
|||
protected TbMsgProcessingCtx(int ruleNodeExecCounter, LinkedList<TbMsgProcessingStackItem> stack) { |
|||
this.ruleNodeExecCounter = new AtomicInteger(ruleNodeExecCounter); |
|||
this.stack = stack; |
|||
} |
|||
|
|||
public int getAndIncrementRuleNodeCounter() { |
|||
return ruleNodeExecCounter.getAndIncrement(); |
|||
} |
|||
|
|||
public TbMsgProcessingCtx copy() { |
|||
if (stack == null || stack.isEmpty()) { |
|||
return new TbMsgProcessingCtx(ruleNodeExecCounter.get()); |
|||
} else { |
|||
return new TbMsgProcessingCtx(ruleNodeExecCounter.get(), new LinkedList<>(stack)); |
|||
} |
|||
} |
|||
|
|||
public void push(RuleChainId ruleChainId, RuleNodeId ruleNodeId) { |
|||
if (stack == null) { |
|||
stack = new LinkedList<>(); |
|||
} |
|||
stack.add(new TbMsgProcessingStackItem(ruleChainId, ruleNodeId)); |
|||
} |
|||
|
|||
public TbMsgProcessingStackItem pop() { |
|||
return !stack.isEmpty() ? stack.removeLast() : null; |
|||
} |
|||
|
|||
public static TbMsgProcessingCtx fromProto(MsgProtos.TbMsgProcessingCtxProto ctx) { |
|||
int ruleNodeExecCounter = ctx.getRuleNodeExecCounter(); |
|||
if (ctx.getStackCount() > 0) { |
|||
LinkedList<TbMsgProcessingStackItem> stack = new LinkedList<>(); |
|||
for (MsgProtos.TbMsgProcessingStackItemProto item : ctx.getStackList()) { |
|||
stack.add(TbMsgProcessingStackItem.fromProto(item)); |
|||
} |
|||
return new TbMsgProcessingCtx(ruleNodeExecCounter, stack); |
|||
} else { |
|||
return new TbMsgProcessingCtx(ruleNodeExecCounter); |
|||
} |
|||
} |
|||
|
|||
public MsgProtos.TbMsgProcessingCtxProto toProto() { |
|||
var ctxBuilder = MsgProtos.TbMsgProcessingCtxProto.newBuilder(); |
|||
ctxBuilder.setRuleNodeExecCounter(ruleNodeExecCounter.get()); |
|||
if (stack != null) { |
|||
for (TbMsgProcessingStackItem item : stack) { |
|||
ctxBuilder.addStack(item.toProto()); |
|||
} |
|||
} |
|||
return ctxBuilder.build(); |
|||
} |
|||
} |
|||
@ -0,0 +1,47 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.RuleNodeId; |
|||
import org.thingsboard.server.common.msg.gen.MsgProtos; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
public class TbMsgProcessingStackItem { |
|||
|
|||
private final RuleChainId ruleChainId; |
|||
private final RuleNodeId ruleNodeId; |
|||
|
|||
MsgProtos.TbMsgProcessingStackItemProto toProto() { |
|||
return MsgProtos.TbMsgProcessingStackItemProto.newBuilder() |
|||
.setRuleChainIdMSB(ruleChainId.getId().getMostSignificantBits()) |
|||
.setRuleChainIdLSB(ruleChainId.getId().getLeastSignificantBits()) |
|||
.setRuleNodeIdMSB(ruleNodeId.getId().getMostSignificantBits()) |
|||
.setRuleNodeIdLSB(ruleNodeId.getId().getLeastSignificantBits()) |
|||
.build(); |
|||
} |
|||
|
|||
static TbMsgProcessingStackItem fromProto(MsgProtos.TbMsgProcessingStackItemProto item){ |
|||
return new TbMsgProcessingStackItem( |
|||
new RuleChainId(new UUID(item.getRuleChainIdMSB(), item.getRuleChainIdLSB())), |
|||
new RuleNodeId(new UUID(item.getRuleNodeIdMSB(), item.getRuleNodeIdLSB())) |
|||
); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,69 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.rule.engine.flow; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.rule.engine.api.RuleNode; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.rule.engine.api.TbNode; |
|||
import org.thingsboard.rule.engine.api.TbNodeConfiguration; |
|||
import org.thingsboard.rule.engine.api.TbNodeException; |
|||
import org.thingsboard.rule.engine.api.TbRelationTypes; |
|||
import org.thingsboard.rule.engine.api.util.TbNodeUtils; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.id.EntityIdFactory; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.plugin.ComponentType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import java.util.UUID; |
|||
|
|||
@Slf4j |
|||
@RuleNode( |
|||
type = ComponentType.FLOW, |
|||
name = "rule chain", |
|||
configClazz = TbRuleChainInputNodeConfiguration.class, |
|||
nodeDescription = "transfers the message to another rule chain", |
|||
nodeDetails = "Allows to nest the rule chain similar to single rule node. " + |
|||
"The incoming message is forwarded to the input node of the specified target rule chain. " + |
|||
"The target rule chain may produce multiple labeled outputs. " + |
|||
"You may use the outputs to forward the results of processing to other rule nodes.", |
|||
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
|||
configDirective = "tbFlowNodeRuleChainInputConfig", |
|||
relationTypes = {}, |
|||
ruleChainNode = true, |
|||
customRelations = true |
|||
) |
|||
public class TbRuleChainInputNode implements TbNode { |
|||
|
|||
private TbRuleChainInputNodeConfiguration config; |
|||
private RuleChainId ruleChainId; |
|||
|
|||
@Override |
|||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|||
this.config = TbNodeUtils.convert(configuration, TbRuleChainInputNodeConfiguration.class); |
|||
this.ruleChainId = new RuleChainId(UUID.fromString(config.getRuleChainId())); |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(TbContext ctx, TbMsg msg) { |
|||
ctx.input(msg, ruleChainId); |
|||
} |
|||
|
|||
@Override |
|||
public void destroy() { |
|||
} |
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.rule.engine.flow; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.rule.engine.api.NodeConfiguration; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
|
|||
@Data |
|||
public class TbRuleChainInputNodeConfiguration implements NodeConfiguration<TbRuleChainInputNodeConfiguration> { |
|||
|
|||
private String ruleChainId; |
|||
|
|||
@Override |
|||
public TbRuleChainInputNodeConfiguration defaultConfiguration() { |
|||
return new TbRuleChainInputNodeConfiguration(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,57 @@ |
|||
/** |
|||
* Copyright © 2016-2021 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.rule.engine.flow; |
|||
|
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.rule.engine.api.EmptyNodeConfiguration; |
|||
import org.thingsboard.rule.engine.api.RuleNode; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.rule.engine.api.TbNode; |
|||
import org.thingsboard.rule.engine.api.TbNodeConfiguration; |
|||
import org.thingsboard.rule.engine.api.TbNodeException; |
|||
import org.thingsboard.rule.engine.api.TbRelationTypes; |
|||
import org.thingsboard.rule.engine.api.util.TbNodeUtils; |
|||
import org.thingsboard.server.common.data.plugin.ComponentType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
@Slf4j |
|||
@RuleNode( |
|||
type = ComponentType.FLOW, |
|||
name = "output", |
|||
configClazz = EmptyNodeConfiguration.class, |
|||
nodeDescription = "transfers the message to the caller rule chain", |
|||
nodeDetails = "Produces output of the rule chain processing. " + |
|||
"The output is forwarded to the caller rule chain, as an output of the corresponding \"input\" rule node. " + |
|||
"The output rule node name corresponds to the relation type of the output message, and it is used to forward messages to other rule nodes in the caller rule chain. ", |
|||
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
|||
configDirective = "tbFlowNodeRuleChainOutputConfig", |
|||
outEnabled = false |
|||
) |
|||
public class TbRuleChainOutputNode implements TbNode { |
|||
|
|||
@Override |
|||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(TbContext ctx, TbMsg msg) { |
|||
ctx.output(msg, ctx.getSelf().getName()); |
|||
} |
|||
|
|||
@Override |
|||
public void destroy() { |
|||
} |
|||
} |
|||
Loading…
Reference in new issue