feat: lazy loading of paging in the adapter connect view

streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java

@@ -41,7 +41,8 @@
 import java.util.NoSuchElementException;
 
 /**
- * This class is responsible for managing all the adapter instances which are executed on worker nodes
+ * This class is responsible for managing all the adapter instances which are
+ * executed on worker nodes
  */
 public class AdapterMasterManagement {
 
@@ -57,8 +58,7 @@ public AdapterMasterManagement(
       IAdapterStorage adapterInstanceStorage,
       AdapterResourceManager adapterResourceManager,
       DataStreamResourceManager dataStreamResourceManager,
-      AdapterMetrics adapterMetrics
-  ) {
+      AdapterMetrics adapterMetrics) {
     this.adapterInstanceStorage = adapterInstanceStorage;
     this.adapterMetrics = adapterMetrics;
     this.adapterResourceManager = adapterResourceManager;
@@ -68,8 +68,7 @@ public AdapterMasterManagement(
   public void addAdapter(
       AdapterDescription adapterDescription,
       String adapterId,
-      String principalSid
-  )
+      String principalSid)
       throws AdapterException {
 
     // Create elementId for datastream
@@ -92,8 +91,7 @@ private void createDataStreamForAdapter(
       AdapterDescription adapterDescription,
       String adapterId,
       String streamId,
-      String principalSid
-  ) throws AdapterException {
+      String principalSid) throws AdapterException {
     var storedDescription = new SourcesManagement()
         .createAdapterDataStream(adapterDescription, streamId);
     storedDescription.setCorrespondingAdapterId(adapterId);
@@ -116,7 +114,8 @@ public AdapterDescription getAdapter(String elementId) throws AdapterException {
   }
 
   /**
-   * First the adapter is stopped removed, then the corresponding data source is deleted
+   * First the adapter is stopped removed, then the corresponding data source is
+   * deleted
    *
    * @param elementId The elementId of the adapter instance
    * @throws AdapterException when adapter can not be stopped
@@ -145,8 +144,20 @@ public List<AdapterDescription> getAllAdapterInstances() {
     return adapterInstanceStorage.findAll();
   }
 
+  public List<AdapterDescription> getPaginatedAdapterInstances(String startKey, String endKey, int limit, String view,
+      boolean descending) {
+    return adapterInstanceStorage.getAdapterPaginator(startKey,endKey, limit, view, descending);
+  }
+
+
+    public List<AdapterDescription> getItemsByCategoryPaginated(String category, String startKey, int limit,
+      boolean descending) {
+    return adapterInstanceStorage.getItemsByCategoryPaginated(category,startKey, limit, descending);
+  }
+
+
   public void stopStreamAdapter(String elementId,
-                                boolean forceStop) throws AdapterException {
+      boolean forceStop) throws AdapterException {
     AdapterDescription ad = adapterInstanceStorage.getElementById(elementId);
 
     try {
@@ -181,8 +192,7 @@ public void startStreamAdapter(String elementId) throws AdapterException {
           ad.getAppId(),
           SpServiceUrlProvider.ADAPTER,
           ad.getDeploymentConfiguration()
-            .getDesiredServiceTags()
-      );
+              .getDesiredServiceTags());
 
       // Update selected endpoint URL of adapter
       ad.setSelectedEndpointUrl(baseUrl);
@@ -191,7 +201,8 @@ public void startStreamAdapter(String elementId) throws AdapterException {
       // Invoke adapter instance
       WorkerRestClient.invokeStreamAdapter(baseUrl, elementId);
 
-      // register the adapter at the metrics manager so that the AdapterHealthCheck can send metrics
+      // register the adapter at the metrics manager so that the AdapterHealthCheck
+      // can send metrics
       adapterMetrics.register(ad.getElementId(), ad.getName());
 
       LOG.info("Started adapter " + elementId + " on: " + baseUrl);
@@ -202,8 +213,7 @@ public void startStreamAdapter(String elementId) throws AdapterException {
 
   private void installDataSource(
       SpDataStream stream,
-      String principalSid
-  ) throws AdapterException {
+      String principalSid) throws AdapterException {
     try {
       new DataStreamVerifier(stream).verifyAndAdd(principalSid, false);
     } catch (SepaParseException e) {

streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java

@@ -52,8 +52,7 @@ public void install() {
     new CreateAssetLinkTypeTask().execute();
     new CreateDefaultAssetTask().execute();
     new AddDefaultPipelineTemplatesTask(
-        StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineTemplateStorage()
-    ).execute();
+        StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineTemplateStorage()).execute();
   }
 
   @Override
@@ -79,6 +78,7 @@ private void createViews() {
     addNotificationView();
     addPipelineView();
     addDataLakeMeasureView();
+    addPaginatorView();
   }
 
   private void addNotificationView() {
@@ -113,8 +113,7 @@ private void addNotificationView() {
           + "}");
       notificationCountTypeViews.put("unread", countFunction);
       notificationCountDocument.setViews(notificationCountTypeViews);
-      Response countResp =
-          Utils.getCouchDbNotificationClient().design().synchronizeWithDb(notificationCountDocument);
+      Response countResp = Utils.getCouchDbNotificationClient().design().synchronizeWithDb(notificationCountDocument);
 
       if (resp.getError() != null && countResp != null) {
         logFailure(PREPARING_NOTIFICATIONS_TEXT);
@@ -126,6 +125,70 @@ private void addNotificationView() {
     }
   }
 
+  private void addPaginatorView() {
+    DesignDocument paginatorDocument = prepareDocument("_design/paginator");
+
+    Map<String, MapReduce> paginatorViews = new HashMap<>();
+
+    // View to paginate documents by creation time
+    MapReduce paginationFunctionByCreate = new MapReduce();
+    paginationFunctionByCreate.setMap(
+        "function (doc) {\n"
+            + "  if (doc.properties && doc.properties.createdAt) {\n"
+            + "    emit(doc.properties.createdAt, doc);\n"
+            + "  }\n"
+            + "}");
+
+    // View to paginate documents by name
+    MapReduce paginationFunctionByName = new MapReduce();
+    paginationFunctionByName.setMap(
+        "function (doc) {\n"
+            + "  if (doc.properties && doc.properties.name && typeof doc.properties.name === 'string') {\n"
+            + "    emit(doc.properties.name, doc);\n"
+            + "  }\n"
+            + "}");
+
+    // View to paginate documents by running
+    MapReduce paginationFunctionByRunning = new MapReduce();
+    paginationFunctionByRunning.setMap(
+        "function (doc) {\n"
+            + "    emit([doc.properties.running, doc._id], doc);\n"
+            + "}");
+
+    MapReduce paginationFunctionByCategory = new MapReduce();
+    paginationFunctionByCategory.setMap(
+        "function (doc) {\n"
+            + "  if (doc.properties && Array.isArray(doc.properties.category)) {\n"
+            + "    doc.properties.category.forEach(function (cat) {\n"
+            + "      emit([cat, doc._id], doc);\n"
+            + "    });\n"
+            + "  }\n"
+            + "}");
+
+    // View to list all non-design documents
+    MapReduce nonDesignDocsView = new MapReduce();
+    nonDesignDocsView.setMap(
+        "function (doc) {\n"
+            + "  if (!doc._id.startsWith(\"_design/\")) {\n"
+            + "    emit(doc._id, doc);\n"
+            + "  }\n"
+            + "}");
+
+    // Add views to the document
+    paginatorViews.put("by_createdAt", paginationFunctionByCreate);
+    paginatorViews.put("by_name", paginationFunctionByName);
+    paginatorViews.put("by_running", paginationFunctionByRunning);
+    paginatorViews.put("by_category", paginationFunctionByCategory);
+    paginatorViews.put("non_design_docs", nonDesignDocsView);
+
+    paginatorDocument.setViews(paginatorViews);
+
+    // Push the design document to CouchDB
+    Utils.getCouchDbAdapterInstanceClient()
+        .design()
+        .synchronizeWithDb(paginatorDocument);
+  }
+
   private void addPipelineView() {
     DesignDocument pipelineDocument = prepareDocument("_design/adapters");
     DesignDocument allPipelinesDocument = prepareDocument("_design/pipelines");
@@ -146,7 +209,6 @@ private void addPipelineView() {
     pipelineDocument.setViews(adapterViews);
     Utils.getCouchDbPipelineClient().design().synchronizeWithDb(pipelineDocument);
 
-
     MapReduce allPipelinesFunction = new MapReduce();
     allPipelinesFunction.setMap("function (doc) {\n"
         + "  emit(doc._id, doc);\n"

streampipes-rest/src/main/java/org/apache/streampipes/rest/ResetManagement.java

@@ -44,7 +44,8 @@
 import java.util.Map;
 
 public class ResetManagement {
-  // This class should be moved into another package. I moved it here because I got a cyclic maven
+  // This class should be moved into another package. I moved it here because I
+  // got a cyclic maven
   // dependency between this package and streampipes-pipeline-management
   // See in issue [STREAMPIPES-405]
 
@@ -104,11 +105,10 @@ private static void stopAndDeleteAllPipelines() {
   private static void stopAndDeleteAllAdapters() {
     AdapterMasterManagement adapterMasterManagement = new AdapterMasterManagement(
         StorageDispatcher.INSTANCE.getNoSqlStore()
-                                  .getAdapterInstanceStorage(),
+            .getAdapterInstanceStorage(),
         new SpResourceManager().manageAdapters(),
         new SpResourceManager().manageDataStreams(),
-        AdapterMetricsManager.INSTANCE.getAdapterMetrics()
-    );
+        AdapterMetricsManager.INSTANCE.getAdapterMetrics());
 
     List<AdapterDescription> allAdapters = adapterMasterManagement.getAllAdapterInstances();
     allAdapters.forEach(adapterDescription -> {
@@ -144,24 +144,22 @@ private static void removeAllDataInDataLake() {
   }
 
   private static void removeAllDataViewWidgets() {
-    var widgetStorage =
-        StorageDispatcher.INSTANCE.getNoSqlStore()
-                                  .getDataExplorerWidgetStorage();
+    var widgetStorage = StorageDispatcher.INSTANCE.getNoSqlStore()
+        .getDataExplorerWidgetStorage();
     widgetStorage.findAll()
-                 .forEach(widget -> widgetStorage.deleteElementById(widget.getElementId()));
+        .forEach(widget -> widgetStorage.deleteElementById(widget.getElementId()));
   }
 
   private static void removeAllDataViews() {
-    var dataLakeDashboardStorage =
-        StorageDispatcher.INSTANCE.getNoSqlStore()
-                                  .getDataExplorerDashboardStorage();
+    var dataLakeDashboardStorage = StorageDispatcher.INSTANCE.getNoSqlStore()
+        .getDataExplorerDashboardStorage();
     dataLakeDashboardStorage.findAll()
-                            .forEach(dashboard -> dataLakeDashboardStorage.deleteElementById(dashboard.getElementId()));
+        .forEach(dashboard -> dataLakeDashboardStorage.deleteElementById(dashboard.getElementId()));
   }
 
   private static void removeAllAssets(String username) {
     IGenericStorage genericStorage = StorageDispatcher.INSTANCE.getNoSqlStore()
-                                                               .getGenericStorage();
+        .getGenericStorage();
     try {
       for (Map<String, Object> asset : genericStorage.findAll("asset-management")) {
         genericStorage.delete((String) asset.get("_id"), (String) asset.get("_rev"));
@@ -172,8 +170,7 @@ private static void removeAllAssets(String username) {
   }
 
   private static void removeAllPipelineTemplates() {
-    var pipelineElementTemplateStorage = StorageDispatcher
-        .INSTANCE
+    var pipelineElementTemplateStorage = StorageDispatcher.INSTANCE
         .getNoSqlStore()
         .getPipelineElementTemplateStorage();
 
@@ -186,21 +183,19 @@ private static void removeAllPipelineTemplates() {
   private static void clearGenericStorage() {
     var appDocTypesToDelete = List.of(
         "asset-management",
-        "asset-sites"
-    );
+        "asset-sites");
     var genericStorage = StorageDispatcher.INSTANCE.getNoSqlStore()
-                                                   .getGenericStorage();
+        .getGenericStorage();
 
     appDocTypesToDelete.forEach(docType -> {
       try {
         var allDocs = genericStorage.findAll(docType);
         for (var doc : allDocs) {
           genericStorage.delete(
               doc.get("_id")
-                 .toString(),
+                  .toString(),
               doc.get("_rev")
-                 .toString()
-          );
+                  .toString());
         }
       } catch (IOException e) {
         throw new RuntimeException(e);