You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by cd...@apache.org on 2019/11/19 15:04:39 UTC
[plc4x] branch feature/knxnet-ip updated: - Added a first
rudimentary ETS5Processor
This is an automated email from the ASF dual-hosted git repository.
cdutz pushed a commit to branch feature/knxnet-ip
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/feature/knxnet-ip by this push:
new 9db7262 - Added a first rudimentary ETS5Processor
9db7262 is described below
commit 9db7262ce852048e4f7730eb6f0ca30c52a96cff
Author: Christofer Dutz <ch...@c-ware.de>
AuthorDate: Tue Nov 19 16:04:29 2019 +0100
- Added a first rudimentary ETS5Processor
---
.../streampipes/knxnetip/KnxNetIpAdapters.java | 4 +--
.../ets5/Ets5DataEnrichmentController.java | 30 +++++++++++++++++++---
.../knxnetip/source/KnxNetIpAdapter.java | 9 ++++++-
3 files changed, 37 insertions(+), 6 deletions(-)
diff --git a/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/KnxNetIpAdapters.java b/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/KnxNetIpAdapters.java
index a71c627..0c248c7 100644
--- a/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/KnxNetIpAdapters.java
+++ b/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/KnxNetIpAdapters.java
@@ -34,9 +34,9 @@ public class KnxNetIpAdapters {
.add(new KnxNetIpAdapter());
// Declare the processors.
- /*DeclarersSingleton
+ DeclarersSingleton
.getInstance()
- .add(new Ets5DataEnrichmentController());*/
+ .add(new Ets5DataEnrichmentController());
// Initialize the adapter itself.
String workerUrl = ConnectWorkerConfig.INSTANCE.getConnectContainerWorkerUrl();
diff --git a/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/processors/ets5/Ets5DataEnrichmentController.java b/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/processors/ets5/Ets5DataEnrichmentController.java
index 0530494..a32f3f9 100644
--- a/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/processors/ets5/Ets5DataEnrichmentController.java
+++ b/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/processors/ets5/Ets5DataEnrichmentController.java
@@ -18,21 +18,45 @@ under the License.
*/
package org.apache.plc4x.java.streampipes.knxnetip.processors.ets5;
+import org.apache.plc4x.java.streampipes.knxnetip.source.KnxNetIpAdapter;
+import org.streampipes.model.DataProcessorType;
import org.streampipes.model.graph.DataProcessorDescription;
import org.streampipes.model.graph.DataProcessorInvocation;
+import org.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.streampipes.sdk.builder.StreamRequirementsBuilder;
import org.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
+import org.streampipes.sdk.helpers.EpRequirements;
+import org.streampipes.sdk.helpers.Labels;
+import org.streampipes.sdk.helpers.Locales;
+import org.streampipes.sdk.helpers.OutputStrategies;
+import org.streampipes.sdk.utils.Assets;
import org.streampipes.wrapper.standalone.ConfiguredEventProcessor;
import org.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
public class Ets5DataEnrichmentController extends StandaloneEventProcessingDeclarer<Ets5DataEnrichmentParameters> {
+ public static final String ID = "http://plc4x.apache.org/streampipes/processor/ets5";
+
@Override
- public ConfiguredEventProcessor<Ets5DataEnrichmentParameters> onInvocation(DataProcessorInvocation dataProcessorInvocation, ProcessingElementParameterExtractor processingElementParameterExtractor) {
- return null;
+ public DataProcessorDescription declareModel() {
+ final DataProcessorDescription description = ProcessingElementBuilder
+ .create(ID, "ETS5", "Processor that interprets a data stream from a KXNnet/IP Datasource according to the settings in the ETS5 'knxproj' file")
+ .category(DataProcessorType.ENRICH)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withLocales(Locales.EN)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+ .requiredProperty(EpRequirements.domainPropertyReq(KnxNetIpAdapter.ID_DESTINATION_ADDRESS))
+ .requiredProperty(EpRequirements.domainPropertyReq(KnxNetIpAdapter.ID_PAYLOAD))
+ .build())
+ .outputStrategy(OutputStrategies.keep())
+ .requiredFile(Labels.from("knxprojFile", "ETS5 Project File", "ETS5 Project File (.knxproj)"))
+ .build();
+ return description;
}
@Override
- public DataProcessorDescription declareModel() {
+ public ConfiguredEventProcessor<Ets5DataEnrichmentParameters> onInvocation(DataProcessorInvocation dataProcessorInvocation, ProcessingElementParameterExtractor processingElementParameterExtractor) {
return null;
}
diff --git a/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/source/KnxNetIpAdapter.java b/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/source/KnxNetIpAdapter.java
index 23c28b6..842519c 100644
--- a/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/source/KnxNetIpAdapter.java
+++ b/sandbox/streampipes-knxnetip/src/main/java/org/apache/plc4x/java/streampipes/knxnetip/source/KnxNetIpAdapter.java
@@ -54,7 +54,10 @@ import java.util.*;
public class KnxNetIpAdapter extends SpecificDataStreamAdapter {
- public static final String ID = "http://plc4x.apache.org/streampipes/adapter/knxnetip";
+ public static final String ID = "http://plc4x.apache.org/streampipes/source/knxnetip";
+ public static final String ID_SOURCE_ADDRESS = ID + "/source-address";
+ public static final String ID_DESTINATION_ADDRESS = ID + "/destination-address";
+ public static final String ID_PAYLOAD = ID + "/payload";
public static final String MAPPING_FIELD_TIME = "time";
public static final String MAPPING_FIELD_SOURCE_ADDRESS = "sourceAddress";
@@ -98,18 +101,21 @@ public class KnxNetIpAdapter extends SpecificDataStreamAdapter {
allProperties.add(
PrimitivePropertyBuilder
.create(Datatypes.Integer, MAPPING_FIELD_SOURCE_ADDRESS)
+ .domainProperty(ID_SOURCE_ADDRESS)
.label("Source Address")
.description("Source address from which the event originated.")
.build());
allProperties.add(
PrimitivePropertyBuilder
.create(Datatypes.Integer, MAPPING_FIELD_DESTINATION_ADDRESS)
+ .domainProperty(ID_DESTINATION_ADDRESS)
.label("Destination Address")
.description("Destination address to which the event is targeted.")
.build());
allProperties.add(
PrimitivePropertyBuilder
.create(Datatypes.String, MAPPING_FIELD_PAYLOAD)
+ .domainProperty(ID_PAYLOAD)
.label("Payload")
.description("Raw payload of the event.")
.build());
@@ -155,6 +161,7 @@ public class KnxNetIpAdapter extends SpecificDataStreamAdapter {
event.put(MAPPING_FIELD_TIME, System.currentTimeMillis());
event.put(MAPPING_FIELD_SOURCE_ADDRESS, addressToString(cemiDataFrame.getSourceAddress()));
event.put(MAPPING_FIELD_DESTINATION_ADDRESS, addressToString(destinationAddress));
+ // Encode the payload as Hex String.
event.put(MAPPING_FIELD_PAYLOAD, Hex.encodeHexString(payload));
// Send it to StreamPipes