Igor Kulikov
8 years ago
53 changed files with 599 additions and 76 deletions
@ -0,0 +1,87 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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; |
|||
|
|||
import com.fasterxml.jackson.core.JsonGenerator; |
|||
import com.fasterxml.jackson.core.JsonProcessingException; |
|||
import com.fasterxml.jackson.core.Version; |
|||
import com.fasterxml.jackson.databind.*; |
|||
import com.fasterxml.jackson.databind.module.SimpleModule; |
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
|
|||
import java.io.IOException; |
|||
|
|||
/** |
|||
* Created by ashvayka on 01.06.18. |
|||
*/ |
|||
@Data |
|||
@AllArgsConstructor |
|||
public class EntityFieldsData { |
|||
|
|||
private static final ObjectMapper mapper = new ObjectMapper(); |
|||
|
|||
static { |
|||
SimpleModule entityFieldsModule = new SimpleModule("EntityFieldsModule", new Version(1, 0, 0, null, null, null)); |
|||
entityFieldsModule.addSerializer(EntityId.class, new EntityIdFieldSerializer()); |
|||
mapper.disable(MapperFeature.USE_ANNOTATIONS); |
|||
mapper.registerModule(entityFieldsModule); |
|||
} |
|||
|
|||
private ObjectNode fieldsData; |
|||
|
|||
public EntityFieldsData(BaseData data) { |
|||
fieldsData = mapper.valueToTree(data); |
|||
} |
|||
|
|||
public String getFieldValue(String field) { |
|||
String[] fieldsTree = field.split("\\."); |
|||
JsonNode current = fieldsData; |
|||
for (String key : fieldsTree) { |
|||
if (current.has(key)) { |
|||
current = current.get(key); |
|||
} else { |
|||
current = null; |
|||
break; |
|||
} |
|||
} |
|||
if (current != null) { |
|||
if (current.isValueNode()) { |
|||
return current.asText(); |
|||
} else { |
|||
try { |
|||
return mapper.writeValueAsString(current); |
|||
} catch (JsonProcessingException e) { |
|||
return null; |
|||
} |
|||
} |
|||
} else { |
|||
return null; |
|||
} |
|||
} |
|||
|
|||
private static class EntityIdFieldSerializer extends JsonSerializer<EntityId> { |
|||
|
|||
@Override |
|||
public void serialize(EntityId value, JsonGenerator gen, SerializerProvider serializers) throws IOException { |
|||
gen.writeObject(value.getId()); |
|||
} |
|||
} |
|||
|
|||
} |
|||
|
|||
@ -0,0 +1,78 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.filter; |
|||
|
|||
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.util.TbNodeUtils; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.EntityIdFactory; |
|||
import org.thingsboard.server.common.data.plugin.ComponentType; |
|||
import org.thingsboard.server.common.data.relation.EntitySearchDirection; |
|||
import org.thingsboard.server.common.data.relation.RelationTypeGroup; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import javax.management.relation.RelationType; |
|||
|
|||
import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS; |
|||
import static org.thingsboard.rule.engine.api.util.DonAsynchron.withCallback; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.01.18. |
|||
*/ |
|||
@Slf4j |
|||
@RuleNode( |
|||
type = ComponentType.FILTER, |
|||
name = "check relation", |
|||
configClazz = TbCheckRelationNodeConfiguration.class, |
|||
relationTypes = {"True", "False"}, |
|||
nodeDescription = "Checks the relation from the selected entity to originator of the message by type and direction", |
|||
nodeDetails = "If relation exists - send Message via <b>True</b> chain, otherwise <b>False</b> chain is used.", |
|||
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
|||
configDirective = "tbFilterNodeCheckRelationConfig") |
|||
public class TbCheckRelationNode implements TbNode { |
|||
|
|||
private TbCheckRelationNodeConfiguration config; |
|||
|
|||
@Override |
|||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|||
this.config = TbNodeUtils.convert(configuration, TbCheckRelationNodeConfiguration.class); |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException { |
|||
EntityId from; |
|||
EntityId to; |
|||
if (EntitySearchDirection.FROM.name().equals(config.getDirection())) { |
|||
from = EntityIdFactory.getByTypeAndId(config.getEntityType(), config.getEntityId()); |
|||
to = msg.getOriginator(); |
|||
} else { |
|||
to = EntityIdFactory.getByTypeAndId(config.getEntityType(), config.getEntityId()); |
|||
from = msg.getOriginator(); |
|||
} |
|||
withCallback(ctx.getRelationService().checkRelation(from, to, config.getRelationType(), RelationTypeGroup.COMMON), |
|||
filterResult -> ctx.tellNext(msg, filterResult ? "True" : "False"), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); |
|||
} |
|||
|
|||
@Override |
|||
public void destroy() { |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,44 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.filter; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.rule.engine.api.NodeConfiguration; |
|||
import org.thingsboard.server.common.data.relation.EntitySearchDirection; |
|||
import org.thingsboard.server.common.msg.session.SessionMsgType; |
|||
|
|||
import java.util.Arrays; |
|||
import java.util.List; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.01.18. |
|||
*/ |
|||
@Data |
|||
public class TbCheckRelationNodeConfiguration implements NodeConfiguration<TbCheckRelationNodeConfiguration> { |
|||
|
|||
private String direction; |
|||
private String entityId; |
|||
private String entityType; |
|||
private String relationType; |
|||
|
|||
@Override |
|||
public TbCheckRelationNodeConfiguration defaultConfiguration() { |
|||
TbCheckRelationNodeConfiguration configuration = new TbCheckRelationNodeConfiguration(); |
|||
configuration.setDirection(EntitySearchDirection.FROM.name()); |
|||
configuration.setRelationType("Contains"); |
|||
return configuration; |
|||
} |
|||
} |
|||
@ -0,0 +1,38 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.metadata; |
|||
|
|||
import lombok.Data; |
|||
import org.thingsboard.rule.engine.api.NodeConfiguration; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.Map; |
|||
|
|||
@Data |
|||
public class TbGetOriginatorFieldsConfiguration implements NodeConfiguration<TbGetOriginatorFieldsConfiguration> { |
|||
|
|||
private Map<String, String> fieldsMapping; |
|||
|
|||
@Override |
|||
public TbGetOriginatorFieldsConfiguration defaultConfiguration() { |
|||
TbGetOriginatorFieldsConfiguration configuration = new TbGetOriginatorFieldsConfiguration(); |
|||
Map<String, String> fieldsMapping = new HashMap<>(); |
|||
fieldsMapping.put("name", "originatorName"); |
|||
fieldsMapping.put("type", "originatorType"); |
|||
configuration.setFieldsMapping(fieldsMapping); |
|||
return configuration; |
|||
} |
|||
} |
|||
@ -0,0 +1,83 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.metadata; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.rule.engine.api.*; |
|||
import org.thingsboard.rule.engine.api.util.TbNodeUtils; |
|||
import org.thingsboard.rule.engine.util.EntitiesFieldsAsyncLoader; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.plugin.ComponentType; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS; |
|||
import static org.thingsboard.rule.engine.api.util.DonAsynchron.withCallback; |
|||
|
|||
/** |
|||
* Created by ashvayka on 19.01.18. |
|||
*/ |
|||
@Slf4j |
|||
@RuleNode(type = ComponentType.ENRICHMENT, |
|||
name = "originator fields", |
|||
configClazz = TbGetOriginatorFieldsConfiguration.class, |
|||
nodeDescription = "Add Message Originator fields values into Message Metadata", |
|||
nodeDetails = "Will fetch fields values specified in mapping. If specified field is not part of originator fields it will be ignored.", |
|||
uiResources = {"static/rulenode/rulenode-core-config.js"}, |
|||
configDirective = "tbEnrichmentNodeOriginatorFieldsConfig") |
|||
public class TbGetOriginatorFieldsNode implements TbNode { |
|||
|
|||
private TbGetOriginatorFieldsConfiguration config; |
|||
|
|||
@Override |
|||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|||
config = TbNodeUtils.convert(configuration, TbGetOriginatorFieldsConfiguration.class); |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(TbContext ctx, TbMsg msg) throws TbNodeException { |
|||
try { |
|||
withCallback(putEntityFields(ctx, msg.getOriginator(), msg), |
|||
i -> ctx.tellNext(msg, SUCCESS), t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); |
|||
} catch (Throwable th) { |
|||
ctx.tellFailure(msg, th); |
|||
} |
|||
} |
|||
|
|||
private ListenableFuture<Void> putEntityFields(TbContext ctx, EntityId entityId, TbMsg msg) { |
|||
if (config.getFieldsMapping().isEmpty()) { |
|||
return Futures.immediateFuture(null); |
|||
} else { |
|||
return Futures.transform(EntitiesFieldsAsyncLoader.findAsync(ctx, entityId), |
|||
data -> { |
|||
config.getFieldsMapping().forEach((field, metaKey) -> { |
|||
String val = data.getFieldValue(field); |
|||
if (val != null) { |
|||
msg.getMetaData().putValue(metaKey, val); |
|||
} |
|||
}); |
|||
return null; |
|||
} |
|||
); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void destroy() { |
|||
|
|||
} |
|||
} |
|||
@ -0,0 +1,71 @@ |
|||
/** |
|||
* Copyright © 2016-2018 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.util; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.rule.engine.api.TbNodeException; |
|||
import org.thingsboard.server.common.data.BaseData; |
|||
import org.thingsboard.server.common.data.EntityFieldsData; |
|||
import org.thingsboard.server.common.data.alarm.AlarmId; |
|||
import org.thingsboard.server.common.data.id.AssetId; |
|||
import org.thingsboard.server.common.data.id.CustomerId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
|
|||
import java.util.function.Function; |
|||
|
|||
public class EntitiesFieldsAsyncLoader { |
|||
|
|||
public static ListenableFuture<EntityFieldsData> findAsync(TbContext ctx, EntityId original) { |
|||
switch (original.getEntityType()) { |
|||
case TENANT: |
|||
return getAsync(ctx.getTenantService().findTenantByIdAsync((TenantId) original), |
|||
t -> new EntityFieldsData(t)); |
|||
case CUSTOMER: |
|||
return getAsync(ctx.getCustomerService().findCustomerByIdAsync((CustomerId) original), |
|||
t -> new EntityFieldsData(t)); |
|||
case USER: |
|||
return getAsync(ctx.getUserService().findUserByIdAsync((UserId) original), |
|||
t -> new EntityFieldsData(t)); |
|||
case ASSET: |
|||
return getAsync(ctx.getAssetService().findAssetByIdAsync((AssetId) original), |
|||
t -> new EntityFieldsData(t)); |
|||
case DEVICE: |
|||
return getAsync(ctx.getDeviceService().findDeviceByIdAsync((DeviceId) original), |
|||
t -> new EntityFieldsData(t)); |
|||
case ALARM: |
|||
return getAsync(ctx.getAlarmService().findAlarmByIdAsync((AlarmId) original), |
|||
t -> new EntityFieldsData(t)); |
|||
case RULE_CHAIN: |
|||
return getAsync(ctx.getRuleChainService().findRuleChainByIdAsync((RuleChainId) original), |
|||
t -> new EntityFieldsData(t)); |
|||
default: |
|||
return Futures.immediateFailedFuture(new TbNodeException("Unexpected original EntityType " + original)); |
|||
} |
|||
} |
|||
|
|||
private static <T extends BaseData> ListenableFuture<EntityFieldsData> getAsync( |
|||
ListenableFuture<T> future, Function<T, EntityFieldsData> converter) { |
|||
return Futures.transformAsync(future, in -> in != null ? |
|||
Futures.immediateFuture(converter.apply(in)) |
|||
: Futures.immediateFailedFuture(new RuntimeException("Entity not found!"))); |
|||
} |
|||
} |
|||
File diff suppressed because one or more lines are too long
Loading…
Reference in new issue