dendi-wang/thingsboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

code fixes

Browse Source
This commit is contained in:
Bohdan Smetaniuk 2020-08-25 14:04:47 +03:00
parent 787c33be73
commit 1d5f5c5d80
5 changed files with 53 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
application/src/main/java/org/thingsboard/server/service/edge/DefaultEdgeNotificationService.java
View File

@ -242,7 +242,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
} }
} }
} else { } else {
ListenableFuture<List<EdgeId>> edgeIdsFuture = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId, dbCallbackExecutorService); ListenableFuture<List<EdgeId>> edgeIdsFuture = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId);
Futures.transform(edgeIdsFuture, edgeIds -> { Futures.transform(edgeIdsFuture, edgeIds -> {
if (edgeIds != null && !edgeIds.isEmpty()) { if (edgeIds != null && !edgeIds.isEmpty()) {
for (EdgeId edgeId : edgeIds) { for (EdgeId edgeId : edgeIds) {
@ -321,7 +321,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
if (alarm != null) { if (alarm != null) {
EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType()); EdgeEventType edgeEventType = getEdgeQueueTypeByEntityType(alarm.getOriginator().getEntityType());
if (edgeEventType != null) { if (edgeEventType != null) {
ListenableFuture<List<EdgeId>> relatedEdgeIdsByEntityIdFuture = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator(), dbCallbackExecutorService); ListenableFuture<List<EdgeId>> relatedEdgeIdsByEntityIdFuture = edgeService.findRelatedEdgeIdsByEntityId(tenantId, alarm.getOriginator());
Futures.transform(relatedEdgeIdsByEntityIdFuture, relatedEdgeIdsByEntityId -> { Futures.transform(relatedEdgeIdsByEntityIdFuture, relatedEdgeIdsByEntityId -> {
if (relatedEdgeIdsByEntityId != null) { if (relatedEdgeIdsByEntityId != null) {
for (EdgeId edgeId : relatedEdgeIdsByEntityId) { for (EdgeId edgeId : relatedEdgeIdsByEntityId) {
@ -346,8 +346,8 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) && if (!relation.getFrom().getEntityType().equals(EntityType.EDGE) &&
!relation.getTo().getEntityType().equals(EntityType.EDGE)) { !relation.getTo().getEntityType().equals(EntityType.EDGE)) {
List<ListenableFuture<List<EdgeId>>> futures = new ArrayList<>(); List<ListenableFuture<List<EdgeId>>> futures = new ArrayList<>();
futures.add(edgeService.findRelatedEdgeIdsByEntityId(tenantId, relation.getTo(), dbCallbackExecutorService)); futures.add(edgeService.findRelatedEdgeIdsByEntityId(tenantId, relation.getTo()));
futures.add(edgeService.findRelatedEdgeIdsByEntityId(tenantId, relation.getFrom(), dbCallbackExecutorService)); futures.add(edgeService.findRelatedEdgeIdsByEntityId(tenantId, relation.getFrom()));
ListenableFuture<List<List<EdgeId>>> combinedFuture = Futures.allAsList(futures); ListenableFuture<List<List<EdgeId>>> combinedFuture = Futures.allAsList(futures);
Futures.transform(combinedFuture, listOfListsEdgeIds -> { Futures.transform(combinedFuture, listOfListsEdgeIds -> {
Set<EdgeId> uniqueEdgeIds = new HashSet<>(); Set<EdgeId> uniqueEdgeIds = new HashSet<>();

48
application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
View File

@ -25,6 +25,7 @@ import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.SettableFuture;
import com.google.gson.Gson; import com.google.gson.Gson;
import com.google.gson.JsonElement; import com.google.gson.JsonElement;
import com.google.gson.JsonObject; import com.google.gson.JsonObject;
@ -937,25 +938,46 @@ public final class EdgeGrpcSession implements Closeable {
} }
private ListenableFuture<Void> processPostTelemetry(EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) { private ListenableFuture<Void> processPostTelemetry(EntityId entityId, TransportProtos.PostTelemetryMsg msg, TbMsgMetaData metaData) {
SettableFuture<Void> futureToSet = SettableFuture.create();
for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) { for (TransportProtos.TsKvListProto tsKv : msg.getTsKvListList()) {
JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList()); JsonObject json = JsonUtils.getJsonObject(tsKv.getKvList());
metaData.putValue("ts", tsKv.getTs() + ""); metaData.putValue("ts", tsKv.getTs() + "");
TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), entityId, metaData, gson.toJson(json)); TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_TELEMETRY_REQUEST.name(), entityId, metaData, gson.toJson(json));
// TODO: voba - verify that null callback is OK ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); @Override
public void onSuccess(TbQueueMsgMetadata metadata) {
futureToSet.set(null);
}
@Override
public void onFailure(Throwable t) {
futureToSet.setException(t);
}
});
} }
return Futures.immediateFuture(null); return futureToSet;
} }
private ListenableFuture<Void> processPostAttributes(EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) { private ListenableFuture<Void> processPostAttributes(EntityId entityId, TransportProtos.PostAttributeMsg msg, TbMsgMetaData metaData) {
SettableFuture<Void> futureToSet = SettableFuture.create();
JsonObject json = JsonUtils.getJsonObject(msg.getKvList()); JsonObject json = JsonUtils.getJsonObject(msg.getKvList());
TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, metaData, gson.toJson(json)); TbMsg tbMsg = TbMsg.newMsg(SessionMsgType.POST_ATTRIBUTES_REQUEST.name(), entityId, metaData, gson.toJson(json));
// TODO: voba - verify that null callback is OK ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, new TbQueueCallback() {
ctx.getTbClusterService().pushMsgToRuleEngine(edge.getTenantId(), tbMsg.getOriginator(), tbMsg, null); @Override
return Futures.immediateFuture(null); public void onSuccess(TbQueueMsgMetadata metadata) {
futureToSet.set(null);
}
@Override
public void onFailure(Throwable t) {
futureToSet.setException(t);
}
});
return futureToSet;
} }
private ListenableFuture<Void> processAttributeDeleteMsg(EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, String entityType) { private ListenableFuture<Void> processAttributeDeleteMsg(EntityId entityId, AttributeDeleteMsg attributeDeleteMsg, String entityType) {
SettableFuture<Void> futureToSet = SettableFuture.create();
try { try {
String scope = attributeDeleteMsg.getScope(); String scope = attributeDeleteMsg.getScope();
List<String> attributeNames = attributeDeleteMsg.getAttributeNamesList(); List<String> attributeNames = attributeDeleteMsg.getAttributeNamesList();
@ -966,13 +988,23 @@ public final class EdgeGrpcSession implements Closeable {
attributeKeys.add(new AttributeKey(scope, attributeName)); attributeKeys.add(new AttributeKey(scope, attributeName));
} }
ctx.getTbClusterService().pushMsgToCore(DeviceAttributesEventNotificationMsg.onDelete( ctx.getTbClusterService().pushMsgToCore(DeviceAttributesEventNotificationMsg.onDelete(
edge.getTenantId(), (DeviceId) entityId, attributeKeys), null); edge.getTenantId(), (DeviceId) entityId, attributeKeys), new TbQueueCallback() {
@Override
public void onSuccess(TbQueueMsgMetadata metadata) {
futureToSet.set(null);
}
@Override
public void onFailure(Throwable t) {
futureToSet.setException(t);
}
});
} }
} catch (Exception e) { } catch (Exception e) {
log.error("Can't process attribute delete msg [{}]", attributeDeleteMsg, e); log.error("Can't process attribute delete msg [{}]", attributeDeleteMsg, e);
return Futures.immediateFailedFuture(new RuntimeException("Can't process attribute delete msg " + attributeDeleteMsg, e)); return Futures.immediateFailedFuture(new RuntimeException("Can't process attribute delete msg " + attributeDeleteMsg, e));
} }
return Futures.immediateFuture(null); return futureToSet;
} }
private ListenableFuture<Void> onDeviceUpdate(DeviceUpdateMsg deviceUpdateMsg) { private ListenableFuture<Void> onDeviceUpdate(DeviceUpdateMsg deviceUpdateMsg) {

3
common/dao-api/src/main/java/org/thingsboard/server/dao/edge/EdgeService.java
View File

@ -30,7 +30,6 @@ import org.thingsboard.server.common.data.page.TextPageLink;
import java.util.List; import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.concurrent.Executor;
public interface EdgeService { public interface EdgeService {
@ -76,7 +75,7 @@ public interface EdgeService {
ListenableFuture<List<Edge>> findEdgesByTenantIdAndDashboardId(TenantId tenantId, DashboardId dashboardId); ListenableFuture<List<Edge>> findEdgesByTenantIdAndDashboardId(TenantId tenantId, DashboardId dashboardId);
ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId, Executor executorService); ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId);
} }

15
dao/src/main/java/org/thingsboard/server/dao/edge/EdgeServiceImpl.java
View File

@ -67,7 +67,6 @@ import java.util.Collections;
import java.util.Comparator; import java.util.Comparator;
import java.util.List; import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.concurrent.Executor;
import java.util.stream.Collectors; import java.util.stream.Collectors;
import static org.thingsboard.server.common.data.CacheConstants.EDGE_CACHE; import static org.thingsboard.server.common.data.CacheConstants.EDGE_CACHE;
@ -429,7 +428,7 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
}; };
@Override @Override
public ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId, Executor executorService) { public ListenableFuture<List<EdgeId>> findRelatedEdgeIdsByEntityId(TenantId tenantId, EntityId entityId) {
switch (entityId.getEntityType()) { switch (entityId.getEntityType()) {
case DEVICE: case DEVICE:
case ASSET: case ASSET:
@ -442,11 +441,11 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
} else { } else {
return Collections.emptyList(); return Collections.emptyList();
} }
}, executorService); }, MoreExecutors.directExecutor());
case DASHBOARD: case DASHBOARD:
return convertToEdgeIds(findEdgesByTenantIdAndDashboardId(tenantId, new DashboardId(entityId.getId())), executorService); return convertToEdgeIds(findEdgesByTenantIdAndDashboardId(tenantId, new DashboardId(entityId.getId())));
case RULE_CHAIN: case RULE_CHAIN:
return convertToEdgeIds(findEdgesByTenantIdAndRuleChainId(tenantId, new RuleChainId(entityId.getId())), executorService); return convertToEdgeIds(findEdgesByTenantIdAndRuleChainId(tenantId, new RuleChainId(entityId.getId())));
case USER: case USER:
User userById = userService.findUserById(tenantId, new UserId(entityId.getId())); User userById = userService.findUserById(tenantId, new UserId(entityId.getId()));
TextPageData<Edge> edges; TextPageData<Edge> edges;
@ -455,20 +454,20 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
} else { } else {
edges = findEdgesByTenantIdAndCustomerId(tenantId, new CustomerId(entityId.getId()), new TextPageLink(Integer.MAX_VALUE)); edges = findEdgesByTenantIdAndCustomerId(tenantId, new CustomerId(entityId.getId()), new TextPageLink(Integer.MAX_VALUE));
} }
return convertToEdgeIds(Futures.immediateFuture(edges.getData()), executorService); return convertToEdgeIds(Futures.immediateFuture(edges.getData()));
default: default:
return Futures.immediateFuture(Collections.emptyList()); return Futures.immediateFuture(Collections.emptyList());
} }
} }
private ListenableFuture<List<EdgeId>> convertToEdgeIds(ListenableFuture<List<Edge>> future, Executor executorService) { private ListenableFuture<List<EdgeId>> convertToEdgeIds(ListenableFuture<List<Edge>> future) {
return Futures.transform(future, edges -> { return Futures.transform(future, edges -> {
if (edges != null && !edges.isEmpty()) { if (edges != null && !edges.isEmpty()) {
return edges.stream().map(IdBased::getId).collect(Collectors.toList()); return edges.stream().map(IdBased::getId).collect(Collectors.toList());
} else { } else {
return Collections.emptyList(); return Collections.emptyList();
} }
}, executorService); }, MoreExecutors.directExecutor());
} }
} }

2
rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/edge/TbMsgPushToEdgeNode.java
View File

@ -232,7 +232,7 @@ public class TbMsgPushToEdgeNode implements TbNode {
TextPageData<Edge> edgesByTenantId = ctx.getEdgeService().findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE)); TextPageData<Edge> edgesByTenantId = ctx.getEdgeService().findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE));
return Futures.immediateFuture(edgesByTenantId.getData().stream().map(IdBased::getId).collect(Collectors.toList())); return Futures.immediateFuture(edgesByTenantId.getData().stream().map(IdBased::getId).collect(Collectors.toList()));
} else { } else {
return ctx.getEdgeService().findRelatedEdgeIdsByEntityId(tenantId, originatorId, ctx.getDbCallbackExecutor()); return ctx.getEdgeService().findRelatedEdgeIdsByEntityId(tenantId, originatorId);
} }
} }