This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch cleanup-persistence-layer-3 in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit ae45c515c33ed3ce5dc61154d362a7cbd38faeae Author: Dominik Riemer <[email protected]> AuthorDate: Fri Jun 28 20:14:58 2024 +0200 chore: Cleanup persistence layer --- .../management/health/AdapterHealthCheck.java | 2 +- .../management/AdapterMasterManagement.java | 12 ++-- .../management/AdapterMigrationManager.java | 2 +- .../management/AdapterUpdateManagement.java | 2 +- .../management/DescriptionManagement.java | 8 +-- .../management/WorkerAdministrationManagement.java | 2 +- .../management/management/WorkerRestClient.java | 6 +- .../management/health/AdapterHealthCheckTest.java | 4 +- .../management/AdapterMasterManagementTest.java | 8 +-- .../export/resolver/AdapterResolver.java | 6 +- .../export/resolver/PipelineResolver.java | 6 +- .../streampipes/model/client/user/Permission.java | 13 +++- .../apache/streampipes/model/VirtualSensor.java | 50 --------------- .../model/pipeline/ElementComposition.java | 74 ---------------------- .../streampipes/model/pipeline/Pipeline.java | 59 +++++++++++++++-- .../execution/task/StorePipelineStatusTask.java | 4 +- .../extensions/AvailableExtensionsProvider.java | 2 +- .../manager/health/PipelineHealthCheck.java | 4 +- .../migration/PipelineElementMigrationManager.java | 10 +-- .../streampipes/manager/operations/Operations.java | 2 +- .../manager/pipeline/PipelineManager.java | 8 +-- .../manager/setup/CouchDbInstallationStep.java | 25 -------- .../manager/storage/PipelineStorageService.java | 4 +- .../PipelineTemplateInvocationHandler.java | 4 +- .../management/AdapterResourceManager.java | 6 +- .../management/PermissionResourceManager.java | 8 +-- .../management/AdapterResourceManagerTest.java | 2 +- .../rest/impl/PipelineElementAsset.java | 2 +- .../streampipes/rest/impl/PipelineResource.java | 2 +- .../rest/impl/connect/AdapterResource.java | 4 +- .../AbstractPipelineExtractionResource.java | 2 +- .../streampipes/service/core/PostStartupTask.java | 6 +- .../service/core/StreamPipesCoreApplication.java | 4 +- .../streampipes/storage/api/IAdapterStorage.java | 12 +--- .../storage/api/IDataProcessorStorage.java | 4 -- .../streampipes/storage/api/IDataSinkStorage.java | 3 - .../storage/api/IPermissionStorage.java | 12 +--- .../api/IPipelineElementDescriptionStorage.java | 23 ------- .../streampipes/storage/api/IPipelineStorage.java | 13 +--- .../storage/couchdb/CouchDbStorageManager.java | 2 +- .../impl/AdapterDescriptionStorageImpl.java | 60 +++--------------- .../couchdb/impl/AdapterInstanceStorageImpl.java | 66 +++---------------- .../couchdb/impl/DataProcessorStorageImpl.java | 9 --- .../storage/couchdb/impl/DataSinkStorageImpl.java | 10 --- .../couchdb/impl/DefaultViewCrudStorage.java | 2 +- .../couchdb/impl/PermissionStorageImpl.java | 34 +--------- .../PipelineElementDescriptionStorageImpl.java | 72 +-------------------- .../storage/couchdb/impl/PipelineStorageImpl.java | 30 +-------- .../streampipes/storage/couchdb/utils/Utils.java | 8 --- 49 files changed, 160 insertions(+), 553 deletions(-) diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java index a8d07530c7..e21d29de0d 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/health/AdapterHealthCheck.java @@ -154,7 +154,7 @@ public class AdapterHealthCheck implements Runnable { */ public Map<String, AdapterDescription> getAllAdaptersSupposedToRun() { Map<String, AdapterDescription> result = new HashMap<>(); - List<AdapterDescription> allRunningInstancesAdapterDescription = this.adapterStorage.getAllAdapters(); + List<AdapterDescription> allRunningInstancesAdapterDescription = this.adapterStorage.findAll(); allRunningInstancesAdapterDescription .stream() .filter(AdapterDescription::isRunning) diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java index 56220fde37..4c613a24f8 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMasterManagement.java @@ -92,7 +92,7 @@ public class AdapterMasterManagement { public AdapterDescription getAdapter(String elementId) throws AdapterException { - List<AdapterDescription> allAdapters = adapterInstanceStorage.getAllAdapters(); + List<AdapterDescription> allAdapters = adapterInstanceStorage.findAll(); if (allAdapters != null && elementId != null) { for (AdapterDescription ad : allAdapters) { @@ -120,7 +120,7 @@ public class AdapterMasterManagement { LOG.info("Could not stop adapter: " + elementId, e); } - AdapterDescription adapter = adapterInstanceStorage.getAdapter(elementId); + AdapterDescription adapter = adapterInstanceStorage.getElementById(elementId); // Delete adapter adapterResourceManager.delete(elementId); ExtensionsLogProvider.INSTANCE.remove(elementId); @@ -133,7 +133,7 @@ public class AdapterMasterManagement { public List<AdapterDescription> getAllAdapterInstances() throws AdapterException { - List<AdapterDescription> allAdapters = adapterInstanceStorage.getAllAdapters(); + List<AdapterDescription> allAdapters = adapterInstanceStorage.findAll(); if (allAdapters == null) { throw new AdapterException("Could not get all adapters"); @@ -143,7 +143,7 @@ public class AdapterMasterManagement { } public void stopStreamAdapter(String elementId) throws AdapterException { - AdapterDescription ad = adapterInstanceStorage.getAdapter(elementId); + AdapterDescription ad = adapterInstanceStorage.getElementById(elementId); WorkerRestClient.stopStreamAdapter(ad.getSelectedEndpointUrl(), ad); ExtensionsLogProvider.INSTANCE.reset(elementId); @@ -159,7 +159,7 @@ public class AdapterMasterManagement { public void startStreamAdapter(String elementId) throws AdapterException { - var ad = adapterInstanceStorage.getAdapter(elementId); + var ad = adapterInstanceStorage.getElementById(elementId); try { // Find endpoint to start adapter on @@ -171,7 +171,7 @@ public class AdapterMasterManagement { // Update selected endpoint URL of adapter ad.setSelectedEndpointUrl(baseUrl); - adapterInstanceStorage.updateAdapter(ad); + adapterInstanceStorage.updateElement(ad); // Invoke adapter instance WorkerRestClient.invokeStreamAdapter(baseUrl, elementId); diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMigrationManager.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMigrationManager.java index 26f158a5b7..e03b4bb168 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMigrationManager.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterMigrationManager.java @@ -83,7 +83,7 @@ public class AdapterMigrationManager extends AbstractMigrationManager implements "Migration was performed by extensions service '{}'", extensionsServiceConfig.getServiceUrl()); - adapterStorage.updateAdapter(migrationResult.element()); + adapterStorage.updateElement(migrationResult.element()); LOG.info("Adapter description is updated - Migration successfully completed at Core."); } else { LOG.error("Migration failed with the following reason: {}", migrationResult.message()); diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java index 0b2b38a6ef..43af954fa2 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/AdapterUpdateManagement.java @@ -91,7 +91,7 @@ public class AdapterUpdateManagement { modifiedPipeline.setPipelineNotifications(toNotification(updateInfo)); modifiedPipeline.setValid(false); } - StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(modifiedPipeline); + StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updateElement(modifiedPipeline); if (shouldRestartPipeline && canAutoMigrate) { Operations.startPipeline(PipelineManager.getPipeline(p.getPipelineId())); } diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/DescriptionManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/DescriptionManagement.java index 661abfaa16..49306186b3 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/DescriptionManagement.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/DescriptionManagement.java @@ -32,7 +32,7 @@ public class DescriptionManagement { public List<AdapterDescription> getAdapters() { IAdapterStorage adapterStorage = CouchDbStorageManager.INSTANCE.getAdapterDescriptionStorage(); - return adapterStorage.getAllAdapters(); + return adapterStorage.findAll(); } public Optional<AdapterDescription> getAdapter(String id) { @@ -43,9 +43,9 @@ public class DescriptionManagement { public void deleteAdapterDescription(String id) throws SpRuntimeException { var adapterStorage = CouchDbStorageManager.INSTANCE.getAdapterDescriptionStorage(); - var adapter = adapterStorage.getAdapter(id); + var adapter = adapterStorage.getElementById(id); if (!isAdapterUsed(adapter)) { - adapterStorage.deleteAdapter(id); + adapterStorage.deleteElementById(id); } else { throw new SpRuntimeException("This adapter is used by an existing instance and cannot be deleted"); } @@ -64,7 +64,7 @@ public class DescriptionManagement { } private boolean isAdapterUsed(AdapterDescription adapter) { - var allAdapters = StorageDispatcher.INSTANCE.getNoSqlStore().getAdapterInstanceStorage().getAllAdapters(); + var allAdapters = StorageDispatcher.INSTANCE.getNoSqlStore().getAdapterInstanceStorage().findAll(); return allAdapters .stream() diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java index e058c40cbf..6cdad54c6b 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerAdministrationManagement.java @@ -91,7 +91,7 @@ public class WorkerAdministrationManagement { } public void performAdapterMigrations(List<SpServiceTag> tags) { - var installedAdapters = CouchDbStorageManager.INSTANCE.getAdapterDescriptionStorage().getAllAdapters(); + var installedAdapters = CouchDbStorageManager.INSTANCE.getAdapterDescriptionStorage().findAll(); var adminSid = new SpResourceManager().manageUsers().getAdminUser().getPrincipalId(); installedAdapters.stream() .filter(adapter -> tags.stream().anyMatch(tag -> tag.getValue().equals(adapter.getAppId()))) diff --git a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java index 91558045e2..854c806a93 100644 --- a/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java +++ b/streampipes-connect-management/src/main/java/org/apache/streampipes/connect/management/management/WorkerRestClient.java @@ -199,7 +199,7 @@ public class WorkerRestClient { private static AdapterDescription getAdapterDescriptionById(AdapterInstanceStorageImpl adapterStorage, String id) { AdapterDescription adapterDescription = null; - List<AdapterDescription> allAdapters = adapterStorage.getAllAdapters(); + List<AdapterDescription> allAdapters = adapterStorage.findAll(); for (AdapterDescription a : allAdapters) { if (a.getElementId().endsWith(id)) { adapterDescription = a; @@ -219,11 +219,11 @@ public class WorkerRestClient { private static void encryptAndUpdateAdapter(AdapterDescription adapter) { AdapterDescription encryptedDescription = new Cloner().adapterDescription(adapter); SecretProvider.getEncryptionService().apply(encryptedDescription); - getAdapterStorage().updateAdapter(encryptedDescription); + getAdapterStorage().updateElement(encryptedDescription); } private static AdapterDescription getAndDecryptAdapter(String adapterId) { - AdapterDescription adapter = getAdapterStorage().getAdapter(adapterId); + AdapterDescription adapter = getAdapterStorage().getElementById(adapterId); SecretProvider.getDecryptionService().apply(adapter); return adapter; } diff --git a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/health/AdapterHealthCheckTest.java b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/health/AdapterHealthCheckTest.java index f4c3a3c2d5..47612f1361 100644 --- a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/health/AdapterHealthCheckTest.java +++ b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/health/AdapterHealthCheckTest.java @@ -44,7 +44,7 @@ public class AdapterHealthCheckTest { @Test public void getAllRunningInstancesAdapterDescriptionsEmpty() { - when(adapterInstanceStorageMock.getAllAdapters()).thenReturn(List.of()); + when(adapterInstanceStorageMock.findAll()).thenReturn(List.of()); var healthCheck = new AdapterHealthCheck( adapterInstanceStorageMock, @@ -74,7 +74,7 @@ public class AdapterHealthCheckTest { runningAdapter.setElementId(nameRunningAdapter); runningAdapter.setRunning(true); - when(adapterInstanceStorageMock.getAllAdapters()).thenReturn(List.of(stoppedAdapter, runningAdapter)); + when(adapterInstanceStorageMock.findAll()).thenReturn(List.of(stoppedAdapter, runningAdapter)); var healthCheck = new AdapterHealthCheck( adapterInstanceStorageMock, diff --git a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java index ae18017699..6d920fb751 100644 --- a/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java +++ b/streampipes-connect-management/src/test/java/org/apache/streampipes/connect/management/management/AdapterMasterManagementTest.java @@ -39,7 +39,7 @@ public class AdapterMasterManagementTest { public void getAdapterFailNull() { AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class); AdapterResourceManager resourceManager = mock(AdapterResourceManager.class); - when(adapterStorage.getAllAdapters()).thenReturn(null); + when(adapterStorage.findAll()).thenReturn(null); AdapterMasterManagement adapterMasterManagement = new AdapterMasterManagement( @@ -57,7 +57,7 @@ public class AdapterMasterManagementTest { List<AdapterDescription> adapterDescriptions = List.of(new AdapterDescription()); AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class); AdapterResourceManager resourceManager = mock(AdapterResourceManager.class); - when(adapterStorage.getAllAdapters()).thenReturn(adapterDescriptions); + when(adapterStorage.findAll()).thenReturn(adapterDescriptions); AdapterMasterManagement adapterMasterManagement = new AdapterMasterManagement( @@ -75,7 +75,7 @@ public class AdapterMasterManagementTest { List<AdapterDescription> adapterDescriptions = List.of(new AdapterDescription()); AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class); AdapterResourceManager resourceManager = mock(AdapterResourceManager.class); - when(adapterStorage.getAllAdapters()).thenReturn(adapterDescriptions); + when(adapterStorage.findAll()).thenReturn(adapterDescriptions); AdapterMasterManagement adapterMasterManagement = new AdapterMasterManagement( @@ -94,7 +94,7 @@ public class AdapterMasterManagementTest { public void getAllAdaptersFail() { AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class); AdapterResourceManager resourceManager = mock(AdapterResourceManager.class); - when(adapterStorage.getAllAdapters()).thenReturn(null); + when(adapterStorage.findAll()).thenReturn(null); AdapterMasterManagement adapterMasterManagement = new AdapterMasterManagement( diff --git a/streampipes-data-export/src/main/java/org/apache/streampipes/export/resolver/AdapterResolver.java b/streampipes-data-export/src/main/java/org/apache/streampipes/export/resolver/AdapterResolver.java index 8a5a445835..d22ae5ba72 100644 --- a/streampipes-data-export/src/main/java/org/apache/streampipes/export/resolver/AdapterResolver.java +++ b/streampipes-data-export/src/main/java/org/apache/streampipes/export/resolver/AdapterResolver.java @@ -29,7 +29,7 @@ public class AdapterResolver extends AbstractResolver<AdapterDescription> { @Override public AdapterDescription findDocument(String resourceId) { - return getNoSqlStore().getAdapterInstanceStorage().getAdapter(resourceId); + return getNoSqlStore().getAdapterInstanceStorage().getElementById(resourceId); } @Override @@ -53,7 +53,7 @@ public class AdapterResolver extends AbstractResolver<AdapterDescription> { @Override public void writeDocument(String document) throws JsonProcessingException { - getNoSqlStore().getAdapterInstanceStorage().storeAdapter(deserializeDocument(document)); + getNoSqlStore().getAdapterInstanceStorage().persist(deserializeDocument(document)); } public void writeDocument(String document, @@ -62,7 +62,7 @@ public class AdapterResolver extends AbstractResolver<AdapterDescription> { if (overrideDocument) { overrideProtocol(adapterDescription.getEventGrounding()); } - getNoSqlStore().getAdapterInstanceStorage().storeAdapter(adapterDescription); + getNoSqlStore().getAdapterInstanceStorage().persist(adapterDescription); } @Override diff --git a/streampipes-data-export/src/main/java/org/apache/streampipes/export/resolver/PipelineResolver.java b/streampipes-data-export/src/main/java/org/apache/streampipes/export/resolver/PipelineResolver.java index bd71f676de..53830c166f 100644 --- a/streampipes-data-export/src/main/java/org/apache/streampipes/export/resolver/PipelineResolver.java +++ b/streampipes-data-export/src/main/java/org/apache/streampipes/export/resolver/PipelineResolver.java @@ -30,7 +30,7 @@ public class PipelineResolver extends AbstractResolver<Pipeline> { @Override public Pipeline findDocument(String resourceId) { - return getNoSqlStore().getPipelineStorageAPI().getPipeline(resourceId); + return getNoSqlStore().getPipelineStorageAPI().getElementById(resourceId); } @Override @@ -58,7 +58,7 @@ public class PipelineResolver extends AbstractResolver<Pipeline> { @Override public void writeDocument(String document) throws JsonProcessingException { - getNoSqlStore().getPipelineStorageAPI().storePipeline(deserializeDocument(document)); + getNoSqlStore().getPipelineStorageAPI().persist(deserializeDocument(document)); } public void writeDocument(String document, @@ -81,7 +81,7 @@ public class PipelineResolver extends AbstractResolver<Pipeline> { }).collect(Collectors.toList())); } - getNoSqlStore().getPipelineStorageAPI().storePipeline(pipeline); + getNoSqlStore().getPipelineStorageAPI().persist(pipeline); } @Override diff --git a/streampipes-model-client/src/main/java/org/apache/streampipes/model/client/user/Permission.java b/streampipes-model-client/src/main/java/org/apache/streampipes/model/client/user/Permission.java index 72b9a1b31e..4e20ca0daa 100644 --- a/streampipes-model-client/src/main/java/org/apache/streampipes/model/client/user/Permission.java +++ b/streampipes-model-client/src/main/java/org/apache/streampipes/model/client/user/Permission.java @@ -18,6 +18,7 @@ package org.apache.streampipes.model.client.user; import org.apache.streampipes.model.shared.annotation.TsModel; +import org.apache.streampipes.model.shared.api.Storable; import com.fasterxml.jackson.annotation.JsonIgnore; import com.google.gson.annotations.SerializedName; @@ -26,7 +27,7 @@ import java.util.ArrayList; import java.util.List; @TsModel -public class Permission { +public class Permission implements Storable { protected @SerializedName("_id") String permissionId; protected @SerializedName("_rev") String rev; @@ -64,6 +65,16 @@ public class Permission { this.rev = rev; } + @Override + public String getElementId() { + return this.permissionId; + } + + @Override + public void setElementId(String elementId) { + this.permissionId = elementId; + } + public String getObjectInstanceId() { return objectInstanceId; } diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/VirtualSensor.java b/streampipes-model/src/main/java/org/apache/streampipes/model/VirtualSensor.java deleted file mode 100644 index 0034a96f84..0000000000 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/VirtualSensor.java +++ /dev/null @@ -1,50 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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.apache.streampipes.model; - -import org.apache.streampipes.model.pipeline.ElementComposition; - -public class VirtualSensor extends ElementComposition { - - protected String createdBy; - - protected int outputIndex; - - public VirtualSensor() { - super(); - } - - public String getCreatedBy() { - return createdBy; - } - - public void setCreatedBy(String createdBy) { - this.createdBy = createdBy; - } - - public int getOutputIndex() { - return outputIndex; - } - - public void setOutputIndex(int outputIndex) { - this.outputIndex = outputIndex; - } - - -} diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/ElementComposition.java b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/ElementComposition.java deleted file mode 100644 index 4e149a2aa7..0000000000 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/ElementComposition.java +++ /dev/null @@ -1,74 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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.apache.streampipes.model.pipeline; - -import org.apache.streampipes.model.SpDataStream; -import org.apache.streampipes.model.graph.DataProcessorInvocation; - -import java.util.ArrayList; -import java.util.List; - - -public abstract class ElementComposition { - - protected List<DataProcessorInvocation> sepas; - - protected List<SpDataStream> streams; - - protected String name; - protected String description; - - public ElementComposition() { - this.sepas = new ArrayList<>(); - this.streams = new ArrayList<>(); - } - - public List<DataProcessorInvocation> getSepas() { - return sepas; - } - - public void setSepas(List<DataProcessorInvocation> sepas) { - this.sepas = sepas; - } - - public List<SpDataStream> getStreams() { - return streams; - } - - public void setStreams(List<SpDataStream> streams) { - this.streams = streams; - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - public String getDescription() { - return description; - } - - public void setDescription(String description) { - this.description = description; - } - -} diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/Pipeline.java b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/Pipeline.java index 169ae4174e..790faf3eb5 100644 --- a/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/Pipeline.java +++ b/streampipes-model/src/main/java/org/apache/streampipes/model/pipeline/Pipeline.java @@ -18,8 +18,11 @@ package org.apache.streampipes.model.pipeline; +import org.apache.streampipes.model.SpDataStream; +import org.apache.streampipes.model.graph.DataProcessorInvocation; import org.apache.streampipes.model.graph.DataSinkInvocation; import org.apache.streampipes.model.shared.annotation.TsModel; +import org.apache.streampipes.model.shared.api.Storable; import com.fasterxml.jackson.annotation.JsonProperty; import com.google.gson.annotations.SerializedName; @@ -28,7 +31,7 @@ import java.util.ArrayList; import java.util.List; @TsModel -public class Pipeline extends ElementComposition { +public class Pipeline implements Storable { private List<DataSinkInvocation> actions; @@ -54,10 +57,19 @@ public class Pipeline extends ElementComposition { private @SerializedName("_rev") String rev; + protected List<DataProcessorInvocation> sepas; + + protected List<SpDataStream> streams; + + protected String name; + protected String description; + public Pipeline() { super(); this.actions = new ArrayList<>(); this.pipelineNotifications = new ArrayList<>(); + this.sepas = new ArrayList<>(); + this.streams = new ArrayList<>(); } public List<DataSinkInvocation> getActions() { @@ -92,7 +104,6 @@ public class Pipeline extends ElementComposition { this.publicElement = publicElement; } - public String getCreatedByUser() { return createdByUser; } @@ -117,6 +128,16 @@ public class Pipeline extends ElementComposition { this.rev = rev; } + @Override + public String getElementId() { + return pipelineId; + } + + @Override + public void setElementId(String elementId) { + this.pipelineId = elementId; + } + public long getCreatedAt() { return createdAt; } @@ -157,6 +178,38 @@ public class Pipeline extends ElementComposition { this.valid = valid; } + public List<DataProcessorInvocation> getSepas() { + return sepas; + } + + public void setSepas(List<DataProcessorInvocation> sepas) { + this.sepas = sepas; + } + + public List<SpDataStream> getStreams() { + return streams; + } + + public void setStreams(List<SpDataStream> streams) { + this.streams = streams; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + public String getDescription() { + return description; + } + + public void setDescription(String description) { + this.description = description; + } + public Pipeline clone() { Pipeline pipeline = new Pipeline(); pipeline.setName(name); @@ -174,6 +227,4 @@ public class Pipeline extends ElementComposition { return pipeline; } - - } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java index df3c09f954..923d6c942b 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/task/StorePipelineStatusTask.java @@ -63,7 +63,7 @@ public class StorePipelineStatusTask implements PipelineExecutionTask { pipeline.setRunning(true); pipeline.setStartedAt(new Date().getTime()); try { - getPipelineStorageApi().updatePipeline(pipeline); + getPipelineStorageApi().updateElement(pipeline); } catch (DocumentConflictException dce) { LOG.error("Could not update pipeline {}", pipeline.getPipelineId(), dce); } @@ -71,7 +71,7 @@ public class StorePipelineStatusTask implements PipelineExecutionTask { private void setPipelineStopped(Pipeline pipeline) { pipeline.setRunning(false); - getPipelineStorageApi().updatePipeline(pipeline); + getPipelineStorageApi().updateElement(pipeline); } private IPipelineStorage getPipelineStorageApi() { diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/extensions/AvailableExtensionsProvider.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/extensions/AvailableExtensionsProvider.java index 6d03aace13..f9112845f3 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/extensions/AvailableExtensionsProvider.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/extensions/AvailableExtensionsProvider.java @@ -107,7 +107,7 @@ public class AvailableExtensionsProvider { } private List<AdapterDescription> getAllAdapters() { - return storage.getAdapterDescriptionStorage().getAllAdapters(); + return storage.getAdapterDescriptionStorage().findAll(); } private List<SpDataStream> getAllDataStreams() { diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java index 84ab806c30..df79c78627 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/health/PipelineHealthCheck.java @@ -126,7 +126,7 @@ public class PipelineHealthCheck implements Runnable { pipelinesStats.attentionRequiredIncrease(); } currentPipeline.setPipelineNotifications(pipelineNotifications); - StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(currentPipeline); + StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updateElement(currentPipeline); } }); int healthNum = pipelinesStats.getRunningPipelines() - pipelinesStats.getFailedPipelines() @@ -240,7 +240,7 @@ public class PipelineHealthCheck implements Runnable { .INSTANCE .getNoSqlStore() .getPipelineStorageAPI() - .getAllPipelines(); + .findAll(); } private int getElementsCount(List<Pipeline> allPipelines){ diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java index 166d24c8e3..40ff88acf3 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/migration/PipelineElementMigrationManager.java @@ -69,7 +69,7 @@ public class PipelineElementMigrationManager extends AbstractMigrationManager im LOG.info("Received {} pipeline element migrations from extension service {}.", migrationConfigs.size(), extensionsServiceConfig.getServiceUrl()); - var availablePipelines = pipelineStorage.getAllPipelines(); + var availablePipelines = pipelineStorage.findAll(); if (!availablePipelines.isEmpty()) { LOG.info("Found {} available pipelines. Checking pipelines for applicable migrations...", availablePipelines.size() @@ -122,13 +122,13 @@ public class PipelineElementMigrationManager extends AbstractMigrationManager im .toList(); pipeline.setActions(migratedDataSinks); - pipelineStorage.updatePipeline(pipeline); + pipelineStorage.updateElement(pipeline); if (failedMigrations.isEmpty()) { LOG.info("Migration for pipeline successfully completed."); } else { // pass most recent version of pipeline - handleFailedMigrations(pipelineStorage.getPipeline(pipeline.getPipelineId()), failedMigrations); + handleFailedMigrations(pipelineStorage.getElementById(pipeline.getPipelineId()), failedMigrations); } } } @@ -166,10 +166,10 @@ public class PipelineElementMigrationManager extends AbstractMigrationManager im ).toList()); pipeline.setHealthStatus(PipelineHealthStatus.REQUIRES_ATTENTION); - pipelineStorage.updatePipeline(pipeline); + pipelineStorage.updateElement(pipeline); // get updated version of pipeline after modification - pipeline = pipelineStorage.getPipeline(pipeline.getPipelineId()); + pipeline = pipelineStorage.getElementById(pipeline.getPipelineId()); stopPipeline(pipeline); } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java index 83a26c7500..81d6215c88 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/operations/Operations.java @@ -93,7 +93,7 @@ public class Operations { public static List<PipelineOperationStatus> stopAllPipelines(boolean forceStop) { List<PipelineOperationStatus> status = new ArrayList<>(); List<Pipeline> pipelines = - StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().getAllPipelines(); + StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().findAll(); pipelines.forEach(p -> { if (p.isRunning()) { diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java index e849aaa7ea..e57675ecba 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/pipeline/PipelineManager.java @@ -44,7 +44,7 @@ public class PipelineManager { * @return all pipelines */ public static List<Pipeline> getAllPipelines() { - return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().getAllPipelines(); + return StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().findAll(); } /** @@ -54,7 +54,7 @@ public class PipelineManager { * @return pipeline resulting pipeline with given id */ public static Pipeline getPipeline(String pipelineId) { - return getPipelineStorage().getPipeline(pipelineId); + return getPipelineStorage().getElementById(pipelineId); } /** @@ -74,7 +74,7 @@ public class PipelineManager { Operations.storePipeline(pipeline); Permission permission = new PermissionManager().makePermission(pipeline, principalSid); - getPermissionStorage().addPermission(permission); + getPermissionStorage().persist(permission); return pipelineId; } @@ -114,7 +114,7 @@ public class PipelineManager { public static void deletePipeline(String pipelineId) { var pipeline = getPipeline(pipelineId); if (Objects.nonNull(pipeline)) { - getPipelineStorage().deletePipeline(pipelineId); + getPipelineStorage().deleteElementById(pipelineId); new NotificationsResourceManager().deleteNotificationsForPipeline(pipeline); } } diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java index 98d94c3ca9..88b03f667c 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/setup/CouchDbInstallationStep.java @@ -63,7 +63,6 @@ public class CouchDbInstallationStep extends InstallationStep { Utils.getCouchDbConnectionClient(); Utils.getCouchDbNotificationClient(); Utils.getCouchDbPipelineCategoriesClient(); - Utils.getCouchDbLabelClient(); logSuccess(getTitle()); } catch (Exception e) { @@ -75,7 +74,6 @@ public class CouchDbInstallationStep extends InstallationStep { addUserView(); addConnectionView(); addNotificationView(); - addLabelView(); addPipelineView(); } @@ -169,29 +167,6 @@ public class CouchDbInstallationStep extends InstallationStep { } } - private void addLabelView() { - try { - DesignDocument labelDocument = prepareDocument("_design/categoryId"); - Map<String, MapReduce> views = new HashMap<>(); - - MapReduce categoryIdFunction = new MapReduce(); - categoryIdFunction.setMap("function(doc) { if(doc.categoryId) { emit(doc.categoryId, doc); } }"); - - views.put("categoryId", categoryIdFunction); - - labelDocument.setViews(views); - Response resp = Utils.getCouchDbLabelClient().design().synchronizeWithDb(labelDocument); - - if (resp.getError() != null) { - logFailure(PREPARING_USERS_TEXT); - } else { - logSuccess(PREPARING_USERS_TEXT); - } - } catch (Exception e) { - logFailure(PREPARING_USERS_TEXT, e); - } - } - private void addConnectionView() { try { DesignDocument connectionDocument = prepareDocument("_design/connection"); diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java index 9d0f2b5824..843164b31a 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/storage/PipelineStorageService.java @@ -41,12 +41,12 @@ public class PipelineStorageService { public void updatePipeline() { preparePipeline(); - StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(pipeline); + StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updateElement(pipeline); } public void addPipeline() { preparePipeline(); - StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().store(pipeline); + StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().persist(pipeline); } private void preparePipeline() { diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineTemplateInvocationHandler.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineTemplateInvocationHandler.java index 519855126d..0848abdac9 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineTemplateInvocationHandler.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/template/PipelineTemplateInvocationHandler.java @@ -59,9 +59,9 @@ public class PipelineTemplateInvocationHandler { replaceStaticProperties(pipeline); Operations.storePipeline(pipeline); Permission permission = new PermissionManager().makePermission(pipeline, username); - StorageDispatcher.INSTANCE.getNoSqlStore().getPermissionStorage().addPermission(permission); + StorageDispatcher.INSTANCE.getNoSqlStore().getPermissionStorage().persist(permission); Pipeline storedPipeline = - StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().getPipeline(pipeline.getPipelineId()); + StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().getElementById(pipeline.getPipelineId()); return Operations.startPipeline(storedPipeline); } diff --git a/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/AdapterResourceManager.java b/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/AdapterResourceManager.java index a115ad829f..72c77121aa 100644 --- a/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/AdapterResourceManager.java +++ b/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/AdapterResourceManager.java @@ -42,7 +42,7 @@ public class AdapterResourceManager extends AbstractResourceManager<IAdapterStor public String encryptAndCreate(AdapterDescription adapterDescription) { AdapterDescription encryptedAdapterDescription = cloneAndEncrypt(adapterDescription); encryptedAdapterDescription.setRev(null); - return db.storeAdapter(encryptedAdapterDescription); + return db.persist(encryptedAdapterDescription).v; } /** @@ -51,11 +51,11 @@ public class AdapterResourceManager extends AbstractResourceManager<IAdapterStor * @param adapterDescription input adapter description */ public void encryptAndUpdate(AdapterDescription adapterDescription) { - db.updateAdapter(cloneAndEncrypt(adapterDescription)); + db.updateElement(cloneAndEncrypt(adapterDescription)); } public void delete(String elementId) { - db.deleteAdapter(elementId); + db.deleteElementById(elementId); } /** diff --git a/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/PermissionResourceManager.java b/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/PermissionResourceManager.java index d04a707aac..dc263ee381 100644 --- a/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/PermissionResourceManager.java +++ b/streampipes-resource-management/src/main/java/org/apache/streampipes/resource/management/PermissionResourceManager.java @@ -31,7 +31,7 @@ public class PermissionResourceManager extends AbstractResourceManager<IPermissi } public List<Permission> findAll() { - return db.getAllPermissions(); + return db.findAll(); } public List<Permission> findForObjectId(String objectInstanceId) { @@ -39,7 +39,7 @@ public class PermissionResourceManager extends AbstractResourceManager<IPermissi } public void create(Permission permission) { - db.addPermission(permission); + db.persist(permission); } public void createDefault(String objectInstanceId, @@ -55,10 +55,10 @@ public class PermissionResourceManager extends AbstractResourceManager<IPermissi } public void update(Permission permission) { - db.updatePermission(permission); + db.updateElement(permission); } public void delete(Permission permission) { - db.deletePermission(permission.getPermissionId()); + db.deleteElementById(permission.getPermissionId()); } } diff --git a/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/AdapterResourceManagerTest.java b/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/AdapterResourceManagerTest.java index 5e026e8ddd..d815855490 100644 --- a/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/AdapterResourceManagerTest.java +++ b/streampipes-resource-management/src/test/java/org/apache/streampipes/resource/management/AdapterResourceManagerTest.java @@ -37,6 +37,6 @@ public class AdapterResourceManagerTest { AdapterResourceManager adapterResourceManager = new AdapterResourceManager(storage); adapterResourceManager.encryptAndUpdate(new AdapterDescription()); - verify(storage, times(1)).updateAdapter(any()); + verify(storage, times(1)).updateElement(any()); } } diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementAsset.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementAsset.java index bd09bb173e..5eb9bf6b78 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementAsset.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineElementAsset.java @@ -61,7 +61,7 @@ public class PipelineElementAsset extends AbstractRestResource { var adapterDescription = StorageDispatcher.INSTANCE .getNoSqlStore() .getAdapterInstanceStorage() - .getAdapter(dataStream.getCorrespondingAdapterId()); + .getElementById(dataStream.getCorrespondingAdapterId()); appId = adapterDescription.getAppId(); } return ok(AssetManager.getAssetDocumentation(appId)); diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java index fae5851172..75e636cd6b 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/PipelineResource.java @@ -234,7 +234,7 @@ public class PipelineResource extends AbstractAuthGuardedRestResource { @PreAuthorize(AuthConstants.HAS_WRITE_PIPELINE_PRIVILEGE) public ResponseEntity<SuccessMessage> overwritePipeline(@PathVariable("pipelineId") String pipelineId, @RequestBody Pipeline pipeline) { - Pipeline storedPipeline = getPipelineStorage().getPipeline(pipelineId); + Pipeline storedPipeline = getPipelineStorage().getElementById(pipelineId); if (!storedPipeline.isRunning()) { storedPipeline.setStreams(pipeline.getStreams()); storedPipeline.setSepas(pipeline.getSepas()); diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java index 39425313a4..6d70b8d6e6 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/AdapterResource.java @@ -176,7 +176,7 @@ public class AdapterResource extends AbstractAdapterResource<AdapterMasterManage } } else if (!deleteAssociatedPipelines) { List<String> namesOfPipelinesUsingAdapter = - pipelinesUsingAdapter.stream().map(pipelineId -> pipelineStorageAPI.getPipeline(pipelineId).getName()) + pipelinesUsingAdapter.stream().map(pipelineId -> pipelineStorageAPI.getElementById(pipelineId).getName()) .collect( Collectors.toList()); return ResponseEntity.status(HttpStatus.SC_CONFLICT).body(String.join(", ", namesOfPipelinesUsingAdapter)); @@ -187,7 +187,7 @@ public class AdapterResource extends AbstractAdapterResource<AdapterMasterManage !permissionResourceManager.findForObjectId(pipelineId).stream().findFirst().map(Permission::getOwnerSid) // if a pipeline has no owner, pretend the owner is the user so the user can delete it .orElse(this.getAuthenticatedUserSid()).equals(this.getAuthenticatedUserSid())) - .map(pipelineId -> pipelineStorageAPI.getPipeline(pipelineId).getName()).collect(Collectors.toList()); + .map(pipelineId -> pipelineStorageAPI.getElementById(pipelineId).getName()).collect(Collectors.toList()); boolean isAdmin = SecurityContextHolder.getContext().getAuthentication().getAuthorities().stream() .anyMatch(r -> r.getAuthority().equals( Role.ROLE_ADMIN.name())); diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/dashboard/AbstractPipelineExtractionResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/dashboard/AbstractPipelineExtractionResource.java index 290953c575..476633090e 100644 --- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/dashboard/AbstractPipelineExtractionResource.java +++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/dashboard/AbstractPipelineExtractionResource.java @@ -46,7 +46,7 @@ public abstract class AbstractPipelineExtractionResource<T> extends AbstractRest protected List<T> extract(List<T> target, String appId) { getPipelineStorage() - .getAllPipelines() + .findAll() .forEach(pipeline -> { List<DataSinkInvocation> sinks = extractSink(pipeline, appId); sinks.forEach(sink -> target.add(convert(pipeline, sink))); diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java index aa378ffae5..f7f0c32d22 100644 --- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java +++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/PostStartupTask.java @@ -88,7 +88,7 @@ public class PostStartupTask implements Runnable { } private void startAllPreviouslyStoppedPipelines() { - var allPipelines = pipelineStorage.getAllPipelines(); + var allPipelines = pipelineStorage.findAll(); LOG.info("Checking for orphaned pipelines..."); List<Pipeline> orphanedPipelines = allPipelines .stream() @@ -123,9 +123,9 @@ public class PostStartupTask implements Runnable { PipelineOperationStatus status = Operations.startPipeline(pipeline); if (status.isSuccess()) { LOG.info("Pipeline {} successfully restarted", status.getPipelineName()); - Pipeline storedPipeline = getPipelineStorage().getPipeline(pipeline.getPipelineId()); + Pipeline storedPipeline = getPipelineStorage().getElementById(pipeline.getPipelineId()); storedPipeline.setRestartOnSystemReboot(restartOnReboot); - getPipelineStorage().updatePipeline(storedPipeline); + getPipelineStorage().updateElement(storedPipeline); } else { storeFailedRestartAttempt(pipeline); int failedAttemptCount = failedPipelines.get(pipeline.getPipelineId()); diff --git a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java index 33b306550a..588b194dfb 100644 --- a/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java +++ b/streampipes-service-core/src/main/java/org/apache/streampipes/service/core/StreamPipesCoreApplication.java @@ -213,7 +213,7 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { pipelinesToStop.forEach(pipeline -> { pipeline.setRestartOnSystemReboot(true); - StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updatePipeline(pipeline); + StorageDispatcher.INSTANCE.getNoSqlStore().getPipelineStorageAPI().updateElement(pipeline); }); LOG.info("Gracefully stopping all running pipelines..."); @@ -231,7 +231,7 @@ public class StreamPipesCoreApplication extends StreamPipesServiceBase { private List<Pipeline> getAllPipelines() { return getPipelineStorage() - .getAllPipelines(); + .findAll(); } private IPipelineStorage getPipelineStorage() { diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IAdapterStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IAdapterStorage.java index f2011bbc0c..36ecb20ac8 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IAdapterStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IAdapterStorage.java @@ -22,17 +22,7 @@ import org.apache.streampipes.model.connect.adapter.AdapterDescription; import java.util.List; -public interface IAdapterStorage extends CRUDStorage<AdapterDescription>{ - - List<AdapterDescription> getAllAdapters(); - - String storeAdapter(AdapterDescription adapter); - - void updateAdapter(AdapterDescription adapter); - - AdapterDescription getAdapter(String adapterId); - - void deleteAdapter(String adapterId); +public interface IAdapterStorage extends CRUDStorage<AdapterDescription> { AdapterDescription getFirstAdapterByAppId(String appId); diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IDataProcessorStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IDataProcessorStorage.java index e0b055f978..9b00f555b6 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IDataProcessorStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IDataProcessorStorage.java @@ -19,11 +19,7 @@ package org.apache.streampipes.storage.api; import org.apache.streampipes.model.graph.DataProcessorDescription; -import java.util.List; - public interface IDataProcessorStorage extends CRUDStorage<DataProcessorDescription> { DataProcessorDescription getFirstDataProcessorByAppId(String appId); - - List<DataProcessorDescription> getDataProcessorsByAppId(String appId); } diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IDataSinkStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IDataSinkStorage.java index d82c02ba6c..1890d1e0a2 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IDataSinkStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IDataSinkStorage.java @@ -19,11 +19,8 @@ package org.apache.streampipes.storage.api; import org.apache.streampipes.model.graph.DataSinkDescription; -import java.util.List; - public interface IDataSinkStorage extends CRUDStorage<DataSinkDescription> { DataSinkDescription getFirstDataSinkByAppId(String appId); - List<DataSinkDescription> getDataSinksByAppId(String appId); } diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPermissionStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPermissionStorage.java index fadcc46a01..8c019b4dc0 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPermissionStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPermissionStorage.java @@ -22,17 +22,7 @@ import org.apache.streampipes.model.client.user.Permission; import java.util.List; import java.util.Set; -public interface IPermissionStorage { - - List<Permission> getAllPermissions(); - - Permission getPermissionById(String permissionId); - - void addPermission(Permission permission); - - void updatePermission(Permission permission); - - void deletePermission(String permissionId); +public interface IPermissionStorage extends CRUDStorage<Permission> { Set<String> getObjectPermissions(List<String> sids); diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPipelineElementDescriptionStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPipelineElementDescriptionStorage.java index b67029147e..4de7465344 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPipelineElementDescriptionStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPipelineElementDescriptionStorage.java @@ -19,7 +19,6 @@ package org.apache.streampipes.storage.api; import org.apache.streampipes.model.SpDataStream; -import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.graph.DataProcessorDescription; import org.apache.streampipes.model.graph.DataSinkDescription; @@ -28,14 +27,10 @@ import java.util.List; public interface IPipelineElementDescriptionStorage { - boolean storeInvocablePipelineElement(InvocableStreamPipesEntity element); - boolean storeDataStream(SpDataStream stream); boolean storeDataProcessor(DataProcessorDescription processorDescription); - SpDataStream getDataStreamByAppId(String appId); - SpDataStream getDataStreamById(String rdfId); DataProcessorDescription getDataProcessorById(String rdfId); @@ -48,36 +43,20 @@ public interface IPipelineElementDescriptionStorage { AdapterDescription getAdapterById(String elementId); - AdapterDescription getAdapterByAppId(String appId); - List<SpDataStream> getAllDataStreams(); List<DataProcessorDescription> getAllDataProcessors(); - List<AdapterDescription> getAllAdapterDescriptions(); - boolean deleteDataStream(SpDataStream sep); - boolean deleteDataStream(String rdfId); - boolean deleteDataProcessor(DataProcessorDescription processorDescription); - boolean deleteDataProcessor(String rdfId); - - boolean deleteAdapterDescription(AdapterDescription adapterDescription); - - boolean deleteAdapterDescription(String elementId); - boolean exists(SpDataStream stream); boolean exists(AdapterDescription adapterDescription); boolean exists(DataProcessorDescription processorDescription); - boolean existsDataProcessorByAppId(String appId); - - boolean existsDataSinkByAppId(String appId); - boolean existsDataProcessor(String elementId); boolean existsDataStream(String elementId); @@ -98,8 +77,6 @@ public interface IPipelineElementDescriptionStorage { boolean deleteDataSink(DataSinkDescription sec); - boolean deleteDataSink(String rdfId); - boolean storeDataSink(DataSinkDescription sec); boolean storeAdapterDescription(AdapterDescription adapterDescription); diff --git a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPipelineStorage.java b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPipelineStorage.java index 155652997c..785c0187a4 100644 --- a/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPipelineStorage.java +++ b/streampipes-storage-api/src/main/java/org/apache/streampipes/storage/api/IPipelineStorage.java @@ -22,19 +22,8 @@ import org.apache.streampipes.model.pipeline.Pipeline; import java.util.List; -public interface IPipelineStorage { +public interface IPipelineStorage extends CRUDStorage<Pipeline> { List<String> getPipelinesUsingAdapter(String adapterId); - List<Pipeline> getAllPipelines(); - - void storePipeline(Pipeline pipeline); - - void updatePipeline(Pipeline pipeline); - - Pipeline getPipeline(String pipelineId); - - void deletePipeline(String pipelineId); - - void store(Pipeline object); } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java index 64e873a8aa..8f9cd5826b 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/CouchDbStorageManager.java @@ -191,7 +191,7 @@ public enum CouchDbStorageManager implements INoSqlStorage { @Override public IPermissionStorage getPermissionStorage() { - return new PermissionStorageImpl(); + return new PermissionStorageImpl("users/permissions"); } @Override diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/AdapterDescriptionStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/AdapterDescriptionStorageImpl.java index 2095bc618e..cd78b5af69 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/AdapterDescriptionStorageImpl.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/AdapterDescriptionStorageImpl.java @@ -20,71 +20,32 @@ package org.apache.streampipes.storage.couchdb.impl; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.storage.api.IAdapterStorage; -import org.apache.streampipes.storage.couchdb.dao.AbstractDao; -import org.apache.streampipes.storage.couchdb.dao.DbCommand; -import org.apache.streampipes.storage.couchdb.dao.FindCommand; import org.apache.streampipes.storage.couchdb.utils.Utils; import java.util.List; import java.util.NoSuchElementException; -import java.util.Optional; -public class AdapterDescriptionStorageImpl extends AbstractDao<AdapterDescription> implements IAdapterStorage { +public class AdapterDescriptionStorageImpl extends DefaultCrudStorage<AdapterDescription> implements IAdapterStorage { public AdapterDescriptionStorageImpl() { super(Utils::getCouchDbAdapterDescriptionClient, AdapterDescription.class); } - @Override - public List<AdapterDescription> getAllAdapters() { - return findAll(); - } - - @Override - public String storeAdapter(AdapterDescription adapter) { - return persist(adapter).v; - } - - @Override - public void updateAdapter(AdapterDescription adapter) { - couchDbClientSupplier.get().update(adapter); - } - - @Override - public AdapterDescription getAdapter(String adapterId) { - DbCommand<Optional<AdapterDescription>, AdapterDescription> cmd = - new FindCommand<>(couchDbClientSupplier, adapterId, AdapterDescription.class); - return cmd.execute().get(); - } - - @Override - public void deleteAdapter(String adapterId) { - - AdapterDescription adapterDescription = getAdapter(adapterId); - couchDbClientSupplier.get().remove(adapterDescription.getElementId(), adapterDescription.getRev()); - - } - @Override public AdapterDescription getFirstAdapterByAppId(String appId) { return this.findAll() - .stream() - .filter(p -> p.getAppId().equals(appId)) - .findFirst() - .orElseThrow(NoSuchElementException::new); + .stream() + .filter(p -> p.getAppId().equals(appId)) + .findFirst() + .orElseThrow(NoSuchElementException::new); } @Override public List<AdapterDescription> getAdaptersByAppId(String appId) { return this.findAll() - .stream() - .filter(p -> p.getAppId().equals(appId)) - .toList(); - } - - @Override - public AdapterDescription getElementById(String id) { - return findWithNullIfEmpty(id); + .stream() + .filter(p -> p.getAppId().equals(appId)) + .toList(); } @Override @@ -95,11 +56,6 @@ public class AdapterDescriptionStorageImpl extends AbstractDao<AdapterDescriptio return getElementById(element.getElementId()); } - @Override - public void deleteElement(AdapterDescription element) { - delete(element.getElementId()); - } - private String getCurrentRev(String elementId) { return find(elementId).get().getRev(); } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/AdapterInstanceStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/AdapterInstanceStorageImpl.java index a744bf7866..9b9f68eb3f 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/AdapterInstanceStorageImpl.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/AdapterInstanceStorageImpl.java @@ -20,81 +20,31 @@ package org.apache.streampipes.storage.couchdb.impl; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.storage.api.IAdapterStorage; -import org.apache.streampipes.storage.couchdb.dao.AbstractDao; -import org.apache.streampipes.storage.couchdb.dao.DbCommand; -import org.apache.streampipes.storage.couchdb.dao.FindCommand; import org.apache.streampipes.storage.couchdb.utils.Utils; import java.util.List; import java.util.NoSuchElementException; -import java.util.Optional; -public class AdapterInstanceStorageImpl extends AbstractDao<AdapterDescription> implements IAdapterStorage { +public class AdapterInstanceStorageImpl extends DefaultCrudStorage<AdapterDescription> implements IAdapterStorage { public AdapterInstanceStorageImpl() { super(Utils::getCouchDbAdapterInstanceClient, AdapterDescription.class); } - @Override - public List<AdapterDescription> getAllAdapters() { - return findAll(); - } - - @Override - public String storeAdapter(AdapterDescription adapter) { - return persist(adapter).v; - } - - @Override - public void updateAdapter(AdapterDescription adapter) { - couchDbClientSupplier.get().update(adapter); - } - - @Override - public AdapterDescription getAdapter(String adapterId) { - DbCommand<Optional<AdapterDescription>, AdapterDescription> cmd = - new FindCommand<>(couchDbClientSupplier, adapterId, AdapterDescription.class); - return cmd.execute().orElse(null); - } - - @Override - public void deleteAdapter(String adapterId) { - - AdapterDescription adapterDescription = getAdapter(adapterId); - couchDbClientSupplier.get().remove(adapterDescription.getElementId(), adapterDescription.getRev()); - - } - - @Override - public AdapterDescription getElementById(String id) { - return findWithNullIfEmpty(id); - } - @Override public AdapterDescription getFirstAdapterByAppId(String appId) { return this.findAll() - .stream() - .filter(p -> p.getAppId().equals(appId)) - .findFirst() - .orElseThrow(NoSuchElementException::new); + .stream() + .filter(p -> p.getAppId().equals(appId)) + .findFirst() + .orElseThrow(NoSuchElementException::new); } @Override public List<AdapterDescription> getAdaptersByAppId(String appId) { return this.findAll() - .stream() - .filter(p -> p.getAppId().equals(appId)) - .toList(); - } - - @Override - public AdapterDescription updateElement(AdapterDescription element) { - update(element); - return getElementById(element.getElementId()); - } - - @Override - public void deleteElement(AdapterDescription element) { - delete(element.getElementId()); + .stream() + .filter(p -> p.getAppId().equals(appId)) + .toList(); } } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataProcessorStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataProcessorStorageImpl.java index c6102fbf6b..f65fe325b4 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataProcessorStorageImpl.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataProcessorStorageImpl.java @@ -22,7 +22,6 @@ import org.apache.streampipes.storage.api.IDataProcessorStorage; import org.apache.streampipes.storage.couchdb.dao.AbstractDao; import org.apache.streampipes.storage.couchdb.utils.Utils; -import java.util.List; import java.util.NoSuchElementException; public class DataProcessorStorageImpl extends AbstractDao<DataProcessorDescription> implements IDataProcessorStorage { @@ -58,14 +57,6 @@ public class DataProcessorStorageImpl extends AbstractDao<DataProcessorDescripti .orElseThrow(NoSuchElementException::new); } - @Override - public List<DataProcessorDescription> getDataProcessorsByAppId(String appId) { - return this.findAll() - .stream() - .filter(p -> p.getAppId().equals(appId)) - .toList(); - } - private String getCurrentRev(String elementId) { return find(elementId).get().getRev(); } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataSinkStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataSinkStorageImpl.java index bfaa065dea..2fe0efe013 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataSinkStorageImpl.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataSinkStorageImpl.java @@ -22,8 +22,6 @@ import org.apache.streampipes.storage.api.IDataSinkStorage; import org.apache.streampipes.storage.couchdb.dao.AbstractDao; import org.apache.streampipes.storage.couchdb.utils.Utils; -import java.util.List; - public class DataSinkStorageImpl extends AbstractDao<DataSinkDescription> implements IDataSinkStorage { @@ -57,14 +55,6 @@ public class DataSinkStorageImpl extends AbstractDao<DataSinkDescription> implem .orElseThrow(IllegalArgumentException::new); } - @Override - public List<DataSinkDescription> getDataSinksByAppId(String appId) { - return this.findAll() - .stream() - .filter(s -> s.getAppId().equals(appId)) - .toList(); - } - private String getCurrentRev(String elementId) { return find(elementId).get().getRev(); } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DefaultViewCrudStorage.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DefaultViewCrudStorage.java index a28c70147f..708682fab5 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DefaultViewCrudStorage.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DefaultViewCrudStorage.java @@ -27,7 +27,7 @@ import java.util.function.Supplier; public class DefaultViewCrudStorage<T extends Storable> extends DefaultCrudStorage<T> { - private final String viewName; + protected final String viewName; public DefaultViewCrudStorage(Supplier<CouchDbClient> couchDbClientSupplier, Class<T> clazz, diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PermissionStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PermissionStorageImpl.java index 401d0a9d61..622f24481d 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PermissionStorageImpl.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PermissionStorageImpl.java @@ -19,44 +19,16 @@ package org.apache.streampipes.storage.couchdb.impl; import org.apache.streampipes.model.client.user.Permission; import org.apache.streampipes.storage.api.IPermissionStorage; -import org.apache.streampipes.storage.couchdb.dao.CrudDao; import org.apache.streampipes.storage.couchdb.utils.Utils; import java.util.List; import java.util.Set; import java.util.stream.Collectors; -public class PermissionStorageImpl extends CrudDao implements IPermissionStorage { +public class PermissionStorageImpl extends DefaultViewCrudStorage<Permission> implements IPermissionStorage { - private String viewName = "users/permissions"; - - public PermissionStorageImpl() { - super(Utils::getCouchDbUserClient); - } - - @Override - public List<Permission> getAllPermissions() { - return findAll(viewName, Permission.class); - } - - @Override - public Permission getPermissionById(String permissionId) { - return findWithNullIfEmpty(permissionId, Permission.class); - } - - @Override - public void addPermission(Permission permission) { - persist(permission, Permission.class); - } - - @Override - public void updatePermission(Permission permission) { - update(permission, Permission.class); - } - - @Override - public void deletePermission(String permissionId) { - delete(permissionId, Permission.class); + public PermissionStorageImpl(String viewName) { + super(Utils::getCouchDbUserClient, Permission.class, viewName); } public Set<String> getObjectPermissions(List<String> principalSids) { diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PipelineElementDescriptionStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PipelineElementDescriptionStorageImpl.java index 7a2a291b52..50a6da5e3e 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PipelineElementDescriptionStorageImpl.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PipelineElementDescriptionStorageImpl.java @@ -18,7 +18,6 @@ package org.apache.streampipes.storage.couchdb.impl; import org.apache.streampipes.model.SpDataStream; -import org.apache.streampipes.model.base.InvocableStreamPipesEntity; import org.apache.streampipes.model.connect.adapter.AdapterDescription; import org.apache.streampipes.model.graph.DataProcessorDescription; import org.apache.streampipes.model.graph.DataSinkDescription; @@ -29,7 +28,6 @@ import org.apache.streampipes.storage.api.IDataStreamStorage; import org.apache.streampipes.storage.api.IPipelineElementDescriptionStorage; import java.util.List; -import java.util.NoSuchElementException; public class PipelineElementDescriptionStorageImpl implements IPipelineElementDescriptionStorage { @@ -45,12 +43,6 @@ public class PipelineElementDescriptionStorageImpl implements IPipelineElementDe this.adapterStorage = new AdapterDescriptionStorageImpl(); } - @Override - public boolean storeInvocablePipelineElement(InvocableStreamPipesEntity element) { - // TODO Check if this is needed - return false; - } - @Override public boolean storeDataStream(SpDataStream stream) { this.dataStreamStorage.persist(stream); @@ -63,11 +55,6 @@ public class PipelineElementDescriptionStorageImpl implements IPipelineElementDe return true; } - @Override - public SpDataStream getDataStreamByAppId(String appId) { - return this.dataStreamStorage.getDataStreamByAppId(appId); - } - @Override public SpDataStream getDataStreamById(String rdfId) { return this.dataStreamStorage.getElementById(rdfId); @@ -98,11 +85,6 @@ public class PipelineElementDescriptionStorageImpl implements IPipelineElementDe return adapterStorage.getElementById(elementId); } - @Override - public AdapterDescription getAdapterByAppId(String appId) { - return this.adapterStorage.getFirstAdapterByAppId(appId); - } - @Override public List<SpDataStream> getAllDataStreams() { return this.dataStreamStorage.findAll(); @@ -113,45 +95,18 @@ public class PipelineElementDescriptionStorageImpl implements IPipelineElementDe return this.dataProcessorStorage.findAll(); } - @Override - public List<AdapterDescription> getAllAdapterDescriptions() { - return null; - } - @Override public boolean deleteDataStream(SpDataStream sep) { this.dataStreamStorage.deleteElement(sep); return true; } - @Override - public boolean deleteDataStream(String rdfId) { - return deleteDataStream(getDataStreamById(rdfId)); - } - @Override public boolean deleteDataProcessor(DataProcessorDescription processorDescription) { this.dataProcessorStorage.deleteElement(processorDescription); return true; } - @Override - public boolean deleteDataProcessor(String rdfId) { - return deleteDataProcessor(getDataProcessorById(rdfId)); - } - - @Override - public boolean deleteAdapterDescription(AdapterDescription adapterDescription) { - adapterStorage.deleteAdapter(adapterDescription.getElementId()); - return true; - } - - @Override - public boolean deleteAdapterDescription(String elementId) { - adapterStorage.deleteAdapter(elementId); - return true; - } - @Override public boolean exists(SpDataStream stream) { return getEventStreamById(stream.getElementId()) != null; @@ -167,26 +122,6 @@ public class PipelineElementDescriptionStorageImpl implements IPipelineElementDe return getDataProcessorById(processorDescription.getElementId()) != null; } - @Override - public boolean existsDataProcessorByAppId(String appId) { - try { - getDataProcessorByAppId(appId); - return true; - } catch (NoSuchElementException e) { - return false; - } - } - - @Override - public boolean existsDataSinkByAppId(String appId) { - try { - getDataSinkByAppId(appId); - return true; - } catch (NoSuchElementException e) { - return false; - } - } - @Override public boolean existsDataProcessor(String elementId) { return getDataProcessorById(elementId) != null; @@ -242,11 +177,6 @@ public class PipelineElementDescriptionStorageImpl implements IPipelineElementDe return true; } - @Override - public boolean deleteDataSink(String rdfId) { - return deleteDataSink(getDataSinkById(rdfId)); - } - @Override public boolean storeDataSink(DataSinkDescription sec) { this.dataSinkStorage.persist(sec); @@ -255,7 +185,7 @@ public class PipelineElementDescriptionStorageImpl implements IPipelineElementDe @Override public boolean storeAdapterDescription(AdapterDescription adapterDescription) { - this.adapterStorage.storeAdapter(adapterDescription); + this.adapterStorage.persist(adapterDescription); return true; } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PipelineStorageImpl.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PipelineStorageImpl.java index 03cae5751b..afc5164c3e 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PipelineStorageImpl.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/PipelineStorageImpl.java @@ -20,7 +20,6 @@ package org.apache.streampipes.storage.couchdb.impl; import org.apache.streampipes.model.pipeline.Pipeline; import org.apache.streampipes.storage.api.IPipelineStorage; -import org.apache.streampipes.storage.couchdb.dao.AbstractDao; import org.apache.streampipes.storage.couchdb.utils.Utils; import com.google.gson.JsonObject; @@ -29,7 +28,7 @@ import java.util.ArrayList; import java.util.List; import java.util.stream.Collectors; -public class PipelineStorageImpl extends AbstractDao<Pipeline> implements IPipelineStorage { +public class PipelineStorageImpl extends DefaultCrudStorage<Pipeline> implements IPipelineStorage { private static final String ADAPTER_VIEW = "adapters/used-adapters"; private static final String ALL_PIPELINES_VIEW = "pipelines/all"; @@ -50,7 +49,7 @@ public class PipelineStorageImpl extends AbstractDao<Pipeline> implements IPipel } @Override - public List<Pipeline> getAllPipelines() { + public List<Pipeline> findAll() { List<Pipeline> pipelines = findAll(ALL_PIPELINES_VIEW); List<Pipeline> result = new ArrayList<>(); @@ -61,29 +60,4 @@ public class PipelineStorageImpl extends AbstractDao<Pipeline> implements IPipel } return result; } - - @Override - public void storePipeline(Pipeline pipeline) { - persist(pipeline); - } - - @Override - public void updatePipeline(Pipeline pipeline) { - update(pipeline); - } - - @Override - public Pipeline getPipeline(String pipelineId) { - return findWithNullIfEmpty(pipelineId); - } - - @Override - public void deletePipeline(String pipelineId) { - delete(pipelineId); - } - - @Override - public void store(Pipeline object) { - persist(object); - } } diff --git a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java index a9e1be34f9..1801df15d8 100644 --- a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java +++ b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/utils/Utils.java @@ -66,14 +66,6 @@ public class Utils { return getCouchDbGsonClient("general-configuration"); } - public static CouchDbClient getCouchDbExtensionsStorage() { - return getCouchDbGsonClient("extensions-services"); - } - - public static CouchDbClient getCouchDbLabelClient() { - return getCouchDbGsonClient("label"); - } - public static CouchDbClient getCouchDbFileMetadataClient() { return getCouchDbGsonClient("filemetadata"); }
