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 2023/08/29 13:39:01 UTC
[plc4x] branch feature/iec-60870-5-104 updated: feat(plc4j/iec-60870): Finished a first rough version of the code to publish incoming IEC 60870-5-104 events to PLC4X Subscription Handlers (Still query expressions and filtering isn't implemented)
This is an automated email from the ASF dual-hosted git repository.
cdutz pushed a commit to branch feature/iec-60870-5-104
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/feature/iec-60870-5-104 by this push:
new 77de2a1234 feat(plc4j/iec-60870): Finished a first rough version of the code to publish incoming IEC 60870-5-104 events to PLC4X Subscription Handlers (Still query expressions and filtering isn't implemented)
77de2a1234 is described below
commit 77de2a1234ffd8f3dca4d31eb7900cf90f52c19d
Author: Christofer Dutz <cd...@apache.org>
AuthorDate: Tue Aug 29 15:38:50 2023 +0200
feat(plc4j/iec-60870): Finished a first rough version of the code to publish incoming IEC 60870-5-104 events to PLC4X Subscription Handlers (Still query expressions and filtering isn't implemented)
---
.../model/Iec608705104SubscriptionHandle.java | 77 +++++++++++++++++++++
.../readwrite/protocol/Iec608705104Protocol.java | 80 +++++++++++++++++-----
.../readwrite/tag/Iec608705104Tag.java | 24 +++++--
.../readwrite/tag/Iec608705104TagHandler.java | 2 +-
.../apache/plc4x/java/iec608705104/ManualTest.java | 10 +++
5 files changed, 172 insertions(+), 21 deletions(-)
diff --git a/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/model/Iec608705104SubscriptionHandle.java b/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/model/Iec608705104SubscriptionHandle.java
new file mode 100644
index 0000000000..b9c456ed1a
--- /dev/null
+++ b/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/model/Iec608705104SubscriptionHandle.java
@@ -0,0 +1,77 @@
+/*
+ * 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
+ *
+ * https://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.plc4x.java.iec608705104.readwrite.model;
+
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.ToStringBuilder;
+import org.apache.plc4x.java.iec608705104.readwrite.tag.Iec608705104Tag;
+import org.apache.plc4x.java.spi.messages.PlcSubscriber;
+import org.apache.plc4x.java.spi.model.DefaultPlcSubscriptionHandle;
+
+public class Iec608705104SubscriptionHandle extends DefaultPlcSubscriptionHandle {
+
+ private final Iec608705104Tag tag;
+
+ public Iec608705104SubscriptionHandle(PlcSubscriber plcSubscriber, Iec608705104Tag tag) {
+ super(plcSubscriber);
+ this.tag = tag;
+ }
+
+ public Iec608705104Tag getTag() {
+ return tag;
+ }
+
+ /*public boolean matches(Iec608705104Tag address) {
+ return tag.matchesGroupAddress(address);
+ }*/
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (!(o instanceof Iec608705104SubscriptionHandle)) {
+ return false;
+ }
+
+ Iec608705104SubscriptionHandle that = (Iec608705104SubscriptionHandle) o;
+
+ return new EqualsBuilder()
+ .append(getTag(), that.getTag())
+ .isEquals();
+ }
+
+ @Override
+ public int hashCode() {
+ return new HashCodeBuilder(17, 37)
+ .appendSuper(super.hashCode())
+ .append(getTag())
+ .toHashCode();
+ }
+
+ @Override
+ public String toString() {
+ return new ToStringBuilder(this)
+ .append("tag", tag)
+ .toString();
+ }
+
+}
diff --git a/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/protocol/Iec608705104Protocol.java b/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/protocol/Iec608705104Protocol.java
index 91bc6ff2f5..7820c3e573 100644
--- a/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/protocol/Iec608705104Protocol.java
+++ b/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/protocol/Iec608705104Protocol.java
@@ -20,35 +20,35 @@
package org.apache.plc4x.java.iec608705104.readwrite.protocol;
import org.apache.plc4x.java.api.messages.PlcSubscriptionEvent;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionResponse;
import org.apache.plc4x.java.api.model.PlcConsumerRegistration;
import org.apache.plc4x.java.api.model.PlcSubscriptionHandle;
+import org.apache.plc4x.java.api.types.PlcResponseCode;
import org.apache.plc4x.java.api.value.PlcValue;
import org.apache.plc4x.java.iec608705104.readwrite.*;
import org.apache.plc4x.java.iec608705104.readwrite.configuration.Iec608705014Configuration;
+import org.apache.plc4x.java.iec608705104.readwrite.model.Iec608705104SubscriptionHandle;
import org.apache.plc4x.java.iec608705104.readwrite.tag.Iec608705104Tag;
import org.apache.plc4x.java.spi.ConversationContext;
import org.apache.plc4x.java.spi.Plc4xProtocolBase;
import org.apache.plc4x.java.spi.configuration.HasConfiguration;
+import org.apache.plc4x.java.spi.messages.DefaultPlcSubscriptionEvent;
+import org.apache.plc4x.java.spi.messages.DefaultPlcSubscriptionResponse;
import org.apache.plc4x.java.spi.messages.PlcBrowser;
import org.apache.plc4x.java.spi.messages.PlcSubscriber;
+import org.apache.plc4x.java.spi.messages.utils.ResponseItem;
+import org.apache.plc4x.java.spi.model.DefaultPlcConsumerRegistration;
+import org.apache.plc4x.java.spi.model.DefaultPlcSubscriptionTag;
import org.apache.plc4x.java.spi.transaction.RequestTransactionManager;
-import java.time.Duration;
-import java.time.LocalDateTime;
-import java.time.LocalTime;
-import java.time.ZoneOffset;
+import java.time.*;
import java.time.format.DateTimeFormatter;
-import java.time.temporal.ChronoField;
-import java.time.temporal.Temporal;
-import java.time.temporal.TemporalAmount;
-import java.util.Calendar;
-import java.util.Collection;
-import java.util.TimeZone;
+import java.util.*;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
-import static java.time.temporal.ChronoField.OFFSET_SECONDS;
-
public class Iec608705104Protocol extends Plc4xProtocolBase<APDU> implements HasConfiguration<Iec608705014Configuration>, PlcSubscriber, PlcBrowser {
private Iec608705014Configuration configuration;
@@ -56,6 +56,8 @@ public class Iec608705104Protocol extends Plc4xProtocolBase<APDU> implements Has
private int unconfirmedPackets;
+ private final Map<DefaultPlcConsumerRegistration, Consumer<PlcSubscriptionEvent>> consumers = new ConcurrentHashMap<>();
+
public Iec608705104Protocol() {
// We're starting with allowing only one message in-flight.
this.tm = new RequestTransactionManager(1);
@@ -129,13 +131,33 @@ public class Iec608705104Protocol extends Plc4xProtocolBase<APDU> implements Has
}
@Override
- public PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer, Collection<PlcSubscriptionHandle> handles) {
- return null;
+ public CompletableFuture<PlcSubscriptionResponse> subscribe(PlcSubscriptionRequest subscriptionRequest) {
+ Map<String, ResponseItem<PlcSubscriptionHandle>> values = new HashMap<>();
+ for (String tagName : subscriptionRequest.getTagNames()) {
+ final DefaultPlcSubscriptionTag tag = (DefaultPlcSubscriptionTag) subscriptionRequest.getTag(tagName);
+ if (!(tag.getTag() instanceof Iec608705104Tag)) {
+ values.put(tagName, new ResponseItem<>(PlcResponseCode.INVALID_ADDRESS, null));
+ } else {
+ values.put(tagName, new ResponseItem<>(PlcResponseCode.OK,
+ new Iec608705104SubscriptionHandle(this, (Iec608705104Tag) tag.getTag())));
+ }
+ }
+ return CompletableFuture.completedFuture(
+ new DefaultPlcSubscriptionResponse(subscriptionRequest, values));
}
@Override
- public void unregister(PlcConsumerRegistration registration) {
+ public PlcConsumerRegistration register(Consumer<PlcSubscriptionEvent> consumer, Collection<PlcSubscriptionHandle> collection) {
+ final DefaultPlcConsumerRegistration consumerRegistration =
+ new DefaultPlcConsumerRegistration(this, consumer, collection.toArray(new PlcSubscriptionHandle[0]));
+ consumers.put(consumerRegistration, consumer);
+ return consumerRegistration;
+ }
+ @Override
+ public void unregister(PlcConsumerRegistration plcConsumerRegistration) {
+ DefaultPlcConsumerRegistration consumerRegistration = (DefaultPlcConsumerRegistration) plcConsumerRegistration;
+ consumers.remove(consumerRegistration);
}
protected void processData(ASDU asdu) {
@@ -159,7 +181,9 @@ public class Iec608705104Protocol extends Plc4xProtocolBase<APDU> implements Has
} else {
eventTime = LocalDateTime.now();
}
- System.out.println(DateTimeFormatter.ISO_DATE_TIME.format(eventTime) + ": " + tag + ": " + plcValue.toString());
+
+ // Send the event out to all subscribed listeners.
+ publishEvent(eventTime, tag, plcValue);
}
}
@@ -180,4 +204,28 @@ public class Iec608705104Protocol extends Plc4xProtocolBase<APDU> implements Has
.minus(localTimeZoneOffsetFromUTC);
}
+ protected void publishEvent(LocalDateTime timeStamp, Iec608705104Tag tag, PlcValue plcValue) {
+ // Create a subscription event from the input.
+ final PlcSubscriptionEvent event = new DefaultPlcSubscriptionEvent(
+ timeStamp.atZone(ZoneId.systemDefault()).toInstant(),
+ Collections.singletonMap(tag.toString(),
+ new ResponseItem<>(PlcResponseCode.OK, plcValue)));
+
+ // Try sending the subscription event to all listeners.
+ for (Map.Entry<DefaultPlcConsumerRegistration, Consumer<PlcSubscriptionEvent>> entry : consumers.entrySet()) {
+ final DefaultPlcConsumerRegistration registration = entry.getKey();
+ final Consumer<PlcSubscriptionEvent> consumer = entry.getValue();
+ // Only if the current data point matches the subscription, publish the event to it.
+ for (PlcSubscriptionHandle handle : registration.getSubscriptionHandles()) {
+ if (handle instanceof Iec608705104SubscriptionHandle) {
+ Iec608705104SubscriptionHandle subscriptionHandle = (Iec608705104SubscriptionHandle) handle;
+ // Check if the subscription matches this current event.
+ if (/*subscriptionHandle.getTag().matchesGroupAddress(groupAddress)*/true) {
+ consumer.accept(event);
+ }
+ }
+ }
+ }
+ }
+
}
diff --git a/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/tag/Iec608705104Tag.java b/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/tag/Iec608705104Tag.java
index f302fe9643..8c48647256 100644
--- a/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/tag/Iec608705104Tag.java
+++ b/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/tag/Iec608705104Tag.java
@@ -19,12 +19,13 @@
package org.apache.plc4x.java.iec608705104.readwrite.tag;
-import org.apache.plc4x.java.api.exceptions.PlcInvalidTagException;
+import org.apache.plc4x.java.api.model.ArrayInfo;
+import org.apache.plc4x.java.api.model.PlcTag;
+import org.apache.plc4x.java.api.types.PlcValueType;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
+import java.util.List;
-public class Iec608705104Tag {
+public class Iec608705104Tag implements PlcTag {
/*private static final Pattern IEC_60870_5_104_ADDRESS =
Pattern.compile("^(?<adsuAddrSingleValueGroup>(\\d|\\*))|((?<adsuAddrDoubleValueGroup1>(\\d|\\*))/(?<adsuAddrDoubleValueGroup2>(\\d|\\*)))/((?<objectAddressSingleValueGroup(\\d|\\*))|((?<objectAddressTrippleValueGroup1>(\\d|\\*))\\.(?<objectAddressTrippleValueGroup2>(\\d|\\*))\\.(?<objectAddressTrippleValueGroup3>(\\d|\\*))))");
@@ -50,6 +51,21 @@ public class Iec608705104Tag {
return objectAddress;
}
+ @Override
+ public String getAddressString() {
+ return null;
+ }
+
+ @Override
+ public PlcValueType getPlcValueType() {
+ return PlcValueType.NULL;
+ }
+
+ @Override
+ public List<ArrayInfo> getArrayInfo() {
+ return PlcTag.super.getArrayInfo();
+ }
+
@Override
public String toString() {
return "Iec608705104Tag{" +
diff --git a/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/tag/Iec608705104TagHandler.java b/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/tag/Iec608705104TagHandler.java
index 27d8004dc8..4de3be91db 100644
--- a/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/tag/Iec608705104TagHandler.java
+++ b/plc4j/drivers/iec-60870/src/main/java/org/apache/plc4x/java/iec608705104/readwrite/tag/Iec608705104TagHandler.java
@@ -27,7 +27,7 @@ public class Iec608705104TagHandler implements PlcTagHandler {
@Override
public PlcTag parseTag(String tagAddress) {
- return null;
+ return new Iec608705104Tag(0, 0);
}
@Override
diff --git a/plc4j/drivers/iec-60870/src/test/java/org/apache/plc4x/java/iec608705104/ManualTest.java b/plc4j/drivers/iec-60870/src/test/java/org/apache/plc4x/java/iec608705104/ManualTest.java
index 9aee98e764..91af0988a6 100644
--- a/plc4j/drivers/iec-60870/src/test/java/org/apache/plc4x/java/iec608705104/ManualTest.java
+++ b/plc4j/drivers/iec-60870/src/test/java/org/apache/plc4x/java/iec608705104/ManualTest.java
@@ -32,6 +32,16 @@ public class ManualTest {
shutdown.complete(null);
}));
try (PlcConnection plcConnection = PlcDriverManager.getDefault().getConnectionManager().getConnection("iec-60870-5-104://192.168.23.10")) {
+ if(!plcConnection.getMetadata().canSubscribe()) {
+ throw new RuntimeException("Subscription not supported");
+ }
+
+ plcConnection.subscriptionRequestBuilder().addChangeOfStateTagAddress("all", "*").addPreRegisteredConsumer("all", plcSubscriptionEvent -> {
+ for (String tagName : plcSubscriptionEvent.getTagNames()) {
+ System.out.println(String.format("TS: %s, Addr: %s, Value; %s", plcSubscriptionEvent.getTimestamp().toString(), tagName, plcSubscriptionEvent.getPlcValue(tagName).toString()));
+ }
+ }).build().execute();
+
// Wait till shutdown.
shutdown.get();
}