Dmytro Skarzhynets
4 years ago
37 changed files with 1277 additions and 369 deletions
@ -0,0 +1,21 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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; |
|||
|
|||
public enum FetchTo { |
|||
DATA, |
|||
METADATA |
|||
} |
|||
@ -0,0 +1,23 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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; |
|||
|
|||
@Data |
|||
public abstract class TbAbstractFetchToNodeConfiguration { |
|||
private FetchTo fetchTo; |
|||
} |
|||
@ -0,0 +1,110 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.fasterxml.jackson.databind.node.ObjectNode; |
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import com.google.common.util.concurrent.MoreExecutors; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.rule.engine.api.util.TbNodeUtils; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.data.kv.KvEntry; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.common.util.DonAsynchron.withCallback; |
|||
import static org.thingsboard.rule.engine.api.TbRelationTypes.FAILURE; |
|||
import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE; |
|||
|
|||
@Slf4j |
|||
public abstract class TbAbstractGetEntityAttrNode<T extends EntityId> extends TbAbstractNodeWithFetchTo<TbGetEntityAttrNodeConfiguration> { |
|||
@Override |
|||
public void onMsg(TbContext ctx, TbMsg msg) { |
|||
ObjectNode msgDataAsJsonNode; |
|||
if (FetchTo.DATA.equals(fetchTo)) { |
|||
msgDataAsJsonNode = getMsgDataAsObjectNode(msg); |
|||
} else { |
|||
msgDataAsJsonNode = null; |
|||
} |
|||
ctx.checkTenantEntity(msg.getOriginator()); |
|||
withCallback(findEntityAsync(ctx, msg.getOriginator()), |
|||
entityId -> safeGetAttributes(ctx, msg, entityId, msgDataAsJsonNode), |
|||
t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); |
|||
} |
|||
|
|||
protected abstract ListenableFuture<T> findEntityAsync(TbContext ctx, EntityId originator); |
|||
|
|||
private void safeGetAttributes(TbContext ctx, TbMsg msg, T entityId, ObjectNode msgDataAsJsonNode) { |
|||
if (entityId == null || entityId.isNullUid()) { |
|||
ctx.tellNext(msg, FAILURE); |
|||
return; |
|||
} |
|||
|
|||
Map<String, String> mappingsMap = new HashMap<>(); |
|||
config.getAttrMapping().forEach((key, value) -> { |
|||
String patternProcessedSourceKey = TbNodeUtils.processPattern(key, msg); |
|||
String patternProcessedTargetKey = TbNodeUtils.processPattern(value, msg); |
|||
mappingsMap.put(patternProcessedSourceKey, patternProcessedTargetKey); |
|||
}); |
|||
|
|||
var sourceKeys = List.copyOf(mappingsMap.keySet()); |
|||
withCallback(config.isTelemetry() ? getLatestTelemetryAsync(ctx, entityId, sourceKeys) : getAttributesAsync(ctx, entityId, sourceKeys), |
|||
data -> putDataAndTell(ctx, msg, data, mappingsMap, msgDataAsJsonNode), |
|||
t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); |
|||
} |
|||
|
|||
private ListenableFuture<List<KvEntry>> getAttributesAsync(TbContext ctx, EntityId entityId, List<String> attrKeys) { |
|||
var latest = ctx.getAttributesService().find(ctx.getTenantId(), entityId, SERVER_SCOPE, attrKeys); |
|||
return Futures.transform(latest, l -> |
|||
l.stream() |
|||
.map(i -> (KvEntry) i) |
|||
.collect(Collectors.toList()), |
|||
MoreExecutors.directExecutor()); |
|||
} |
|||
|
|||
private ListenableFuture<List<KvEntry>> getLatestTelemetryAsync(TbContext ctx, EntityId entityId, List<String> timeseriesKeys) { |
|||
var latest = ctx.getTimeseriesService().findLatest(ctx.getTenantId(), entityId, timeseriesKeys); |
|||
return Futures.transform(latest, l -> |
|||
l.stream() |
|||
.map(i -> (KvEntry) i) |
|||
.collect(Collectors.toList()), |
|||
MoreExecutors.directExecutor()); |
|||
} |
|||
|
|||
private void putDataAndTell(TbContext ctx, TbMsg msg, List<? extends KvEntry> data, Map<String, String> map, ObjectNode msgDataAsJsonNode) { |
|||
for (KvEntry entry : data) { |
|||
String targetKey = map.get(entry.getKey()); |
|||
String value = entry.getValueAsString(); |
|||
if (FetchTo.DATA.equals(fetchTo)) { |
|||
msgDataAsJsonNode.put(targetKey, value); |
|||
} else if (FetchTo.METADATA.equals(fetchTo)) { |
|||
msg.getMetaData().putValue(targetKey, value); |
|||
} |
|||
} |
|||
if (FetchTo.DATA.equals(fetchTo)) { |
|||
ctx.tellSuccess(TbMsg.transformMsgData(msg, JacksonUtil.toString(msgDataAsJsonNode))); |
|||
} else if (FetchTo.METADATA.equals(fetchTo)) { |
|||
ctx.tellSuccess(msg); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,50 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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.fasterxml.jackson.databind.JsonNode; |
|||
import com.fasterxml.jackson.databind.node.ObjectNode; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
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.server.common.msg.TbMsg; |
|||
|
|||
public abstract class TbAbstractNodeWithFetchTo<C extends TbAbstractFetchToNodeConfiguration> implements TbNode { |
|||
protected C config; |
|||
protected FetchTo fetchTo; |
|||
|
|||
@Override |
|||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|||
config = loadNodeConfiguration(configuration); |
|||
if (config.getFetchTo() == null) { |
|||
throw new TbNodeException("FetchTo cannot be NULL!"); |
|||
} else { |
|||
fetchTo = config.getFetchTo(); |
|||
} |
|||
} |
|||
|
|||
protected abstract C loadNodeConfiguration(TbNodeConfiguration configuration) throws TbNodeException; |
|||
|
|||
protected ObjectNode getMsgDataAsObjectNode(TbMsg msg) { |
|||
JsonNode msgDataNode = JacksonUtil.toJsonNode(msg.getData()); |
|||
if (!msgDataNode.isObject()) { |
|||
throw new IllegalArgumentException("Message body is not an object!"); |
|||
} |
|||
return (ObjectNode) msgDataNode; |
|||
} |
|||
} |
|||
@ -1,109 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2023 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 com.google.common.util.concurrent.MoreExecutors; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
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.kv.AttributeKvEntry; |
|||
import org.thingsboard.server.common.data.kv.KvEntry; |
|||
import org.thingsboard.server.common.data.kv.TsKvEntry; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
|
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.common.util.DonAsynchron.withCallback; |
|||
import static org.thingsboard.rule.engine.api.TbRelationTypes.FAILURE; |
|||
import static org.thingsboard.server.common.data.DataConstants.SERVER_SCOPE; |
|||
|
|||
@Slf4j |
|||
public abstract class TbEntityGetAttrNode<T extends EntityId> implements TbNode { |
|||
|
|||
private TbGetEntityAttrNodeConfiguration config; |
|||
|
|||
@Override |
|||
public void init(TbContext ctx, TbNodeConfiguration configuration) throws TbNodeException { |
|||
this.config = TbNodeUtils.convert(configuration, TbGetEntityAttrNodeConfiguration.class); |
|||
} |
|||
|
|||
@Override |
|||
public void onMsg(TbContext ctx, TbMsg msg) { |
|||
try { |
|||
withCallback(findEntityAsync(ctx, msg.getOriginator()), |
|||
entityId -> safeGetAttributes(ctx, msg, entityId), |
|||
t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); |
|||
} catch (Throwable th) { |
|||
ctx.tellFailure(msg, th); |
|||
} |
|||
} |
|||
|
|||
private void safeGetAttributes(TbContext ctx, TbMsg msg, T entityId) { |
|||
if (entityId == null || entityId.isNullUid()) { |
|||
ctx.tellNext(msg, FAILURE); |
|||
return; |
|||
} |
|||
|
|||
Map<String, String> mappingsMap = new HashMap<>(); |
|||
config.getAttrMapping().forEach((key, value) -> { |
|||
String processPatternKey = TbNodeUtils.processPattern(key, msg); |
|||
String processPatternValue = TbNodeUtils.processPattern(value, msg); |
|||
mappingsMap.put(processPatternKey, processPatternValue); |
|||
}); |
|||
|
|||
List<String> keys = List.copyOf(mappingsMap.keySet()); |
|||
withCallback(config.isTelemetry() ? getLatestTelemetry(ctx, entityId, keys) : getAttributesAsync(ctx, entityId, keys), |
|||
attributes -> putAttributesAndTell(ctx, msg, attributes, mappingsMap), |
|||
t -> ctx.tellFailure(msg, t), ctx.getDbCallbackExecutor()); |
|||
} |
|||
|
|||
private ListenableFuture<List<KvEntry>> getAttributesAsync(TbContext ctx, EntityId entityId, List<String> attrKeys) { |
|||
ListenableFuture<List<AttributeKvEntry>> latest = ctx.getAttributesService().find(ctx.getTenantId(), entityId, SERVER_SCOPE, attrKeys); |
|||
return Futures.transform(latest, l -> |
|||
l.stream().map(i -> (KvEntry) i).collect(Collectors.toList()), MoreExecutors.directExecutor()); |
|||
} |
|||
|
|||
private ListenableFuture<List<KvEntry>> getLatestTelemetry(TbContext ctx, EntityId entityId, List<String> timeseriesKeys) { |
|||
ListenableFuture<List<TsKvEntry>> latest = ctx.getTimeseriesService().findLatest(ctx.getTenantId(), entityId, timeseriesKeys); |
|||
return Futures.transform(latest, l -> |
|||
l.stream().map(i -> (KvEntry) i).collect(Collectors.toList()), MoreExecutors.directExecutor()); |
|||
} |
|||
|
|||
|
|||
private void putAttributesAndTell(TbContext ctx, TbMsg msg, List<? extends KvEntry> attributes, Map<String, String> map) { |
|||
attributes.forEach(r -> { |
|||
String attrName = map.get(r.getKey()); |
|||
msg.getMetaData().putValue(attrName, r.getValueAsString()); |
|||
}); |
|||
ctx.tellSuccess(msg); |
|||
} |
|||
|
|||
protected abstract ListenableFuture<T> findEntityAsync(TbContext ctx, EntityId originator); |
|||
|
|||
public void setConfig(TbGetEntityAttrNodeConfiguration config) { |
|||
this.config = config; |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,376 @@ |
|||
/** |
|||
* Copyright © 2016-2023 The Thingsboard Authors |
|||
* <p> |
|||
* 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 |
|||
* <p> |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* <p> |
|||
* 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 org.jetbrains.annotations.NotNull; |
|||
import org.junit.jupiter.api.BeforeEach; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.ArgumentCaptor; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.thingsboard.common.util.JacksonUtil; |
|||
import org.thingsboard.common.util.ListeningExecutor; |
|||
import org.thingsboard.rule.engine.api.TbContext; |
|||
import org.thingsboard.rule.engine.api.TbNodeConfiguration; |
|||
import org.thingsboard.rule.engine.api.TbNodeException; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.id.DashboardId; |
|||
import org.thingsboard.server.common.data.id.DeviceId; |
|||
import org.thingsboard.server.common.data.id.EntityId; |
|||
import org.thingsboard.server.common.msg.TbMsg; |
|||
import org.thingsboard.server.common.msg.TbMsgMetaData; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
|
|||
import java.util.Collections; |
|||
import java.util.Map; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.Callable; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.junit.jupiter.api.Assertions.assertThrows; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.Mockito.doThrow; |
|||
import static org.mockito.Mockito.never; |
|||
import static org.mockito.Mockito.times; |
|||
import static org.mockito.Mockito.verify; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class TbGetOriginatorFieldsNodeTest { |
|||
private static final EntityId DUMMY_ENTITY_ID = new DeviceId(UUID.randomUUID()); |
|||
public static final ListeningExecutor DB_EXECUTOR = new ListeningExecutor() { |
|||
@Override |
|||
public <T> ListenableFuture<T> executeAsync(Callable<T> task) { |
|||
try { |
|||
return Futures.immediateFuture(task.call()); |
|||
} catch (Exception e) { |
|||
throw new RuntimeException(e); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void execute(@NotNull Runnable command) { |
|||
command.run(); |
|||
} |
|||
}; |
|||
@Mock |
|||
private TbContext ctxMock; |
|||
@Mock |
|||
private DeviceService deviceService; |
|||
private TbGetOriginatorFieldsNode node; |
|||
private TbGetOriginatorFieldsConfiguration config; |
|||
private TbNodeConfiguration nodeConfiguration; |
|||
private TbMsg msg; |
|||
|
|||
@BeforeEach |
|||
public void setUp() { |
|||
config = new TbGetOriginatorFieldsConfiguration(); |
|||
node = new TbGetOriginatorFieldsNode(); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
} |
|||
|
|||
@Test |
|||
public void givenConfigWithNullFetchTo_whenOnInit_thenException() { |
|||
// GIVEN
|
|||
config = config.defaultConfiguration(); |
|||
config.setFetchTo(null); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
|
|||
// WHEN
|
|||
var exception = assertThrows(TbNodeException.class, () -> node.init(ctxMock, nodeConfiguration)); |
|||
|
|||
// THEN
|
|||
assertThat(exception.getMessage()).isEqualTo("FetchTo cannot be NULL!"); |
|||
verify(ctxMock, never()).tellSuccess(any()); |
|||
} |
|||
|
|||
@Test |
|||
public void givenDefaultConfig_whenInit_thenOK() throws TbNodeException { |
|||
// GIVEN
|
|||
config = config.defaultConfiguration(); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
|
|||
// WHEN
|
|||
node.init(ctxMock, nodeConfiguration); |
|||
|
|||
// THEN
|
|||
assertThat(node.config).isEqualTo(config); |
|||
assertThat(config.getFieldsMapping()).isEqualTo(Map.of( |
|||
"name", "originatorName", |
|||
"type", "originatorType")); |
|||
assertThat(config.isIgnoreNullStrings()).isEqualTo(false); |
|||
assertThat(node.fetchTo).isEqualTo(FetchTo.METADATA); |
|||
} |
|||
|
|||
@Test |
|||
public void givenCustomConfig_whenInit_thenOK() throws TbNodeException { |
|||
// GIVEN
|
|||
config.setFieldsMapping(Map.of( |
|||
"sourceField1", "targetKey1", |
|||
"sourceField2", "targetKey2", |
|||
"sourceField3", "targetKey3")); |
|||
config.setIgnoreNullStrings(true); |
|||
config.setFetchTo(FetchTo.DATA); |
|||
nodeConfiguration = new TbNodeConfiguration(JacksonUtil.valueToTree(config)); |
|||
|
|||
// WHEN
|
|||
node.init(ctxMock, nodeConfiguration); |
|||
|
|||
// THEN
|
|||
assertThat(node.config).isEqualTo(config); |
|||
assertThat(config.getFieldsMapping()).isEqualTo(Map.of( |
|||
"sourceField1", "targetKey1", |
|||
"sourceField2", "targetKey2", |
|||
"sourceField3", "targetKey3")); |
|||
assertThat(config.isIgnoreNullStrings()).isEqualTo(true); |
|||
assertThat(node.fetchTo).isEqualTo(FetchTo.DATA); |
|||
} |
|||
|
|||
@Test |
|||
public void givenMsgDataIsNotAnJsonObjectAndFetchToData_whenOnMsg_thenException() { |
|||
// GIVEN
|
|||
node.fetchTo = FetchTo.DATA; |
|||
msg = TbMsg.newMsg("SOME_MESSAGE_TYPE", DUMMY_ENTITY_ID, new TbMsgMetaData(), "[]"); |
|||
|
|||
// WHEN
|
|||
var exception = assertThrows(IllegalArgumentException.class, () -> node.onMsg(ctxMock, msg)); |
|||
|
|||
// THEN
|
|||
assertThat(exception.getMessage()).isEqualTo("Message body is not an object!"); |
|||
verify(ctxMock, never()).tellSuccess(any()); |
|||
} |
|||
|
|||
@Test |
|||
public void givenEntityThatDoesNotBelongToTheCurrentTenant_whenOnMsg_thenException() { |
|||
// SETUP
|
|||
var expectedExceptionMessage = "Entity with id: '" + DUMMY_ENTITY_ID + |
|||
"' specified in the configuration doesn't belong to the current tenant."; |
|||
|
|||
// GIVEN
|
|||
doThrow(new RuntimeException(expectedExceptionMessage)).when(ctxMock).checkTenantEntity(DUMMY_ENTITY_ID); |
|||
msg = TbMsg.newMsg("SOME_MESSAGE_TYPE", DUMMY_ENTITY_ID, new TbMsgMetaData(), "{}"); |
|||
|
|||
// WHEN
|
|||
var exception = assertThrows(RuntimeException.class, () -> node.onMsg(ctxMock, msg)); |
|||
|
|||
// THEN
|
|||
assertThat(exception.getMessage()).isEqualTo(expectedExceptionMessage); |
|||
verify(ctxMock, never()).tellSuccess(any()); |
|||
} |
|||
|
|||
@Test |
|||
public void givenValidMsgAndFetchToData_whenOnMsg_thenShouldTellSuccessAndFetchToData() { |
|||
// GIVEN
|
|||
var device = new Device(); |
|||
device.setId((DeviceId) DUMMY_ENTITY_ID); |
|||
device.setName("Test device"); |
|||
device.setType("Test device type"); |
|||
|
|||
config.setFieldsMapping(Map.of( |
|||
"name", "originatorName", |
|||
"type", "originatorType", |
|||
"label", "originatorLabel")); |
|||
config.setIgnoreNullStrings(true); |
|||
config.setFetchTo(FetchTo.DATA); |
|||
|
|||
node.config = config; |
|||
node.fetchTo = FetchTo.DATA; |
|||
var msgMetaData = new TbMsgMetaData(); |
|||
var msgData = "{\"temp\":42,\"humidity\":77}"; |
|||
msg = TbMsg.newMsg("POST_TELEMETRY_REQUEST", DUMMY_ENTITY_ID, msgMetaData, msgData); |
|||
|
|||
when(ctxMock.getDeviceService()).thenReturn(deviceService); |
|||
when(deviceService.findDeviceByIdAsync(any(), eq(device.getId()))).thenReturn(Futures.immediateFuture(device)); |
|||
|
|||
when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); |
|||
|
|||
// WHEN
|
|||
node.onMsg(ctxMock, msg); |
|||
|
|||
// THEN
|
|||
var actualMessageCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
verify(ctxMock, times(1)).tellSuccess(actualMessageCaptor.capture()); |
|||
verify(ctxMock, never()).tellFailure(any(), any()); |
|||
|
|||
var expectedMsgData = "{\"temp\":42,\"humidity\":77,\"originatorName\":\"Test device\",\"originatorType\":\"Test device type\"}"; |
|||
|
|||
assertThat(actualMessageCaptor.getValue().getData()).isEqualTo(expectedMsgData); |
|||
assertThat(actualMessageCaptor.getValue().getMetaData()).isEqualTo(msgMetaData); |
|||
} |
|||
|
|||
@Test |
|||
public void givenValidMsgAndFetchToMetaData_whenOnMsg_thenShouldTellSuccessAndFetchToMetaData() { |
|||
// GIVEN
|
|||
var device = new Device(); |
|||
device.setId((DeviceId) DUMMY_ENTITY_ID); |
|||
device.setName("Test device"); |
|||
device.setType("Test device type"); |
|||
|
|||
config.setFieldsMapping(Map.of( |
|||
"name", "originatorName", |
|||
"type", "originatorType", |
|||
"label", "originatorLabel")); |
|||
config.setIgnoreNullStrings(true); |
|||
config.setFetchTo(FetchTo.METADATA); |
|||
|
|||
node.config = config; |
|||
node.fetchTo = FetchTo.METADATA; |
|||
var msgMetaData = new TbMsgMetaData(Map.of( |
|||
"testKey1", "testValue1", |
|||
"testKey2", "123")); |
|||
var msgData = "[\"value1\",\"value2\"]"; |
|||
msg = TbMsg.newMsg("POST_TELEMETRY_REQUEST", DUMMY_ENTITY_ID, msgMetaData, msgData); |
|||
|
|||
when(ctxMock.getDeviceService()).thenReturn(deviceService); |
|||
when(deviceService.findDeviceByIdAsync(any(), eq(device.getId()))).thenReturn(Futures.immediateFuture(device)); |
|||
|
|||
when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); |
|||
|
|||
// WHEN
|
|||
node.onMsg(ctxMock, msg); |
|||
|
|||
// THEN
|
|||
var actualMessageCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
verify(ctxMock, times(1)).tellSuccess(actualMessageCaptor.capture()); |
|||
verify(ctxMock, never()).tellFailure(any(), any()); |
|||
|
|||
var expectedMsgMetaData = new TbMsgMetaData(Map.of( |
|||
"testKey1", "testValue1", |
|||
"testKey2", "123", |
|||
"originatorName", "Test device", |
|||
"originatorType", "Test device type" |
|||
)); |
|||
|
|||
assertThat(actualMessageCaptor.getValue().getData()).isEqualTo(msgData); |
|||
assertThat(actualMessageCaptor.getValue().getMetaData()).isEqualTo(expectedMsgMetaData); |
|||
} |
|||
|
|||
@Test |
|||
public void givenNullEntityFieldsAndIgnoreNullStringsFalse_whenOnMsg_thenShouldTellSuccessAndFetchNullField() { |
|||
// GIVEN
|
|||
var device = new Device(); |
|||
device.setId((DeviceId) DUMMY_ENTITY_ID); |
|||
device.setName("Test device"); |
|||
device.setType("Test device type"); |
|||
|
|||
config.setFieldsMapping(Map.of( |
|||
"name", "originatorName", |
|||
"type", "originatorType", |
|||
"label", "originatorLabel")); |
|||
config.setIgnoreNullStrings(false); |
|||
config.setFetchTo(FetchTo.METADATA); |
|||
|
|||
node.config = config; |
|||
node.fetchTo = FetchTo.METADATA; |
|||
var msgMetaData = new TbMsgMetaData(Map.of( |
|||
"testKey1", "testValue1", |
|||
"testKey2", "123")); |
|||
var msgData = "[\"value1\",\"value2\"]"; |
|||
msg = TbMsg.newMsg("POST_TELEMETRY_REQUEST", DUMMY_ENTITY_ID, msgMetaData, msgData); |
|||
|
|||
when(ctxMock.getDeviceService()).thenReturn(deviceService); |
|||
when(deviceService.findDeviceByIdAsync(any(), eq(device.getId()))).thenReturn(Futures.immediateFuture(device)); |
|||
|
|||
when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); |
|||
|
|||
// WHEN
|
|||
node.onMsg(ctxMock, msg); |
|||
|
|||
// THEN
|
|||
var actualMessageCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
verify(ctxMock, times(1)).tellSuccess(actualMessageCaptor.capture()); |
|||
verify(ctxMock, never()).tellFailure(any(), any()); |
|||
|
|||
var expectedMsgMetaData = new TbMsgMetaData(Map.of( |
|||
"testKey1", "testValue1", |
|||
"testKey2", "123", |
|||
"originatorName", "Test device", |
|||
"originatorType", "Test device type", |
|||
"originatorLabel", "null" |
|||
)); |
|||
|
|||
assertThat(actualMessageCaptor.getValue().getData()).isEqualTo(msgData); |
|||
assertThat(actualMessageCaptor.getValue().getMetaData()).isEqualTo(expectedMsgMetaData); |
|||
} |
|||
|
|||
@Test |
|||
public void givenEmptyFieldsMapping_whenOnMsg_thenShouldTellSuccessWithSameMsg() { |
|||
// GIVEN
|
|||
config.setFieldsMapping(Collections.emptyMap()); |
|||
config.setIgnoreNullStrings(false); |
|||
config.setFetchTo(FetchTo.METADATA); |
|||
|
|||
node.config = config; |
|||
node.fetchTo = FetchTo.METADATA; |
|||
var msgMetaData = new TbMsgMetaData(Map.of( |
|||
"testKey1", "testValue1", |
|||
"testKey2", "123")); |
|||
var msgData = "[\"value1\",\"value2\"]"; |
|||
msg = TbMsg.newMsg("POST_TELEMETRY_REQUEST", DUMMY_ENTITY_ID, msgMetaData, msgData); |
|||
|
|||
// WHEN
|
|||
node.onMsg(ctxMock, msg); |
|||
|
|||
// THEN
|
|||
var actualMessageCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
verify(ctxMock, times(1)).tellSuccess(actualMessageCaptor.capture()); |
|||
verify(ctxMock, never()).tellFailure(any(), any()); |
|||
|
|||
var expectedMsgMetaData = new TbMsgMetaData(Map.of( |
|||
"testKey1", "testValue1", |
|||
"testKey2", "123" |
|||
)); |
|||
|
|||
assertThat(actualMessageCaptor.getValue().getData()).isEqualTo(msgData); |
|||
assertThat(actualMessageCaptor.getValue().getMetaData()).isEqualTo(expectedMsgMetaData); |
|||
} |
|||
|
|||
@Test |
|||
public void givenUnsupportedEntityType_whenOnMsg_thenShouldTellFailureWithSameMsg() { |
|||
// GIVEN
|
|||
config.setFieldsMapping(Map.of( |
|||
"name", "originatorName", |
|||
"type", "originatorType", |
|||
"label", "originatorLabel")); |
|||
config.setIgnoreNullStrings(false); |
|||
config.setFetchTo(FetchTo.METADATA); |
|||
|
|||
node.config = config; |
|||
node.fetchTo = FetchTo.METADATA; |
|||
var msgMetaData = new TbMsgMetaData(Map.of( |
|||
"testKey1", "testValue1", |
|||
"testKey2", "123")); |
|||
var msgData = "[\"value1\",\"value2\"]"; |
|||
msg = TbMsg.newMsg("POST_TELEMETRY_REQUEST", new DashboardId(UUID.randomUUID()), msgMetaData, msgData); |
|||
|
|||
when(ctxMock.getDbCallbackExecutor()).thenReturn(DB_EXECUTOR); |
|||
|
|||
// WHEN
|
|||
node.onMsg(ctxMock, msg); |
|||
|
|||
// THEN
|
|||
var actualMessageCaptor = ArgumentCaptor.forClass(TbMsg.class); |
|||
verify(ctxMock, times(1)).tellFailure(actualMessageCaptor.capture(), any()); |
|||
verify(ctxMock, never()).tellSuccess(any()); |
|||
|
|||
assertThat(actualMessageCaptor.getValue().getData()).isEqualTo(msgData); |
|||
assertThat(actualMessageCaptor.getValue().getMetaData()).isEqualTo(msgMetaData); |
|||
} |
|||
} |
|||
@ -0,0 +1,231 @@ |
|||
package org.thingsboard.rule.engine.util; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import org.junit.jupiter.api.Assertions; |
|||
import org.junit.jupiter.api.Test; |
|||
import org.junit.jupiter.api.BeforeAll; |
|||
import org.junit.jupiter.api.extension.ExtendWith; |
|||
import org.mockito.Mock; |
|||
import org.mockito.junit.jupiter.MockitoExtension; |
|||
import org.thingsboard.rule.engine.api.RuleEngineAlarmService; |
|||
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.Customer; |
|||
import org.thingsboard.server.common.data.Device; |
|||
import org.thingsboard.server.common.data.EntityFieldsData; |
|||
import org.thingsboard.server.common.data.EntityType; |
|||
import org.thingsboard.server.common.data.EntityView; |
|||
import org.thingsboard.server.common.data.Tenant; |
|||
import org.thingsboard.server.common.data.User; |
|||
import org.thingsboard.server.common.data.alarm.Alarm; |
|||
import org.thingsboard.server.common.data.asset.Asset; |
|||
import org.thingsboard.server.common.data.id.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.EntityIdFactory; |
|||
import org.thingsboard.server.common.data.id.EntityViewId; |
|||
import org.thingsboard.server.common.data.id.RuleChainId; |
|||
import org.thingsboard.server.common.data.id.TenantId; |
|||
import org.thingsboard.server.common.data.id.UUIDBased; |
|||
import org.thingsboard.server.common.data.id.UserId; |
|||
import org.thingsboard.server.common.data.rule.RuleChain; |
|||
import org.thingsboard.server.dao.asset.AssetService; |
|||
import org.thingsboard.server.dao.customer.CustomerService; |
|||
import org.thingsboard.server.dao.device.DeviceService; |
|||
import org.thingsboard.server.dao.entityview.EntityViewService; |
|||
import org.thingsboard.server.dao.rule.RuleChainService; |
|||
import org.thingsboard.server.dao.tenant.TenantService; |
|||
import org.thingsboard.server.dao.user.UserService; |
|||
|
|||
import java.util.EnumSet; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.ExecutionException; |
|||
|
|||
import static org.assertj.core.api.Assertions.assertThat; |
|||
import static org.junit.jupiter.api.Assertions.assertInstanceOf; |
|||
import static org.junit.jupiter.api.Assertions.assertThrows; |
|||
import static org.mockito.ArgumentMatchers.any; |
|||
import static org.mockito.ArgumentMatchers.eq; |
|||
import static org.mockito.Mockito.doReturn; |
|||
import static org.mockito.Mockito.when; |
|||
|
|||
@ExtendWith(MockitoExtension.class) |
|||
public class EntitiesFieldsAsyncLoaderTest { |
|||
private static EnumSet<EntityType> SUPPORTED_ENTITY_TYPES; |
|||
private static UUID RANDOM_UUID; |
|||
private static TenantId TENANT_ID; |
|||
@Mock |
|||
private TbContext ctxMock; |
|||
@Mock |
|||
private TenantService tenantServiceMock; |
|||
@Mock |
|||
private CustomerService customerServiceMock; |
|||
@Mock |
|||
private UserService userServiceMock; |
|||
@Mock |
|||
private AssetService assetServiceMock; |
|||
@Mock |
|||
private DeviceService deviceServiceMock; |
|||
@Mock |
|||
private RuleEngineAlarmService ruleEngineAlarmServiceMock; |
|||
@Mock |
|||
private RuleChainService ruleChainServiceMock; |
|||
@Mock |
|||
private EntityViewService entityViewServiceMock; |
|||
|
|||
@BeforeAll |
|||
public static void setup() { |
|||
RANDOM_UUID = UUID.randomUUID(); |
|||
TENANT_ID = new TenantId(UUID.randomUUID()); |
|||
SUPPORTED_ENTITY_TYPES = EnumSet.of( |
|||
EntityType.TENANT, |
|||
EntityType.CUSTOMER, |
|||
EntityType.USER, |
|||
EntityType.ASSET, |
|||
EntityType.DEVICE, |
|||
EntityType.ALARM, |
|||
EntityType.RULE_CHAIN, |
|||
EntityType.ENTITY_VIEW |
|||
); |
|||
} |
|||
|
|||
@Test |
|||
public void givenSupportedEntityTypes_whenFindAsync_thenOK() throws ExecutionException, InterruptedException { |
|||
for (var entityType : SUPPORTED_ENTITY_TYPES) { |
|||
var entityId = EntityIdFactory.getByTypeAndUuid(entityType, RANDOM_UUID); |
|||
|
|||
initMocks(entityType, false); |
|||
|
|||
when(ctxMock.getTenantId()).thenReturn(TENANT_ID); |
|||
|
|||
var actualEntityFieldsData = EntitiesFieldsAsyncLoader.findAsync(ctxMock, entityId).get(); |
|||
var expectedEntityFieldsData = new EntityFieldsData(getEntityFromEntityId(entityId)); |
|||
|
|||
Assertions.assertEquals(expectedEntityFieldsData, actualEntityFieldsData); |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
public void givenUnsupportedEntityTypes_whenFindAsync_thenException() { |
|||
for (var entityType : EntityType.values()) { |
|||
if (!SUPPORTED_ENTITY_TYPES.contains(entityType)) { |
|||
var entityId = EntityIdFactory.getByTypeAndUuid(entityType, RANDOM_UUID); |
|||
|
|||
var expectedExceptionMsg = "org.thingsboard.rule.engine.api.TbNodeException: Unexpected originator EntityType: " + entityType; |
|||
|
|||
var exception = assertThrows(ExecutionException.class, |
|||
() -> EntitiesFieldsAsyncLoader.findAsync(ctxMock, entityId).get()); |
|||
|
|||
assertInstanceOf(TbNodeException.class, exception.getCause()); |
|||
assertThat(exception.getMessage()).isEqualTo(expectedExceptionMsg); |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Test |
|||
public void givenSupportedTypeButEntityDoesNotExist_whenFindAsync_thenException() { |
|||
for (var entityType : SUPPORTED_ENTITY_TYPES) { |
|||
var entityId = EntityIdFactory.getByTypeAndUuid(entityType, RANDOM_UUID); |
|||
|
|||
initMocks(entityType, true); |
|||
when(ctxMock.getTenantId()).thenReturn(TENANT_ID); |
|||
|
|||
var expectedExceptionMsg = "org.thingsboard.rule.engine.api.TbNodeException: Entity not found!"; |
|||
|
|||
var exception = assertThrows(ExecutionException.class, |
|||
() -> EntitiesFieldsAsyncLoader.findAsync(ctxMock, entityId).get()); |
|||
|
|||
assertInstanceOf(TbNodeException.class, exception.getCause()); |
|||
assertThat(exception.getMessage()).isEqualTo(expectedExceptionMsg); |
|||
} |
|||
} |
|||
|
|||
private void initMocks(EntityType entityType, boolean entityDoesNotExist) { |
|||
switch (entityType) { |
|||
case TENANT: |
|||
var tenant = Futures.immediateFuture(entityDoesNotExist ? null : new Tenant(new TenantId(RANDOM_UUID))); |
|||
|
|||
when(ctxMock.getTenantService()).thenReturn(tenantServiceMock); |
|||
doReturn(tenant).when(tenantServiceMock).findTenantByIdAsync(eq(TENANT_ID), any()); |
|||
|
|||
break; |
|||
case CUSTOMER: |
|||
var customer = Futures.immediateFuture(entityDoesNotExist ? null : new Customer(new CustomerId(RANDOM_UUID))); |
|||
|
|||
when(ctxMock.getCustomerService()).thenReturn(customerServiceMock); |
|||
doReturn(customer).when(customerServiceMock).findCustomerByIdAsync(eq(TENANT_ID), any()); |
|||
|
|||
break; |
|||
case USER: |
|||
var user = Futures.immediateFuture(entityDoesNotExist ? null : new User(new UserId(RANDOM_UUID))); |
|||
|
|||
when(ctxMock.getUserService()).thenReturn(userServiceMock); |
|||
doReturn(user).when(userServiceMock).findUserByIdAsync(eq(TENANT_ID), any()); |
|||
|
|||
break; |
|||
case ASSET: |
|||
var asset = Futures.immediateFuture(entityDoesNotExist ? null : new Asset(new AssetId(RANDOM_UUID))); |
|||
|
|||
when(ctxMock.getAssetService()).thenReturn(assetServiceMock); |
|||
doReturn(asset).when(assetServiceMock).findAssetByIdAsync(eq(TENANT_ID), any()); |
|||
|
|||
break; |
|||
case DEVICE: |
|||
var device = Futures.immediateFuture(entityDoesNotExist ? null : new Device(new DeviceId(RANDOM_UUID))); |
|||
|
|||
when(ctxMock.getDeviceService()).thenReturn(deviceServiceMock); |
|||
doReturn(device).when(deviceServiceMock).findDeviceByIdAsync(eq(TENANT_ID), any()); |
|||
|
|||
break; |
|||
case ALARM: |
|||
var alarm = Futures.immediateFuture(entityDoesNotExist ? null : new Alarm(new AlarmId(RANDOM_UUID))); |
|||
|
|||
when(ctxMock.getAlarmService()).thenReturn(ruleEngineAlarmServiceMock); |
|||
doReturn(alarm).when(ruleEngineAlarmServiceMock).findAlarmByIdAsync(eq(TENANT_ID), any()); |
|||
|
|||
break; |
|||
case RULE_CHAIN: |
|||
var ruleChain = Futures.immediateFuture(entityDoesNotExist ? null : new RuleChain(new RuleChainId(RANDOM_UUID))); |
|||
|
|||
when(ctxMock.getRuleChainService()).thenReturn(ruleChainServiceMock); |
|||
doReturn(ruleChain).when(ruleChainServiceMock).findRuleChainByIdAsync(eq(TENANT_ID), any()); |
|||
|
|||
break; |
|||
case ENTITY_VIEW: |
|||
var entityView = Futures.immediateFuture(entityDoesNotExist ? null : new EntityView(new EntityViewId(RANDOM_UUID))); |
|||
|
|||
when(ctxMock.getEntityViewService()).thenReturn(entityViewServiceMock); |
|||
doReturn(entityView).when(entityViewServiceMock).findEntityViewByIdAsync(eq(TENANT_ID), any()); |
|||
|
|||
break; |
|||
default: |
|||
throw new RuntimeException("Unexpected EntityType: " + entityType); |
|||
} |
|||
} |
|||
|
|||
private BaseData<? extends UUIDBased> getEntityFromEntityId(EntityId entityId) { |
|||
switch (entityId.getEntityType()) { |
|||
case TENANT: |
|||
return new Tenant((TenantId) entityId); |
|||
case CUSTOMER: |
|||
return new Customer((CustomerId) entityId); |
|||
case USER: |
|||
return new User((UserId) entityId); |
|||
case ASSET: |
|||
return new Asset((AssetId) entityId); |
|||
case DEVICE: |
|||
return new Device((DeviceId) entityId); |
|||
case ALARM: |
|||
return new Alarm((AlarmId) entityId); |
|||
case RULE_CHAIN: |
|||
return new RuleChain((RuleChainId) entityId); |
|||
case ENTITY_VIEW: |
|||
return new EntityView((EntityViewId) entityId); |
|||
default: |
|||
throw new RuntimeException("Unexpected EntityType: " + entityId.getEntityType()); |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue