You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2022/07/03 13:12:57 UTC
[incubator-streampipes] branch dev updated: [hotfix] Avoid update conflict when starting adapters from different service groups
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/incubator-streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 20be221e6 [hotfix] Avoid update conflict when starting adapters from different service groups
20be221e6 is described below
commit 20be221e65d2d264353eca176a225c5eb017f3b7
Author: Dominik Riemer <do...@gmail.com>
AuthorDate: Sun Jul 3 15:12:49 2022 +0200
[hotfix] Avoid update conflict when starting adapters from different service groups
---
.../management/WorkerAdministrationManagement.java | 2 +-
.../connect/container/master/util/WorkerPaths.java | 3 +--
.../TransformSchemaAdapterPipelineElement.java | 3 ++-
.../TransformValueAdapterPipelineElement.java | 15 +++++++--------
.../model/connect/adapter/AdapterDescription.java | 21 +++++++++++++++------
5 files changed, 26 insertions(+), 18 deletions(-)
diff --git a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/WorkerAdministrationManagement.java b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/WorkerAdministrationManagement.java
index 9d3b5e3a6..11249295d 100644
--- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/WorkerAdministrationManagement.java
+++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/management/WorkerAdministrationManagement.java
@@ -49,7 +49,7 @@ public class WorkerAdministrationManagement {
// only install once adapter description per service group
boolean alreadyInstalled = alreadyRegisteredAdapters
.stream()
- .anyMatch(a -> a.getAppId().equals(adapterDescription.getAppId()) && a.getCorrespondingServiceGroup().equals(adapterDescription.getCorrespondingServiceGroup()));
+ .anyMatch(a -> a.getAppId().equals(adapterDescription.getAppId()));
if (!alreadyInstalled) {
this.adapterDescriptionStorage.storeAdapter(adapterDescription);
}
diff --git a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/WorkerPaths.java b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/WorkerPaths.java
index f3af79b6a..1a4882893 100644
--- a/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/WorkerPaths.java
+++ b/streampipes-connect-container-master/src/main/java/org/apache/streampipes/connect/container/master/util/WorkerPaths.java
@@ -60,8 +60,7 @@ public class WorkerPaths {
SpServiceUrlProvider serviceUrlProvider = SpServiceUrlProvider.ADAPTER;
String endpointUrl = new ExtensionsServiceEndpointGenerator(appId, serviceUrlProvider).getEndpointResourceUrl();
URI uri = new URI(endpointUrl);
- String baseUrl = uri.getScheme() + "://" + uri.getAuthority();
- return baseUrl;
+ return uri.getScheme() + "://" + uri.getAuthority();
}
diff --git a/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/preprocessing/elements/TransformSchemaAdapterPipelineElement.java b/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/preprocessing/elements/TransformSchemaAdapterPipelineElement.java
index 735933f48..e795af37d 100644
--- a/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/preprocessing/elements/TransformSchemaAdapterPipelineElement.java
+++ b/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/preprocessing/elements/TransformSchemaAdapterPipelineElement.java
@@ -27,6 +27,7 @@ import org.apache.streampipes.connect.adapter.preprocessing.transform.schema.*;
import org.apache.streampipes.model.connect.rules.*;
import org.apache.streampipes.model.connect.rules.schema.*;
+import javax.xml.crypto.dsig.Transform;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -36,7 +37,7 @@ public class TransformSchemaAdapterPipelineElement implements IAdapterPipelineEl
private SchemaEventTransformer eventTransformer;
Logger logger = LoggerFactory.getLogger(TransformSchemaAdapterPipelineElement.class);
- public TransformSchemaAdapterPipelineElement(List<SchemaTransformationRuleDescription> transformationRuleDescriptions) {
+ public TransformSchemaAdapterPipelineElement(List<? extends TransformationRuleDescription> transformationRuleDescriptions) {
List<TransformationRule> rules = new ArrayList<>();
// transforms description to actual rules
diff --git a/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/preprocessing/elements/TransformValueAdapterPipelineElement.java b/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/preprocessing/elements/TransformValueAdapterPipelineElement.java
index 137966c7e..25d357518 100644
--- a/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/preprocessing/elements/TransformValueAdapterPipelineElement.java
+++ b/streampipes-connect/src/main/java/org/apache/streampipes/connect/adapter/preprocessing/elements/TransformValueAdapterPipelineElement.java
@@ -18,16 +18,15 @@
package org.apache.streampipes.connect.adapter.preprocessing.elements;
-import org.apache.streampipes.model.connect.rules.value.CorrectionValueTransformationRuleDescription;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.streampipes.connect.api.IAdapterPipelineElement;
import org.apache.streampipes.connect.adapter.preprocessing.Util;
import org.apache.streampipes.connect.adapter.preprocessing.transform.value.*;
+import org.apache.streampipes.connect.api.IAdapterPipelineElement;
import org.apache.streampipes.model.connect.rules.TransformationRuleDescription;
+import org.apache.streampipes.model.connect.rules.value.CorrectionValueTransformationRuleDescription;
import org.apache.streampipes.model.connect.rules.value.TimestampTranfsformationRuleDescription;
import org.apache.streampipes.model.connect.rules.value.UnitTransformRuleDescription;
-import org.apache.streampipes.model.connect.rules.value.ValueTransformationRuleDescription;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.List;
@@ -35,10 +34,10 @@ import java.util.Map;
public class TransformValueAdapterPipelineElement implements IAdapterPipelineElement {
- private ValueEventTransformer eventTransformer;
- private Logger logger = LoggerFactory.getLogger(TransformValueAdapterPipelineElement.class);
+ private final ValueEventTransformer eventTransformer;
+ private final static Logger logger = LoggerFactory.getLogger(TransformValueAdapterPipelineElement.class);
- public TransformValueAdapterPipelineElement(List<ValueTransformationRuleDescription> transformationRuleDescriptions) {
+ public TransformValueAdapterPipelineElement(List<? extends TransformationRuleDescription> transformationRuleDescriptions) {
List<ValueTransformationRule> rules = new ArrayList<>();
// transforms description to actual rules
diff --git a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
index 5647f3f02..24eb04a32 100644
--- a/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
+++ b/streampipes-model/src/main/java/org/apache/streampipes/model/connect/adapter/AdapterDescription.java
@@ -162,8 +162,8 @@ public abstract class AdapterDescription extends NamedStreamPipesEntity {
this.adapterType = adapterType;
}
- public List getValueRules() {
- List tmp = new ArrayList<>();
+ public List<TransformationRuleDescription> getValueRules() {
+ var tmp = new ArrayList<TransformationRuleDescription>();
rules.forEach(rule -> {
if(rule instanceof ValueTransformationRuleDescription)
tmp.add(rule);
@@ -171,8 +171,8 @@ public abstract class AdapterDescription extends NamedStreamPipesEntity {
return tmp;
}
- public List getStreamRules() {
- List tmp = new ArrayList<>();
+ public List<TransformationRuleDescription> getStreamRules() {
+ var tmp = new ArrayList<TransformationRuleDescription>();
rules.forEach(rule -> {
if(rule instanceof StreamTransformationRuleDescription)
tmp.add(rule);
@@ -180,8 +180,8 @@ public abstract class AdapterDescription extends NamedStreamPipesEntity {
return tmp;
}
- public List getSchemaRules() {
- List tmp = new ArrayList<>();
+ public List<TransformationRuleDescription> getSchemaRules() {
+ var tmp = new ArrayList<TransformationRuleDescription>();
rules.forEach(rule -> {
if(rule instanceof SchemaTransformationRuleDescription)
tmp.add(rule);
@@ -231,10 +231,19 @@ public abstract class AdapterDescription extends NamedStreamPipesEntity {
this.selectedEndpointUrl = selectedEndpointUrl;
}
+ /**
+ * @deprecated check if the service group can be removed as a single pipeline element
+ * can correspond to different service groups
+ */
+ @Deprecated
public String getCorrespondingServiceGroup() {
return correspondingServiceGroup;
}
+ /**
+ * @deprecated check if the service group can be removed as a single pipeline element
+ * can correspond to different service groups
+ */
public void setCorrespondingServiceGroup(String correspondingServiceGroup) {
this.correspondingServiceGroup = correspondingServiceGroup;
}