简柏林 / thingskit · Commits

Commit a76d3a2b70287eec95dcb76b0da20941a3d7357c

Authored by Volodymyr Babak
1 parent 612a17af

Cover cases when entities added during edge disconnect. Refactoring

Inline Side-by-side
Showing 31 changed files with 1195 additions and 505 deletions
... ... @@ -245,11 +245,11 @@ public class TenantActor extends RuleChainManagerActor {
245 245 EdgeId edgeId = new EdgeId(msg.getEntityId().getId());
246 246 EdgeRpcService edgeRpcService = systemContext.getEdgeRpcService();
247 247 if (msg.getEvent() == ComponentLifecycleEvent.DELETED) {
248   - edgeRpcService.deleteEdge(edgeId);
  248 + edgeRpcService.deleteEdge(tenantId, edgeId);
249 249 } else {
250 250 Edge edge = systemContext.getEdgeService().findEdgeById(tenantId, edgeId);
251 251 if (msg.getEvent() == ComponentLifecycleEvent.UPDATED) {
252   - edgeRpcService.updateEdge(edge);
  252 + edgeRpcService.updateEdge(tenantId, edge);
253 253 }
254 254 }
255 255 } else if (isRuleEngineForCurrentTenant) {
... ... @@ -277,7 +277,7 @@ public class TenantActor extends RuleChainManagerActor {
277 277
278 278 private void onToEdgeSessionMsg(EdgeEventUpdateMsg msg) {
279 279 log.trace("[{}] onToEdgeSessionMsg [{}]", msg.getTenantId(), msg);
280   - systemContext.getEdgeRpcService().onEdgeEvent(msg.getEdgeId());
  280 + systemContext.getEdgeRpcService().onEdgeEvent(tenantId, msg.getEdgeId());
281 281 }
282 282
283 283 public static class ActorCreator extends ContextBasedCreator {
... ...
... ... @@ -43,8 +43,8 @@ import org.thingsboard.server.common.data.Firmware;
43 43 import org.thingsboard.server.common.data.FirmwareInfo;
44 44 import org.thingsboard.server.common.data.HasName;
45 45 import org.thingsboard.server.common.data.HasTenantId;
46   -import org.thingsboard.server.common.data.TbResourceInfo;
47 46 import org.thingsboard.server.common.data.TbResource;
  47 +import org.thingsboard.server.common.data.TbResourceInfo;
48 48 import org.thingsboard.server.common.data.Tenant;
49 49 import org.thingsboard.server.common.data.TenantInfo;
50 50 import org.thingsboard.server.common.data.TenantProfile;
... ... @@ -71,9 +71,9 @@ import org.thingsboard.server.common.data.id.EntityId;
71 71 import org.thingsboard.server.common.data.id.EntityIdFactory;
72 72 import org.thingsboard.server.common.data.id.EntityViewId;
73 73 import org.thingsboard.server.common.data.id.FirmwareId;
74   -import org.thingsboard.server.common.data.id.TbResourceId;
75 74 import org.thingsboard.server.common.data.id.RuleChainId;
76 75 import org.thingsboard.server.common.data.id.RuleNodeId;
  76 +import org.thingsboard.server.common.data.id.TbResourceId;
77 77 import org.thingsboard.server.common.data.id.TenantId;
78 78 import org.thingsboard.server.common.data.id.TenantProfileId;
79 79 import org.thingsboard.server.common.data.id.UserId;
... ... @@ -128,10 +128,9 @@ import org.thingsboard.server.queue.discovery.PartitionService;
128 128 import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
129 129 import org.thingsboard.server.queue.util.TbCoreComponent;
130 130 import org.thingsboard.server.service.component.ComponentDiscoveryService;
131   -import org.thingsboard.server.service.firmware.FirmwareStateService;
132 131 import org.thingsboard.server.service.edge.EdgeNotificationService;
133   -import org.thingsboard.server.service.edge.rpc.EdgeGrpcService;
134   -import org.thingsboard.server.service.edge.rpc.init.SyncEdgeService;
  132 +import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
  133 +import org.thingsboard.server.service.firmware.FirmwareStateService;
135 134 import org.thingsboard.server.service.lwm2m.LwM2MServerSecurityInfoRepository;
136 135 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
137 136 import org.thingsboard.server.service.queue.TbClusterService;
... ... @@ -276,10 +275,7 @@ public abstract class BaseController {
276 275 protected EdgeNotificationService edgeNotificationService;
277 276
278 277 @Autowired(required = false)
279   - protected SyncEdgeService syncEdgeService;
280   -
281   - @Autowired(required = false)
282   - protected EdgeGrpcService edgeGrpcService;
  278 + protected EdgeRpcService edgeGrpcService;
283 279
284 280 @Value("${server.log_controller_error_stack_trace}")
285 281 @Getter
... ...
... ... @@ -49,7 +49,6 @@ import org.thingsboard.server.dao.exception.DataValidationException;
49 49 import org.thingsboard.server.dao.exception.IncorrectParameterException;
50 50 import org.thingsboard.server.dao.model.ModelConstants;
51 51 import org.thingsboard.server.queue.util.TbCoreComponent;
52   -import org.thingsboard.server.service.edge.rpc.EdgeGrpcSession;
53 52 import org.thingsboard.server.service.security.model.SecurityUser;
54 53 import org.thingsboard.server.service.security.permission.Operation;
55 54 import org.thingsboard.server.service.security.permission.Resource;
... ... @@ -536,9 +535,7 @@ public class EdgeController extends BaseController {
536 535 edgeId = checkNotNull(edgeId);
537 536 SecurityUser user = getCurrentUser();
538 537 TenantId tenantId = user.getTenantId();
539   - EdgeGrpcSession session = edgeGrpcService.getEdgeGrpcSessionById(tenantId, edgeId);
540   - Edge edge = session.getEdge();
541   - syncEdgeService.sync(tenantId, edge);
  538 + edgeGrpcService.startSyncProcess(tenantId, edgeId);
542 539 } else {
543 540 throw new ThingsboardException("Edges support disabled", ThingsboardErrorCode.GENERAL);
544 541 }
... ...
... ... @@ -326,7 +326,7 @@ public class DefaultEdgeNotificationService implements EdgeNotificationService {
326 326 case ASSIGNED_TO_CUSTOMER:
327 327 case UNASSIGNED_FROM_CUSTOMER:
328 328 edgeIdsFuture = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId);
329   - Futures.addCallback(edgeIdsFuture, new FutureCallback<List<EdgeId>>() {
  329 + Futures.addCallback(edgeIdsFuture, new FutureCallback<>() {
330 330 @Override
331 331 public void onSuccess(@Nullable List<EdgeId> edgeIds) {
332 332 if (edgeIds != null && !edgeIds.isEmpty()) {
... ...
... ... @@ -54,7 +54,7 @@ import org.thingsboard.server.service.edge.rpc.constructor.RuleChainMsgConstruct
54 54 import org.thingsboard.server.service.edge.rpc.constructor.UserMsgConstructor;
55 55 import org.thingsboard.server.service.edge.rpc.constructor.WidgetTypeMsgConstructor;
56 56 import org.thingsboard.server.service.edge.rpc.constructor.WidgetsBundleMsgConstructor;
57   -import org.thingsboard.server.service.edge.rpc.init.SyncEdgeService;
  57 +import org.thingsboard.server.service.edge.rpc.sync.EdgeRequestsService;
58 58 import org.thingsboard.server.service.edge.rpc.processor.AlarmProcessor;
59 59 import org.thingsboard.server.service.edge.rpc.processor.DeviceProcessor;
60 60 import org.thingsboard.server.service.edge.rpc.processor.RelationProcessor;
... ... @@ -149,7 +149,7 @@ public class EdgeContextComponent {
149 149
150 150 @Lazy
151 151 @Autowired
152   - private SyncEdgeService syncEdgeService;
  152 + private EdgeRequestsService edgeRequestsService;
153 153
154 154 @Lazy
155 155 @Autowired
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc;
  17 +
  18 +import com.fasterxml.jackson.databind.JsonNode;
  19 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  20 +import org.thingsboard.server.common.data.edge.EdgeEventActionType;
  21 +import org.thingsboard.server.common.data.edge.EdgeEventType;
  22 +import org.thingsboard.server.common.data.id.EdgeId;
  23 +import org.thingsboard.server.common.data.id.EntityId;
  24 +import org.thingsboard.server.common.data.id.TenantId;
  25 +
  26 +public final class EdgeEventUtils {
  27 +
  28 + private EdgeEventUtils() {
  29 + }
  30 +
  31 + public static EdgeEvent constructEdgeEvent(TenantId tenantId,
  32 + EdgeId edgeId,
  33 + EdgeEventType type,
  34 + EdgeEventActionType action,
  35 + EntityId entityId,
  36 + JsonNode body) {
  37 + EdgeEvent edgeEvent = new EdgeEvent();
  38 + edgeEvent.setTenantId(tenantId);
  39 + edgeEvent.setEdgeId(edgeId);
  40 + edgeEvent.setType(type);
  41 + edgeEvent.setAction(action);
  42 + if (entityId != null) {
  43 + edgeEvent.setEntityId(entityId.getId());
  44 + }
  45 + edgeEvent.setBody(body);
  46 + return edgeEvent;
  47 + }
  48 +}
... ...
... ... @@ -53,6 +53,7 @@ import java.util.Map;
53 53 import java.util.UUID;
54 54 import java.util.concurrent.ConcurrentHashMap;
55 55 import java.util.concurrent.ConcurrentMap;
  56 +import java.util.concurrent.ExecutorService;
56 57 import java.util.concurrent.Executors;
57 58 import java.util.concurrent.ScheduledExecutorService;
58 59 import java.util.concurrent.ScheduledFuture;
... ... @@ -95,6 +96,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
95 96
96 97 private ScheduledExecutorService scheduler;
97 98
  99 + private ExecutorService syncExecutorService;
  100 +
98 101 @PostConstruct
99 102 public void init() {
100 103 log.info("Initializing Edge RPC service!");
... ... @@ -120,6 +123,8 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
120 123 throw new RuntimeException("Failed to start Edge RPC server!");
121 124 }
122 125 this.scheduler = Executors.newScheduledThreadPool(schedulerPoolSize, ThingsBoardThreadFactory.forName("edge-scheduler"));
  126 + this.syncExecutorService = Executors.newFixedThreadPool(
  127 + Runtime.getRuntime().availableProcessors(), ThingsBoardThreadFactory.forName("edge-sync"));
123 128 log.info("Edge RPC service initialized!");
124 129 }
125 130
... ... @@ -139,29 +144,32 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
139 144 if (scheduler != null) {
140 145 scheduler.shutdownNow();
141 146 }
  147 + if (syncExecutorService != null) {
  148 + syncExecutorService.shutdownNow();
  149 + }
142 150 }
143 151
144 152 @Override
145 153 public StreamObserver<RequestMsg> handleMsgs(StreamObserver<ResponseMsg> outputStream) {
146   - return new EdgeGrpcSession(ctx, outputStream, this::onEdgeConnect, this::onEdgeDisconnect, mapper).getInputStream();
  154 + return new EdgeGrpcSession(ctx, outputStream, this::onEdgeConnect, this::onEdgeDisconnect, mapper, syncExecutorService).getInputStream();
147 155 }
148 156
149 157 @Override
150   - public void updateEdge(Edge edge) {
  158 + public void updateEdge(TenantId tenantId, Edge edge) {
151 159 EdgeGrpcSession session = sessions.get(edge.getId());
152 160 if (session != null && session.isConnected()) {
153   - log.debug("[{}] Updating configuration for edge [{}] [{}]", edge.getTenantId(), edge.getName(), edge.getId());
  161 + log.debug("[{}] Updating configuration for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
154 162 session.onConfigurationUpdate(edge);
155 163 } else {
156   - log.debug("[{}] Session doesn't exist for edge [{}] [{}]", edge.getTenantId(), edge.getName(), edge.getId());
  164 + log.debug("[{}] Session doesn't exist for edge [{}] [{}]", tenantId, edge.getName(), edge.getId());
157 165 }
158 166 }
159 167
160 168 @Override
161   - public void deleteEdge(EdgeId edgeId) {
  169 + public void deleteEdge(TenantId tenantId, EdgeId edgeId) {
162 170 EdgeGrpcSession session = sessions.get(edgeId);
163 171 if (session != null && session.isConnected()) {
164   - log.info("Closing and removing session for edge [{}]", edgeId);
  172 + log.info("[{}] Closing and removing session for edge [{}]", tenantId, edgeId);
165 173 session.close();
166 174 sessions.remove(edgeId);
167 175 sessionNewEvents.remove(edgeId);
... ... @@ -170,10 +178,10 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
170 178 }
171 179
172 180 @Override
173   - public void onEdgeEvent(EdgeId edgeId) {
174   - log.trace("[{}] onEdgeEvent", edgeId.getId());
  181 + public void onEdgeEvent(TenantId tenantId, EdgeId edgeId) {
  182 + log.trace("[{}] onEdgeEvent [{}]", tenantId, edgeId.getId());
175 183 if (!sessionNewEvents.get(edgeId)) {
176   - log.trace("[{}] set session new events flag to true", edgeId.getId());
  184 + log.trace("[{}] set session new events flag to true [{}]", tenantId, edgeId.getId());
177 185 sessionNewEvents.put(edgeId, true);
178 186 }
179 187 }
... ... @@ -188,10 +196,11 @@ public class EdgeGrpcService extends EdgeRpcServiceGrpc.EdgeRpcServiceImplBase i
188 196 scheduleEdgeEventsCheck(edgeGrpcSession);
189 197 }
190 198
191   - public EdgeGrpcSession getEdgeGrpcSessionById(TenantId tenantId, EdgeId edgeId) {
  199 + @Override
  200 + public void startSyncProcess(TenantId tenantId, EdgeId edgeId) {
192 201 EdgeGrpcSession session = sessions.get(edgeId);
193 202 if (session != null && session.isConnected()) {
194   - return session;
  203 + session.startSyncProcess();
195 204 } else {
196 205 log.error("[{}] Edge is not connected [{}]", tenantId, edgeId);
197 206 throw new RuntimeException("Edge is not connected");
... ...
... ... @@ -16,9 +16,7 @@
16 16 package org.thingsboard.server.service.edge.rpc;
17 17
18 18 import com.datastax.oss.driver.api.core.uuid.Uuids;
19   -import com.fasterxml.jackson.core.JsonProcessingException;
20 19 import com.fasterxml.jackson.databind.ObjectMapper;
21   -import com.fasterxml.jackson.databind.node.ObjectNode;
22 20 import com.google.common.util.concurrent.FutureCallback;
23 21 import com.google.common.util.concurrent.Futures;
24 22 import com.google.common.util.concurrent.ListenableFuture;
... ... @@ -28,6 +26,7 @@ import io.grpc.stub.StreamObserver;
28 26 import lombok.Data;
29 27 import lombok.extern.slf4j.Slf4j;
30 28 import org.checkerframework.checker.nullness.qual.Nullable;
  29 +import org.thingsboard.common.util.JacksonUtil;
31 30 import org.thingsboard.server.common.data.AdminSettings;
32 31 import org.thingsboard.server.common.data.Customer;
33 32 import org.thingsboard.server.common.data.Dashboard;
... ... @@ -61,8 +60,7 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry;
61 60 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
62 61 import org.thingsboard.server.common.data.kv.LongDataEntry;
63 62 import org.thingsboard.server.common.data.page.PageData;
64   -import org.thingsboard.server.common.data.page.SortOrder;
65   -import org.thingsboard.server.common.data.page.TimePageLink;
  63 +import org.thingsboard.server.common.data.page.PageLink;
66 64 import org.thingsboard.server.common.data.relation.EntityRelation;
67 65 import org.thingsboard.server.common.data.rule.RuleChain;
68 66 import org.thingsboard.server.common.data.rule.RuleChainMetaData;
... ... @@ -71,7 +69,6 @@ import org.thingsboard.server.common.data.security.UserCredentials;
71 69 import org.thingsboard.server.common.data.widget.WidgetType;
72 70 import org.thingsboard.server.common.data.widget.WidgetsBundle;
73 71 import org.thingsboard.server.common.transport.util.JsonUtils;
74   -import org.thingsboard.common.util.JacksonUtil;
75 72 import org.thingsboard.server.gen.edge.AdminSettingsUpdateMsg;
76 73 import org.thingsboard.server.gen.edge.AlarmUpdateMsg;
77 74 import org.thingsboard.server.gen.edge.AssetUpdateMsg;
... ... @@ -83,6 +80,7 @@ import org.thingsboard.server.gen.edge.CustomerUpdateMsg;
83 80 import org.thingsboard.server.gen.edge.DashboardUpdateMsg;
84 81 import org.thingsboard.server.gen.edge.DeviceCredentialsRequestMsg;
85 82 import org.thingsboard.server.gen.edge.DeviceCredentialsUpdateMsg;
  83 +import org.thingsboard.server.gen.edge.DeviceProfileDevicesRequestMsg;
86 84 import org.thingsboard.server.gen.edge.DeviceProfileUpdateMsg;
87 85 import org.thingsboard.server.gen.edge.DeviceRpcCallMsg;
88 86 import org.thingsboard.server.gen.edge.DeviceUpdateMsg;
... ... @@ -92,6 +90,7 @@ import org.thingsboard.server.gen.edge.EdgeConfiguration;
92 90 import org.thingsboard.server.gen.edge.EdgeUpdateMsg;
93 91 import org.thingsboard.server.gen.edge.EntityDataProto;
94 92 import org.thingsboard.server.gen.edge.EntityViewUpdateMsg;
  93 +import org.thingsboard.server.gen.edge.EntityViewsRequestMsg;
95 94 import org.thingsboard.server.gen.edge.RelationRequestMsg;
96 95 import org.thingsboard.server.gen.edge.RelationUpdateMsg;
97 96 import org.thingsboard.server.gen.edge.RequestMsg;
... ... @@ -100,14 +99,26 @@ import org.thingsboard.server.gen.edge.ResponseMsg;
100 99 import org.thingsboard.server.gen.edge.RuleChainMetadataRequestMsg;
101 100 import org.thingsboard.server.gen.edge.RuleChainMetadataUpdateMsg;
102 101 import org.thingsboard.server.gen.edge.RuleChainUpdateMsg;
  102 +import org.thingsboard.server.gen.edge.SyncCompletedMsg;
103 103 import org.thingsboard.server.gen.edge.UpdateMsgType;
104 104 import org.thingsboard.server.gen.edge.UplinkMsg;
105 105 import org.thingsboard.server.gen.edge.UplinkResponseMsg;
106 106 import org.thingsboard.server.gen.edge.UserCredentialsRequestMsg;
107 107 import org.thingsboard.server.gen.edge.UserCredentialsUpdateMsg;
  108 +import org.thingsboard.server.gen.edge.WidgetBundleTypesRequestMsg;
108 109 import org.thingsboard.server.gen.edge.WidgetTypeUpdateMsg;
109 110 import org.thingsboard.server.gen.edge.WidgetsBundleUpdateMsg;
110 111 import org.thingsboard.server.service.edge.EdgeContextComponent;
  112 +import org.thingsboard.server.service.edge.rpc.fetch.AssetsEdgeEventFetcher;
  113 +import org.thingsboard.server.service.edge.rpc.fetch.CustomerUsersEdgeEventFetcher;
  114 +import org.thingsboard.server.service.edge.rpc.fetch.DashboardsEdgeEventFetcher;
  115 +import org.thingsboard.server.service.edge.rpc.fetch.DeviceProfilesEdgeEventFetcher;
  116 +import org.thingsboard.server.service.edge.rpc.fetch.EdgeEventFetcher;
  117 +import org.thingsboard.server.service.edge.rpc.fetch.GeneralEdgeEventFetcher;
  118 +import org.thingsboard.server.service.edge.rpc.fetch.RuleChainsEdgeEventFetcher;
  119 +import org.thingsboard.server.service.edge.rpc.fetch.SystemWidgetsBundlesEdgeEventFetcher;
  120 +import org.thingsboard.server.service.edge.rpc.fetch.TenantAdminUsersEdgeEventFetcher;
  121 +import org.thingsboard.server.service.edge.rpc.fetch.TenantWidgetsBundlesEdgeEventFetcher;
111 122
112 123 import java.io.Closeable;
113 124 import java.util.ArrayList;
... ... @@ -117,6 +128,7 @@ import java.util.Optional;
117 128 import java.util.UUID;
118 129 import java.util.concurrent.CountDownLatch;
119 130 import java.util.concurrent.ExecutionException;
  131 +import java.util.concurrent.ExecutorService;
120 132 import java.util.concurrent.TimeUnit;
121 133 import java.util.concurrent.locks.ReentrantLock;
122 134 import java.util.function.BiConsumer;
... ... @@ -140,22 +152,26 @@ public final class EdgeGrpcSession implements Closeable {
140 152 private StreamObserver<RequestMsg> inputStream;
141 153 private StreamObserver<ResponseMsg> outputStream;
142 154 private boolean connected;
  155 + private boolean syncCompleted;
  156 +
  157 + private ExecutorService syncExecutorService;
143 158
144 159 private CountDownLatch latch;
145 160
146 161 EdgeGrpcSession(EdgeContextComponent ctx, StreamObserver<ResponseMsg> outputStream, BiConsumer<EdgeId, EdgeGrpcSession> sessionOpenListener,
147   - Consumer<EdgeId> sessionCloseListener, ObjectMapper mapper) {
  162 + Consumer<EdgeId> sessionCloseListener, ObjectMapper mapper, ExecutorService syncExecutorService) {
148 163 this.sessionId = UUID.randomUUID();
149 164 this.ctx = ctx;
150 165 this.outputStream = outputStream;
151 166 this.sessionOpenListener = sessionOpenListener;
152 167 this.sessionCloseListener = sessionCloseListener;
153 168 this.mapper = mapper;
  169 + this.syncExecutorService = syncExecutorService;
154 170 initInputStream();
155 171 }
156 172
157 173 private void initInputStream() {
158   - this.inputStream = new StreamObserver<RequestMsg>() {
  174 + this.inputStream = new StreamObserver<>() {
159 175 @Override
160 176 public void onNext(RequestMsg requestMsg) {
161 177 if (!connected && requestMsg.getMsgType().equals(RequestMsgType.CONNECT_RPC_MESSAGE)) {
... ... @@ -170,7 +186,10 @@ public final class EdgeGrpcSession implements Closeable {
170 186 }
171 187 }
172 188 if (connected && requestMsg.getMsgType().equals(RequestMsgType.SYNC_REQUEST_RPC_MESSAGE)) {
173   - ctx.getSyncEdgeService().sync(edge.getTenantId(), edge);
  189 + if (requestMsg.getSyncRequestMsg().getSyncRequired()) {
  190 + startSyncProcess();
  191 + }
  192 + syncCompleted = true;
174 193 }
175 194 if (connected) {
176 195 if (requestMsg.getMsgType().equals(RequestMsgType.UPLINK_RPC_MESSAGE) && requestMsg.hasUplinkMsg()) {
... ... @@ -198,18 +217,55 @@ public final class EdgeGrpcSession implements Closeable {
198 217 if (edge != null) {
199 218 try {
200 219 sessionCloseListener.accept(edge.getId());
201   - } catch (Exception ignored) {}
  220 + } catch (Exception ignored) {
  221 + }
202 222 }
203 223 try {
204 224 outputStream.onCompleted();
205   - } catch (Exception ignored) {}
  225 + } catch (Exception ignored) {
  226 + }
206 227 }
207 228 };
208 229 }
209 230
  231 + public void startSyncProcess() {
  232 + log.trace("[{}][{}] Staring edge sync process", edge.getTenantId(), edge.getId());
  233 + syncExecutorService.submit(() -> {
  234 + try {
  235 + startProcessingEdgeEvents(new TenantWidgetsBundlesEdgeEventFetcher(ctx.getWidgetsBundleService()));
  236 + startProcessingEdgeEvents(new SystemWidgetsBundlesEdgeEventFetcher(ctx.getWidgetsBundleService()));
  237 + startProcessingEdgeEvents(new DeviceProfilesEdgeEventFetcher(ctx.getDeviceProfileService()));
  238 + startProcessingEdgeEvents(new RuleChainsEdgeEventFetcher(ctx.getRuleChainService()));
  239 +
  240 + startProcessingEdgeEvents(new TenantAdminUsersEdgeEventFetcher(ctx.getUserService()));
  241 + if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) {
  242 + EdgeEvent customerEdgeEvent = EdgeEventUtils.constructEdgeEvent(edge.getTenantId(), edge.getId(),
  243 + EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, edge.getCustomerId(), null);
  244 + DownlinkMsg customerDownlinkMsg = convertToDownlinkMsg(customerEdgeEvent);
  245 + sendSingleDownlinkMsg(customerDownlinkMsg);
  246 +
  247 + startProcessingEdgeEvents(new CustomerUsersEdgeEventFetcher(ctx.getUserService(), edge.getCustomerId()));
  248 + }
  249 +
  250 + // TODO: voba - implement this functionality
  251 + // syncAdminSettings(edge);
  252 +
  253 + startProcessingEdgeEvents(new AssetsEdgeEventFetcher(ctx.getAssetService()));
  254 + startProcessingEdgeEvents(new DashboardsEdgeEventFetcher(ctx.getDashboardService()));
  255 +
  256 + DownlinkMsg syncCompleteDownlinkMsg = DownlinkMsg.newBuilder()
  257 + .setSyncCompletedMsg(SyncCompletedMsg.newBuilder().build())
  258 + .build();
  259 + sendSingleDownlinkMsg(syncCompleteDownlinkMsg);
  260 + } catch (Exception e) {
  261 + log.error("[{}][{}] Exception during sync process", edge.getTenantId(), edge.getId(), e);
  262 + }
  263 + });
  264 + }
  265 +
210 266 private void onUplinkMsg(UplinkMsg uplinkMsg) {
211 267 ListenableFuture<List<Void>> future = processUplinkMsg(uplinkMsg);
212   - Futures.addCallback(future, new FutureCallback<List<Void>>() {
  268 + Futures.addCallback(future, new FutureCallback<>() {
213 269 @Override
214 270 public void onSuccess(@Nullable List<Void> result) {
215 271 UplinkResponseMsg uplinkResponseMsg = UplinkResponseMsg.newBuilder().setSuccess(true).build();
... ... @@ -271,101 +327,136 @@ public final class EdgeGrpcSession implements Closeable {
271 327
272 328 void processEdgeEvents() throws ExecutionException, InterruptedException {
273 329 log.trace("[{}] processHandleMessages started", this.sessionId);
274   - if (isConnected()) {
  330 + if (isConnected() && isSyncCompleted()) {
275 331 Long queueStartTs = getQueueStartTs().get();
276   - TimePageLink pageLink = new TimePageLink(
  332 + GeneralEdgeEventFetcher fetcher = new GeneralEdgeEventFetcher(
277 333 ctx.getEdgeEventStorageSettings().getMaxReadRecordsCount(),
278   - 0,
279   - null,
280   - new SortOrder("createdTime", SortOrder.Direction.ASC),
281 334 queueStartTs,
282   - null);
283   - PageData<EdgeEvent> pageData;
284   - UUID ifOffset = null;
285   - boolean success = true;
286   - do {
287   - pageData = ctx.getEdgeEventService().findEdgeEvents(edge.getTenantId(), edge.getId(), pageLink, true);
288   - if (isConnected() && !pageData.getData().isEmpty()) {
289   - log.trace("[{}] [{}] event(s) are going to be processed.", this.sessionId, pageData.getData().size());
290   - List<DownlinkMsg> downlinkMsgsPack = convertToDownlinkMsgsPack(pageData.getData());
291   - log.trace("[{}] [{}] downlink msg(s) are going to be send.", this.sessionId, downlinkMsgsPack.size());
292   -
293   - latch = new CountDownLatch(downlinkMsgsPack.size());
294   - for (DownlinkMsg downlinkMsg : downlinkMsgsPack) {
295   - sendDownlinkMsg(ResponseMsg.newBuilder()
296   - .setDownlinkMsg(downlinkMsg)
297   - .build());
298   - }
  335 + ctx.getEdgeEventService());
  336 + UUID ifOffset = startProcessingEdgeEvents(fetcher);
  337 + if (ifOffset != null) {
  338 + Long newStartTs = Uuids.unixTimestamp(ifOffset);
  339 + updateQueueStartTs(newStartTs);
  340 + }
  341 + }
  342 + log.trace("[{}] processHandleMessages finished", this.sessionId);
  343 + }
299 344
300   - ifOffset = pageData.getData().get(pageData.getData().size() - 1).getUuidId();
  345 + private UUID startProcessingEdgeEvents(EdgeEventFetcher fetcher) throws InterruptedException {
  346 + PageLink pageLink = fetcher.getPageLink();
  347 + PageData<EdgeEvent> pageData;
  348 + UUID ifOffset = null;
  349 + boolean success = true;
  350 + do {
  351 + pageData = fetcher.fetchEdgeEvents(edge.getTenantId(), edge.getId(), pageLink);
  352 + if (isConnected() && !pageData.getData().isEmpty()) {
  353 + log.trace("[{}] [{}] event(s) are going to be processed.", this.sessionId, pageData.getData().size());
301 354
302   - success = latch.await(10, TimeUnit.SECONDS);
303   - if (!success) {
304   - log.warn("[{}] Failed to deliver the batch: {}", this.sessionId, downlinkMsgsPack);
305   - }
  355 + success = processEdgeEventsPack(pageData.getData());
  356 +
  357 + ifOffset = pageData.getData().get(pageData.getData().size() - 1).getUuidId();
  358 + }
  359 + if (isConnected() && (!success || pageData.hasNext())) {
  360 + try {
  361 + Thread.sleep(ctx.getEdgeEventStorageSettings().getSleepIntervalBetweenBatches());
  362 + } catch (InterruptedException e) {
  363 + log.error("[{}] Error during sleep between batches", this.sessionId, e);
306 364 }
307   - if (isConnected() && (!success || pageData.hasNext())) {
308   - try {
309   - Thread.sleep(ctx.getEdgeEventStorageSettings().getSleepIntervalBetweenBatches());
310   - } catch (InterruptedException e) {
311   - log.error("[{}] Error during sleep between batches", this.sessionId, e);
312   - }
313   - if (success) {
314   - pageLink = pageLink.nextPageLink();
315   - }
  365 + if (success) {
  366 + pageLink = pageLink.nextPageLink();
316 367 }
317   - } while (isConnected() && (!success || pageData.hasNext()));
  368 + }
  369 + } while (isConnected() && (!success || pageData.hasNext()));
  370 + return ifOffset;
  371 + }
318 372
319   - if (ifOffset != null) {
320   - Long newStartTs = Uuids.unixTimestamp(ifOffset);
321   - updateQueueStartTs(newStartTs);
  373 + private boolean processEdgeEventsPack(List<EdgeEvent> edgeEvents) throws InterruptedException {
  374 + List<DownlinkMsg> downlinkMsgsPack = convertToDownlinkMsgsPack(edgeEvents);
  375 +
  376 + log.trace("[{}] [{}] downlink msg(s) are going to be send.", this.sessionId, downlinkMsgsPack.size());
  377 +
  378 + latch = new CountDownLatch(downlinkMsgsPack.size());
  379 + for (DownlinkMsg downlinkMsg : downlinkMsgsPack) {
  380 + sendDownlinkMsg(ResponseMsg.newBuilder()
  381 + .setDownlinkMsg(downlinkMsg)
  382 + .build());
  383 + }
  384 +
  385 + boolean success = latch.await(10, TimeUnit.SECONDS);
  386 + if (!success) {
  387 + log.warn("[{}] Failed to deliver the batch: {}", this.sessionId, downlinkMsgsPack);
  388 + }
  389 + return success;
  390 + }
  391 +
  392 + private void sendSingleDownlinkMsg(DownlinkMsg downlinkMsg) throws InterruptedException {
  393 + boolean success;
  394 + do {
  395 + latch = new CountDownLatch(1);
  396 + sendDownlinkMsg(ResponseMsg.newBuilder()
  397 + .setDownlinkMsg(downlinkMsg)
  398 + .build());
  399 + success = latch.await(10, TimeUnit.SECONDS);
  400 + if (!success) {
  401 + log.warn("[{}] Failed to deliver single downlink msg!", this.sessionId);
  402 + }
  403 + if (isConnected() && !success) {
  404 + try {
  405 + Thread.sleep(ctx.getEdgeEventStorageSettings().getSleepIntervalBetweenBatches());
  406 + } catch (InterruptedException e) {
  407 + log.error("[{}] Error during sleep between next send of single downlink msg", this.sessionId, e);
  408 + }
322 409 }
  410 + } while (isConnected() && !success);
  411 + }
  412 +
  413 + private DownlinkMsg convertToDownlinkMsg(EdgeEvent edgeEvent) {
  414 + log.trace("[{}] converting edge event to downlink msg [{}]", this.sessionId, edgeEvent);
  415 + DownlinkMsg downlinkMsg = null;
  416 + try {
  417 + switch (edgeEvent.getAction()) {
  418 + case UPDATED:
  419 + case ADDED:
  420 + case DELETED:
  421 + case ASSIGNED_TO_EDGE:
  422 + case UNASSIGNED_FROM_EDGE:
  423 + case ALARM_ACK:
  424 + case ALARM_CLEAR:
  425 + case CREDENTIALS_UPDATED:
  426 + case RELATION_ADD_OR_UPDATE:
  427 + case RELATION_DELETED:
  428 + case ASSIGNED_TO_CUSTOMER:
  429 + case UNASSIGNED_FROM_CUSTOMER:
  430 + downlinkMsg = processEntityMessage(edgeEvent, edgeEvent.getAction());
  431 + break;
  432 + case ATTRIBUTES_UPDATED:
  433 + case POST_ATTRIBUTES:
  434 + case ATTRIBUTES_DELETED:
  435 + case TIMESERIES_UPDATED:
  436 + downlinkMsg = processTelemetryMessage(edgeEvent);
  437 + break;
  438 + case CREDENTIALS_REQUEST:
  439 + downlinkMsg = processCredentialsRequestMessage(edgeEvent);
  440 + break;
  441 + case ENTITY_MERGE_REQUEST:
  442 + downlinkMsg = processEntityMergeRequestMessage(edgeEvent);
  443 + break;
  444 + case RPC_CALL:
  445 + downlinkMsg = processRpcCallMsg(edgeEvent);
  446 + break;
  447 + }
  448 + } catch (Exception e) {
  449 + log.error("Exception during converting edge event to downlink msg", e);
323 450 }
324   - log.trace("[{}] processHandleMessages finished", this.sessionId);
  451 + return downlinkMsg;
325 452 }
326 453
327 454 private List<DownlinkMsg> convertToDownlinkMsgsPack(List<EdgeEvent> edgeEvents) {
328 455 List<DownlinkMsg> result = new ArrayList<>();
329 456 for (EdgeEvent edgeEvent : edgeEvents) {
330   - log.trace("[{}] Processing edge event [{}]", this.sessionId, edgeEvent);
331   - try {
332   - DownlinkMsg downlinkMsg = null;
333   - switch (edgeEvent.getAction()) {
334   - case UPDATED:
335   - case ADDED:
336   - case DELETED:
337   - case ASSIGNED_TO_EDGE:
338   - case UNASSIGNED_FROM_EDGE:
339   - case ALARM_ACK:
340   - case ALARM_CLEAR:
341   - case CREDENTIALS_UPDATED:
342   - case RELATION_ADD_OR_UPDATE:
343   - case RELATION_DELETED:
344   - case ASSIGNED_TO_CUSTOMER:
345   - case UNASSIGNED_FROM_CUSTOMER:
346   - downlinkMsg = processEntityMessage(edgeEvent, edgeEvent.getAction());
347   - break;
348   - case ATTRIBUTES_UPDATED:
349   - case POST_ATTRIBUTES:
350   - case ATTRIBUTES_DELETED:
351   - case TIMESERIES_UPDATED:
352   - downlinkMsg = processTelemetryMessage(edgeEvent);
353   - break;
354   - case CREDENTIALS_REQUEST:
355   - downlinkMsg = processCredentialsRequestMessage(edgeEvent);
356   - break;
357   - case ENTITY_MERGE_REQUEST:
358   - downlinkMsg = processEntityMergeRequestMessage(edgeEvent);
359   - break;
360   - case RPC_CALL:
361   - downlinkMsg = processRpcCallMsg(edgeEvent);
362   - break;
363   - }
364   - if (downlinkMsg != null) {
365   - result.add(downlinkMsg);
366   - }
367   - } catch (Exception e) {
368   - log.error("Exception during processing records from queue", e);
  457 + DownlinkMsg downlinkMsg = convertToDownlinkMsg(edgeEvent);
  458 + if (downlinkMsg != null) {
  459 + result.add(downlinkMsg);
369 460 }
370 461 }
371 462 return result;
... ... @@ -945,27 +1036,27 @@ public final class EdgeGrpcSession implements Closeable {
945 1036 }
946 1037 if (uplinkMsg.getRuleChainMetadataRequestMsgCount() > 0) {
947 1038 for (RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg : uplinkMsg.getRuleChainMetadataRequestMsgList()) {
948   - result.add(ctx.getSyncEdgeService().processRuleChainMetadataRequestMsg(edge.getTenantId(), edge, ruleChainMetadataRequestMsg));
  1039 + result.add(ctx.getEdgeRequestsService().processRuleChainMetadataRequestMsg(edge.getTenantId(), edge, ruleChainMetadataRequestMsg));
949 1040 }
950 1041 }
951 1042 if (uplinkMsg.getAttributesRequestMsgCount() > 0) {
952 1043 for (AttributesRequestMsg attributesRequestMsg : uplinkMsg.getAttributesRequestMsgList()) {
953   - result.add(ctx.getSyncEdgeService().processAttributesRequestMsg(edge.getTenantId(), edge, attributesRequestMsg));
  1044 + result.add(ctx.getEdgeRequestsService().processAttributesRequestMsg(edge.getTenantId(), edge, attributesRequestMsg));
954 1045 }
955 1046 }
956 1047 if (uplinkMsg.getRelationRequestMsgCount() > 0) {
957 1048 for (RelationRequestMsg relationRequestMsg : uplinkMsg.getRelationRequestMsgList()) {
958   - result.add(ctx.getSyncEdgeService().processRelationRequestMsg(edge.getTenantId(), edge, relationRequestMsg));
  1049 + result.add(ctx.getEdgeRequestsService().processRelationRequestMsg(edge.getTenantId(), edge, relationRequestMsg));
959 1050 }
960 1051 }
961 1052 if (uplinkMsg.getUserCredentialsRequestMsgCount() > 0) {
962 1053 for (UserCredentialsRequestMsg userCredentialsRequestMsg : uplinkMsg.getUserCredentialsRequestMsgList()) {
963   - result.add(ctx.getSyncEdgeService().processUserCredentialsRequestMsg(edge.getTenantId(), edge, userCredentialsRequestMsg));
  1054 + result.add(ctx.getEdgeRequestsService().processUserCredentialsRequestMsg(edge.getTenantId(), edge, userCredentialsRequestMsg));
964 1055 }
965 1056 }
966 1057 if (uplinkMsg.getDeviceCredentialsRequestMsgCount() > 0) {
967 1058 for (DeviceCredentialsRequestMsg deviceCredentialsRequestMsg : uplinkMsg.getDeviceCredentialsRequestMsgList()) {
968   - result.add(ctx.getSyncEdgeService().processDeviceCredentialsRequestMsg(edge.getTenantId(), edge, deviceCredentialsRequestMsg));
  1059 + result.add(ctx.getEdgeRequestsService().processDeviceCredentialsRequestMsg(edge.getTenantId(), edge, deviceCredentialsRequestMsg));
969 1060 }
970 1061 }
971 1062 if (uplinkMsg.getDeviceRpcCallMsgCount() > 0) {
... ... @@ -973,6 +1064,21 @@ public final class EdgeGrpcSession implements Closeable {
973 1064 result.add(ctx.getDeviceProcessor().processDeviceRpcCallResponseMsg(edge.getTenantId(), deviceRpcCallMsg));
974 1065 }
975 1066 }
  1067 + if (uplinkMsg.getDeviceProfileDevicesRequestMsgCount() > 0) {
  1068 + for (DeviceProfileDevicesRequestMsg deviceProfileDevicesRequestMsg : uplinkMsg.getDeviceProfileDevicesRequestMsgList()) {
  1069 + result.add(ctx.getEdgeRequestsService().processDeviceProfileDevicesRequestMsg(edge.getTenantId(), edge, deviceProfileDevicesRequestMsg));
  1070 + }
  1071 + }
  1072 + if (uplinkMsg.getWidgetBundleTypesRequestMsgCount() > 0) {
  1073 + for (WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg : uplinkMsg.getWidgetBundleTypesRequestMsgList()) {
  1074 + result.add(ctx.getEdgeRequestsService().processWidgetBundleTypesRequestMsg(edge.getTenantId(), edge, widgetBundleTypesRequestMsg));
  1075 + }
  1076 + }
  1077 + if (uplinkMsg.getEntityViewsRequestMsgCount() > 0) {
  1078 + for (EntityViewsRequestMsg entityViewRequestMsg : uplinkMsg.getEntityViewsRequestMsgList()) {
  1079 + result.add(ctx.getEdgeRequestsService().processEntityViewsRequestMsg(edge.getTenantId(), edge, entityViewRequestMsg));
  1080 + }
  1081 + }
976 1082 } catch (Exception e) {
977 1083 log.error("[{}] Can't process uplink msg [{}]", this.sessionId, uplinkMsg, e);
978 1084 }
... ...
... ... @@ -17,12 +17,15 @@ package org.thingsboard.server.service.edge.rpc;
17 17
18 18 import org.thingsboard.server.common.data.edge.Edge;
19 19 import org.thingsboard.server.common.data.id.EdgeId;
  20 +import org.thingsboard.server.common.data.id.TenantId;
20 21
21 22 public interface EdgeRpcService {
22 23
23   - void updateEdge(Edge edge);
  24 + void updateEdge(TenantId tenantId, Edge edge);
24 25
25   - void deleteEdge(EdgeId edgeId);
  26 + void deleteEdge(TenantId tenantId, EdgeId edgeId);
26 27
27   - void onEdgeEvent(EdgeId edgeId);
  28 + void onEdgeEvent(TenantId tenantId, EdgeId edgeId);
  29 +
  30 + void startSyncProcess(TenantId tenantId, EdgeId edgeId);
28 31 }
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  21 +import org.thingsboard.server.common.data.id.EdgeId;
  22 +import org.thingsboard.server.common.data.id.TenantId;
  23 +import org.thingsboard.server.common.data.page.PageData;
  24 +import org.thingsboard.server.common.data.page.PageLink;
  25 +
  26 +@AllArgsConstructor
  27 +@Slf4j
  28 +public class AdminSettingsEdgeEventFetcher implements EdgeEventFetcher {
  29 +
  30 + @Override
  31 + public PageLink getPageLink() {
  32 + return new PageLink(DEFAULT_LIMIT);
  33 + }
  34 +
  35 + @Override
  36 + public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink) {
  37 + return null;
  38 + }
  39 +
  40 +
  41 +//
  42 +// private void syncAdminSettings(TenantId tenantId, Edge edge) {
  43 +// log.trace("[{}] syncAdminSettings [{}]", tenantId, edge.getName());
  44 +// try {
  45 +// AdminSettings systemMailSettings = adminSettingsService.findAdminSettingsByKey(TenantId.SYS_TENANT_ID, "mail");
  46 +// saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(systemMailSettings));
  47 +// AdminSettings tenantMailSettings = convertToTenantAdminSettings(systemMailSettings.getKey(), (ObjectNode) systemMailSettings.getJsonValue());
  48 +// saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(tenantMailSettings));
  49 +// AdminSettings systemMailTemplates = loadMailTemplates();
  50 +// saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(systemMailTemplates));
  51 +// AdminSettings tenantMailTemplates = convertToTenantAdminSettings(systemMailTemplates.getKey(), (ObjectNode) systemMailTemplates.getJsonValue());
  52 +// saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(tenantMailTemplates));
  53 +// } catch (Exception e) {
  54 +// log.error("Can't load admin settings", e);
  55 +// }
  56 +// }
  57 +//
  58 +// private AdminSettings loadMailTemplates() throws Exception {
  59 +// Map<String, Object> mailTemplates = new HashMap<>();
  60 +// Pattern startPattern = Pattern.compile("<div class=\"content\".*?>");
  61 +// Pattern endPattern = Pattern.compile("<div class=\"footer\".*?>");
  62 +// File[] files = new DefaultResourceLoader().getResource("classpath:/templates/").getFile().listFiles();
  63 +// for (File file : files) {
  64 +// Map<String, String> mailTemplate = new HashMap<>();
  65 +// String name = validateName(file.getName());
  66 +// String stringTemplate = FileUtils.readFileToString(file, StandardCharsets.UTF_8);
  67 +// Matcher start = startPattern.matcher(stringTemplate);
  68 +// Matcher end = endPattern.matcher(stringTemplate);
  69 +// if (start.find() && end.find()) {
  70 +// String body = StringUtils.substringBetween(stringTemplate, start.group(), end.group()).replaceAll("\t", "");
  71 +// String subject = StringUtils.substringBetween(body, "<h2>", "</h2>");
  72 +// mailTemplate.put("subject", subject);
  73 +// mailTemplate.put("body", body);
  74 +// mailTemplates.put(name, mailTemplate);
  75 +// } else {
  76 +// log.error("Can't load mail template from file {}", file.getName());
  77 +// }
  78 +// }
  79 +// AdminSettings adminSettings = new AdminSettings();
  80 +// adminSettings.setId(new AdminSettingsId(Uuids.timeBased()));
  81 +// adminSettings.setKey("mailTemplates");
  82 +// adminSettings.setJsonValue(mapper.convertValue(mailTemplates, JsonNode.class));
  83 +// return adminSettings;
  84 +// }
  85 +//
  86 +// private String validateName(String name) throws Exception {
  87 +// StringBuilder nameBuilder = new StringBuilder();
  88 +// name = name.replace(".vm", "");
  89 +// String[] nameParts = name.split("\\.");
  90 +// if (nameParts.length >= 1) {
  91 +// nameBuilder.append(nameParts[0]);
  92 +// for (int i = 1; i < nameParts.length; i++) {
  93 +// String word = WordUtils.capitalize(nameParts[i]);
  94 +// nameBuilder.append(word);
  95 +// }
  96 +// return nameBuilder.toString();
  97 +// } else {
  98 +// throw new Exception("Error during filename validation");
  99 +// }
  100 +// }
  101 +//
  102 +// private AdminSettings convertToTenantAdminSettings(String key, ObjectNode jsonValue) {
  103 +// AdminSettings tenantMailSettings = new AdminSettings();
  104 +// jsonValue.put("useSystemMailSettings", true);
  105 +// tenantMailSettings.setJsonValue(jsonValue);
  106 +// tenantMailSettings.setKey(key);
  107 +// return tenantMailSettings;
  108 +// }
  109 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.thingsboard.server.common.data.asset.Asset;
  21 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  22 +import org.thingsboard.server.common.data.edge.EdgeEventActionType;
  23 +import org.thingsboard.server.common.data.edge.EdgeEventType;
  24 +import org.thingsboard.server.common.data.id.EdgeId;
  25 +import org.thingsboard.server.common.data.id.TenantId;
  26 +import org.thingsboard.server.common.data.page.PageData;
  27 +import org.thingsboard.server.common.data.page.PageLink;
  28 +import org.thingsboard.server.dao.asset.AssetService;
  29 +import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
  30 +
  31 +import java.util.ArrayList;
  32 +import java.util.List;
  33 +
  34 +@AllArgsConstructor
  35 +@Slf4j
  36 +public class AssetsEdgeEventFetcher implements EdgeEventFetcher {
  37 +
  38 + private final AssetService assetService;
  39 +
  40 + @Override
  41 + public PageLink getPageLink() {
  42 + return new PageLink(DEFAULT_LIMIT);
  43 + }
  44 +
  45 + @Override
  46 + public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink) {
  47 + log.trace("[{}] start fetching edge events [{}]", tenantId, edgeId);
  48 + PageData<Asset> pageData = assetService.findAssetsByTenantIdAndEdgeId(tenantId, edgeId, pageLink);
  49 + List<EdgeEvent> result = new ArrayList<>();
  50 + if (!pageData.getData().isEmpty()) {
  51 + for (Asset asset : pageData.getData()) {
  52 + result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edgeId, EdgeEventType.ASSET,
  53 + EdgeEventActionType.ADDED, asset.getId(), null));
  54 + }
  55 + }
  56 + return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
  57 + }
  58 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.thingsboard.server.common.data.User;
  21 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  22 +import org.thingsboard.server.common.data.edge.EdgeEventActionType;
  23 +import org.thingsboard.server.common.data.edge.EdgeEventType;
  24 +import org.thingsboard.server.common.data.id.EdgeId;
  25 +import org.thingsboard.server.common.data.id.TenantId;
  26 +import org.thingsboard.server.common.data.page.PageData;
  27 +import org.thingsboard.server.common.data.page.PageLink;
  28 +import org.thingsboard.server.common.data.widget.WidgetsBundle;
  29 +import org.thingsboard.server.dao.user.UserService;
  30 +import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
  31 +
  32 +import java.util.ArrayList;
  33 +import java.util.List;
  34 +
  35 +@Slf4j
  36 +public abstract class BaseUsersEdgeEventFetcher implements EdgeEventFetcher {
  37 +
  38 + @Override
  39 + public PageLink getPageLink() {
  40 + return new PageLink(DEFAULT_LIMIT);
  41 + }
  42 +
  43 + @Override
  44 + public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink) {
  45 + log.trace("[{}] start fetching edge events [{}]", tenantId, edgeId);
  46 + PageData<User> pageData = findUsers(tenantId, pageLink);
  47 + List<EdgeEvent> result = new ArrayList<>();
  48 + if (!pageData.getData().isEmpty()) {
  49 + for (User user : pageData.getData()) {
  50 + result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edgeId, EdgeEventType.USER,
  51 + EdgeEventActionType.ADDED, user.getId(), null));
  52 + }
  53 + }
  54 + return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
  55 + }
  56 +
  57 + protected abstract PageData<User> findUsers(TenantId tenantId, PageLink pageLink);
  58 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.extern.slf4j.Slf4j;
  19 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  20 +import org.thingsboard.server.common.data.edge.EdgeEventActionType;
  21 +import org.thingsboard.server.common.data.edge.EdgeEventType;
  22 +import org.thingsboard.server.common.data.id.EdgeId;
  23 +import org.thingsboard.server.common.data.id.TenantId;
  24 +import org.thingsboard.server.common.data.page.PageData;
  25 +import org.thingsboard.server.common.data.page.PageLink;
  26 +import org.thingsboard.server.common.data.widget.WidgetsBundle;
  27 +import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
  28 +
  29 +import java.util.ArrayList;
  30 +import java.util.List;
  31 +
  32 +@Slf4j
  33 +public abstract class BaseWidgetsBundlesEdgeEventFetcher implements EdgeEventFetcher {
  34 +
  35 + @Override
  36 + public PageLink getPageLink() {
  37 + return new PageLink(DEFAULT_LIMIT);
  38 + }
  39 +
  40 + @Override
  41 + public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink) {
  42 + log.trace("[{}] start fetching edge events [{}]", tenantId, edgeId);
  43 + PageData<WidgetsBundle> pageData = findWidgetsBundles(tenantId, pageLink);
  44 + List<EdgeEvent> result = new ArrayList<>();
  45 + if (!pageData.getData().isEmpty()) {
  46 + for (WidgetsBundle widgetsBundle : pageData.getData()) {
  47 + result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edgeId, EdgeEventType.WIDGETS_BUNDLE,
  48 + EdgeEventActionType.ADDED, widgetsBundle.getId(), null));
  49 + }
  50 + }
  51 + return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
  52 + }
  53 +
  54 + protected abstract PageData<WidgetsBundle> findWidgetsBundles(TenantId tenantId, PageLink pageLink);
  55 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import org.thingsboard.server.common.data.User;
  20 +import org.thingsboard.server.common.data.id.CustomerId;
  21 +import org.thingsboard.server.common.data.id.TenantId;
  22 +import org.thingsboard.server.common.data.page.PageData;
  23 +import org.thingsboard.server.common.data.page.PageLink;
  24 +import org.thingsboard.server.dao.user.UserService;
  25 +
  26 +@AllArgsConstructor
  27 +public class CustomerUsersEdgeEventFetcher extends BaseUsersEdgeEventFetcher {
  28 +
  29 + private final UserService userService;
  30 + private final CustomerId customerId;
  31 +
  32 + @Override
  33 + protected PageData<User> findUsers(TenantId tenantId, PageLink pageLink) {
  34 + return userService.findCustomerUsers(tenantId, customerId, pageLink);
  35 + }
  36 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.thingsboard.server.common.data.DashboardInfo;
  21 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  22 +import org.thingsboard.server.common.data.edge.EdgeEventActionType;
  23 +import org.thingsboard.server.common.data.edge.EdgeEventType;
  24 +import org.thingsboard.server.common.data.id.EdgeId;
  25 +import org.thingsboard.server.common.data.id.TenantId;
  26 +import org.thingsboard.server.common.data.page.PageData;
  27 +import org.thingsboard.server.common.data.page.PageLink;
  28 +import org.thingsboard.server.dao.dashboard.DashboardService;
  29 +import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
  30 +
  31 +import java.util.ArrayList;
  32 +import java.util.List;
  33 +
  34 +@AllArgsConstructor
  35 +@Slf4j
  36 +public class DashboardsEdgeEventFetcher implements EdgeEventFetcher {
  37 +
  38 + private final DashboardService dashboardService;
  39 +
  40 + @Override
  41 + public PageLink getPageLink() {
  42 + return new PageLink(DEFAULT_LIMIT);
  43 + }
  44 +
  45 + @Override
  46 + public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink) {
  47 + log.trace("[{}] start fetching edge events [{}]", tenantId, edgeId);
  48 + PageData<DashboardInfo> pageData = dashboardService.findDashboardsByTenantIdAndEdgeId(tenantId, edgeId, pageLink);
  49 + List<EdgeEvent> result = new ArrayList<>();
  50 + if (!pageData.getData().isEmpty()) {
  51 + for (DashboardInfo dashboardInfo : pageData.getData()) {
  52 + result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edgeId, EdgeEventType.DASHBOARD,
  53 + EdgeEventActionType.ADDED, dashboardInfo.getId(), null));
  54 + }
  55 + }
  56 + return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
  57 + }
  58 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.thingsboard.server.common.data.DeviceProfile;
  21 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  22 +import org.thingsboard.server.common.data.edge.EdgeEventActionType;
  23 +import org.thingsboard.server.common.data.edge.EdgeEventType;
  24 +import org.thingsboard.server.common.data.id.EdgeId;
  25 +import org.thingsboard.server.common.data.id.TenantId;
  26 +import org.thingsboard.server.common.data.page.PageData;
  27 +import org.thingsboard.server.common.data.page.PageLink;
  28 +import org.thingsboard.server.dao.device.DeviceProfileService;
  29 +import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
  30 +
  31 +import java.util.ArrayList;
  32 +import java.util.List;
  33 +
  34 +@AllArgsConstructor
  35 +@Slf4j
  36 +public class DeviceProfilesEdgeEventFetcher implements EdgeEventFetcher {
  37 +
  38 + private final DeviceProfileService deviceProfileService;
  39 +
  40 + @Override
  41 + public PageLink getPageLink() {
  42 + return new PageLink(DEFAULT_LIMIT);
  43 + }
  44 +
  45 + @Override
  46 + public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink) {
  47 + log.trace("[{}] start fetching edge events [{}]", tenantId, edgeId);
  48 + PageData<DeviceProfile> pageData = deviceProfileService.findDeviceProfiles(tenantId, pageLink);
  49 + List<EdgeEvent> result = new ArrayList<>();
  50 + if (!pageData.getData().isEmpty()) {
  51 + for (DeviceProfile deviceProfile : pageData.getData()) {
  52 + result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edgeId, EdgeEventType.DEVICE_PROFILE,
  53 + EdgeEventActionType.ADDED, deviceProfile.getId(), null));
  54 + }
  55 + }
  56 + return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
  57 + }
  58 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  19 +import org.thingsboard.server.common.data.id.EdgeId;
  20 +import org.thingsboard.server.common.data.id.TenantId;
  21 +import org.thingsboard.server.common.data.page.PageData;
  22 +import org.thingsboard.server.common.data.page.PageLink;
  23 +
  24 +public interface EdgeEventFetcher {
  25 +
  26 + final int DEFAULT_LIMIT = 100;
  27 +
  28 + PageLink getPageLink();
  29 +
  30 + PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink);
  31 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  20 +import org.thingsboard.server.common.data.id.EdgeId;
  21 +import org.thingsboard.server.common.data.id.TenantId;
  22 +import org.thingsboard.server.common.data.page.PageData;
  23 +import org.thingsboard.server.common.data.page.PageLink;
  24 +import org.thingsboard.server.common.data.page.SortOrder;
  25 +import org.thingsboard.server.common.data.page.TimePageLink;
  26 +import org.thingsboard.server.dao.edge.EdgeEventService;
  27 +
  28 +@AllArgsConstructor
  29 +public class GeneralEdgeEventFetcher implements EdgeEventFetcher {
  30 +
  31 + private final int maxReadRecordsCount;
  32 + private final Long queueStartTs;
  33 + private final EdgeEventService edgeEventService;
  34 +
  35 + @Override
  36 + public PageLink getPageLink() {
  37 + return new TimePageLink(
  38 + maxReadRecordsCount,
  39 + 0,
  40 + null,
  41 + new SortOrder("createdTime", SortOrder.Direction.ASC),
  42 + queueStartTs,
  43 + null);
  44 + }
  45 +
  46 + @Override
  47 + public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink) {
  48 + return edgeEventService.findEdgeEvents(tenantId, edgeId, (TimePageLink) pageLink, true);
  49 + }
  50 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.thingsboard.server.common.data.edge.EdgeEvent;
  21 +import org.thingsboard.server.common.data.edge.EdgeEventActionType;
  22 +import org.thingsboard.server.common.data.edge.EdgeEventType;
  23 +import org.thingsboard.server.common.data.id.EdgeId;
  24 +import org.thingsboard.server.common.data.id.TenantId;
  25 +import org.thingsboard.server.common.data.page.PageData;
  26 +import org.thingsboard.server.common.data.page.PageLink;
  27 +import org.thingsboard.server.common.data.rule.RuleChain;
  28 +import org.thingsboard.server.dao.rule.RuleChainService;
  29 +import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
  30 +
  31 +import java.util.ArrayList;
  32 +import java.util.List;
  33 +
  34 +@Slf4j
  35 +@AllArgsConstructor
  36 +public class RuleChainsEdgeEventFetcher implements EdgeEventFetcher {
  37 +
  38 + private final RuleChainService ruleChainService;
  39 +
  40 + @Override
  41 + public PageLink getPageLink() {
  42 + return new PageLink(DEFAULT_LIMIT);
  43 + }
  44 +
  45 + @Override
  46 + public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, EdgeId edgeId, PageLink pageLink) {
  47 + log.trace("[{}] start fetching edge events [{}]", tenantId, edgeId);
  48 + PageData<RuleChain> pageData = ruleChainService.findRuleChainsByTenantIdAndEdgeId(tenantId, edgeId, pageLink);
  49 + List<EdgeEvent> result = new ArrayList<>();
  50 + if (!pageData.getData().isEmpty()) {
  51 + for (RuleChain ruleChain : pageData.getData()) {
  52 + result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edgeId, EdgeEventType.RULE_CHAIN,
  53 + EdgeEventActionType.ADDED, ruleChain.getId(), null));
  54 + }
  55 + }
  56 + return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
  57 + }
  58 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.thingsboard.server.common.data.id.TenantId;
  21 +import org.thingsboard.server.common.data.page.PageData;
  22 +import org.thingsboard.server.common.data.page.PageLink;
  23 +import org.thingsboard.server.common.data.widget.WidgetsBundle;
  24 +import org.thingsboard.server.dao.widget.WidgetsBundleService;
  25 +
  26 +@Slf4j
  27 +@AllArgsConstructor
  28 +public class SystemWidgetsBundlesEdgeEventFetcher extends BaseWidgetsBundlesEdgeEventFetcher {
  29 +
  30 + private final WidgetsBundleService widgetsBundleService;
  31 +
  32 + @Override
  33 + protected PageData<WidgetsBundle> findWidgetsBundles(TenantId tenantId, PageLink pageLink) {
  34 + return widgetsBundleService.findSystemWidgetsBundlesByPageLink(tenantId, pageLink);
  35 + }
  36 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import org.thingsboard.server.common.data.User;
  20 +import org.thingsboard.server.common.data.id.TenantId;
  21 +import org.thingsboard.server.common.data.page.PageData;
  22 +import org.thingsboard.server.common.data.page.PageLink;
  23 +import org.thingsboard.server.dao.user.UserService;
  24 +
  25 +@AllArgsConstructor
  26 +public class TenantAdminUsersEdgeEventFetcher extends BaseUsersEdgeEventFetcher {
  27 +
  28 + private final UserService userService;
  29 +
  30 + @Override
  31 + protected PageData<User> findUsers(TenantId tenantId, PageLink pageLink) {
  32 + return userService.findTenantAdmins(tenantId, pageLink);
  33 + }
  34 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2021 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.service.edge.rpc.fetch;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.extern.slf4j.Slf4j;
  20 +import org.thingsboard.server.common.data.id.TenantId;
  21 +import org.thingsboard.server.common.data.page.PageData;
  22 +import org.thingsboard.server.common.data.page.PageLink;
  23 +import org.thingsboard.server.common.data.widget.WidgetsBundle;
  24 +import org.thingsboard.server.dao.widget.WidgetsBundleService;
  25 +
  26 +@Slf4j
  27 +@AllArgsConstructor
  28 +public class TenantWidgetsBundlesEdgeEventFetcher extends BaseWidgetsBundlesEdgeEventFetcher implements EdgeEventFetcher {
  29 +
  30 + private final WidgetsBundleService widgetsBundleService;
  31 +
  32 + @Override
  33 + protected PageData<WidgetsBundle> findWidgetsBundles(TenantId tenantId, PageLink pageLink) {
  34 + return widgetsBundleService.findAllTenantWidgetsBundlesByTenantIdAndPageLink(tenantId, pageLink);
  35 + }
  36 +}
... ...
application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/DefaultEdgeRequestsService.java renamed from application/src/main/java/org/thingsboard/server/service/edge/rpc/init/DefaultSyncEdgeService.java
... ... @@ -13,9 +13,8 @@
13 13 * See the License for the specific language governing permissions and
14 14 * limitations under the License.
15 15 */
16   -package org.thingsboard.server.service.edge.rpc.init;
  16 +package org.thingsboard.server.service.edge.rpc.sync;
17 17
18   -import com.datastax.oss.driver.api.core.uuid.Uuids;
19 18 import com.fasterxml.jackson.databind.JsonNode;
20 19 import com.fasterxml.jackson.databind.ObjectMapper;
21 20 import com.fasterxml.jackson.databind.node.ObjectNode;
... ... @@ -24,34 +23,27 @@ import com.google.common.util.concurrent.Futures;
24 23 import com.google.common.util.concurrent.ListenableFuture;
25 24 import com.google.common.util.concurrent.SettableFuture;
26 25 import lombok.extern.slf4j.Slf4j;
27   -import org.apache.commons.io.FileUtils;
28   -import org.apache.commons.lang3.StringUtils;
29   -import org.apache.commons.lang3.text.WordUtils;
30 26 import org.checkerframework.checker.nullness.qual.Nullable;
31 27 import org.springframework.beans.factory.annotation.Autowired;
32   -import org.springframework.core.io.DefaultResourceLoader;
33 28 import org.springframework.stereotype.Service;
34   -import org.thingsboard.server.common.data.AdminSettings;
35   -import org.thingsboard.server.common.data.DashboardInfo;
36 29 import org.thingsboard.server.common.data.Device;
37 30 import org.thingsboard.server.common.data.DeviceProfile;
38 31 import org.thingsboard.server.common.data.EdgeUtils;
39 32 import org.thingsboard.server.common.data.EntityType;
40 33 import org.thingsboard.server.common.data.EntityView;
41   -import org.thingsboard.server.common.data.User;
42   -import org.thingsboard.server.common.data.asset.Asset;
43 34 import org.thingsboard.server.common.data.edge.Edge;
44 35 import org.thingsboard.server.common.data.edge.EdgeEvent;
45 36 import org.thingsboard.server.common.data.edge.EdgeEventActionType;
46 37 import org.thingsboard.server.common.data.edge.EdgeEventType;
47   -import org.thingsboard.server.common.data.id.AdminSettingsId;
48 38 import org.thingsboard.server.common.data.id.DeviceId;
  39 +import org.thingsboard.server.common.data.id.DeviceProfileId;
49 40 import org.thingsboard.server.common.data.id.EdgeId;
50 41 import org.thingsboard.server.common.data.id.EntityId;
51 42 import org.thingsboard.server.common.data.id.EntityIdFactory;
52 43 import org.thingsboard.server.common.data.id.RuleChainId;
53 44 import org.thingsboard.server.common.data.id.TenantId;
54 45 import org.thingsboard.server.common.data.id.UserId;
  46 +import org.thingsboard.server.common.data.id.WidgetsBundleId;
55 47 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
56 48 import org.thingsboard.server.common.data.kv.DataType;
57 49 import org.thingsboard.server.common.data.page.PageData;
... ... @@ -59,44 +51,39 @@ import org.thingsboard.server.common.data.page.PageLink;
59 51 import org.thingsboard.server.common.data.relation.EntityRelation;
60 52 import org.thingsboard.server.common.data.relation.EntityRelationsQuery;
61 53 import org.thingsboard.server.common.data.relation.EntitySearchDirection;
  54 +import org.thingsboard.server.common.data.relation.RelationTypeGroup;
62 55 import org.thingsboard.server.common.data.relation.RelationsSearchParameters;
63   -import org.thingsboard.server.common.data.rule.RuleChain;
64 56 import org.thingsboard.server.common.data.widget.WidgetType;
65 57 import org.thingsboard.server.common.data.widget.WidgetsBundle;
66   -import org.thingsboard.server.dao.asset.AssetService;
67 58 import org.thingsboard.server.dao.attributes.AttributesService;
68   -import org.thingsboard.server.dao.dashboard.DashboardService;
69 59 import org.thingsboard.server.dao.device.DeviceProfileService;
70 60 import org.thingsboard.server.dao.device.DeviceService;
71 61 import org.thingsboard.server.dao.edge.EdgeEventService;
72 62 import org.thingsboard.server.dao.entityview.EntityViewService;
73 63 import org.thingsboard.server.dao.relation.RelationService;
74   -import org.thingsboard.server.dao.rule.RuleChainService;
75   -import org.thingsboard.server.dao.settings.AdminSettingsService;
76   -import org.thingsboard.server.dao.user.UserService;
77 64 import org.thingsboard.server.dao.widget.WidgetTypeService;
78 65 import org.thingsboard.server.dao.widget.WidgetsBundleService;
79 66 import org.thingsboard.server.gen.edge.AttributesRequestMsg;
80 67 import org.thingsboard.server.gen.edge.DeviceCredentialsRequestMsg;
  68 +import org.thingsboard.server.gen.edge.DeviceProfileDevicesRequestMsg;
  69 +import org.thingsboard.server.gen.edge.EntityViewsRequestMsg;
81 70 import org.thingsboard.server.gen.edge.RelationRequestMsg;
82 71 import org.thingsboard.server.gen.edge.RuleChainMetadataRequestMsg;
83 72 import org.thingsboard.server.gen.edge.UserCredentialsRequestMsg;
  73 +import org.thingsboard.server.gen.edge.WidgetBundleTypesRequestMsg;
  74 +import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
84 75 import org.thingsboard.server.service.executors.DbCallbackExecutorService;
85 76 import org.thingsboard.server.service.queue.TbClusterService;
86 77
87   -import java.io.File;
88   -import java.nio.charset.StandardCharsets;
89 78 import java.util.ArrayList;
90 79 import java.util.HashMap;
91 80 import java.util.List;
92 81 import java.util.Map;
93 82 import java.util.UUID;
94   -import java.util.regex.Matcher;
95   -import java.util.regex.Pattern;
96 83
97 84 @Service
98 85 @Slf4j
99   -public class DefaultSyncEdgeService implements SyncEdgeService {
  86 +public class DefaultEdgeRequestsService implements EdgeRequestsService {
100 87
101 88 private static final ObjectMapper mapper = new ObjectMapper();
102 89
... ... @@ -109,28 +96,16 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
109 96 private AttributesService attributesService;
110 97
111 98 @Autowired
112   - private RuleChainService ruleChainService;
113   -
114   - @Autowired
115 99 private RelationService relationService;
116 100
117 101 @Autowired
118 102 private DeviceService deviceService;
119 103
120 104 @Autowired
121   - private DeviceProfileService deviceProfileService;
122   -
123   - @Autowired
124   - private AssetService assetService;
125   -
126   - @Autowired
127 105 private EntityViewService entityViewService;
128 106
129 107 @Autowired
130   - private DashboardService dashboardService;
131   -
132   - @Autowired
133   - private UserService userService;
  108 + private DeviceProfileService deviceProfileService;
134 109
135 110 @Autowired
136 111 private WidgetsBundleService widgetsBundleService;
... ... @@ -139,316 +114,20 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
139 114 private WidgetTypeService widgetTypeService;
140 115
141 116 @Autowired
142   - private AdminSettingsService adminSettingsService;
143   -
144   - @Autowired
145 117 private DbCallbackExecutorService dbCallbackExecutorService;
146 118
147 119 @Autowired
148 120 private TbClusterService tbClusterService;
149 121
150 122 @Override
151   - public void sync(TenantId tenantId, Edge edge) {
152   - log.trace("[{}][{}] Staring edge sync process", tenantId, edge.getId());
153   - try {
154   - syncWidgetsBundles(tenantId, edge);
155   - // TODO: voba - implement this functionality
156   - // syncAdminSettings(edge);
157   - syncDeviceProfiles(tenantId, edge);
158   - syncRuleChains(tenantId, edge);
159   - syncUsers(tenantId, edge);
160   - syncAssets(tenantId, edge);
161   - syncEntityViews(tenantId, edge);
162   - syncDashboards(tenantId, edge);
163   - syncWidgetsTypes(tenantId, edge);
164   - syncDevices(tenantId, edge);
165   - } catch (Exception e) {
166   - log.error("[{}][{}] Exception during sync process", tenantId, edge.getId(), e);
167   - }
168   - }
169   -
170   - private void syncRuleChains(TenantId tenantId, Edge edge) {
171   - log.trace("[{}] syncRuleChains [{}]", tenantId, edge.getName());
172   - try {
173   - PageLink pageLink = new PageLink(DEFAULT_LIMIT);
174   - PageData<RuleChain> pageData;
175   - do {
176   - pageData = ruleChainService.findRuleChainsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
177   - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
178   - log.trace("[{}] [{}] rule chains(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
179   - for (RuleChain ruleChain : pageData.getData()) {
180   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN, EdgeEventActionType.ADDED, ruleChain.getId(), null);
181   - }
182   - if (pageData.hasNext()) {
183   - pageLink = pageLink.nextPageLink();
184   - }
185   - }
186   - } while (pageData != null && pageData.hasNext());
187   - } catch (Exception e) {
188   - log.error("Exception during loading edge rule chain(s) on sync!", e);
189   - }
190   - }
191   -
192   - private void syncDevices(TenantId tenantId, Edge edge) {
193   - log.trace("[{}] syncDevices [{}]", tenantId, edge.getName());
194   - try {
195   - PageLink pageLink = new PageLink(DEFAULT_LIMIT);
196   - PageData<Device> pageData;
197   - do {
198   - pageData = deviceService.findDevicesByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
199   - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
200   - log.trace("[{}] [{}] device(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
201   - for (Device device : pageData.getData()) {
202   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ADDED, device.getId(), null);
203   - }
204   - if (pageData.hasNext()) {
205   - pageLink = pageLink.nextPageLink();
206   - }
207   - }
208   - } while (pageData != null && pageData.hasNext());
209   - } catch (Exception e) {
210   - log.error("Exception during loading edge device(s) on sync!", e);
211   - }
212   - }
213   -
214   - private void syncDeviceProfiles(TenantId tenantId, Edge edge) {
215   - log.trace("[{}] syncDeviceProfiles [{}]", tenantId, edge.getName());
216   - try {
217   - PageLink pageLink = new PageLink(DEFAULT_LIMIT);
218   - PageData<DeviceProfile> pageData;
219   - do {
220   - pageData = deviceProfileService.findDeviceProfiles(tenantId, pageLink);
221   - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
222   - log.trace("[{}] [{}] user(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
223   - for (DeviceProfile deviceProfile : pageData.getData()) {
224   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE_PROFILE, EdgeEventActionType.ADDED, deviceProfile.getId(), null);
225   - }
226   - if (pageData.hasNext()) {
227   - pageLink = pageLink.nextPageLink();
228   - }
229   - }
230   - } while (pageData != null && pageData.hasNext());
231   - } catch (Exception e) {
232   - log.error("Exception during loading device profile(s) on sync!", e);
233   - }
234   - }
235   -
236   - private void syncAssets(TenantId tenantId, Edge edge) {
237   - log.trace("[{}] syncAssets [{}]", tenantId, edge.getName());
238   - try {
239   - PageLink pageLink = new PageLink(DEFAULT_LIMIT);
240   - PageData<Asset> pageData;
241   - do {
242   - pageData = assetService.findAssetsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
243   - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
244   - log.trace("[{}] [{}] asset(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
245   - for (Asset asset : pageData.getData()) {
246   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ASSET, EdgeEventActionType.ADDED, asset.getId(), null);
247   - }
248   - if (pageData.hasNext()) {
249   - pageLink = pageLink.nextPageLink();
250   - }
251   - }
252   - } while (pageData != null && pageData.hasNext());
253   - } catch (Exception e) {
254   - log.error("Exception during loading edge asset(s) on sync!", e);
255   - }
256   - }
257   -
258   - private void syncEntityViews(TenantId tenantId, Edge edge) {
259   - log.trace("[{}] syncEntityViews [{}]", tenantId, edge.getName());
260   - try {
261   - PageLink pageLink = new PageLink(DEFAULT_LIMIT);
262   - PageData<EntityView> pageData;
263   - do {
264   - pageData = entityViewService.findEntityViewsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
265   - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
266   - log.trace("[{}] [{}] entity view(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
267   - for (EntityView entityView : pageData.getData()) {
268   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ENTITY_VIEW, EdgeEventActionType.ADDED, entityView.getId(), null);
269   - }
270   - if (pageData.hasNext()) {
271   - pageLink = pageLink.nextPageLink();
272   - }
273   - }
274   - } while (pageData != null && pageData.hasNext());
275   - } catch (Exception e) {
276   - log.error("Exception during loading edge entity view(s) on sync!", e);
277   - }
278   - }
279   -
280   - private void syncDashboards(TenantId tenantId, Edge edge) {
281   - log.trace("[{}] syncDashboards [{}]", tenantId, edge.getName());
282   - try {
283   - PageLink pageLink = new PageLink(DEFAULT_LIMIT);
284   - PageData<DashboardInfo> pageData;
285   - do {
286   - pageData = dashboardService.findDashboardsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
287   - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
288   - log.trace("[{}] [{}] dashboard(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
289   - for (DashboardInfo dashboardInfo : pageData.getData()) {
290   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DASHBOARD, EdgeEventActionType.ADDED, dashboardInfo.getId(), null);
291   - }
292   - if (pageData.hasNext()) {
293   - pageLink = pageLink.nextPageLink();
294   - }
295   - }
296   - } while (pageData != null && pageData.hasNext());
297   - } catch (Exception e) {
298   - log.error("Exception during loading edge dashboard(s) on sync!", e);
299   - }
300   - }
301   -
302   - private void syncUsers(TenantId tenantId, Edge edge) {
303   - log.trace("[{}] syncUsers [{}]", tenantId, edge.getName());
304   - try {
305   - PageLink pageLink = new PageLink(DEFAULT_LIMIT);
306   - PageData<User> pageData;
307   - do {
308   - pageData = userService.findTenantAdmins(tenantId, pageLink);
309   - pushUsersToEdge(tenantId, pageData, edge);
310   - if (pageData.hasNext()) {
311   - pageLink = pageLink.nextPageLink();
312   - }
313   - } while (pageData.hasNext());
314   - syncCustomerUsers(tenantId, edge);
315   - } catch (Exception e) {
316   - log.error("Exception during loading edge user(s) on sync!", e);
317   - }
318   - }
319   -
320   - private void syncCustomerUsers(TenantId tenantId, Edge edge) {
321   - if (edge.getCustomerId() != null && !EntityId.NULL_UUID.equals(edge.getCustomerId().getId())) {
322   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.CUSTOMER, EdgeEventActionType.ADDED, edge.getCustomerId(), null);
323   - PageLink pageLink = new PageLink(DEFAULT_LIMIT);
324   - PageData<User> pageData;
325   - do {
326   - pageData = userService.findCustomerUsers(tenantId, edge.getCustomerId(), pageLink);
327   - pushUsersToEdge(tenantId, pageData, edge);
328   - if (pageData != null && pageData.hasNext()) {
329   - pageLink = pageLink.nextPageLink();
330   - }
331   - } while (pageData != null && pageData.hasNext());
332   - }
333   - }
334   -
335   - private void pushUsersToEdge(TenantId tenantId, PageData<User> pageData, Edge edge) {
336   - if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
337   - log.trace("[{}] [{}] user(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
338   - for (User user : pageData.getData()) {
339   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER, EdgeEventActionType.ADDED, user.getId(), null);
340   - }
341   - }
342   - }
343   -
344   - private void syncWidgetsBundles(TenantId tenantId, Edge edge) {
345   - log.trace("[{}] syncWidgetsBundles [{}]", tenantId, edge.getName());
346   - List<WidgetsBundle> widgetsBundlesToPush = new ArrayList<>();
347   - widgetsBundlesToPush.addAll(widgetsBundleService.findAllTenantWidgetsBundlesByTenantId(tenantId));
348   - widgetsBundlesToPush.addAll(widgetsBundleService.findSystemWidgetsBundles(tenantId));
349   - try {
350   - for (WidgetsBundle widgetsBundle : widgetsBundlesToPush) {
351   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.WIDGETS_BUNDLE, EdgeEventActionType.ADDED, widgetsBundle.getId(), null);
352   - }
353   - } catch (Exception e) {
354   - log.error("Exception during loading widgets bundle(s) on sync!", e);
355   - }
356   - }
357   -
358   - private void syncWidgetsTypes(TenantId tenantId, Edge edge) {
359   - log.trace("[{}] syncWidgetsTypes [{}]", tenantId, edge.getName());
360   - List<WidgetsBundle> widgetsBundlesToPush = new ArrayList<>();
361   - widgetsBundlesToPush.addAll(widgetsBundleService.findAllTenantWidgetsBundlesByTenantId(tenantId));
362   - widgetsBundlesToPush.addAll(widgetsBundleService.findSystemWidgetsBundles(tenantId));
363   - try {
364   - for (WidgetsBundle widgetsBundle : widgetsBundlesToPush) {
365   - List<WidgetType> widgetTypesToPush =
366   - widgetTypeService.findWidgetTypesByTenantIdAndBundleAlias(widgetsBundle.getTenantId(), widgetsBundle.getAlias());
367   - for (WidgetType widgetType : widgetTypesToPush) {
368   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.WIDGET_TYPE, EdgeEventActionType.ADDED, widgetType.getId(), null);
369   - }
370   - }
371   - } catch (Exception e) {
372   - log.error("Exception during loading widgets type(s) on sync!", e);
373   - }
374   - }
375   -
376   - private void syncAdminSettings(TenantId tenantId, Edge edge) {
377   - log.trace("[{}] syncAdminSettings [{}]", tenantId, edge.getName());
378   - try {
379   - AdminSettings systemMailSettings = adminSettingsService.findAdminSettingsByKey(TenantId.SYS_TENANT_ID, "mail");
380   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(systemMailSettings));
381   - AdminSettings tenantMailSettings = convertToTenantAdminSettings(systemMailSettings.getKey(), (ObjectNode) systemMailSettings.getJsonValue());
382   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(tenantMailSettings));
383   - AdminSettings systemMailTemplates = loadMailTemplates();
384   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(systemMailTemplates));
385   - AdminSettings tenantMailTemplates = convertToTenantAdminSettings(systemMailTemplates.getKey(), (ObjectNode) systemMailTemplates.getJsonValue());
386   - saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ADMIN_SETTINGS, EdgeEventActionType.UPDATED, null, mapper.valueToTree(tenantMailTemplates));
387   - } catch (Exception e) {
388   - log.error("Can't load admin settings", e);
389   - }
390   - }
391   -
392   - private AdminSettings loadMailTemplates() throws Exception {
393   - Map<String, Object> mailTemplates = new HashMap<>();
394   - Pattern startPattern = Pattern.compile("<div class=\"content\".*?>");
395   - Pattern endPattern = Pattern.compile("<div class=\"footer\".*?>");
396   - File[] files = new DefaultResourceLoader().getResource("classpath:/templates/").getFile().listFiles();
397   - for (File file : files) {
398   - Map<String, String> mailTemplate = new HashMap<>();
399   - String name = validateName(file.getName());
400   - String stringTemplate = FileUtils.readFileToString(file, StandardCharsets.UTF_8);
401   - Matcher start = startPattern.matcher(stringTemplate);
402   - Matcher end = endPattern.matcher(stringTemplate);
403   - if (start.find() && end.find()) {
404   - String body = StringUtils.substringBetween(stringTemplate, start.group(), end.group()).replaceAll("\t", "");
405   - String subject = StringUtils.substringBetween(body, "<h2>", "</h2>");
406   - mailTemplate.put("subject", subject);
407   - mailTemplate.put("body", body);
408   - mailTemplates.put(name, mailTemplate);
409   - } else {
410   - log.error("Can't load mail template from file {}", file.getName());
411   - }
412   - }
413   - AdminSettings adminSettings = new AdminSettings();
414   - adminSettings.setId(new AdminSettingsId(Uuids.timeBased()));
415   - adminSettings.setKey("mailTemplates");
416   - adminSettings.setJsonValue(mapper.convertValue(mailTemplates, JsonNode.class));
417   - return adminSettings;
418   - }
419   -
420   - private String validateName(String name) throws Exception {
421   - StringBuilder nameBuilder = new StringBuilder();
422   - name = name.replace(".vm", "");
423   - String[] nameParts = name.split("\\.");
424   - if (nameParts.length >= 1) {
425   - nameBuilder.append(nameParts[0]);
426   - for (int i = 1; i < nameParts.length; i++) {
427   - String word = WordUtils.capitalize(nameParts[i]);
428   - nameBuilder.append(word);
429   - }
430   - return nameBuilder.toString();
431   - } else {
432   - throw new Exception("Error during filename validation");
433   - }
434   - }
435   -
436   - private AdminSettings convertToTenantAdminSettings(String key, ObjectNode jsonValue) {
437   - AdminSettings tenantMailSettings = new AdminSettings();
438   - jsonValue.put("useSystemMailSettings", true);
439   - tenantMailSettings.setJsonValue(jsonValue);
440   - tenantMailSettings.setKey(key);
441   - return tenantMailSettings;
442   - }
443   -
444   - @Override
445 123 public ListenableFuture<Void> processRuleChainMetadataRequestMsg(TenantId tenantId, Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg) {
446 124 log.trace("[{}] processRuleChainMetadataRequestMsg [{}][{}]", tenantId, edge.getName(), ruleChainMetadataRequestMsg);
447 125 SettableFuture<Void> futureToSet = SettableFuture.create();
448 126 if (ruleChainMetadataRequestMsg.getRuleChainIdMSB() != 0 && ruleChainMetadataRequestMsg.getRuleChainIdLSB() != 0) {
449 127 RuleChainId ruleChainId =
450 128 new RuleChainId(new UUID(ruleChainMetadataRequestMsg.getRuleChainIdMSB(), ruleChainMetadataRequestMsg.getRuleChainIdLSB()));
451   - ListenableFuture<EdgeEvent> future = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN_METADATA, EdgeEventActionType.ADDED, ruleChainId, null);
  129 + ListenableFuture<EdgeEvent> future = saveEdgeEvent(tenantId, edge.getId(),
  130 + EdgeEventType.RULE_CHAIN_METADATA, EdgeEventActionType.ADDED, ruleChainId, null);
452 131 Futures.addCallback(future, new FutureCallback<EdgeEvent>() {
453 132 @Override
454 133 public void onSuccess(@Nullable EdgeEvent result) {
... ... @@ -584,7 +263,8 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
584 263 return futureToSet;
585 264 }
586 265
587   - private ListenableFuture<List<EntityRelation>> findRelationByQuery(TenantId tenantId, Edge edge, EntityId entityId, EntitySearchDirection direction) {
  266 + private ListenableFuture<List<EntityRelation>> findRelationByQuery(TenantId tenantId, Edge edge,
  267 + EntityId entityId, EntitySearchDirection direction) {
588 268 EntityRelationsQuery query = new EntityRelationsQuery();
589 269 query.setParameters(new RelationsSearchParameters(entityId, direction, -1, false));
590 270 return relationService.findByQuery(tenantId, query);
... ... @@ -596,7 +276,8 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
596 276 SettableFuture<Void> futureToSet = SettableFuture.create();
597 277 if (deviceCredentialsRequestMsg.getDeviceIdMSB() != 0 && deviceCredentialsRequestMsg.getDeviceIdLSB() != 0) {
598 278 DeviceId deviceId = new DeviceId(new UUID(deviceCredentialsRequestMsg.getDeviceIdMSB(), deviceCredentialsRequestMsg.getDeviceIdLSB()));
599   - ListenableFuture<EdgeEvent> future = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.CREDENTIALS_UPDATED, deviceId, null);
  279 + ListenableFuture<EdgeEvent> future = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE,
  280 + EdgeEventActionType.CREDENTIALS_UPDATED, deviceId, null);
600 281 Futures.addCallback(future, new FutureCallback<EdgeEvent>() {
601 282 @Override
602 283 public void onSuccess(@Nullable EdgeEvent result) {
... ... @@ -619,8 +300,9 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
619 300 SettableFuture<Void> futureToSet = SettableFuture.create();
620 301 if (userCredentialsRequestMsg.getUserIdMSB() != 0 && userCredentialsRequestMsg.getUserIdLSB() != 0) {
621 302 UserId userId = new UserId(new UUID(userCredentialsRequestMsg.getUserIdMSB(), userCredentialsRequestMsg.getUserIdLSB()));
622   - ListenableFuture<EdgeEvent> future = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER, EdgeEventActionType.CREDENTIALS_UPDATED, userId, null);
623   - Futures.addCallback(future, new FutureCallback<EdgeEvent>() {
  303 + ListenableFuture<EdgeEvent> future = saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER,
  304 + EdgeEventActionType.CREDENTIALS_UPDATED, userId, null);
  305 + Futures.addCallback(future, new FutureCallback<>() {
624 306 @Override
625 307 public void onSuccess(@Nullable EdgeEvent result) {
626 308 futureToSet.set(null);
... ... @@ -636,6 +318,139 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
636 318 return futureToSet;
637 319 }
638 320
  321 + @Override
  322 + public ListenableFuture<Void> processDeviceProfileDevicesRequestMsg(TenantId tenantId, Edge edge, DeviceProfileDevicesRequestMsg deviceProfileDevicesRequestMsg) {
  323 + log.trace("[{}] processDeviceProfileDevicesRequestMsg [{}][{}]", tenantId, edge.getName(), deviceProfileDevicesRequestMsg);
  324 + SettableFuture<Void> futureToSet = SettableFuture.create();
  325 + if (deviceProfileDevicesRequestMsg.getDeviceProfileIdMSB() != 0 && deviceProfileDevicesRequestMsg.getDeviceProfileIdLSB() != 0) {
  326 + DeviceProfileId deviceProfileId = new DeviceProfileId(new UUID(deviceProfileDevicesRequestMsg.getDeviceProfileIdMSB(), deviceProfileDevicesRequestMsg.getDeviceProfileIdLSB()));
  327 + DeviceProfile deviceProfileById = deviceProfileService.findDeviceProfileById(tenantId, deviceProfileId);
  328 + List<ListenableFuture<EdgeEvent>> futures;
  329 + if (deviceProfileById != null) {
  330 + futures = syncDevices(tenantId, edge, deviceProfileById.getName());
  331 + } else {
  332 + futures = new ArrayList<>();
  333 + }
  334 + Futures.addCallback(Futures.allAsList(futures), new FutureCallback<>() {
  335 + @Override
  336 + public void onSuccess(@Nullable List<EdgeEvent> result) {
  337 + futureToSet.set(null);
  338 + }
  339 +
  340 + @Override
  341 + public void onFailure(Throwable t) {
  342 + log.error("Can't sync devices by device profile [{}]", deviceProfileDevicesRequestMsg, t);
  343 + futureToSet.setException(t);
  344 + }
  345 + }, dbCallbackExecutorService);
  346 + }
  347 + return futureToSet;
  348 + }
  349 +
  350 + private List<ListenableFuture<EdgeEvent>> syncDevices(TenantId tenantId, Edge edge, String deviceType) {
  351 + List<ListenableFuture<EdgeEvent>> futures = new ArrayList<>();
  352 + log.trace("[{}] syncDevices [{}][{}]", tenantId, edge.getName(), deviceType);
  353 + try {
  354 + PageLink pageLink = new PageLink(DEFAULT_LIMIT);
  355 + PageData<Device> pageData;
  356 + do {
  357 + pageData = deviceService.findDevicesByTenantIdAndEdgeIdAndType(tenantId, edge.getId(), deviceType, pageLink);
  358 + if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
  359 + log.trace("[{}] [{}] device(s) are going to be pushed to edge.", edge.getId(), pageData.getData().size());
  360 + for (Device device : pageData.getData()) {
  361 + futures.add(saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE, EdgeEventActionType.ADDED, device.getId(), null));
  362 + }
  363 + if (pageData.hasNext()) {
  364 + pageLink = pageLink.nextPageLink();
  365 + }
  366 + }
  367 + } while (pageData != null && pageData.hasNext());
  368 + } catch (Exception e) {
  369 + log.error("Exception during loading edge device(s) on sync!", e);
  370 + }
  371 + return futures;
  372 + }
  373 +
  374 + @Override
  375 + public ListenableFuture<Void> processWidgetBundleTypesRequestMsg(TenantId tenantId, Edge edge,
  376 + WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg) {
  377 + log.trace("[{}] processWidgetBundleTypesRequestMsg [{}][{}]", tenantId, edge.getName(), widgetBundleTypesRequestMsg);
  378 + SettableFuture<Void> futureToSet = SettableFuture.create();
  379 + if (widgetBundleTypesRequestMsg.getWidgetBundleIdMSB() != 0 && widgetBundleTypesRequestMsg.getWidgetBundleIdLSB() != 0) {
  380 + WidgetsBundleId widgetsBundleId = new WidgetsBundleId(new UUID(widgetBundleTypesRequestMsg.getWidgetBundleIdMSB(), widgetBundleTypesRequestMsg.getWidgetBundleIdLSB()));
  381 + WidgetsBundle widgetsBundleById = widgetsBundleService.findWidgetsBundleById(tenantId, widgetsBundleId);
  382 + List<ListenableFuture<EdgeEvent>> futures = new ArrayList<>();
  383 + if (widgetsBundleById != null) {
  384 + List<WidgetType> widgetTypesToPush =
  385 + widgetTypeService.findWidgetTypesByTenantIdAndBundleAlias(widgetsBundleById.getTenantId(), widgetsBundleById.getAlias());
  386 +
  387 + for (WidgetType widgetType : widgetTypesToPush) {
  388 + futures.add(saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.WIDGET_TYPE, EdgeEventActionType.ADDED, widgetType.getId(), null));
  389 + }
  390 + }
  391 + Futures.addCallback(Futures.allAsList(futures), new FutureCallback<>() {
  392 + @Override
  393 + public void onSuccess(@Nullable List<EdgeEvent> result) {
  394 + futureToSet.set(null);
  395 + }
  396 +
  397 + @Override
  398 + public void onFailure(Throwable t) {
  399 + log.error("Can't sync widget types by widget bundle [{}]", widgetBundleTypesRequestMsg, t);
  400 + futureToSet.setException(t);
  401 + }
  402 + }, dbCallbackExecutorService);
  403 + }
  404 + return futureToSet;
  405 + }
  406 +
  407 + @Override
  408 + public ListenableFuture<Void> processEntityViewsRequestMsg(TenantId tenantId, Edge edge, EntityViewsRequestMsg entityViewsRequestMsg) {
  409 + log.trace("[{}] processEntityViewsRequestMsg [{}][{}]", tenantId, edge.getName(), entityViewsRequestMsg);
  410 + EntityId entityId = EntityIdFactory.getByTypeAndUuid(
  411 + EntityType.valueOf(entityViewsRequestMsg.getEntityType()),
  412 + new UUID(entityViewsRequestMsg.getEntityIdMSB(), entityViewsRequestMsg.getEntityIdLSB()));
  413 + SettableFuture<Void> futureToSet = SettableFuture.create();
  414 + Futures.addCallback(entityViewService.findEntityViewsByTenantIdAndEntityIdAsync(tenantId, entityId), new FutureCallback<>() {
  415 + @Override
  416 + public void onSuccess(@Nullable List<EntityView> entityViews) {
  417 + try {
  418 + if (entityViews != null && !entityViews.isEmpty()) {
  419 + for (EntityView entityView : entityViews) {
  420 + Futures.addCallback(relationService.checkRelation(tenantId, edge.getId(), entityView.getId(),
  421 + EntityRelation.CONTAINS_TYPE, RelationTypeGroup.EDGE), new FutureCallback<>() {
  422 + @Override
  423 + public void onSuccess(@Nullable Boolean result) {
  424 + if (Boolean.TRUE.equals(result)) {
  425 + saveEdgeEvent(tenantId, edge.getId(), EdgeEventType.ENTITY_VIEW,
  426 + EdgeEventActionType.ADDED, entityView.getId(), null);
  427 + }
  428 + }
  429 +
  430 + @Override
  431 + public void onFailure(Throwable t) {
  432 + log.error("Exception during loading relation [{}] to edge on sync!", t, t);
  433 + futureToSet.setException(t);
  434 + }
  435 + }, dbCallbackExecutorService);
  436 + }
  437 + }
  438 + futureToSet.set(null);
  439 + } catch (Exception e) {
  440 + log.error("Exception during loading relation(s) to edge on sync!", e);
  441 + futureToSet.setException(e);
  442 + }
  443 + }
  444 +
  445 + @Override
  446 + public void onFailure(Throwable t) {
  447 + log.error("[{}] Can't find entity views by entity id [{}]", tenantId, entityId, t);
  448 + futureToSet.setException(t);
  449 + }
  450 + }, dbCallbackExecutorService);
  451 + return futureToSet;
  452 + }
  453 +
639 454 private ListenableFuture<EdgeEvent> saveEdgeEvent(TenantId tenantId,
640 455 EdgeId edgeId,
641 456 EdgeEventType type,
... ... @@ -645,17 +460,10 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
645 460 log.trace("Pushing edge event to edge queue. tenantId [{}], edgeId [{}], type [{}], action[{}], entityId [{}], body [{}]",
646 461 tenantId, edgeId, type, action, entityId, body);
647 462
648   - EdgeEvent edgeEvent = new EdgeEvent();
649   - edgeEvent.setTenantId(tenantId);
650   - edgeEvent.setEdgeId(edgeId);
651   - edgeEvent.setType(type);
652   - edgeEvent.setAction(action);
653   - if (entityId != null) {
654   - edgeEvent.setEntityId(entityId.getId());
655   - }
656   - edgeEvent.setBody(body);
  463 + EdgeEvent edgeEvent = EdgeEventUtils.constructEdgeEvent(tenantId, edgeId, type, action, entityId, body);
  464 +
657 465 ListenableFuture<EdgeEvent> future = edgeEventService.saveAsync(edgeEvent);
658   - Futures.addCallback(future, new FutureCallback<EdgeEvent>() {
  466 + Futures.addCallback(future, new FutureCallback<>() {
659 467 @Override
660 468 public void onSuccess(@Nullable EdgeEvent result) {
661 469 tbClusterService.onEdgeEventUpdate(tenantId, edgeId);
... ... @@ -668,4 +476,5 @@ public class DefaultSyncEdgeService implements SyncEdgeService {
668 476 }, dbCallbackExecutorService);
669 477 return future;
670 478 }
  479 +
671 480 }
... ...
application/src/main/java/org/thingsboard/server/service/edge/rpc/sync/EdgeRequestsService.java renamed from application/src/main/java/org/thingsboard/server/service/edge/rpc/init/SyncEdgeService.java
... ... @@ -13,20 +13,21 @@
13 13 * See the License for the specific language governing permissions and
14 14 * limitations under the License.
15 15 */
16   -package org.thingsboard.server.service.edge.rpc.init;
  16 +package org.thingsboard.server.service.edge.rpc.sync;
17 17
18 18 import com.google.common.util.concurrent.ListenableFuture;
19 19 import org.thingsboard.server.common.data.edge.Edge;
20 20 import org.thingsboard.server.common.data.id.TenantId;
21 21 import org.thingsboard.server.gen.edge.AttributesRequestMsg;
22 22 import org.thingsboard.server.gen.edge.DeviceCredentialsRequestMsg;
  23 +import org.thingsboard.server.gen.edge.DeviceProfileDevicesRequestMsg;
  24 +import org.thingsboard.server.gen.edge.EntityViewsRequestMsg;
23 25 import org.thingsboard.server.gen.edge.RelationRequestMsg;
24 26 import org.thingsboard.server.gen.edge.RuleChainMetadataRequestMsg;
25 27 import org.thingsboard.server.gen.edge.UserCredentialsRequestMsg;
  28 +import org.thingsboard.server.gen.edge.WidgetBundleTypesRequestMsg;
26 29
27   -public interface SyncEdgeService {
28   -
29   - void sync(TenantId tenantId, Edge edge);
  30 +public interface EdgeRequestsService {
30 31
31 32 ListenableFuture<Void> processRuleChainMetadataRequestMsg(TenantId tenantId, Edge edge, RuleChainMetadataRequestMsg ruleChainMetadataRequestMsg);
32 33
... ... @@ -37,4 +38,10 @@ public interface SyncEdgeService {
37 38 ListenableFuture<Void> processDeviceCredentialsRequestMsg(TenantId tenantId, Edge edge, DeviceCredentialsRequestMsg deviceCredentialsRequestMsg);
38 39
39 40 ListenableFuture<Void> processUserCredentialsRequestMsg(TenantId tenantId, Edge edge, UserCredentialsRequestMsg userCredentialsRequestMsg);
  41 +
  42 + ListenableFuture<Void> processDeviceProfileDevicesRequestMsg(TenantId tenantId, Edge edge, DeviceProfileDevicesRequestMsg deviceProfileDevicesRequestMsg);
  43 +
  44 + ListenableFuture<Void> processWidgetBundleTypesRequestMsg(TenantId tenantId, Edge edge, WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg);
  45 +
  46 + ListenableFuture<Void> processEntityViewsRequestMsg(TenantId tenantId, Edge edge, EntityViewsRequestMsg entityViewsRequestMsg);
40 47 }
... ...
... ... @@ -27,6 +27,7 @@
27 27
28 28 <logger name="org.thingsboard.server" level="INFO" />
29 29 <logger name="org.thingsboard.server.transport.snmp" level="TRACE" />
  30 +<!-- <logger name="org.thingsboard.server.service.edge.rpc" level="TRACE" />-->
30 31
31 32 <!-- <logger name="org.thingsboard.server.service.queue" level="TRACE" />-->
32 33 <!-- <logger name="org.thingsboard.server.service.transport" level="TRACE" />-->
... ...
... ... @@ -191,7 +191,7 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
191 191 @Test
192 192 public void test() throws Exception {
193 193 testReceivedInitialData();
194   - int expectedDownlinkSize = 10;
  194 + int expectedDownlinkSize = 9;
195 195 Assert.assertEquals(expectedDownlinkSize, edgeImitator.getDownlinkMsgs().size());
196 196
197 197 testDevices();
... ... @@ -308,7 +308,8 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
308 308
309 309 private void testReceivedInitialData() throws Exception {
310 310 log.info("Checking received data");
311   - Assert.assertTrue(edgeImitator.waitForMessages());
  311 + boolean condition = edgeImitator.waitForMessages();
  312 + Assert.assertTrue(condition);
312 313
313 314 EdgeConfiguration configuration = edgeImitator.getConfiguration();
314 315 Assert.assertNotNull(configuration);
... ... @@ -1523,6 +1524,10 @@ abstract public class BaseEdgeTest extends AbstractControllerTest {
1523 1524 private void installation() throws Exception {
1524 1525 edge = doPost("/api/edge", constructEdge("Test Edge", "test"), Edge.class);
1525 1526
  1527 + Device savedDeviceDefault = saveDevice("Edge Device Default", "Default");
  1528 + doPost("/api/edge/" + edge.getId().getId().toString()
  1529 + + "/device/" + savedDeviceDefault.getId().getId().toString(), Device.class);
  1530 +
1526 1531 DeviceProfile deviceProfile = this.createDeviceProfile(CUSTOM_DEVICE_PROFILE_NAME, null);
1527 1532 extendDeviceProfileData(deviceProfile);
1528 1533 doPost("/api/deviceProfile", deviceProfile, DeviceProfile.class);
... ...
... ... @@ -109,7 +109,7 @@ public class EdgeImitator {
109 109 this::onDownlink,
110 110 this::onClose);
111 111
112   - edgeRpcClient.sendSyncRequestMsg();
  112 + edgeRpcClient.sendSyncRequestMsg(false);
113 113 }
114 114
115 115 public void disconnect() throws InterruptedException {
... ...
... ... @@ -34,6 +34,7 @@ import org.thingsboard.server.gen.edge.EdgeRpcServiceGrpc;
34 34 import org.thingsboard.server.gen.edge.RequestMsg;
35 35 import org.thingsboard.server.gen.edge.RequestMsgType;
36 36 import org.thingsboard.server.gen.edge.ResponseMsg;
  37 +import org.thingsboard.server.gen.edge.SyncRequestMsg;
37 38 import org.thingsboard.server.gen.edge.UplinkMsg;
38 39 import org.thingsboard.server.gen.edge.UplinkResponseMsg;
39 40
... ... @@ -102,7 +103,7 @@ public class EdgeGrpcClient implements EdgeRpcClient {
102 103 Consumer<EdgeConfiguration> onEdgeUpdate,
103 104 Consumer<DownlinkMsg> onDownlink,
104 105 Consumer<Exception> onError) {
105   - return new StreamObserver<ResponseMsg>() {
  106 + return new StreamObserver<>() {
106 107 @Override
107 108 public void onNext(ResponseMsg responseMsg) {
108 109 if (responseMsg.hasConnectResponseMsg()) {
... ... @@ -195,11 +196,13 @@ public class EdgeGrpcClient implements EdgeRpcClient {
195 196 }
196 197
197 198 @Override
198   - public void sendSyncRequestMsg() {
  199 + public void sendSyncRequestMsg(boolean syncRequired) {
199 200 try {
200 201 uplinkMsgLock.lock();
  202 + SyncRequestMsg syncRequestMsg = SyncRequestMsg.newBuilder().setSyncRequired(syncRequired).build();
201 203 this.inputStream.onNext(RequestMsg.newBuilder()
202 204 .setMsgType(RequestMsgType.SYNC_REQUEST_RPC_MESSAGE)
  205 + .setSyncRequestMsg(syncRequestMsg)
203 206 .build());
204 207 } finally {
205 208 uplinkMsgLock.unlock();
... ...
... ... @@ -34,7 +34,7 @@ public interface EdgeRpcClient {
34 34
35 35 void disconnect(boolean onError) throws InterruptedException;
36 36
37   - void sendSyncRequestMsg();
  37 + void sendSyncRequestMsg(boolean syncRequired);
38 38
39 39 void sendUplinkMsg(UplinkMsg uplinkMsg);
40 40
... ...
... ... @@ -38,6 +38,7 @@ message RequestMsg {
38 38 ConnectRequestMsg connectRequestMsg = 2;
39 39 UplinkMsg uplinkMsg = 3;
40 40 DownlinkResponseMsg downlinkResponseMsg = 4;
  41 + SyncRequestMsg syncRequestMsg = 5;
41 42 }
42 43
43 44 message ResponseMsg {
... ... @@ -74,6 +75,13 @@ message ConnectResponseMsg {
74 75 EdgeConfiguration configuration = 3;
75 76 }
76 77
  78 +message SyncRequestMsg {
  79 + bool syncRequired = 1;
  80 +}
  81 +
  82 +message SyncCompletedMsg {
  83 +}
  84 +
77 85 message EdgeConfiguration {
78 86 int64 edgeIdMSB = 1;
79 87 int64 edgeIdLSB = 2;
... ... @@ -358,6 +366,22 @@ message DeviceCredentialsRequestMsg {
358 366 int64 deviceIdLSB = 2;
359 367 }
360 368
  369 +message DeviceProfileDevicesRequestMsg {
  370 + int64 deviceProfileIdMSB = 1;
  371 + int64 deviceProfileIdLSB = 2;
  372 +}
  373 +
  374 +message WidgetBundleTypesRequestMsg {
  375 + int64 widgetBundleIdMSB = 1;
  376 + int64 widgetBundleIdLSB = 2;
  377 +}
  378 +
  379 +message EntityViewsRequestMsg {
  380 + int64 entityIdMSB = 1;
  381 + int64 entityIdLSB = 2;
  382 + string entityType = 3;
  383 +}
  384 +
361 385 message DeviceRpcCallMsg {
362 386 int64 deviceIdMSB = 1;
363 387 int64 deviceIdLSB = 2;
... ... @@ -402,6 +426,9 @@ message UplinkMsg {
402 426 repeated UserCredentialsRequestMsg userCredentialsRequestMsg = 10;
403 427 repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 11;
404 428 repeated DeviceRpcCallMsg deviceRpcCallMsg = 12;
  429 + repeated DeviceProfileDevicesRequestMsg deviceProfileDevicesRequestMsg = 13;
  430 + repeated WidgetBundleTypesRequestMsg widgetBundleTypesRequestMsg = 14;
  431 + repeated EntityViewsRequestMsg entityViewsRequestMsg = 15;
405 432 }
406 433
407 434 message UplinkResponseMsg {
... ... @@ -416,24 +443,25 @@ message DownlinkResponseMsg {
416 443
417 444 message DownlinkMsg {
418 445 int32 downlinkMsgId = 1;
419   - repeated EntityDataProto entityData = 2;
420   - repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 3;
421   - repeated DeviceUpdateMsg deviceUpdateMsg = 4;
422   - repeated DeviceProfileUpdateMsg deviceProfileUpdateMsg = 5;
423   - repeated DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = 6;
424   - repeated RuleChainUpdateMsg ruleChainUpdateMsg = 7;
425   - repeated RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = 8;
426   - repeated DashboardUpdateMsg dashboardUpdateMsg = 9;
427   - repeated AssetUpdateMsg assetUpdateMsg = 10;
428   - repeated EntityViewUpdateMsg entityViewUpdateMsg = 11;
429   - repeated AlarmUpdateMsg alarmUpdateMsg = 12;
430   - repeated UserUpdateMsg userUpdateMsg = 13;
431   - repeated UserCredentialsUpdateMsg userCredentialsUpdateMsg = 14;
432   - repeated CustomerUpdateMsg customerUpdateMsg = 15;
433   - repeated RelationUpdateMsg relationUpdateMsg = 16;
434   - repeated WidgetsBundleUpdateMsg widgetsBundleUpdateMsg = 17;
435   - repeated WidgetTypeUpdateMsg widgetTypeUpdateMsg = 18;
436   - repeated AdminSettingsUpdateMsg adminSettingsUpdateMsg = 19;
437   - repeated DeviceRpcCallMsg deviceRpcCallMsg = 20;
  446 + SyncCompletedMsg syncCompletedMsg = 2;
  447 + repeated EntityDataProto entityData = 3;
  448 + repeated DeviceCredentialsRequestMsg deviceCredentialsRequestMsg = 4;
  449 + repeated DeviceUpdateMsg deviceUpdateMsg = 5;
  450 + repeated DeviceProfileUpdateMsg deviceProfileUpdateMsg = 6;
  451 + repeated DeviceCredentialsUpdateMsg deviceCredentialsUpdateMsg = 7;
  452 + repeated RuleChainUpdateMsg ruleChainUpdateMsg = 8;
  453 + repeated RuleChainMetadataUpdateMsg ruleChainMetadataUpdateMsg = 9;
  454 + repeated DashboardUpdateMsg dashboardUpdateMsg = 10;
  455 + repeated AssetUpdateMsg assetUpdateMsg = 11;
  456 + repeated EntityViewUpdateMsg entityViewUpdateMsg = 12;
  457 + repeated AlarmUpdateMsg alarmUpdateMsg = 13;
  458 + repeated UserUpdateMsg userUpdateMsg = 14;
  459 + repeated UserCredentialsUpdateMsg userCredentialsUpdateMsg = 15;
  460 + repeated CustomerUpdateMsg customerUpdateMsg = 16;
  461 + repeated RelationUpdateMsg relationUpdateMsg = 17;
  462 + repeated WidgetsBundleUpdateMsg widgetsBundleUpdateMsg = 18;
  463 + repeated WidgetTypeUpdateMsg widgetTypeUpdateMsg = 19;
  464 + repeated AdminSettingsUpdateMsg adminSettingsUpdateMsg = 20;
  465 + repeated DeviceRpcCallMsg deviceRpcCallMsg = 21;
438 466 }
439 467
... ...
... ... @@ -558,6 +558,7 @@ public class EdgeServiceImpl extends AbstractEntityService implements EdgeServic
558 558 if (userById.getCustomerId() == null || userById.getCustomerId().isNullUid()) {
559 559 pageData = findEdgesByTenantId(tenantId, pageLink);
560 560 } else {
  561 +// pageData = findEdgesByTenantIdAndCustomerId(tenantId, userById.getCustomerId(), pageLink);
561 562 pageData = findEdgesByTenantIdAndCustomerId(tenantId, new CustomerId(entityId.getId()), pageLink);
562 563 }
563 564 if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
... ...