简柏林 / thingskit · Commits

Commit 4e167d8566121f191709896abb3c700b3eff83c5

Authored by Bohdan Smetaniuk
2 parents b5659fd3 869a492c

Merge remote-tracking branch 'upstream/feature/edge' into feature/widgets_bundle_fetch

Inline Side-by-side
Showing 5 changed files with 68 additions and 34 deletions
... ... @@ -304,6 +304,12 @@ public final class EdgeGrpcSession implements Closeable {
304 304 case DASHBOARD:
305 305 entityId = new DashboardId(edgeEvent.getEntityId());
306 306 break;
  307 + case TENANT:
  308 + entityId = new TenantId(edgeEvent.getEntityId());
  309 + break;
  310 + case CUSTOMER:
  311 + entityId = new CustomerId(edgeEvent.getEntityId());
  312 + break;
307 313 }
308 314 if (entityId != null) {
309 315 log.debug("Sending telemetry data msg, entityId [{}], body [{}]", edgeEvent.getEntityId(), edgeEvent.getEntityBody());
... ... @@ -820,6 +826,10 @@ public final class EdgeGrpcSession implements Closeable {
820 826 return new EntityViewId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB()));
821 827 case DASHBOARD:
822 828 return new DashboardId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB()));
  829 + case TENANT:
  830 + return new TenantId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB()));
  831 + case CUSTOMER:
  832 + return new CustomerId(new UUID(entityData.getEntityIdMSB(), entityData.getEntityIdLSB()));
823 833 default:
824 834 log.warn("Unsupported entity type [{}] during construct of entity id. EntityDataProto [{}]", entityData.getEntityType(), entityData);
825 835 return null;
... ...
... ... @@ -36,6 +36,10 @@ public final class EdgeUtils {
36 36 return EdgeEventType.USER;
37 37 case ALARM:
38 38 return EdgeEventType.ALARM;
  39 + case TENANT:
  40 + return EdgeEventType.TENANT;
  41 + case CUSTOMER:
  42 + return EdgeEventType.CUSTOMER;
39 43 case WIDGETS_BUNDLE:
40 44 return EdgeEventType.WIDGETS_BUNDLE;
41 45 case WIDGET_TYPE:
... ...
... ... @@ -27,6 +27,8 @@ public enum EdgeEventType {
27 27 USER,
28 28 CUSTOMER,
29 29 RELATION,
  30 + TENANT
  31 + RELATION,
30 32 WIDGETS_BUNDLE,
31 33 WIDGET_TYPE
32 34 }
... ...
... ... @@ -155,6 +155,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
155 155 @Override
156 156 public void onCompleted() {
157 157 log.debug("[{}] The rpc session was closed!", edgeKey);
  158 + onError.accept(new EdgeConnectionException("[" + edgeKey + "] The rpc session was closed!"));
158 159 }
159 160 };
160 161 }
... ...
... ... @@ -33,11 +33,15 @@ import org.thingsboard.server.common.data.DataConstants;
33 33 import org.thingsboard.server.common.data.EdgeUtils;
34 34 import org.thingsboard.server.common.data.EntityType;
35 35 import org.thingsboard.server.common.data.audit.ActionType;
  36 +import org.thingsboard.server.common.data.edge.Edge;
36 37 import org.thingsboard.server.common.data.edge.EdgeEvent;
37 38 import org.thingsboard.server.common.data.edge.EdgeEventType;
38 39 import org.thingsboard.server.common.data.id.EdgeId;
39 40 import org.thingsboard.server.common.data.id.EntityId;
  41 +import org.thingsboard.server.common.data.id.IdBased;
40 42 import org.thingsboard.server.common.data.id.TenantId;
  43 +import org.thingsboard.server.common.data.page.TextPageData;
  44 +import org.thingsboard.server.common.data.page.TextPageLink;
41 45 import org.thingsboard.server.common.data.plugin.ComponentType;
42 46 import org.thingsboard.server.common.data.relation.EntityRelation;
43 47 import org.thingsboard.server.common.data.relation.RelationTypeGroup;
... ... @@ -46,10 +50,12 @@ import org.thingsboard.server.common.msg.TbMsg;
46 50 import org.thingsboard.server.common.msg.session.SessionMsgType;
47 51
48 52 import javax.annotation.Nullable;
  53 +import java.util.ArrayList;
49 54 import java.util.HashMap;
50 55 import java.util.List;
51 56 import java.util.Map;
52 57 import java.util.UUID;
  58 +import java.util.stream.Collectors;
53 59
54 60 import static org.thingsboard.rule.engine.api.TbRelationTypes.SUCCESS;
55 61
... ... @@ -84,34 +90,38 @@ public class TbMsgPushToEdgeNode implements TbNode {
84 90 }
85 91 if (isSupportedOriginator(msg.getOriginator().getEntityType())) {
86 92 if (isSupportedMsgType(msg.getType())) {
87   - ListenableFuture<EdgeId> getEdgeIdFuture = getEdgeIdByOriginatorId(ctx, ctx.getTenantId(), msg.getOriginator());
88   - Futures.addCallback(getEdgeIdFuture, new FutureCallback<EdgeId>() {
  93 + ListenableFuture<List<EdgeId>> getEdgeIdsFuture = getEdgeIdsByOriginatorId(ctx, ctx.getTenantId(), msg.getOriginator());
  94 + Futures.addCallback(getEdgeIdsFuture, new FutureCallback<List<EdgeId>>() {
89 95 @Override
90   - public void onSuccess(@Nullable EdgeId edgeId) {
91   - try {
92   - EdgeEvent edgeEvent = buildEdgeEvent(msg, ctx);
93   - if (edgeEvent == null) {
94   - log.debug("Edge event type is null. Entity Type {}", msg.getOriginator().getEntityType());
95   - ctx.tellFailure(msg, new RuntimeException("Edge event type is null. Entity Type '" + msg.getOriginator().getEntityType() + "'"));
96   - } else {
97   - edgeEvent.setEdgeId(edgeId);
98   - ListenableFuture<EdgeEvent> saveFuture = ctx.getEdgeEventService().saveAsync(edgeEvent);
99   - Futures.addCallback(saveFuture, new FutureCallback<EdgeEvent>() {
100   - @Override
101   - public void onSuccess(@Nullable EdgeEvent event) {
102   - ctx.tellNext(msg, SUCCESS);
  96 + public void onSuccess(@Nullable List<EdgeId> edgeIds) {
  97 + if (edgeIds != null && !edgeIds.isEmpty()) {
  98 + for (EdgeId edgeId : edgeIds) {
  99 + try {
  100 + EdgeEvent edgeEvent = buildEdgeEvent(msg, ctx);
  101 + if (edgeEvent == null) {
  102 + log.debug("Edge event type is null. Entity Type {}", msg.getOriginator().getEntityType());
  103 + ctx.tellFailure(msg, new RuntimeException("Edge event type is null. Entity Type '" + msg.getOriginator().getEntityType() + "'"));
  104 + } else {
  105 + edgeEvent.setEdgeId(edgeId);
  106 + ListenableFuture<EdgeEvent> saveFuture = ctx.getEdgeEventService().saveAsync(edgeEvent);
  107 + Futures.addCallback(saveFuture, new FutureCallback<EdgeEvent>() {
  108 + @Override
  109 + public void onSuccess(@Nullable EdgeEvent event) {
  110 + ctx.tellNext(msg, SUCCESS);
  111 + }
  112 +
  113 + @Override
  114 + public void onFailure(Throwable th) {
  115 + log.error("Could not save edge event", th);
  116 + ctx.tellFailure(msg, th);
  117 + }
  118 + }, ctx.getDbCallbackExecutor());
103 119 }
104   -
105   - @Override
106   - public void onFailure(Throwable th) {
107   - log.error("Could not save edge event", th);
108   - ctx.tellFailure(msg, th);
109   - }
110   - }, ctx.getDbCallbackExecutor());
  120 + } catch (JsonProcessingException e) {
  121 + log.error("Failed to build edge event", e);
  122 + ctx.tellFailure(msg, e);
  123 + }
111 124 }
112   - } catch (JsonProcessingException e) {
113   - log.error("Failed to build edge event", e);
114   - ctx.tellFailure(msg, e);
115 125 }
116 126 }
117 127
... ... @@ -201,6 +211,8 @@ public class TbMsgPushToEdgeNode implements TbNode {
201 211 case ASSET:
202 212 case ENTITY_VIEW:
203 213 case DASHBOARD:
  214 + case TENANT:
  215 + case CUSTOMER:
204 216 return true;
205 217 default:
206 218 return false;
... ... @@ -215,15 +227,20 @@ public class TbMsgPushToEdgeNode implements TbNode {
215 227 || DataConstants.ALARM.equals(msgType);
216 228 }
217 229
218   - private ListenableFuture<EdgeId> getEdgeIdByOriginatorId(TbContext ctx, TenantId tenantId, EntityId originatorId) {
219   - ListenableFuture<List<EntityRelation>> future = ctx.getRelationService().findByToAndTypeAsync(tenantId, originatorId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE);
220   - return Futures.transform(future, relations -> {
221   - if (relations != null && relations.size() > 0) {
222   - return new EdgeId(relations.get(0).getFrom().getId());
223   - } else {
224   - return null;
225   - }
226   - }, ctx.getDbCallbackExecutor());
  230 + private ListenableFuture<List<EdgeId>> getEdgeIdsByOriginatorId(TbContext ctx, TenantId tenantId, EntityId originatorId) {
  231 + if (EntityType.TENANT.equals(originatorId.getEntityType())) {
  232 + TextPageData<Edge> edgesByTenantId = ctx.getEdgeService().findEdgesByTenantId(tenantId, new TextPageLink(Integer.MAX_VALUE));
  233 + return Futures.immediateFuture(edgesByTenantId.getData().stream().map(IdBased::getId).collect(Collectors.toList()));
  234 + } else {
  235 + ListenableFuture<List<EntityRelation>> future = ctx.getRelationService().findByToAndTypeAsync(tenantId, originatorId, EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE);
  236 + return Futures.transform(future, relations -> {
  237 + List<EdgeId> result = new ArrayList<>();
  238 + if (relations != null && relations.size() > 0) {
  239 + result.add(new EdgeId(relations.get(0).getFrom().getId()));
  240 + }
  241 + return result;
  242 + }, ctx.getDbCallbackExecutor());
  243 + }
227 244 }
228 245
229 246 @Override
... ...