This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 30edcbb341 chore: Cleanup persistence layer (#2974)
30edcbb341 is described below
commit 30edcbb341e5e9dcf4321127478c1b5c50142b44
Author: Dominik Riemer <[email protected]>
AuthorDate: Sat Jun 29 13:28:19 2024 +0200
chore: Cleanup persistence layer (#2974)
* chore: Cleanup persistence layer
* Further cleanup
---
.../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 | 23 +------
.../storage/couchdb/impl/DataSinkStorageImpl.java | 23 +------
.../couchdb/impl/DataStreamStorageImpl.java | 13 +---
.../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 ---
50 files changed, 164 insertions(+), 589 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..3c02040549 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
@@ -19,24 +19,18 @@ package org.apache.streampipes.storage.couchdb.impl;
import org.apache.streampipes.model.graph.DataProcessorDescription;
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 {
+public class DataProcessorStorageImpl extends
DefaultCrudStorage<DataProcessorDescription>
+ implements IDataProcessorStorage {
public DataProcessorStorageImpl() {
super(Utils::getCouchDbDataProcessorDescriptionClient,
DataProcessorDescription.class);
}
- @Override
- public DataProcessorDescription getElementById(String s) {
- return findWithNullIfEmpty(s);
- }
-
@Override
public DataProcessorDescription updateElement(DataProcessorDescription
element) {
element.setRev(getCurrentRev(element.getElementId()));
@@ -44,11 +38,6 @@ public class DataProcessorStorageImpl extends
AbstractDao<DataProcessorDescripti
return getElementById(element.getElementId());
}
- @Override
- public void deleteElement(DataProcessorDescription element) {
- delete(element.getElementId());
- }
-
@Override
public DataProcessorDescription getFirstDataProcessorByAppId(String appId) {
return this.findAll()
@@ -58,14 +47,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..1e3587fd2f 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
@@ -19,23 +19,15 @@ package org.apache.streampipes.storage.couchdb.impl;
import org.apache.streampipes.model.graph.DataSinkDescription;
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 {
+public class DataSinkStorageImpl extends
DefaultCrudStorage<DataSinkDescription> implements IDataSinkStorage {
public DataSinkStorageImpl() {
super(Utils::getCouchDbDataSinkDescriptionClient,
DataSinkDescription.class);
}
- @Override
- public DataSinkDescription getElementById(String s) {
- return findWithNullIfEmpty(s);
- }
-
@Override
public DataSinkDescription updateElement(DataSinkDescription element) {
element.setRev(getCurrentRev(element.getElementId()));
@@ -43,11 +35,6 @@ public class DataSinkStorageImpl extends
AbstractDao<DataSinkDescription> implem
return getElementById(element.getElementId());
}
- @Override
- public void deleteElement(DataSinkDescription element) {
- delete(element.getElementId());
- }
-
@Override
public DataSinkDescription getFirstDataSinkByAppId(String appId) {
return this.findAll()
@@ -57,14 +44,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/DataStreamStorageImpl.java
b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataStreamStorageImpl.java
index d035f1f6b1..121f40c53d 100644
---
a/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataStreamStorageImpl.java
+++
b/streampipes-storage-couchdb/src/main/java/org/apache/streampipes/storage/couchdb/impl/DataStreamStorageImpl.java
@@ -19,20 +19,14 @@ package org.apache.streampipes.storage.couchdb.impl;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.storage.api.IDataStreamStorage;
-import org.apache.streampipes.storage.couchdb.dao.AbstractDao;
import org.apache.streampipes.storage.couchdb.utils.Utils;
-public class DataStreamStorageImpl extends AbstractDao<SpDataStream>
implements IDataStreamStorage {
+public class DataStreamStorageImpl extends DefaultCrudStorage<SpDataStream>
implements IDataStreamStorage {
public DataStreamStorageImpl() {
super(Utils::getCouchDbDataStreamDescriptionClient, SpDataStream.class);
}
- @Override
- public SpDataStream getElementById(String s) {
- return findWithNullIfEmpty(s);
- }
-
@Override
public SpDataStream updateElement(SpDataStream element) {
element.setRev(getCurrentRev(element.getElementId()));
@@ -40,11 +34,6 @@ public class DataStreamStorageImpl extends
AbstractDao<SpDataStream> implements
return getElementById(element.getElementId());
}
- @Override
- public void deleteElement(SpDataStream element) {
- delete(element.getElementId());
- }
-
@Override
public SpDataStream getDataStreamByAppId(String appId) {
return this.findAll()
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");
}
