You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by bo...@apache.org on 2022/09/27 14:39:35 UTC
[incubator-streampipes] branch explore/websocket-adapter updated: implement EventSchemas
This is an automated email from the ASF dual-hosted git repository.
bossenti pushed a commit to branch explore/websocket-adapter
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/explore/websocket-adapter by this push:
new 76041b365 implement EventSchemas
76041b365 is described below
commit 76041b36517b91c60d3d1a33305ad2242fcba7cd
Author: bossenti <bo...@posteo.de>
AuthorDate: Tue Sep 27 16:39:27 2022 +0200
implement EventSchemas
---
installer/cli/environments/lite | 2 +-
.../stream/websocket/WebsocketAdapter.java | 28 ++++++----
.../stream/websocket/WebsocketSchemaUtils.java | 64 +++++++++++++++++-----
.../strings.en | 8 +--
4 files changed, 71 insertions(+), 31 deletions(-)
diff --git a/installer/cli/environments/lite b/installer/cli/environments/lite
index 3e6053a8a..776d21dfd 100644
--- a/installer/cli/environments/lite
+++ b/installer/cli/environments/lite
@@ -16,7 +16,7 @@
[environment:lite]
ui
backend
-connect-adapters-iiot
+# connect-adapters-iiot
consul
couchdb
kafka
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/websocket/WebsocketAdapter.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/websocket/WebsocketAdapter.java
index 6d287315b..466d6c130 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/websocket/WebsocketAdapter.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/websocket/WebsocketAdapter.java
@@ -52,8 +52,8 @@ public class WebsocketAdapter extends SpecificDataStreamAdapter {
public static final String DEVICE_ID = "device-id";
public static final String AUTH_KEY = "auth-key";
public static final String MESSAGE_TYPE = "message-type";
- public static final String TEMPERATURE = "temperature-alternative";
- public static final String EULER = "euler-alternative";
+ public static final String TEMPERATURE = "temperature";
+ public static final String ACCELEROMETER = "accelerometer";
private String authToken;
@@ -84,7 +84,7 @@ public class WebsocketAdapter extends SpecificDataStreamAdapter {
.requiredSecret(Labels.withId(AUTH_KEY))
.requiredAlternatives(Labels.withId(MESSAGE_TYPE),
Alternatives.from(Labels.withId(TEMPERATURE)),
- Alternatives.from(Labels.withId(EULER))).build();
+ Alternatives.from(Labels.withId(ACCELEROMETER))).build();
}
@@ -104,7 +104,7 @@ public class WebsocketAdapter extends SpecificDataStreamAdapter {
if (this.messageType.equals(TEMPERATURE)) {
return WebsocketSchemaUtils.getTemperatureSchema();
} else {
- return WebsocketSchemaUtils.getEulerSchema();
+ return WebsocketSchemaUtils.getAccelerometerSchema();
}
}
@@ -112,6 +112,8 @@ public class WebsocketAdapter extends SpecificDataStreamAdapter {
@Override
public void startAdapter() {
+ this.getConfigurations(adapterDescription);
+
URI connectionUri = URI.create(BROKER_URL + this.deviceId);
HashMap<String, String> httpHeader = new HashMap<>();
httpHeader.put("Authorization", "Bearer " + this.authToken);
@@ -133,26 +135,28 @@ public class WebsocketAdapter extends SpecificDataStreamAdapter {
public void onEvent(String eventPayload) {
- if (eventPayload.contains(this.messageType)) {
- try {
- Map<String, Object> event = objectMapper.readValue(eventPayload, typeRef);
+ try {
+ Map<String, Object> event = objectMapper.readValue(eventPayload, typeRef);
+ event = (Map<String, Object>) event.get("Message");
+
+ if (event.containsKey(this.messageType)) {
if (this.messageType.equals(TEMPERATURE)) {
event = WebsocketSchemaUtils.prepareTemperatureEvent(event);
- } else if (this.messageType.equals(EULER)) {
- event = WebsocketSchemaUtils.prepareEulerEvent(event);
+ } else if (this.messageType.equals(ACCELEROMETER)) {
+ event = WebsocketSchemaUtils.prepareAccelerometerEvent(event);
}
adapterPipeline.process(event);
- } catch (JsonProcessingException e) {
- LOG.error("Event e: {} of websocket adapter could not be parsed.", eventPayload, e);
}
+ } catch (JsonProcessingException e) {
+ LOG.error("Event e: {} of websocket adapter could not be parsed.", eventPayload, e);
}
}
@Override
public IAdapter getInstance(SpecificAdapterStreamDescription adapterDescription) {
- return new WebsocketAdapter();
+ return new WebsocketAdapter(adapterDescription);
}
@Override
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/websocket/WebsocketSchemaUtils.java b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/websocket/WebsocketSchemaUtils.java
index 1486ecd99..224ee2b12 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/websocket/WebsocketSchemaUtils.java
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/java/org/apache/streampipes/connect/iiot/protocol/stream/websocket/WebsocketSchemaUtils.java
@@ -27,19 +27,35 @@ import org.apache.streampipes.sdk.helpers.EpProperties;
import org.apache.streampipes.sdk.utils.Datatypes;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
public class WebsocketSchemaUtils {
public static Map<String, Object> prepareTemperatureEvent(Map<String, Object> input) {
- // TODO implement
- return input;
+
+ Map<String, Object> temperatureEvent = new HashMap<>();
+
+ temperatureEvent.put("temperature", input.get("temperature"));
+ temperatureEvent.put("timestamp", input.get("timestamp"));
+
+ return temperatureEvent;
}
- public static Map<String, Object> prepareEulerEvent(Map<String, Object> input) {
- // TODO implement
- return input;
+ public static Map<String, Object> prepareAccelerometerEvent(Map<String, Object> input) {
+
+ Map<String, Object> accelerationEvent = new HashMap<>();
+
+ HashMap<String, Object> accelerometerEvent = (HashMap<String, Object>) input.get("accelerometer");
+
+ accelerationEvent.put("x", accelerometerEvent.get("x"));
+ accelerationEvent.put("y", accelerometerEvent.get("y"));
+ accelerationEvent.put("z", accelerometerEvent.get("z"));
+ accelerationEvent.put("timestamp", input.get("timestamp"));
+
+
+ return accelerationEvent;
}
public static GuessSchema getTemperatureSchema() {
@@ -53,7 +69,7 @@ public class WebsocketSchemaUtils {
allProperties.add(
PrimitivePropertyBuilder
- .create(Datatypes.String, "temperature")
+ .create(Datatypes.Double, "temperature")
.scope(PropertyScope.MEASUREMENT_PROPERTY)
.label("Temperature")
.description("")
@@ -65,7 +81,7 @@ public class WebsocketSchemaUtils {
return guessSchema;
}
- public static GuessSchema getEulerSchema() {
+ public static GuessSchema getAccelerometerSchema() {
GuessSchema guessSchema = new GuessSchema();
@@ -75,13 +91,33 @@ public class WebsocketSchemaUtils {
allProperties.add(EpProperties.timestampProperty("timestamp"));
// TODO implement
-// allProperties.add(
-// PrimitivePropertyBuilder
-// .create(Datatypes.String, "temperature")
-// .scope(PropertyScope.MEASUREMENT_PROPERTY)
-// .label("Temperature")
-// .description("")
-// .build());
+ allProperties.add(
+ PrimitivePropertyBuilder
+ .create(Datatypes.Double, "x")
+ .scope(PropertyScope.MEASUREMENT_PROPERTY)
+ .label("accelerationX")
+ .domainProperty("http://streampipes.org/hmi/accelerationX")
+ .description("")
+ .build()
+ );
+ allProperties.add(
+ PrimitivePropertyBuilder
+ .create(Datatypes.Double, "y")
+ .scope(PropertyScope.MEASUREMENT_PROPERTY)
+ .label("accelerationY")
+ .domainProperty("http://streampipes.org/hmi/accelerationY")
+ .description("")
+ .build()
+ );
+ allProperties.add(
+ PrimitivePropertyBuilder
+ .create(Datatypes.Double, "z")
+ .scope(PropertyScope.MEASUREMENT_PROPERTY)
+ .label("accelerationZ")
+ .domainProperty("http://streampipes.org/hmi/accelerationZ")
+ .description("")
+ .build()
+ );
eventSchema.setEventProperties(allProperties);
diff --git a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.websocket/strings.en b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.websocket/strings.en
index 6020dce29..9a3d601b9 100644
--- a/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.websocket/strings.en
+++ b/streampipes-extensions/streampipes-connect-adapters-iiot/src/main/resources/org.apache.streampipes.connect.iiot.protocol.stream.websocket/strings.en
@@ -10,8 +10,8 @@ auth-key.description=Token to authenticate against Azure Websocket
message-type.title=Message Type
message-type.description=Select the kind of messages the adapter should process
-temperature-alternative.title=Temperature
-temperature-alternative.description=
+temperature.title=Temperature
+temperature.description=
-euler-alternative.title=Euler
-euler-alternative.description=
+accelerometer.title=Accelerometer
+accelerometer.description=