You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2021/10/11 14:26:09 UTC
[incubator-streampipes] 02/02: [STREAMPIPES-438] Harmonize Model
Submitter
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch STREAMPIPES-438
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
commit 3e4aec4515ff967a75b6023d446f9ef6bfb01639
Author: Philipp Zehnder <ze...@fzi.de>
AuthorDate: Mon Oct 11 16:25:38 2021 +0200
[STREAMPIPES-438] Harmonize Model Submitter
---
.../master/management/AdapterMasterManagement.java | 10 ----
.../master/management/SourcesManagement.java | 54 +++++-----------------
.../master/management/WorkerRestClient.java | 2 +-
.../master/management/SourcesManagementTest.java | 39 ----------------
.../manager/execution/http/GraphSubmitter.java | 2 +-
.../rest/impl/connect/SourcesResource.java | 9 ++--
ui/cypress/tests/adapter/InfluxDbAdapter.ts | 4 +-
ui/cypress/tests/adapter/KafkaAdapter.ts | 3 +-
ui/cypress/tests/adapter/MqttAdapter.ts | 3 +-
ui/cypress/tests/adapter/MySQLDbAdapter.ts | 5 +-
ui/cypress/tests/adapter/fileStream.ts | 1 -
ui/cypress/tests/adapter/machineDataSimulator.ts | 14 +-----
ui/cypress/tests/adapter/persistInDataLake.ts | 18 +++++---
13 files changed, 35 insertions(+), 129 deletions(-)
diff --git a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/AdapterMasterManagement.java b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/AdapterMasterManagement.java
index a6d28f6..f7ed6bb 100644
--- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/AdapterMasterManagement.java
+++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/AdapterMasterManagement.java
@@ -28,7 +28,6 @@ import org.apache.streampipes.manager.storage.UserService;
import org.apache.streampipes.manager.verification.DataStreamVerifier;
import org.apache.streampipes.model.SpDataStream;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
-import org.apache.streampipes.model.connect.adapter.AdapterSetDescription;
import org.apache.streampipes.model.connect.adapter.AdapterStreamDescription;
import org.apache.streampipes.model.grounding.EventGrounding;
import org.apache.streampipes.model.util.Cloner;
@@ -176,15 +175,6 @@ public class AdapterMasterManagement {
return allAdapters;
}
- public void stopSetAdapter(String elementId,
- String baseUrl,
- AdapterInstanceStorageImpl adapterStorage) throws AdapterException {
-
- AdapterSetDescription ad = (AdapterSetDescription) adapterStorage.getAdapter(elementId);
-
- WorkerRestClient.stopSetAdapter(baseUrl, ad);
- }
-
public void stopStreamAdapter(String elementId) throws AdapterException {
AdapterDescription ad = adapterInstanceStorage.getAdapter(elementId);
diff --git a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java
index 8e3e284..1a71453 100644
--- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java
+++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/SourcesManagement.java
@@ -37,18 +37,15 @@ import org.slf4j.LoggerFactory;
import java.net.URISyntaxException;
import java.util.Arrays;
-import java.util.List;
public class SourcesManagement {
private Logger logger = LoggerFactory.getLogger(SourcesManagement.class);
private AdapterInstanceStorageImpl adapterInstanceStorage;
- private WorkerUrlProvider workerUrlProvider;
public SourcesManagement(AdapterInstanceStorageImpl adapterStorage) {
this.adapterInstanceStorage = adapterStorage;
- this.workerUrlProvider = new WorkerUrlProvider();
}
public SourcesManagement() {
@@ -70,48 +67,19 @@ public class SourcesManagement {
}
/**
- * @param streamId
* @param runningInstanceId
* @throws AdapterException
* @throws NoServiceEndpointsAvailableException
*/
- public void detachAdapter(String streamId,
- String runningInstanceId) throws AdapterException, NoServiceEndpointsAvailableException {
- AdapterSetDescription adapterDescription = (AdapterSetDescription) getAdapterDescriptionById(streamId);
-
- String newId = adapterDescription.getElementId() + "/streams/" + runningInstanceId;
- adapterDescription.setElementId(newId);
-
- String newUrl = getAdapterUrl(streamId);
- WorkerRestClient.stopSetAdapter(newUrl, adapterDescription);
- }
-
- private String getAdapterUrl(String streamId) throws NoServiceEndpointsAvailableException {
- String appId = "";
- List<AdapterDescription> adapterDescriptions = this.adapterInstanceStorage.getAllAdapters();
- for (AdapterDescription ad : adapterDescriptions) {
- if (ad.getElementId().contains(streamId)) {
- appId = ad.getAppId();
- }
- }
-
- return workerUrlProvider.getWorkerBaseUrl(appId);
-
- }
-
- private AdapterDescription getAdapterDescriptionById(String id) {
- AdapterDescription adapterDescription = null;
- List<AdapterDescription> allAdapters = adapterInstanceStorage.getAllAdapters();
- for (AdapterDescription a : allAdapters) {
- if (a.getElementId().equals(id)) {
- adapterDescription = a;
- }
+ public void detachAdapter(String elementId, String runningInstanceId) throws AdapterException, NoServiceEndpointsAvailableException {
+ AdapterSetDescription ad = (AdapterSetDescription) getAndDecryptAdapter(elementId);
+ try {
+ String baseUrl = WorkerPaths.findEndpointUrl(ad.getAppId());
+ ad.setElementId(ad.getElementId() + "/streams/" + runningInstanceId);
+ WorkerRestClient.stopSetAdapter(baseUrl, ad);
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
}
- AdapterDescription decryptedAdapterDescription =
- new AdapterEncryptionService(new Cloner()
- .adapterDescription(adapterDescription)).decrypt();
-
- return decryptedAdapterDescription;
}
public SpDataStream createAdapterDataStream(AdapterDescription adapterDescription) {
@@ -120,8 +88,10 @@ public class SourcesManagement {
if (adapterDescription instanceof AdapterSetDescription) {
ds = ((AdapterSetDescription) adapterDescription).getDataSet();
EventGrounding eg = new EventGrounding();
- eg.setTransportProtocols(Arrays.asList(SupportedProtocols.kafka(), SupportedProtocols.jms(),
- SupportedProtocols.mqtt()));
+ eg.setTransportProtocols(
+ Arrays.asList(SupportedProtocols.kafka(),
+ SupportedProtocols.jms(),
+ SupportedProtocols.mqtt()));
eg.setTransportFormats(Arrays.asList(TransportFormatGenerator.getTransportFormat()));
((SpDataSet) ds).setSupportedGrounding(eg);
} else {
diff --git a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/WorkerRestClient.java b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/WorkerRestClient.java
index 2240797..74bdfe2 100644
--- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/WorkerRestClient.java
+++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/WorkerRestClient.java
@@ -124,7 +124,7 @@ public class WorkerRestClient {
// Stop execution of adapter
try {
- logger.info("Trying to stopAdapter adapter on endpoint: " + url);
+ logger.info("Trying to stop adapter on endpoint: " + url);
String adapterDescription = JacksonSerializer.getObjectMapper().writeValueAsString(ad);
diff --git a/streampipes-connect-container-master/src/test/java/org/apache/streampipes/connect/container/master/management/SourcesManagementTest.java b/streampipes-connect-container-master/src/test/java/org/apache/streampipes/connect/container/master/management/SourcesManagementTest.java
index fb9db01..d9feb52 100644
--- a/streampipes-connect-container-master/src/test/java/org/apache/streampipes/connect/container/master/management/SourcesManagementTest.java
+++ b/streampipes-connect-container-master/src/test/java/org/apache/streampipes/connect/container/master/management/SourcesManagementTest.java
@@ -18,7 +18,6 @@
package org.apache.streampipes.connect.container.master.management;
-import org.apache.streampipes.connect.api.exception.AdapterException;
import org.apache.streampipes.model.SpDataSet;
import org.apache.streampipes.model.connect.adapter.AdapterDescription;
import org.apache.streampipes.model.connect.adapter.GenericAdapterSetDescription;
@@ -68,44 +67,6 @@ public class SourcesManagementTest {
}
- @Ignore
- @Test(expected = AdapterException.class)
- public void addAdapterFail() throws Exception {
- AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class);
- when(adapterStorage.getAllAdapters()).thenReturn(getAdapterDescriptionList());
- SourcesManagement sourcesManagement = new SourcesManagement(adapterStorage);
-
- org.powermock.api.mockito.PowerMockito.doThrow(new AdapterException()).when(WorkerRestClient.class, "stopSetAdapter", anyString(), any());
-
- sourcesManagement.detachAdapter( ID, "id1");
- }
-
- @Ignore
- @Test
- public void detachAdapterSuccess() throws Exception {
- AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class);
- when(adapterStorage.getAllAdapters()).thenReturn(getAdapterDescriptionList());
- SourcesManagement sourcesManagement = new SourcesManagement(adapterStorage);
- doNothing().when(WorkerRestClient.class, "stopSetAdapter", anyString(), any());
-
- sourcesManagement.detachAdapter(ID, "id1");
-
- verify(adapterStorage, times(1)).getAllAdapters();
- verifyStatic(WorkerRestClient.class, times(1));
- WorkerRestClient.stopSetAdapter(eq("/"), any());
- }
-
- @Ignore
- @Test(expected = AdapterException.class)
- public void detachAdapterFail() throws Exception {
- AdapterInstanceStorageImpl adapterStorage = mock(AdapterInstanceStorageImpl.class);
- when(adapterStorage.getAllAdapters()).thenReturn(getAdapterDescriptionList());
- SourcesManagement sourcesManagement = new SourcesManagement(adapterStorage);
- org.powermock.api.mockito.PowerMockito.doThrow(new AdapterException()).when(WorkerRestClient.class, "stopSetAdapter", anyString(), any());
-
- sourcesManagement.detachAdapter( ID, "id1");
- }
-
private List<AdapterDescription> getAdapterDescriptionList() {
GenericAdapterSetDescription adapterSetDescription = new GenericAdapterSetDescription();
diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java
index 8074b4d..a215f6c 100644
--- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java
+++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/execution/http/GraphSubmitter.java
@@ -124,7 +124,7 @@ public class GraphSubmitter {
}
private PipelineElementStatus performDetach(SpDataSet dataset) {
- String endpointUrl = dataset.getSelectedEndpointUrl() + "/" + dataset.getDatasetInvocationId();
+ String endpointUrl = dataset.getSelectedEndpointUrl() + "/" + dataset.getCorrespondingAdapterId() + "/" + dataset.getDatasetInvocationId();
return new HttpRequestBuilder(dataset, endpointUrl).detach();
}
}
diff --git a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/SourcesResource.java b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/SourcesResource.java
index 00293d4..1729e31 100644
--- a/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/SourcesResource.java
+++ b/streampipes-rest/src/main/java/org/apache/streampipes/rest/impl/connect/SourcesResource.java
@@ -58,16 +58,15 @@ public class SourcesResource extends AbstractAdapterResource<SourcesManagement>
}
@DELETE
- @Path("/{streamId}/{runningInstanceId}")
+ @Path("{adapterId}/{runningInstanceId}")
@Produces(MediaType.APPLICATION_JSON)
- public Response detach(@PathParam("streamId") String elementId,
- @PathParam("runningInstanceId") String runningInstanceId) {
- String responseMessage = "Instance of set id: " + elementId + " with instance id: "+ runningInstanceId + " successfully started";
+ public Response detach(@PathParam("adapterId") String elementId, @PathParam("runningInstanceId") String runningInstanceId) {
+ String responseMessage = "Instance id: "+ runningInstanceId + " successfully started";
try {
managementService.detachAdapter(elementId, runningInstanceId);
} catch (AdapterException | NoServiceEndpointsAvailableException e) {
- LOG.error("Could not set set id "+ elementId + " with instance id: "+ runningInstanceId, e);
+ LOG.error("Could not detach instance id: "+ runningInstanceId, e);
return fail();
}
diff --git a/ui/cypress/tests/adapter/InfluxDbAdapter.ts b/ui/cypress/tests/adapter/InfluxDbAdapter.ts
index d1018ce..48a0400 100644
--- a/ui/cypress/tests/adapter/InfluxDbAdapter.ts
+++ b/ui/cypress/tests/adapter/InfluxDbAdapter.ts
@@ -22,7 +22,7 @@ import { PipelineElementBuilder } from '../../support/builder/PipelineElementBui
import { AdapterUtils } from '../../support/utils/AdapterUtils';
import { PipelineUtils } from '../../support/utils/PipelineUtils';
-describe('Test Random Data Simulator Stream Adapter', () => {
+describe('Test InfluxDB Integration', () => {
before('Setup Test', () => {
cy.initStreamPipesTest();
});
@@ -59,8 +59,6 @@ describe('Test Random Data Simulator Stream Adapter', () => {
const adapterInput = SpecificAdapterBuilder
.create('InfluxDB_Stream_Adapter')
.setName('InfluxDB Adapter')
- .setTimestampProperty('time')
- .setStoreInDataLake()
.addInput('input', 'influxDbHost', 'http://localhost')
.addInput('input', 'influxDbPort', '8086')
.addInput('input', 'influxDbDatabase', 'sp')
diff --git a/ui/cypress/tests/adapter/KafkaAdapter.ts b/ui/cypress/tests/adapter/KafkaAdapter.ts
index 089ff23..1125c73 100644
--- a/ui/cypress/tests/adapter/KafkaAdapter.ts
+++ b/ui/cypress/tests/adapter/KafkaAdapter.ts
@@ -23,7 +23,7 @@ import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
import { PipelineUtils } from '../../support/utils/PipelineUtils';
-describe('Test Random Data Simulator Stream Adapter', () => {
+describe('Test Kafka Integration', () => {
before('Setup Test', () => {
cy.initStreamPipesTest();
});
@@ -58,7 +58,6 @@ describe('Test Random Data Simulator Stream Adapter', () => {
.create('Apache_Kafka')
.setName('Kafka4')
.setTimestampProperty('timestamp')
- .setStoreInDataLake()
.addProtocolInput('select', 'Unauthenticated', 'check')
.addProtocolInput('input', 'host', 'localhost')
.addProtocolInput('input', 'port', '9094')
diff --git a/ui/cypress/tests/adapter/MqttAdapter.ts b/ui/cypress/tests/adapter/MqttAdapter.ts
index 1fdd730..c5ecf4c 100644
--- a/ui/cypress/tests/adapter/MqttAdapter.ts
+++ b/ui/cypress/tests/adapter/MqttAdapter.ts
@@ -23,7 +23,7 @@ import { PipelineBuilder } from '../../support/builder/PipelineBuilder';
import { PipelineElementBuilder } from '../../support/builder/PipelineElementBuilder';
import { PipelineUtils } from '../../support/utils/PipelineUtils';
-describe('Test Random Data Simulator Stream Adapter', () => {
+describe('Test MQTT Integration', () => {
before('Setup Test', () => {
cy.initStreamPipesTest();
});
@@ -56,7 +56,6 @@ describe('Test Random Data Simulator Stream Adapter', () => {
.create('MQTT')
.setName('Adapter Mqtt')
.setTimestampProperty('timestamp')
- .setStoreInDataLake()
.addProtocolInput('select', 'Unauthenticated', 'check')
.addProtocolInput('input', 'broker_url', 'tcp://localhost:1883')
.addProtocolInput('input', 'topic', topicname)
diff --git a/ui/cypress/tests/adapter/MySQLDbAdapter.ts b/ui/cypress/tests/adapter/MySQLDbAdapter.ts
index 39f2cb1..0055e68 100644
--- a/ui/cypress/tests/adapter/MySQLDbAdapter.ts
+++ b/ui/cypress/tests/adapter/MySQLDbAdapter.ts
@@ -22,7 +22,7 @@ import { PipelineElementBuilder } from '../../support/builder/PipelineElementBui
import { AdapterUtils } from '../../support/utils/AdapterUtils';
import { PipelineUtils } from '../../support/utils/PipelineUtils';
-describe('Test Random Data Simulator Stream Adapter', () => {
+describe('Test MySQL Integration', () => {
before('Setup Test', () => {
cy.initStreamPipesTest();
});
@@ -32,7 +32,7 @@ describe('Test Random Data Simulator Stream Adapter', () => {
AdapterUtils.addMachineDataSimulator(simulatorAdapterName);
- const topicname = 'cypresstopic';
+ const topicname = 'cypresstopic1';
const pipelineInput = PipelineBuilder.create('Pipeline Test')
.addSource(simulatorAdapterName)
.addSink(
@@ -51,7 +51,6 @@ describe('Test Random Data Simulator Stream Adapter', () => {
.create('MySql_Stream_Adapter')
.setName('MySQL Adapter')
.setTimestampProperty('timestamp')
- .setStoreInDataLake()
.addInput('input', 'mysqlHost', 'localhost')
.addInput('input', 'mysqlUser', 'root')
.addInput('input', 'mysqlPassword', '7uc4rAymrPhxv6a5')
diff --git a/ui/cypress/tests/adapter/fileStream.ts b/ui/cypress/tests/adapter/fileStream.ts
index 22ebcc5..2c84ae3 100644
--- a/ui/cypress/tests/adapter/fileStream.ts
+++ b/ui/cypress/tests/adapter/fileStream.ts
@@ -31,7 +31,6 @@ describe('Test File Stream Adapter', () => {
.create('File_Stream')
.setName('File Stream Adapter Test')
.setTimestampProperty('timestamp')
- .setStoreInDataLake()
.addProtocolInput('input', 'speed', '1')
.addProtocolInput('checkbox', 'replaceTimestamp', 'check')
.setFormat('csv')
diff --git a/ui/cypress/tests/adapter/machineDataSimulator.ts b/ui/cypress/tests/adapter/machineDataSimulator.ts
index f2849f6..6ff206f 100644
--- a/ui/cypress/tests/adapter/machineDataSimulator.ts
+++ b/ui/cypress/tests/adapter/machineDataSimulator.ts
@@ -29,23 +29,11 @@ describe('Test Random Data Simulator Stream Adapter', () => {
.create('Machine_Data_Simulator')
.setName('Machine Data Simulator Test')
.addInput('input', 'wait-time-ms', '1000')
- // .setTimestampProperty('Timestamp')
- // .setStoreInDataLake()
.build();
AdapterUtils.testSpecificStreamAdapter(adapterInput);
+ AdapterUtils.deleteAdapter();
- // const adapterInput1 = SpecificAdapterBuilder
- // .create('Machine_Data_Simulator')
- // .setName('Machine Data Simulator Test 2')
- // .addInput('input', 'wait-time-ms', '1000')
- // .addInput('radio', 'selected-simulator-option', 'pressure')
- // .setTimestampProperty('timestamp')
- // .setStoreInDataLake()
- // .build();
- //
- // AdapterUtils.testSpecificStreamAdapter(adapterInput1);
- // AdapterUtils.deleteAdapter();
});
});
diff --git a/ui/cypress/tests/adapter/persistInDataLake.ts b/ui/cypress/tests/adapter/persistInDataLake.ts
index 6745cf4..5ef07b9 100644
--- a/ui/cypress/tests/adapter/persistInDataLake.ts
+++ b/ui/cypress/tests/adapter/persistInDataLake.ts
@@ -17,8 +17,8 @@
*/
import { AdapterUtils } from '../../support/utils/AdapterUtils';
-import { SpecificAdapterBuilder } from '../../support/builder/SpecificAdapterBuilder';
import { PipelineUtils } from '../../support/utils/PipelineUtils';
+import { GenericAdapterBuilder } from '../../support/builder/GenericAdapterBuilder';
describe('Test File Stream Adapter', () => {
before('Setup Test', () => {
@@ -27,15 +27,19 @@ describe('Test File Stream Adapter', () => {
it('Perform Test', () => {
- const adapterInput = SpecificAdapterBuilder
- .create('Machine_Data_Simulator')
- .setName('Machine Data Simulator Test')
- .addInput('input', 'wait-time-ms', '1000')
- .setTimestampProperty('Timestamp')
+ const adapterInput = GenericAdapterBuilder
+ .create('File_Stream')
+ .setName('File Stream Adapter Test')
+ .setTimestampProperty('timestamp')
.setStoreInDataLake()
+ .addProtocolInput('input', 'speed', '1')
+ .addProtocolInput('checkbox', 'replaceTimestamp', 'check')
+ .setFormat('csv')
+ .addFormatInput('input', 'delimiter', ';')
+ .addFormatInput('checkbox', 'header', 'check')
.build();
- AdapterUtils.testSpecificStreamAdapter(adapterInput);
+ AdapterUtils.testGenericStreamAdapter(adapterInput);
PipelineUtils.checkAmountOfPipelinesPipeline(1);
});