Use generics to remove duplication for event fetcher. Use page data iterable

@@ -37,10 +37,10 @@ import org.thingsboard.server.common.data.EdgeUtils;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.EntityView;
 import org.thingsboard.server.common.data.EntityViewInfo;
-import org.thingsboard.server.common.data.OtaPackage;
-import org.thingsboard.server.common.data.OtaPackageInfo;
 import org.thingsboard.server.common.data.HasName;
 import org.thingsboard.server.common.data.HasTenantId;
+import org.thingsboard.server.common.data.OtaPackage;
+import org.thingsboard.server.common.data.OtaPackageInfo;
 import org.thingsboard.server.common.data.TbResource;
 import org.thingsboard.server.common.data.TbResourceInfo;
 import org.thingsboard.server.common.data.Tenant;
@@ -70,7 +70,6 @@ import org.thingsboard.server.common.data.id.EntityIdFactory;
 import org.thingsboard.server.common.data.id.EntityViewId;
 import org.thingsboard.server.common.data.id.OtaPackageId;
 import org.thingsboard.server.common.data.id.RpcId;
-import org.thingsboard.server.common.data.id.TbResourceId;
 import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TbResourceId;
@@ -79,7 +78,7 @@ import org.thingsboard.server.common.data.id.TenantProfileId;
 import org.thingsboard.server.common.data.id.UserId;
 import org.thingsboard.server.common.data.id.WidgetTypeId;
 import org.thingsboard.server.common.data.id.WidgetsBundleId;
-import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.page.PageDataIterableByTenantIdEntityId;
 import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.page.SortOrder;
 import org.thingsboard.server.common.data.page.TimePageLink;
@@ -105,10 +104,10 @@ import org.thingsboard.server.dao.edge.EdgeService;
 import org.thingsboard.server.dao.entityview.EntityViewService;
 import org.thingsboard.server.dao.exception.DataValidationException;
 import org.thingsboard.server.dao.exception.IncorrectParameterException;
-import org.thingsboard.server.dao.ota.OtaPackageService;
 import org.thingsboard.server.dao.model.ModelConstants;
 import org.thingsboard.server.dao.oauth2.OAuth2ConfigTemplateService;
 import org.thingsboard.server.dao.oauth2.OAuth2Service;
+import org.thingsboard.server.dao.ota.OtaPackageService;
 import org.thingsboard.server.dao.relation.RelationService;
 import org.thingsboard.server.dao.rpc.RpcService;
 import org.thingsboard.server.dao.rule.RuleChainService;
@@ -125,11 +124,10 @@ import org.thingsboard.server.queue.provider.TbQueueProducerProvider;
 import org.thingsboard.server.queue.util.TbCoreComponent;
 import org.thingsboard.server.service.action.RuleEngineEntityActionService;
 import org.thingsboard.server.service.component.ComponentDiscoveryService;
-import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
-import org.thingsboard.server.service.ota.OtaPackageStateService;
 import org.thingsboard.server.service.edge.EdgeNotificationService;
-import org.thingsboard.server.service.edge.rpc.EdgeGrpcService;
+import org.thingsboard.server.service.edge.rpc.EdgeRpcService;
 import org.thingsboard.server.service.lwm2m.LwM2MServerSecurityInfoRepository;
+import org.thingsboard.server.service.ota.OtaPackageStateService;
 import org.thingsboard.server.service.profile.TbDeviceProfileCache;
 import org.thingsboard.server.service.queue.TbClusterService;
 import org.thingsboard.server.service.resource.TbResourceService;
@@ -923,18 +921,12 @@ public abstract class BaseController {
         if (EntityType.EDGE.equals(entityId.getEntityType())) {
             return Collections.singletonList(new EdgeId(entityId.getId()));
         }
+        PageDataIterableByTenantIdEntityId<EdgeId> relatedEdgeIdsIterator =
+                new PageDataIterableByTenantIdEntityId<>(edgeService::findRelatedEdgeIdsByEntityId, tenantId, entityId, DEFAULT_PAGE_SIZE);
         List<EdgeId> result = new ArrayList<>();
-        PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
-        PageData<EdgeId> pageData;
-        do {
-            pageData = edgeService.findRelatedEdgeIdsByEntityId(tenantId, entityId, pageLink);
-            if (pageData != null && pageData.getData() != null && !pageData.getData().isEmpty()) {
-                result.addAll(pageData.getData());
-                if (pageData.hasNext()) {
-                    pageLink = pageLink.nextPageLink();
-                }
-            }
-        } while (pageData != null && pageData.hasNext());
+        for(EdgeId edgeId : relatedEdgeIdsIterator) {
+            result.add(edgeId);
+        }
         return result;
     }
 

@@ -50,6 +50,7 @@ import org.thingsboard.server.common.data.id.RuleChainId;
 import org.thingsboard.server.common.data.id.RuleNodeId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
+import org.thingsboard.server.common.data.page.PageDataIterableByTenant;
 import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent;
 import org.thingsboard.server.common.data.rule.DefaultRuleChainCreateRequest;
@@ -645,17 +646,11 @@ public class RuleChainController extends BaseController {
         try {
             TenantId tenantId = getCurrentUser().getTenantId();
             List<RuleChain> result = new ArrayList<>();
-            PageLink pageLink = new PageLink(DEFAULT_PAGE_SIZE);
-            PageData<RuleChain> pageData;
-            do {
-                pageData = ruleChainService.findAutoAssignToEdgeRuleChainsByTenantId(tenantId, pageLink);
-                if (pageData.getData().size() > 0) {
-                    result.addAll(pageData.getData());
-                    if (pageData.hasNext()) {
-                        pageLink = pageLink.nextPageLink();
-                    }
-                }
-            } while (pageData.hasNext());
+            PageDataIterableByTenant<RuleChain> autoAssignRuleChainsIterator =
+                    new PageDataIterableByTenant<>(ruleChainService::findAutoAssignToEdgeRuleChainsByTenantId, tenantId, DEFAULT_PAGE_SIZE);
+            for (RuleChain ruleChain : autoAssignRuleChainsIterator) {
+                result.add(ruleChain);
+            }
             return checkNotNull(result);
         } catch (Exception e) {
             throw handleException(e);

@@ -16,7 +16,6 @@
 package org.thingsboard.server.service.edge;
 
 import lombok.Data;
-import lombok.Getter;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
@@ -53,117 +52,90 @@ import org.thingsboard.server.service.executors.DbCallbackExecutorService;
 @Component
 @TbCoreComponent
 @Data
+@Lazy
 public class EdgeContextComponent {
 
-    @Lazy
     @Autowired
     private EdgeService edgeService;
 
-    @Lazy
     @Autowired
     private EdgeEventService edgeEventService;
 
-    @Lazy
     @Autowired
     private AdminSettingsService adminSettingsService;
 
-    @Lazy
     @Autowired
     private AssetService assetService;
 
-    @Lazy
     @Autowired
     private DeviceProfileService deviceProfileService;
 
-    @Lazy
     @Autowired
     private AttributesService attributesService;
 
-    @Lazy
     @Autowired
     private DashboardService dashboardService;
 
-    @Lazy
     @Autowired
     private RuleChainService ruleChainService;
 
-    @Lazy
     @Autowired
     private UserService userService;
 
-    @Lazy
     @Autowired
     private WidgetsBundleService widgetsBundleService;
 
-    @Lazy
     @Autowired
     private EdgeRequestsService edgeRequestsService;
 
-    @Lazy
     @Autowired
     private AlarmEdgeProcessor alarmProcessor;
 
-    @Lazy
     @Autowired
     private DeviceProfileEdgeProcessor deviceProfileProcessor;
 
-    @Lazy
     @Autowired
     private DeviceEdgeProcessor deviceProcessor;
 
-    @Lazy
     @Autowired
     private EntityEdgeProcessor entityProcessor;
 
-    @Lazy
     @Autowired
     private AssetEdgeProcessor assetProcessor;
 
-    @Lazy
     @Autowired
     private EntityViewEdgeProcessor entityViewProcessor;
 
-    @Lazy
     @Autowired
     private UserEdgeProcessor userProcessor;
 
-    @Lazy
     @Autowired
     private RelationEdgeProcessor relationProcessor;
 
-    @Lazy
     @Autowired
     private TelemetryEdgeProcessor telemetryProcessor;
 
-    @Lazy
     @Autowired
     private DashboardEdgeProcessor dashboardProcessor;
 
-    @Lazy
     @Autowired
     private RuleChainEdgeProcessor ruleChainProcessor;
 
-    @Lazy
     @Autowired
     private CustomerEdgeProcessor customerProcessor;
 
-    @Lazy
     @Autowired
     private WidgetBundleEdgeProcessor widgetBundleProcessor;
 
-    @Lazy
     @Autowired
     private WidgetTypeEdgeProcessor widgetTypeProcessor;
 
-    @Lazy
     @Autowired
     private AdminSettingsEdgeProcessor adminSettingsProcessor;
 
-    @Lazy
     @Autowired
     private EdgeEventStorageSettings edgeEventStorageSettings;
 
     @Autowired
-    @Getter
     private DbCallbackExecutorService dbCallbackExecutor;
 }

@@ -316,6 +316,7 @@ public final class EdgeGrpcSession implements Closeable {
             if (ifOffset != null) {
                 Long newStartTs = Uuids.unixTimestamp(ifOffset);
                 updateQueueStartTs(newStartTs);
+                log.debug("[{}] queue offset was updated [{}][{}]", this.sessionId, ifOffset, newStartTs);
             }
         }
         log.trace("[{}] processHandleMessages finished", this.sessionId);

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -17,6 +17,7 @@ package org.thingsboard.server.service.edge.rpc.fetch;
 
 import com.datastax.oss.driver.api.core.uuid.Uuids;
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -47,11 +48,18 @@ import java.util.regex.Pattern;
 
 @AllArgsConstructor
 @Slf4j
-public class AdminSettingsEdgeEventFetcher extends BasePageableEdgeEventFetcher {
+public class AdminSettingsEdgeEventFetcher implements EdgeEventFetcher {
+
+    private static final ObjectMapper mapper = new ObjectMapper();
 
     private final AdminSettingsService adminSettingsService;
 
     @Override
+    public PageLink getPageLink(int pageSize) {
+        return new PageLink(pageSize);
+    }
+
+    @Override
     public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) throws Exception {
         List<EdgeEvent> result = new ArrayList<>();
 

@@ -28,26 +28,20 @@ import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.dao.asset.AssetService;
 import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-
 @AllArgsConstructor
 @Slf4j
-public class AssetsEdgeEventFetcher extends BasePageableEdgeEventFetcher {
+public class AssetsEdgeEventFetcher extends BasePageableEdgeEventFetcher<Asset> {
 
     private final AssetService assetService;
 
     @Override
-    public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) {
-        log.trace("[{}] start fetching edge events [{}]", tenantId, edge.getId());
-        PageData<Asset> pageData = assetService.findAssetsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
-        List<EdgeEvent> result = new ArrayList<>();
-        if (!pageData.getData().isEmpty()) {
-            for (Asset asset : pageData.getData()) {
-                result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ASSET,
-                        EdgeEventActionType.ADDED, asset.getId(), null));
-            }
-        }
-        return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
+    PageData<Asset> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
+        return assetService.findAssetsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
+    }
+
+    @Override
+    EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, Asset asset) {
+        return EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.ASSET,
+                EdgeEventActionType.ADDED, asset.getId(), null);
     }
 }

@@ -16,14 +16,50 @@
 package org.thingsboard.server.service.edge.rpc.fetch;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.common.data.BaseData;
+import org.thingsboard.server.common.data.edge.Edge;
+import org.thingsboard.server.common.data.edge.EdgeEvent;
+import org.thingsboard.server.common.data.edge.EdgeEventActionType;
+import org.thingsboard.server.common.data.edge.EdgeEventType;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.EventId;
+import org.thingsboard.server.common.data.id.HasId;
+import org.thingsboard.server.common.data.id.HasUUID;
+import org.thingsboard.server.common.data.id.IdBased;
+import org.thingsboard.server.common.data.id.RuleChainId;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.id.UUIDBased;
+import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
+import org.thingsboard.server.common.data.rule.RuleChain;
+import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
 
-public abstract class BasePageableEdgeEventFetcher implements EdgeEventFetcher {
+import java.util.ArrayList;
+import java.util.List;
 
-    protected static final ObjectMapper mapper = new ObjectMapper();
+@Slf4j
+public abstract class BasePageableEdgeEventFetcher<T> implements EdgeEventFetcher {
 
     @Override
     public PageLink getPageLink(int pageSize) {
         return new PageLink(pageSize);
     }
+
+    @Override
+    public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) {
+        log.trace("[{}] start fetching edge events [{}]", tenantId, edge.getId());
+        PageData<T> pageData = fetchPageData(tenantId, edge, pageLink);
+        List<EdgeEvent> result = new ArrayList<>();
+        if (!pageData.getData().isEmpty()) {
+            for (T entity : pageData.getData()) {
+                result.add(constructEdgeEvent(tenantId, edge, entity));
+            }
+        }
+        return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
+    }
+
+    abstract PageData<T> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink);
+
+    abstract EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, T entity);
 }

@@ -28,27 +28,21 @@ import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.dao.user.UserService;
 import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-
 @Slf4j
 @AllArgsConstructor
-public abstract class BaseUsersEdgeEventFetcher extends BasePageableEdgeEventFetcher {
+public abstract class BaseUsersEdgeEventFetcher extends BasePageableEdgeEventFetcher<User> {
 
     protected final UserService userService;
 
     @Override
-    public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) {
-        log.trace("[{}] start fetching edge events [{}]", tenantId, edge.getId());
-        PageData<User> pageData = findUsers(tenantId, pageLink);
-        List<EdgeEvent> result = new ArrayList<>();
-        if (!pageData.getData().isEmpty()) {
-            for (User user : pageData.getData()) {
-                result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER,
-                        EdgeEventActionType.ADDED, user.getId(), null));
-            }
-        }
-        return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
+    PageData<User> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
+        return findUsers(tenantId, pageLink);
+    }
+
+    @Override
+    EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, User user) {
+        return EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.USER,
+                EdgeEventActionType.ADDED, user.getId(), null);
     }
 
     protected abstract PageData<User> findUsers(TenantId tenantId, PageLink pageLink);

@@ -28,27 +28,21 @@ import org.thingsboard.server.common.data.widget.WidgetsBundle;
 import org.thingsboard.server.dao.widget.WidgetsBundleService;
 import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-
 @Slf4j
 @AllArgsConstructor
-public abstract class BaseWidgetsBundlesEdgeEventFetcher extends BasePageableEdgeEventFetcher {
+public abstract class BaseWidgetsBundlesEdgeEventFetcher extends BasePageableEdgeEventFetcher<WidgetsBundle> {
 
     protected final WidgetsBundleService widgetsBundleService;
 
     @Override
-    public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) {
-        log.trace("[{}] start fetching edge events [{}]", tenantId, edge.getId());
-        PageData<WidgetsBundle> pageData = findWidgetsBundles(tenantId, pageLink);
-        List<EdgeEvent> result = new ArrayList<>();
-        if (!pageData.getData().isEmpty()) {
-            for (WidgetsBundle widgetsBundle : pageData.getData()) {
-                result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.WIDGETS_BUNDLE,
-                        EdgeEventActionType.ADDED, widgetsBundle.getId(), null));
-            }
-        }
-        return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
+    PageData<WidgetsBundle> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
+        return findWidgetsBundles(tenantId, pageLink);
+    }
+
+    @Override
+    EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, WidgetsBundle widgetsBundle) {
+        return EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.WIDGETS_BUNDLE,
+                EdgeEventActionType.ADDED, widgetsBundle.getId(), null);
     }
 
     protected abstract PageData<WidgetsBundle> findWidgetsBundles(TenantId tenantId, PageLink pageLink);

@@ -16,6 +16,8 @@
 package org.thingsboard.server.service.edge.rpc.fetch;
 
 import org.thingsboard.server.common.data.User;
+import org.thingsboard.server.common.data.edge.Edge;
+import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.id.CustomerId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
@@ -35,4 +37,5 @@ public class CustomerUsersEdgeEventFetcher extends BaseUsersEdgeEventFetcher {
     protected PageData<User> findUsers(TenantId tenantId, PageLink pageLink) {
         return userService.findCustomerUsers(tenantId, customerId, pageLink);
     }
+
 }

@@ -28,26 +28,20 @@ import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.dao.dashboard.DashboardService;
 import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-
 @AllArgsConstructor
 @Slf4j
-public class DashboardsEdgeEventFetcher extends BasePageableEdgeEventFetcher {
+public class DashboardsEdgeEventFetcher extends BasePageableEdgeEventFetcher<DashboardInfo> {
 
     private final DashboardService dashboardService;
 
     @Override
-    public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) {
-        log.trace("[{}] start fetching edge events [{}]", tenantId, edge.getId());
-        PageData<DashboardInfo> pageData = dashboardService.findDashboardsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
-        List<EdgeEvent> result = new ArrayList<>();
-        if (!pageData.getData().isEmpty()) {
-            for (DashboardInfo dashboardInfo : pageData.getData()) {
-                result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.DASHBOARD,
-                        EdgeEventActionType.ADDED, dashboardInfo.getId(), null));
-            }
-        }
-        return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
+    PageData<DashboardInfo> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
+        return dashboardService.findDashboardsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
+    }
+
+    @Override
+    EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, DashboardInfo dashboardInfo) {
+        return EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.DASHBOARD,
+                EdgeEventActionType.ADDED, dashboardInfo.getId(), null);
     }
 }

@@ -28,26 +28,20 @@ import org.thingsboard.server.common.data.page.PageLink;
 import org.thingsboard.server.dao.device.DeviceProfileService;
 import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-
 @AllArgsConstructor
 @Slf4j
-public class DeviceProfilesEdgeEventFetcher extends BasePageableEdgeEventFetcher {
+public class DeviceProfilesEdgeEventFetcher extends BasePageableEdgeEventFetcher<DeviceProfile> {
 
     private final DeviceProfileService deviceProfileService;
 
     @Override
-    public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) {
-        log.trace("[{}] start fetching edge events [{}]", tenantId, edge.getId());
-        PageData<DeviceProfile> pageData = deviceProfileService.findDeviceProfiles(tenantId, pageLink);
-        List<EdgeEvent> result = new ArrayList<>();
-        if (!pageData.getData().isEmpty()) {
-            for (DeviceProfile deviceProfile : pageData.getData()) {
-                result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE_PROFILE,
-                        EdgeEventActionType.ADDED, deviceProfile.getId(), null));
-            }
-        }
-        return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
+    PageData<DeviceProfile> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
+        return deviceProfileService.findDeviceProfiles(tenantId, pageLink);
+    }
+
+    @Override
+    EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, DeviceProfile deviceProfile) {
+        return EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.DEVICE_PROFILE,
+                EdgeEventActionType.ADDED, deviceProfile.getId(), null);
     }
 }

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -28,26 +28,20 @@ import org.thingsboard.server.common.data.rule.RuleChain;
 import org.thingsboard.server.dao.rule.RuleChainService;
 import org.thingsboard.server.service.edge.rpc.EdgeEventUtils;
 
-import java.util.ArrayList;
-import java.util.List;
-
 @Slf4j
 @AllArgsConstructor
-public class RuleChainsEdgeEventFetcher extends BasePageableEdgeEventFetcher {
+public class RuleChainsEdgeEventFetcher extends BasePageableEdgeEventFetcher<RuleChain> {
 
     private final RuleChainService ruleChainService;
 
     @Override
-    public PageData<EdgeEvent> fetchEdgeEvents(TenantId tenantId, Edge edge, PageLink pageLink) {
-        log.trace("[{}] start fetching edge events [{}]", tenantId, edge.getId());
-        PageData<RuleChain> pageData = ruleChainService.findRuleChainsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
-        List<EdgeEvent> result = new ArrayList<>();
-        if (!pageData.getData().isEmpty()) {
-            for (RuleChain ruleChain : pageData.getData()) {
-                result.add(EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN,
-                        EdgeEventActionType.ADDED, ruleChain.getId(), null));
-            }
-        }
-        return new PageData<>(result, pageData.getTotalPages(), pageData.getTotalElements(), pageData.hasNext());
+    PageData<RuleChain> fetchPageData(TenantId tenantId, Edge edge, PageLink pageLink) {
+        return ruleChainService.findRuleChainsByTenantIdAndEdgeId(tenantId, edge.getId(), pageLink);
+    }
+
+    @Override
+    EdgeEvent constructEdgeEvent(TenantId tenantId, Edge edge, RuleChain ruleChain) {
+        return EdgeEventUtils.constructEdgeEvent(tenantId, edge.getId(), EdgeEventType.RULE_CHAIN,
+                EdgeEventActionType.ADDED, ruleChain.getId(), null);
     }
 }

@@ -16,6 +16,8 @@
 package org.thingsboard.server.service.edge.rpc.fetch;
 
 import lombok.extern.slf4j.Slf4j;
+import org.thingsboard.server.common.data.edge.Edge;
+import org.thingsboard.server.common.data.edge.EdgeEvent;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.page.PageData;
 import org.thingsboard.server.common.data.page.PageLink;
@@ -23,7 +25,7 @@ import org.thingsboard.server.common.data.widget.WidgetsBundle;
 import org.thingsboard.server.dao.widget.WidgetsBundleService;
 
 @Slf4j
-public class TenantWidgetsBundlesEdgeEventFetcher extends BaseWidgetsBundlesEdgeEventFetcher implements EdgeEventFetcher {
+public class TenantWidgetsBundlesEdgeEventFetcher extends BaseWidgetsBundlesEdgeEventFetcher {
 
     public TenantWidgetsBundlesEdgeEventFetcher(WidgetsBundleService widgetsBundleService) {
         super(widgetsBundleService);

+package org.thingsboard.server.common.data.page;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.NoSuchElementException;
+
+public abstract class BasePageDataIterable<T> implements Iterable<T>, Iterator<T> {
+
+    private final int fetchSize;
+
+    private List<T> currentItems;
+    private int currentIdx;
+    private boolean hasNextPack;
+    private PageLink nextPackLink;
+    private boolean initialized;
+
+    public BasePageDataIterable(int fetchSize) {
+        super();
+        this.fetchSize = fetchSize;
+    }
+
+    @Override
+    public Iterator<T> iterator() {
+        return this;
+    }
+
+    @Override
+    public boolean hasNext() {
+        if (!initialized) {
+            fetch(new PageLink(fetchSize));
+            initialized = true;
+        }
+        if (currentIdx == currentItems.size()) {
+            if (hasNextPack) {
+                fetch(nextPackLink);
+            }
+        }
+        return currentIdx < currentItems.size();
+    }
+
+    @Override
+    public T next() {
+        if (!hasNext()) {
+            throw new NoSuchElementException();
+        }
+        return currentItems.get(currentIdx++);
+    }
+
+    private void fetch(PageLink link) {
+        PageData<T> pageData = fetchPageData(link);
+        currentIdx = 0;
+        currentItems = pageData.getData();
+        hasNextPack = pageData.hasNext();
+        nextPackLink = link.nextPageLink();
+    }
+
+    abstract PageData<T> fetchPageData(PageLink link);
+}

@@ -5,7 +5,7 @@
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
- *     http://www.apache.org/licenses/LICENSE-2.0
+ * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
@@ -15,70 +15,21 @@
  */
 package org.thingsboard.server.common.data.page;
 
-import java.util.Iterator;
-import java.util.List;
-import java.util.NoSuchElementException;
-
-import org.thingsboard.server.common.data.BaseData;
-import org.thingsboard.server.common.data.SearchTextBased;
-import org.thingsboard.server.common.data.id.EntityId;
-import org.thingsboard.server.common.data.id.UUIDBased;
-
-public class PageDataIterable<T> implements Iterable<T>, Iterator<T> {
+public class PageDataIterable<T> extends BasePageDataIterable<T> {
 
     private final FetchFunction<T> function;
-    private final int fetchSize;
-
-    private List<T> currentItems;
-    private int currentIdx;
-    private boolean hasNextPack;
-    private PageLink nextPackLink;
-    private boolean initialized;
 
     public PageDataIterable(FetchFunction<T> function, int fetchSize) {
-        super();
+        super(fetchSize);
         this.function = function;
-        this.fetchSize = fetchSize;
     }
 
     @Override
-    public Iterator<T> iterator() {
-        return this;
+    PageData<T> fetchPageData(PageLink link) {
+        return function.fetch(link);
     }
 
-    @Override
-    public boolean hasNext() {
-        if(!initialized){
-            fetch(new PageLink(fetchSize));
-            initialized = true;
-        }
-        if(currentIdx == currentItems.size()){
-            if(hasNextPack){
-                fetch(nextPackLink);
-            }
-        }
-        return currentIdx < currentItems.size();
-    }
-
-    private void fetch(PageLink link) {
-        PageData<T> pageData = function.fetch(link);
-        currentIdx = 0;
-        currentItems = pageData.getData();
-        hasNextPack = pageData.hasNext();
-        nextPackLink = link.nextPageLink();
-    }
-
-    @Override
-    public T next() {
-        if(!hasNext()){
-            throw new NoSuchElementException();
-        }
-        return currentItems.get(currentIdx++);
-    }
-
-    public static interface FetchFunction<T> {
-
+    public interface FetchFunction<T> {
         PageData<T> fetch(PageLink link);
-
     }
 }

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.page;
+
+import org.thingsboard.server.common.data.id.TenantId;
+
+public class PageDataIterableByTenant<T> extends BasePageDataIterable<T> {
+
+    private final FetchFunction<T> function;
+    private final TenantId tenantId;
+
+    public PageDataIterableByTenant(FetchFunction<T> function, TenantId tenantId, int fetchSize) {
+        super(fetchSize);
+        this.function = function;
+        this.tenantId = tenantId;
+    }
+
+    @Override
+    PageData<T> fetchPageData(PageLink link) {
+        return function.fetch(tenantId, link);
+    }
+
+    public interface FetchFunction<T> {
+        PageData<T> fetch(TenantId tenantId, PageLink link);
+    }
+}

+/**
+ * Copyright © 2016-2021 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.common.data.page;
+
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.TenantId;
+
+public class PageDataIterableByTenantIdEntityId<T> extends BasePageDataIterable<T> {
+
+    private final FetchFunction<T> function;
+    private final TenantId tenantId;
+    private final EntityId entityId;
+
+    public PageDataIterableByTenantIdEntityId(FetchFunction<T> function, TenantId tenantId, EntityId entityId, int fetchSize) {
+        super(fetchSize);
+        this.function = function;
+        this.tenantId = tenantId;
+        this.entityId = entityId;
+
+    }
+
+    @Override
+    PageData<T> fetchPageData(PageLink link) {
+        return function.fetch(tenantId, entityId, link);
+    }
+
+    public interface FetchFunction<T> {
+        PageData<T> fetch(TenantId tenantId, EntityId entityId, PageLink link);
+    }
+}