This is an automated email from the ASF dual-hosted git repository.
cgarcia pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 5a1f768f1c Fix/s7async (#1451)
5a1f768f1c is described below
commit 5a1f768f1cd6b85ac98d8c650e000acde075a564
Author: César José García León <cesarjos...@gmail.com>
AuthorDate: Fri Mar 15 09:13:09 2024 -0400
Fix/s7async (#1451)
* Change type transfer in cyclic subscription from byte[] to PlcValue.
Accepts cyclic subscription to bits.
* Add short pattern to tag subscription.
* Corrects short tag handling in CYC subscriptions. In observation./karaf
* Modified the cyclical subscription system. TODO time base management fpr
CYC.
* Fixed time base handling for cyclical subscriptions. Subscription routine
for changes is added experimentally.
---------
Co-authored-by: Cesar Garcia <cesar.gar...@ceos.com.ve>
---
.../java/knxnetip/readwrite/KnxManufacturer.java | 1 +
.../apache/plc4x/java/s7/events/S7CyclicEvent.java | 282 +++++++++++-----
.../s7/readwrite/protocol/S7HPlcConnection.java | 12 +
.../readwrite/protocol/S7ProtocolEventLogic.java | 10 +
.../s7/readwrite/protocol/S7ProtocolLogic.java | 41 ++-
.../java/s7/readwrite/tag/S7SubscriptionTag.java | 37 ++-
.../readwrite/utils/S7PlcSubscriptionRequest.java | 363 +++++++++++++++++++++
.../java/s7/readwrite/utils/StaticHelper.java | 3 +-
.../org/apache/plc4x/java/spi/values/PlcBINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcBOOL.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcBREAL.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcBYTE.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcCHAR.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcDINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcDWORD.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcLINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcLREAL.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcLWORD.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcREAL.java | 5 +
.../plc4x/java/spi/values/PlcRawByteArray.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcSINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcUBINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcUDINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcUINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcULINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcUSINT.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcWCHAR.java | 5 +
.../org/apache/plc4x/java/spi/values/PlcWORD.java | 5 +
.../protocols/knxnetip/knx-master-data.mspec | 3 +-
30 files changed, 761 insertions(+), 96 deletions(-)
diff --git
a/plc4j/drivers/knxnetip/src/main/generated/org/apache/plc4x/java/knxnetip/readwrite/KnxManufacturer.java
b/plc4j/drivers/knxnetip/src/main/generated/org/apache/plc4x/java/knxnetip/readwrite/KnxManufacturer.java
index 7d8a793c30..21830b7202 100644
---
a/plc4j/drivers/knxnetip/src/main/generated/org/apache/plc4x/java/knxnetip/readwrite/KnxManufacturer.java
+++
b/plc4j/drivers/knxnetip/src/main/generated/org/apache/plc4x/java/knxnetip/readwrite/KnxManufacturer.java
@@ -776,6 +776,7 @@ public enum KnxManufacturer {
M_ABB___RESERVED((int) 670, (int) 43954, (String) "ABB - reserved"),
M_BUSCH_JAEGER_ELEKTRO___RESERVED(
(int) 671, (int) 43959, (String) "Busch-Jaeger Elektro - reserved");
+
private static final Map<Integer, KnxManufacturer> map;
static {
diff --git
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/events/S7CyclicEvent.java
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/events/S7CyclicEvent.java
index 5a40e5037b..42e96ce34a 100644
---
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/events/S7CyclicEvent.java
+++
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/events/S7CyclicEvent.java
@@ -20,6 +20,7 @@ package org.apache.plc4x.java.s7.events;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
+import static io.netty.buffer.Unpooled.wrappedBuffer;
import org.apache.plc4x.java.api.messages.PlcReadRequest;
import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
import org.apache.plc4x.java.api.model.PlcTag;
@@ -38,6 +39,37 @@ import java.time.*;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import org.apache.plc4x.java.api.model.PlcSubscriptionTag;
+import org.apache.plc4x.java.api.types.PlcValueType;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.BOOL;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.CHAR;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.DATE;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.DATE_AND_TIME;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.DINT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.DT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.DWORD;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.INT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.LDT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.LINT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.LREAL;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.LTIME;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.LTOD;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.LWORD;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.REAL;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.S5TIME;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.SINT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.TIME_OF_DAY;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.TOD;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.UDINT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.ULINT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.USINT;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.WCHAR;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.WORD;
+import static org.apache.plc4x.java.s7.readwrite.TransportSize.WSTRING;
+import org.apache.plc4x.java.s7.readwrite.tag.S7SubscriptionTag;
+import org.apache.plc4x.java.s7.readwrite.tag.S7Tag;
+import org.apache.plc4x.java.spi.model.DefaultPlcSubscriptionTag;
+import org.apache.plc4x.java.spi.values.PlcValueHandler;
public class S7CyclicEvent implements S7Event {
@@ -68,34 +100,16 @@ public class S7CyclicEvent implements S7Event {
map.put(Fields.TIMESTAMP.name(), this.timeStamp);
map.put(Fields.JOBID.name(), jobid);
map.put(Fields.ITEMSCOUNT.name(), event.getItemsCount());
- int[] n = new int[1];
+ int[] n = new int[1];
+
request.getTagNames().forEach(tagname -> {
int i = n[0];
map.put(Fields.RETURNCODE_.name() + i,
event.getItems().get(i).getReturnCode().getValue());
map.put(Fields.TRANSPORTSIZE_.name() + i,
event.getItems().get(i).getTransportSize().getValue());
- byte[] buffer = new byte[event.getItems().get(i).getData().size()];
- j = 0;
- event.getItems().get(i).getData().forEach(s -> {
- buffer[j] = s.byteValue();
- j++;
- });
- map.put(tagname, buffer);
- n[0]++;
+ map.put(tagname, DataToPlcValue(tagname, request,
event.getItems().get(i).getData()));
+ n[0]++;
});
-
-// for (int i=0; i<event.getItemsCount(); i++){
-// //map.put(Fields.RETURNCODE_.name()+i,
event.getItems()[i].getReturnCode().getValue());
-// map.put(Fields.RETURNCODE_.name()+i,
event.getItems().get(i).getReturnCode().getValue());
-// map.put(Fields.TRANSPORTSIZE_.name()+i,
event.getItems().get(i).getTransportSize().getValue());
-// byte[] buffer = new
byte[event.getItems().get(i).getData().size()];
-// j = 0;
-// event.getItems().get(i).getData().forEach(s->{
-// buffer[j] = s.byteValue();
-// j ++;
-// });
-// map.put(Fields.DATA_.name()+i, buffer);
-// }
}
public S7CyclicEvent(PlcSubscriptionRequest request, short jobid,
S7PayloadUserDataItemCyclicServicesChangeDrivenPush event) {
@@ -107,30 +121,15 @@ public class S7CyclicEvent implements S7Event {
map.put(Fields.JOBID.name(), jobid);
map.put(Fields.ITEMSCOUNT.name(), event.getItemsCount());
int[] n = new int[1];
+
request.getTagNames().forEach(tagname -> {
int i = n[0];
map.put(Fields.RETURNCODE_.name() + i,
event.getItems().get(i).getReturnCode().getValue());
map.put(Fields.TRANSPORTSIZE_.name() + i,
event.getItems().get(i).getTransportSize().getValue());
- byte[] buffer = new byte[event.getItems().get(i).getData().size()];
- j = 0;
- event.getItems().get(i).getData().forEach(s -> {
- buffer[j] = s.byteValue();
- j++;
- });
- map.put(tagname, buffer);
- n[0]++;
+ map.put(tagname, DataToPlcValue(tagname, request,
event.getItems().get(i).getData()));
+ n[0]++;
});
-// for (int i=0; i<event.getItemsCount(); i++){
-// map.put(Fields.RETURNCODE_.name()+i,
event.getItems().get(i).getReturnCode().getValue());
-// map.put(Fields.TRANSPORTSIZE_.name()+i,
event.getItems().get(i).getTransportSize().getValue());
-// byte[] buffer = new
byte[event.getItems().get(i).getData().size()];
-// j = 0;
-// event.getItems().get(i).getData().forEach(s->{
-// buffer[j] = s.byteValue();
-// j ++;
-// });
-// map.put(Fields.DATA_.name()+i, buffer);
-// }
+
}
public S7CyclicEvent(PlcSubscriptionRequest request, short jobid,
S7PayloadUserDataItemCyclicServicesSubscribeResponse event) {
@@ -142,30 +141,14 @@ public class S7CyclicEvent implements S7Event {
map.put(Fields.JOBID.name(), jobid);
map.put(Fields.ITEMSCOUNT.name(), event.getItemsCount());
int[] n = new int[1];
+
request.getTagNames().forEach(tagname -> {
int i = n[0];
map.put(Fields.RETURNCODE_.name() + i,
event.getItems().get(i).getReturnCode().getValue());
map.put(Fields.TRANSPORTSIZE_.name() + i,
event.getItems().get(i).getTransportSize().getValue());
- byte[] buffer = new byte[event.getItems().get(i).getData().size()];
- j = 0;
- event.getItems().get(i).getData().forEach(s -> {
- buffer[j] = s.byteValue();
- j++;
- });
- map.put(tagname, buffer);
- n[0]++;
- });
-// for (int i=0; i<event.getItemsCount(); i++){
-// map.put(Fields.RETURNCODE_.name()+i,
event.getItems().get(i).getReturnCode().getValue());
-// map.put(Fields.TRANSPORTSIZE_.name()+i,
event.getItems().get(i).getTransportSize().getValue());
-// byte[] buffer = new
byte[event.getItems().get(i).getData().size()];
-// j = 0;
-// event.getItems().get(i).getData().forEach(s->{
-// buffer[j] = s.byteValue();
-// j ++;
-// });
-// map.put(Fields.DATA_.name()+i, buffer);
-// }
+ map.put(tagname, DataToPlcValue(tagname, request,
event.getItems().get(i).getData()));
+ n[0]++;
+ });
}
public S7CyclicEvent(PlcSubscriptionRequest request, short jobid,
S7PayloadUserDataItemCyclicServicesChangeDrivenSubscribeResponse event) {
@@ -177,30 +160,14 @@ public class S7CyclicEvent implements S7Event {
map.put(Fields.JOBID.name(), jobid);
map.put(Fields.ITEMSCOUNT.name(), event.getItemsCount());
int[] n = new int[1];
+
request.getTagNames().forEach(tagname -> {
int i = n[0];
map.put(Fields.RETURNCODE_.name() + i,
event.getItems().get(i).getReturnCode().getValue());
map.put(Fields.TRANSPORTSIZE_.name() + i,
event.getItems().get(i).getTransportSize().getValue());
- byte[] buffer = new byte[event.getItems().get(i).getData().size()];
- j = 0;
- event.getItems().get(i).getData().forEach(s -> {
- buffer[j] = s.byteValue();
- j++;
- });
- map.put(tagname, buffer);
- n[0]++;
- });
-// for (int i=0; i<event.getItemsCount(); i++){
-// map.put(Fields.RETURNCODE_.name()+i,
event.getItems().get(i).getReturnCode().getValue());
-// map.put(Fields.TRANSPORTSIZE_.name()+i,
event.getItems().get(i).getTransportSize().getValue());
-// byte[] buffer = new
byte[event.getItems().get(i).getData().size()];
-// j = 0;
-// event.getItems().get(i).getData().forEach(s->{
-// buffer[j] = s.byteValue();
-// j ++;
-// });
-// map.put(Fields.DATA_.name()+i, buffer);
-// }
+ map.put(tagname, DataToPlcValue(tagname, request,
event.getItems().get(i).getData()));
+ n[0]++;
+ });
}
@Override
@@ -225,7 +192,12 @@ public class S7CyclicEvent implements S7Event {
@Override
public PlcValue getPlcValue(String name) {
- throw new UnsupportedOperationException("Not supported yet.");
+ if (request.getTagNames().contains(name)) {
+ PlcValue plcvalue = (PlcValue) map.get(name);
+ plcvalue.getRaw();
+ return plcvalue;
+ }
+ return null;
}
@Override
@@ -775,4 +747,150 @@ public class S7CyclicEvent implements S7Event {
throw new UnsupportedOperationException("Not supported yet.");
}
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null) {
+ return false;
+ }
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+ final S7CyclicEvent other = (S7CyclicEvent) obj;
+
+ for (String tag:request.getTagNames()) {
+ final PlcValue othervalue = other.getPlcValue(tag);
+ if (othervalue == null) return false;
+ final PlcValue localvalue = (PlcValue) getPlcValue(tag);
+ if (Arrays.equals(localvalue.getRaw(), othervalue.getRaw()) ==
false){
+ return false;
+ }
+ };
+
+ return true;
+ }
+
+
+ private static PlcValue DataToPlcValue(String tagname,
PlcSubscriptionRequest request, List<Short> data){
+
+ int[] i = new int[1];
+
+ final byte[] buffer = new byte[data.size()];
+ i[0] = 0;
+
+ data.forEach( b -> {
+ buffer[i[0]] = b.byteValue();
+ i[0]++;
+ });
+
+ ByteBuf bb = wrappedBuffer(buffer);
+
+
+ final DefaultPlcSubscriptionTag dpst = (DefaultPlcSubscriptionTag)
request.getTag(tagname);
+ final S7SubscriptionTag subTag = (S7SubscriptionTag) dpst.getTag();
+ final S7Tag[] s7Tags = subTag.getS7Tags();
+
+ PlcValue plcValue = null;
+
+ switch(s7Tags[0].getDataType()){
+ case BOOL:;
+
+ Boolean[] bools = new
Boolean[s7Tags[0].getNumberOfElements()];
+ for (int iter = 0; iter < s7Tags[0].getNumberOfElements();
iter++ )
+ bools[iter] = bb.readBoolean();
+ plcValue = PlcValueHandler.of(bools);
+ break;
+ case BYTE:;
+ Byte[] bytes = new Byte[bb.capacity()];
+ for (Byte b:bytes)
+ b = Byte.valueOf(bb.readByte());
+ plcValue = PlcValueHandler.of(bytes);
+ break;
+ case WORD:;
+ break;
+ case DWORD:;
+ break;
+ case LWORD:;
+ break;
+ case INT:;
+ Short[] shorts = new
Short[s7Tags[0].getNumberOfElements()];
+ for (int iter = 0; iter < s7Tags[0].getNumberOfElements();
iter ++)
+ shorts[iter] = bb.readShort();
+ plcValue = PlcValueHandler.of(shorts);
+ break;
+ case UINT:;
+ break;
+ case SINT:;
+ break;
+ case USINT:;
+ break;
+ case DINT:;
+ Integer[] integers = new Integer[bb.capacity() /
Integer.SIZE];
+ for (Integer di:integers) di =
Integer.valueOf(bb.readInt());
+ plcValue = PlcValueHandler.of(integers);
+ break;
+ case UDINT:;
+ break;
+ case LINT:;
+ Long[] longs = new Long[bb.capacity() / Long.SIZE];
+ for (Long l:longs) l = bb.readLong();
+ plcValue = PlcValueHandler.of(longs);
+ break;
+ case ULINT:;
+ break;
+ case REAL:;
+ Float[] floats = new Float[bb.capacity() / Float.SIZE];
+ for (Float f:floats) f = bb.readFloat();
+ plcValue = PlcValueHandler.of(floats);
+ break;
+ case LREAL:;
+ Double[] doubles = new Double[bb.capacity() / Double.SIZE];
+ for (Double d:doubles) d = bb.readDouble();
+ plcValue = PlcValueHandler.of(doubles);
+ break;
+ case CHAR:;
+ break;
+ case WCHAR:;
+ break;
+ case STRING:;
+ break;
+ case WSTRING:;
+ break;
+ case S5TIME:;
+ break;
+ case TIME:;
+ break;
+ case LTIME:;
+ break;
+ case DATE:;
+ break;
+ case TIME_OF_DAY:;
+ break;
+ case TOD:;
+ break;
+ case LTIME_OF_DAY:;
+ break;
+ case LTOD:;
+ break;
+ case DATE_AND_TIME:;
+ break;
+ case DT:;
+ break;
+ case DATE_AND_LTIME:;
+ break;
+ case LDT:;
+ break;
+ case DTL:;
+ break;
+ }
+
+ return plcValue;
+
+ };
+
+
+
+
}
diff --git
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HPlcConnection.java
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HPlcConnection.java
index bd1f3bd789..8e75bd2e59 100644
---
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HPlcConnection.java
+++
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7HPlcConnection.java
@@ -49,6 +49,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.*;
+import org.apache.plc4x.java.api.exceptions.PlcUnsupportedOperationException;
+import org.apache.plc4x.java.api.messages.PlcSubscriptionRequest;
+import org.apache.plc4x.java.s7.readwrite.utils.S7PlcSubscriptionRequest;
+import org.apache.plc4x.java.spi.messages.DefaultPlcSubscriptionRequest;
/**
* This object generates the main connection and includes the management
@@ -430,4 +434,12 @@ public class S7HPlcConnection extends
DefaultNettyPlcConnection implements Runna
return null;
}
+ @Override
+ public PlcSubscriptionRequest.Builder subscriptionRequestBuilder() {
+ if (!isSubscribeSupported()) {
+ throw new PlcUnsupportedOperationException("The connection does
not support subscription");
+ }
+ return new S7PlcSubscriptionRequest.Builder(this, getPlcTagHandler());
+ }
+
}
diff --git
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolEventLogic.java
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolEventLogic.java
index 0191660ceb..3288a2d77b 100644
---
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolEventLogic.java
+++
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolEventLogic.java
@@ -114,6 +114,8 @@ public class S7ProtocolEventLogic implements PlcSubscriber {
mapConsumers.remove(registration);
}
+
+ //TODO: Replace with disruptor
private class ObjectProcessor implements Runnable {
private final BlockingQueue<S7Event> eventQueue;
@@ -154,6 +156,7 @@ public class S7ProtocolEventLogic implements PlcSubscriber {
}
}
+ //TODO: Replace with disruptor
private class EventDispatcher implements Runnable {
private final BlockingQueue<S7Event> dispatchQueue;
private boolean shutdown = false;
@@ -173,6 +176,10 @@ public class S7ProtocolEventLogic implements PlcSubscriber
{
while (!shutdown) {
try {
S7Event s7Event = dispatchQueue.poll(DEFAULT_DELAY,
TimeUnit.MILLISECONDS);
+ if ((s7Event == null) && (cycDelayedObject != null)) {
+ s7Event = cycDelayedObject;
+ cycDelayedObject = null;
+ }
if (s7Event != null) {
if (s7Event instanceof S7ModeEvent) {
S7ModeEvent modeEvent = (S7ModeEvent) s7Event;
@@ -202,10 +209,13 @@ public class S7ProtocolEventLogic implements
PlcSubscriber {
S7CyclicEvent cyclicEvent = (S7CyclicEvent)
s7Event;
if (mapIndex.containsKey(EventType.CYC)) {
Map<PlcConsumerRegistration,
Consumer<PlcSubscriptionEvent>> mapConsumers = mapIndex.get(EventType.CYC);
+
if (cycDelayedObject != null) {
mapConsumers.forEach((x, y) ->
y.accept(cycDelayedObject));
cycDelayedObject = null;
}
+ if (mapConsumers.isEmpty()) cycDelayedObject =
s7Event;
+
mapConsumers.forEach((x, y) -> {
S7PlcSubscriptionHandle sh =
(S7PlcSubscriptionHandle) x.getSubscriptionHandles().get(0);
Short id =
Short.parseShort(sh.getEventId());
diff --git
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java
index ee64e9b0ec..b161770abf 100644
---
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java
+++
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java
@@ -63,10 +63,12 @@ import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import org.apache.plc4x.java.api.types.PlcSubscriptionType;
import org.apache.plc4x.java.s7.events.S7AlarmEvent;
import org.apache.plc4x.java.s7.events.S7ModeEvent;
import org.apache.plc4x.java.s7.events.S7SysEvent;
import org.apache.plc4x.java.s7.events.S7UserEvent;
+import org.apache.plc4x.java.s7.readwrite.utils.S7PlcSubscriptionRequest;
/**
* The S7 Protocol states that there can not be more then {min(maxAmqCaller,
maxAmqCallee} "ongoing" requests.
@@ -113,7 +115,14 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
* the values sent PUSH from the PLC to the driver refer to this JobID.
*/
private final Map<Short, PlcSubscriptionRequest> cycRequests = new
HashMap<>();
-
+
+ /*
+ * This data structure stores the last value associated with a cyclic
+ * subscription request. In each event received, the values of the internal
+ * PlcValue are compared and if any of them are different, the new value is
+ * transferred to the event stack and the value is updated in this HashMap.
+ */
+ private final Map<Short, S7CyclicEvent> cycChangeValueEvents = new
HashMap<>();
private S7DriverContext s7DriverContext;
private RequestTransactionManager tm;
@@ -397,7 +406,7 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
futures.put("DATA_", new CompletableFuture<>());
- DefaultPlcSubscriptionRequest request =
(DefaultPlcSubscriptionRequest) subscriptionRequest;
+ S7PlcSubscriptionRequest request = (S7PlcSubscriptionRequest)
subscriptionRequest;
int tpduId = getTpduId();
@@ -541,7 +550,7 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
return future;
}
- private S7Message
encodeEventSubscriptionRequest(DefaultPlcSubscriptionRequest request, int
tpduId) {
+ private S7Message encodeEventSubscriptionRequest(S7PlcSubscriptionRequest
request, int tpduId) {
List<S7ParameterUserDataItem> parameterItems = new
ArrayList<>(request.getNumberOfTags());
List<S7PayloadUserDataItem> payloadItems = new
ArrayList<>(request.getNumberOfTags());
@@ -799,6 +808,10 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
S7CyclicEvent cycEvent = new S7CyclicEvent(plcSubscriptionRequest,
msgParameter.getSequenceNumber(),
(S7PayloadUserDataItemCyclicServicesSubscribeResponse)
payloadItems.get(0));
+
+ if
(plcSubscriptionRequest.getTags().get(0).getPlcSubscriptionType() ==
PlcSubscriptionType.CHANGE_OF_STATE) {
+ cycChangeValueEvents.put(msgParameter.getSequenceNumber(),
cycEvent);
+ }
eventQueue.add(cycEvent);
@@ -924,7 +937,7 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
new S7PayloadUserData(payloadItems)));
}
- private S7Message encodeAlarmQueryRequest(DefaultPlcSubscriptionRequest
request, int tpduId) {
+ private S7Message encodeAlarmQueryRequest(S7PlcSubscriptionRequest
request, int tpduId) {
List<S7ParameterUserDataItem> parameterItems = new
ArrayList<>(request.getNumberOfTags());
List<S7PayloadUserDataItem> payloadItems = new
ArrayList<>(request.getNumberOfTags());
@@ -956,11 +969,11 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
new S7PayloadUserData(payloadItems));
}
- private void encodeCycledSubscriptionRequest(DefaultPlcSubscriptionRequest
request, int tpduId) {
+ private void encodeCycledSubscriptionRequest(S7PlcSubscriptionRequest
request, int tpduId) {
}
- private S7Message
encodeCycledS7ANYSubscriptionRequest(DefaultPlcSubscriptionRequest request, int
tpduId) {
+ private S7Message
encodeCycledS7ANYSubscriptionRequest(S7PlcSubscriptionRequest request, int
tpduId) {
List<S7ParameterUserDataItem> parameterItems = new
ArrayList<>(request.getNumberOfTags());
List<S7PayloadUserDataItem> payloadItems = new
ArrayList<>(request.getNumberOfTags());
@@ -1040,7 +1053,7 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
}
- private S7Message
encodeCycledDBREADSubscriptionRequest(DefaultPlcSubscriptionRequest request,
int tpduId) {
+ private S7Message
encodeCycledDBREADSubscriptionRequest(S7PlcSubscriptionRequest request, int
tpduId) {
List<S7ParameterUserDataItem> parameterItems = new
ArrayList<>(request.getNumberOfTags());
List<S7PayloadUserDataItem> payloadItems = new
ArrayList<>(request.getNumberOfTags());
@@ -1525,7 +1538,7 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
} else if ((myParameter.getCpuFunctionType() == 0x00) &&
(myParameter.getCpuSubfunction() == 0x13)) {
//TODO: Requires reverse engineering.
- } else if ((myParameter.getCpuFunctionGroup() == 0x02) &&
(myParameter.getCpuFunctionType() == 0x00) && (myParameter.getCpuSubfunction()
== 0x01)) { //(05)
+ } else if (((myParameter.getCpuFunctionGroup() == 0x02) &&
(myParameter.getCpuFunctionType() == 0x00) && (myParameter.getCpuSubfunction()
== 0x01))) { //(05)
S7ParameterUserDataItemCPUFunctions parameterItem =
(S7ParameterUserDataItemCPUFunctions)
@@ -1538,7 +1551,17 @@ public class S7ProtocolLogic extends
Plc4xProtocolBase<TPKTPacket> {
S7CyclicEvent cycEvent = new
S7CyclicEvent(cycRequests.get(parameterItem.getSequenceNumber()),
parameterItem.getSequenceNumber(),
payloadItem);
- eventQueue.add(cycEvent);
+
+ if
(cycChangeValueEvents.containsKey(parameterItem.getSequenceNumber())){
+ S7CyclicEvent lastCycEvent =
cycChangeValueEvents.get(parameterItem.getSequenceNumber());
+ if (cycEvent.equals(lastCycEvent ) == false) {
+
cycChangeValueEvents.replace(parameterItem.getSequenceNumber(), cycEvent);
+ eventQueue.add(cycEvent);
+ }
+
+ } else {
+ eventQueue.add(cycEvent);
+ }
} else if ((myParameter.getCpuFunctionGroup() == 0x02) &&
(myParameter.getCpuFunctionType() == 0x00) && (myParameter.getCpuSubfunction()
== 0x05)) { //(06)
diff --git
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/tag/S7SubscriptionTag.java
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/tag/S7SubscriptionTag.java
index cbb0d96d78..ccc4acf658 100644
---
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/tag/S7SubscriptionTag.java
+++
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/tag/S7SubscriptionTag.java
@@ -55,10 +55,20 @@ public class S7SubscriptionTag implements PlcTag {
//blockNumber usually has its max hat around 64000 --> 5digits
private static final Pattern DATA_BLOCK_ADDRESS_PATTERN =
Pattern.compile("%DB(?<blockNumber>\\d{1,5}).DB(?<transferDBSizeCode>[XBWD]?)(?<byteDBOffset>\\d{1,7})(.(?<bitDBOffset>[0-7]))?:(?<dataDBType>[a-zA-Z_]+)(\\[(?<numDBElements>\\d+)])?");
+
+ private static final Pattern DATA_BLOCK_SHORT_PATTERN =
+
Pattern.compile("^%DB(?<blockNumber>\\d{1,5}):(?<byteOffset>\\d{1,7})(.(?<bitOffset>[0-7]))?:(?<dataType>(S5)?[a-zA-Z_]+)(\\[(?<numElements>\\d+)])?");
//All fields index 9
private static final Pattern EVENT_SUBSCRIPTION_S7ANY_QUERY_PATTERN =
-
Pattern.compile("(^CYC(\\((?<timeBase>((B01SEC)|(B1SEC)|(B10SEC))):(?<multiplier>[1-99])\\)):)(((?:,{0,1})(("
+ ADDRESS_PATTERN + ")|(" + DATA_BLOCK_ADDRESS_PATTERN + ")))+)");
+
Pattern.compile("(^CYC(\\((?<timeBase>((B01SEC)|(B1SEC)|(B10SEC))):(?<multiplier>[1-99])\\)):)(((?:,{0,1})(("
+
+ ADDRESS_PATTERN + ")|(" +
+ DATA_BLOCK_ADDRESS_PATTERN + ")))+)");
+
+ private static final Pattern EVENT_SUBSCRIPTION_S7ANY_QUERY_PATTERN_SHORT =
+
Pattern.compile("(^CYC(\\((?<timeBase>((B01SEC)|(B1SEC)|(B10SEC))):(?<multiplier>[1-99])\\)):)(((?:,{0,1})(("
+ DATA_BLOCK_SHORT_PATTERN + ")))+)");
+
+
private static final Pattern EVENT_SUBSCRIPTION_DB_QUERY_PATTERN =
Pattern.compile("(^CYC(\\((?<timeBase>((B01SEC)|(B1SEC)|(B10SEC))):(?<multiplier>[1-99])\\)):)(((?:,{0,1})(%DB(?<blockNumber>\\d{1,5}).DB(?<transferDBSizeCode>[B]?)(?<byteDBOffset>\\d{1,7})(\\[(?<numDBElements>\\d+)]))?)+)");
@@ -199,7 +209,8 @@ public class S7SubscriptionTag implements PlcTag {
EVENT_ALARM_QUERY_PATTERN.matcher(tagString).matches() ||
EVENT_SUBSCRIPTION_S7ANY_QUERY_PATTERN.matcher(tagString).matches() ||
EVENT_SUBSCRIPTION_DB_QUERY_PATTERN.matcher(tagString).matches() ||
- EVENT_CANCEL_JOB_QUERY_PATTERN.matcher(tagString).matches();
+ EVENT_CANCEL_JOB_QUERY_PATTERN.matcher(tagString).matches() ||
+
EVENT_SUBSCRIPTION_S7ANY_QUERY_PATTERN_SHORT.matcher(tagString).matches();
// return EVENT_SUBSCRIPTION_TYPE_PATTERN.matcher(tagString).matches()
||
// EVENT_ALARM_ACK_PATTERN.matcher(tagString).matches() ||
@@ -287,6 +298,27 @@ public class S7SubscriptionTag implements PlcTag {
multi);
}
}
+
+ {
+ Matcher matcher =
EVENT_SUBSCRIPTION_S7ANY_QUERY_PATTERN_SHORT.matcher(tagString);
+ if (matcher.matches()) {
+ TimeBase tb = TimeBase.valueOf(matcher.group(TIME_BASE));
+ short multi =
Short.parseShort(matcher.group(TIME_BASE_MULTIPLIER));
+ S7Tag[] myTags;
+ String strAddress = matcher.group(9);
+ String[] fieldAddress = strAddress.split(",");
+ myTags = new S7Tag[fieldAddress.length];
+ int i = 0;
+ for (String address : fieldAddress) {
+ myTags[i] = S7Tag.of(address);
+ i++;
+ }
+ return new
S7SubscriptionTag(S7SubscriptionType.CYCLIC_SUBSCRIPTION,
+ myTags,
+ tb,
+ multi);
+ }
+ }
{
Matcher matcher =
EVENT_CANCEL_JOB_QUERY_PATTERN.matcher(tagString);
@@ -302,6 +334,7 @@ public class S7SubscriptionTag implements PlcTag {
}
}
+
throw new PlcInvalidTagException("Unable to parse address: " +
tagString);
diff --git
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/S7PlcSubscriptionRequest.java
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/S7PlcSubscriptionRequest.java
new file mode 100644
index 0000000000..a0c06cb6cf
--- /dev/null
+++
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/S7PlcSubscriptionRequest.java
@@ -0,0 +1,363 @@
+/*
+ * 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.s7.readwrite.utils;
+
+import org.apache.plc4x.java.api.exceptions.PlcRuntimeException;
+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.PlcSubscriptionTag;
+import org.apache.plc4x.java.api.model.PlcTag;
+import org.apache.plc4x.java.api.types.PlcSubscriptionType;
+import org.apache.plc4x.java.spi.connection.PlcTagHandler;
+import org.apache.plc4x.java.spi.generation.SerializationException;
+import org.apache.plc4x.java.spi.generation.WriteBuffer;
+import org.apache.plc4x.java.spi.model.DefaultPlcSubscriptionTag;
+import org.apache.plc4x.java.spi.utils.Serializable;
+
+import java.time.Duration;
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import org.apache.plc4x.java.s7.readwrite.TimeBase;
+import static org.apache.plc4x.java.s7.readwrite.TimeBase.B01SEC;
+import org.apache.plc4x.java.s7.readwrite.tag.S7SubscriptionTag;
+import org.apache.plc4x.java.s7.readwrite.tag.S7Tag;
+import org.apache.plc4x.java.s7.readwrite.types.S7SubscriptionType;
+import org.apache.plc4x.java.spi.messages.DefaultPlcSubscriptionRequest;
+import org.apache.plc4x.java.spi.messages.PlcSubscriber;
+
+public class S7PlcSubscriptionRequest implements PlcSubscriptionRequest,
Serializable {
+
+ private final PlcSubscriber subscriber;
+
+ private final LinkedHashMap<String, PlcSubscriptionTag> tags;
+
+ private final LinkedHashMap<String, List<Consumer<PlcSubscriptionEvent>>>
preRegisteredConsumers;
+
+ public S7PlcSubscriptionRequest(PlcSubscriber subscriber,
+ LinkedHashMap<String,
PlcSubscriptionTag> tags,
+ LinkedHashMap<String,
List<Consumer<PlcSubscriptionEvent>>> preRegisteredConsumers) {
+ this.subscriber = subscriber;
+ this.tags = tags;
+ this.preRegisteredConsumers = preRegisteredConsumers;
+ }
+
+ @Override
+ public CompletableFuture<PlcSubscriptionResponse> execute() {
+ return subscriber.subscribe(this);
+ }
+
+ @Override
+ public int getNumberOfTags() {
+ return tags.size();
+ }
+
+ @Override
+ public LinkedHashSet<String> getTagNames() {
+ return new LinkedHashSet<>(tags.keySet());
+ }
+
+ @Override
+ public PlcSubscriptionTag getTag(String name) {
+ return tags.get(name);
+ }
+
+ @Override
+ public List<PlcSubscriptionTag> getTags() {
+ return new ArrayList<>(tags.values());
+ }
+
+ @Override
+ public Map<String, List<Consumer<PlcSubscriptionEvent>>>
getPreRegisteredConsumers() {
+ return new LinkedHashMap<>(preRegisteredConsumers);
+ }
+
+ @Override
+ public void serialize(WriteBuffer writeBuffer) throws
SerializationException {
+ writeBuffer.pushContext("PlcSubscriptionRequest");
+
+ writeBuffer.pushContext("tags");
+ for (Map.Entry<String, PlcSubscriptionTag> tagEntry : tags.entrySet())
{
+ String tagName = tagEntry.getKey();
+ writeBuffer.pushContext(tagName);
+ PlcTag tag = tagEntry.getValue();
+ if (!(tag instanceof Serializable)) {
+ throw new RuntimeException("Error serializing. Tag doesn't
implement XmlSerializable");
+ }
+ ((Serializable) tag).serialize(writeBuffer);
+ writeBuffer.popContext(tagName);
+ }
+ writeBuffer.popContext("tags");
+
+ writeBuffer.popContext("PlcSubscriptionRequest");
+ }
+
+ public static class Builder implements PlcSubscriptionRequest.Builder {
+
+ private final PlcSubscriber subscriber;
+ private final PlcTagHandler tagHandler;
+ private final Map<String, BuilderItem> tags;
+ private final LinkedHashMap<String,
List<Consumer<PlcSubscriptionEvent>>> preRegisteredConsumers;
+
+ public Builder(PlcSubscriber subscriber, PlcTagHandler tagHandler) {
+ this.subscriber = subscriber;
+ this.tagHandler = tagHandler;
+ this.tags = new TreeMap<>();
+ this.preRegisteredConsumers = new LinkedHashMap<>();
+ }
+
+ /*
+ * This method receives a String that describes an S7Tag and the
+ * interval required for its sampling.
+ * The value of the "pollingInterval" parameter is adapted to the
+ * cyclical subscription requirements of an S7-300/S7-400,
+ * for which multiples of the time base given by TimeBase
+ * must be handled. To say:
+ *
+ * . B01SEC -> 100, 200, 300, 400, 500, 600, 700, 800, 900 msec
+ * . B1SEC -> 1, 2, 3, 4, 5, 6, 7, 8, 9 sec
+ * . B10SEC -> 10, 20, 30, 40, 50, 60, 70, 80, 90 sec
+ *
+ * As you can see there are no intermediate values, for example 513
msec,
+ * it will actually be 500 msec, or its nearest rounding.
+ *
+ * @param name Name of the subscription Tag.
+ * @param tagAddress String representing an S7Tag
+ * @param pollingInterval Required sampling rate based on the
"TimeBase"
+ * @return PlcSubscriptionRequest.Builder S7SubscriptonTag type
constructor
+ *
+ */
+ @Override
+ public PlcSubscriptionRequest.Builder addCyclicTagAddress(String name,
String tagAddress, Duration pollingInterval) {
+ if (tags.containsKey(name)) {
+ throw new PlcRuntimeException("Duplicate tag definition '" +
name + "'");
+ }
+ TimeBase tb = getTimeBase(pollingInterval);
+ short multiplier = getMultiplier(tb, pollingInterval);
+ S7Tag[] s7tags = new S7Tag[]{S7Tag.of(tagAddress)};
+ S7SubscriptionTag tag = new
S7SubscriptionTag(S7SubscriptionType.CYCLIC_SUBSCRIPTION, s7tags, tb,
multiplier);
+ tags.put(name, new BuilderItem(() -> tag,
PlcSubscriptionType.CYCLIC, pollingInterval));
+ return this;
+ }
+
+ /*
+ * This method receives an S7Tag built by the user, he is responsible
+ * for the construction of the object, so no additional verification
+ * is included.
+ *
+ * @param name Name of the subscription Tag.
+ * @param tag Tag of S7SubscriptionTag type.
+ * @param pollingInterval Required sampling rate based on the "TimeBase"
+ * @return PlcSubscriptionRequest.Builder S7SubscriptonTag type
constructor
+ */
+ @Override
+ public PlcSubscriptionRequest.Builder addCyclicTag(String name, PlcTag
tag, Duration pollingInterval) {
+ if (tags.containsKey(name)) {
+ throw new PlcRuntimeException("Duplicate tag definition '" +
name + "'");
+ }
+ if ((tag instanceof S7SubscriptionTag) == false){
+ throw new PlcRuntimeException("Tag is not of type
S7SubcriptionTag");
+ }
+ tags.put(name, new BuilderItem(() -> tag,
PlcSubscriptionType.CYCLIC, pollingInterval));
+ return this;
+ }
+
+ /*
+ *
+ */
+ @Override
+ public PlcSubscriptionRequest.Builder
addChangeOfStateTagAddress(String name, String tagAddress) {
+ if (tags.containsKey(name)) {
+ throw new PlcRuntimeException("Duplicate tag definition '" +
name + "'");
+ }
+ S7Tag[] s7tags = new S7Tag[]{S7Tag.of(tagAddress)};
+ S7SubscriptionTag tag = new
S7SubscriptionTag(S7SubscriptionType.CYCLIC_SUBSCRIPTION, s7tags,
TimeBase.B01SEC, (short) 1);
+ tags.put(name, new BuilderItem(() -> tag,
PlcSubscriptionType.CHANGE_OF_STATE));
+ return this;
+ }
+
+ /*
+ *
+ */
+ @Override
+ public PlcSubscriptionRequest.Builder addChangeOfStateTag(String name,
PlcTag tag) {
+ if (tags.containsKey(name)) {
+ throw new PlcRuntimeException("Duplicate tag definition '" +
name + "'");
+ }
+ if ((tag instanceof S7SubscriptionTag) == false){
+ throw new PlcRuntimeException("Tag is not of type
S7SubcriptionTag");
+ }
+ tags.put(name, new BuilderItem(() -> tag,
PlcSubscriptionType.CHANGE_OF_STATE));
+ return this;
+ }
+
+ /*
+ * This method is responsible for the subscription to Events associated
+ * with the PLC as well as the preliminary version of cyclical
+ * subscription of values.
+ *
+ * The type of function performed by the tag is given by the definition
+ * of the "tagAddress", for example:
+ *
+ * "ACK:16#12345678"
+ *
+ * Represents an acknowledgment of an alarm whose ID is 16#12345678.
+ * The following functions are defined:
+ *
+ * . MODE
+ * . SYS
+ * . USR
+ * . ALM
+ * . ACK
+ * . QUERY
+ * . CYC
+ * . CANCEL
+ *
+ * Go to the driver manual for a complete description.
+ *
+ * @param name Name of the subscription Tag.
+ * @param tag Tag of S7SubscriptionTag type.
+ * @return PlcSubscriptionRequest.Builder S7SubscriptonTag type
constructor
+ */
+ @Override
+ public PlcSubscriptionRequest.Builder addEventTagAddress(String name,
String tagAddress) {
+ if (tags.containsKey(name)) {
+ throw new PlcRuntimeException("Duplicate tag definition '" +
name + "'");
+ }
+ PlcTag tag = tagHandler.parseTag(tagAddress);
+ if ((tag instanceof S7SubscriptionTag) == false){
+ throw new PlcRuntimeException("Tag address is not of type
S7SubcriptionTag");
+ }
+ tags.put(name, new BuilderItem(() ->
tagHandler.parseTag(tagAddress), PlcSubscriptionType.EVENT));
+ return this;
+ }
+
+ /*
+ * This method receives an S7Tag built by the user, he is responsible
+ * for the construction of the object, so no additional verification
+ * is included.
+ *
+ * @param name Name of the subscription Tag.
+ * @param tag Tag of S7SubscriptionTag type.
+ * @return PlcSubscriptionRequest.Builder S7SubscriptonTag type
constructor
+ */
+ @Override
+ public PlcSubscriptionRequest.Builder addEventTag(String name, PlcTag
tag) {
+ if (tags.containsKey(name)) {
+ throw new PlcRuntimeException("Duplicate tag definition '" +
name + "'");
+ }
+ if ((tag instanceof S7SubscriptionTag) == false){
+ throw new PlcRuntimeException("Tag is not of type
S7SubcriptionTag");
+ }
+ tags.put(name, new BuilderItem(() -> tag,
PlcSubscriptionType.EVENT));
+ return this;
+ }
+
+ @Override
+ public PlcSubscriptionRequest.Builder addPreRegisteredConsumer(String
name, Consumer<PlcSubscriptionEvent> consumer) {
+ preRegisteredConsumers.putIfAbsent(name, new LinkedList<>());
+ preRegisteredConsumers.get(name).add(consumer);
+ return this;
+ }
+
+ @Override
+ public PlcSubscriptionRequest build() {
+ LinkedHashMap<String, PlcSubscriptionTag> parsedTags = new
LinkedHashMap<>();
+
+ tags.forEach((name, builderItem) -> {
+ PlcTag parsedTag = builderItem.tag.get();
+ parsedTags.put(name, new
DefaultPlcSubscriptionTag(builderItem.plcSubscriptionType, parsedTag,
builderItem.duration));
+ });
+ preRegisteredConsumers.forEach((tagName, ignored) -> {
+ if (!tags.containsKey(tagName)) {
+ throw new RuntimeException("tagName " + tagName + "for
preRegisteredConsumer not found");
+ }
+ });
+ return new S7PlcSubscriptionRequest(subscriber, parsedTags,
preRegisteredConsumers);
+ }
+
+ private static class BuilderItem {
+ private final Supplier<PlcTag> tag;
+ private final PlcSubscriptionType plcSubscriptionType;
+ private final Duration duration;
+
+ private BuilderItem(Supplier<PlcTag> tag, PlcSubscriptionType
plcSubscriptionType) {
+ this(tag, plcSubscriptionType, null);
+ }
+
+ private BuilderItem(Supplier<PlcTag> tag, PlcSubscriptionType
plcSubscriptionType, Duration duration) {
+ this.tag = tag;
+ this.plcSubscriptionType = plcSubscriptionType;
+ this.duration = duration;
+ }
+
+ }
+
+ private TimeBase getTimeBase(Duration duration) {
+ if (duration.equals(Duration.ZERO)) {
+ throw new PlcRuntimeException("Subscription time cannot be
zero.");
+ }
+ long millis = duration.toMillis();
+ if (millis < 1000) {
+ return TimeBase.B01SEC;
+ } if (millis < 10000) {
+ return TimeBase.B1SEC;
+ } if (millis < 100000) {
+ return TimeBase.B10SEC;
+ }
+
+ throw new PlcRuntimeException("The maximum subscription time is 90
sec.");
+ }
+
+ //TODO: Chek multiplier is 1-99 in BCD??
+ private short getMultiplier(TimeBase tbase, Duration duration) {
+ short multiplier = 1;
+ if (duration.equals(Duration.ZERO)) {
+ throw new PlcRuntimeException("Subscription time cannot be
zero.");
+ }
+ long millis = duration.toMillis();
+ switch(tbase) {
+ case B01SEC:;
+ if (millis > 100) {
+ multiplier = (short) (millis / 100);
+ }
+ break;
+ case B1SEC:;
+ multiplier = (short) (millis / 1000);
+ break;
+ case B10SEC:;
+ multiplier = (short) (millis / 10000);
+ break;
+
+ }
+ return multiplier;
+ }
+
+ }
+
+ @Override
+ public String toString() {
+ return "DefaultPlcSubscriptionRequest{" +
+ "subscriber=" + subscriber +
+ ", tags=" + tags +
+ '}';
+ }
+}
diff --git
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/StaticHelper.java
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/StaticHelper.java
index 76c65b5e8f..709cfc7fcf 100644
---
a/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/StaticHelper.java
+++
b/plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/utils/StaticHelper.java
@@ -2139,7 +2139,8 @@ public class StaticHelper {
public static int rightShift3(final ReadBuffer buffer, DataTransportSize
tsize) throws ParseException {
int value = 0;
if ((tsize == DataTransportSize.OCTET_STRING) ||
- (tsize == DataTransportSize.REAL)) {
+ (tsize == DataTransportSize.REAL) ||
+ (tsize == DataTransportSize.BIT)) {
value = buffer.readUnsignedInt(16);
} else {
value = buffer.readUnsignedInt(16) >> 3;
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBINT.java
index cd8b739a96..e3295121f8 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBINT.java
@@ -210,6 +210,11 @@ public class PlcBINT extends PlcIECValue<BigInteger> {
public String toString() {
return value.toString();
}
+
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
public byte[] getBytes() {
byte[] tmp = value.toByteArray();
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBOOL.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBOOL.java
index b1f900a9ad..229c805de9 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBOOL.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBOOL.java
@@ -253,6 +253,11 @@ public class PlcBOOL extends PlcIECValue<Boolean> {
return toString();
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
return ((value != null) && value) ? new byte[]{0x01} : new
byte[]{0x00};
}
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBREAL.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBREAL.java
index 6b67ea4f3c..6f73a709e6 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBREAL.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBREAL.java
@@ -211,6 +211,11 @@ public class PlcBREAL extends PlcIECValue<BigDecimal> {
return value.toString();
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
// TODO: Not sure if this is correct ...
byte[] tmp = value.unscaledValue().toByteArray();
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBYTE.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBYTE.java
index 76f67e75c1..72370dee5a 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBYTE.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcBYTE.java
@@ -266,6 +266,11 @@ public class PlcBYTE extends PlcIECValue<Short> {
public String toString() {
return Short.toString(value);
}
+
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
public byte[] getBytes() {
byte[] bytes = new byte[1];
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcCHAR.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcCHAR.java
index 7f24576d0d..19a4727e0f 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcCHAR.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcCHAR.java
@@ -300,6 +300,11 @@ public class PlcCHAR extends PlcIECValue<Short> {
return Character.toString((char) ((short) value));
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
byte[] bytes = new byte[1];
bytes[0] = (byte) (value & 0xff);
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcDINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcDINT.java
index b0f2abd170..dc587043a9 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcDINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcDINT.java
@@ -240,6 +240,11 @@ public class PlcDINT extends PlcIECValue<Integer> {
return Integer.toString(value);
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
return new byte[]{
(byte) ((value >> 24) & 0xff),
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcDWORD.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcDWORD.java
index 64ac2e0575..e33a6f9f25 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcDWORD.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcDWORD.java
@@ -267,6 +267,11 @@ public class PlcDWORD extends PlcIECValue<Long> {
return Long.toString(value);
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
byte[] bytes = new byte[4];
bytes[0] = (byte) ((value >> 24) & 0xff);
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcINT.java
index 874ad6fb2a..93caa2ce75 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcINT.java
@@ -244,6 +244,11 @@ public class PlcINT extends PlcIECValue<Short> {
return Integer.toString(value);
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
return new byte[]{
(byte) ((value >> 8) & 0xff),
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLINT.java
index 3b6e4d5ebf..1e1ecfac0f 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLINT.java
@@ -237,6 +237,11 @@ public class PlcLINT extends PlcIECValue<Long> {
return Long.toString(value);
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
return new byte[]{
(byte) ((value >> 56) & 0xff),
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLREAL.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLREAL.java
index a407405f10..2d440d847b 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLREAL.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLREAL.java
@@ -227,6 +227,11 @@ public class PlcLREAL extends PlcIECValue<Double> {
return Double.toString(value);
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
long longBits = Double.doubleToRawLongBits(value);
return new byte[]{
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLWORD.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLWORD.java
index acba299118..c0b055a6a0 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLWORD.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcLWORD.java
@@ -277,6 +277,11 @@ public class PlcLWORD extends PlcIECValue<BigInteger> {
return value.toString();
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
byte[] tmp = value.toByteArray();
byte[] bytes = new byte[8];
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcREAL.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcREAL.java
index bef0e744c5..f919cacecb 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcREAL.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcREAL.java
@@ -229,6 +229,11 @@ public class PlcREAL extends PlcIECValue<Float> {
public String toString() {
return Float.toString(value);
}
+
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
public byte[] getBytes() {
int intBits = Float.floatToIntBits(value);
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcRawByteArray.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcRawByteArray.java
index 7111783054..0310ba3a33 100644
---
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcRawByteArray.java
+++
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcRawByteArray.java
@@ -46,6 +46,11 @@ public class PlcRawByteArray extends PlcIECValue<byte[]> {
public byte[] getRaw() {
return value;
}
+
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
@Override
public String toString() {
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcSINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcSINT.java
index 852da57090..162f430caa 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcSINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcSINT.java
@@ -246,6 +246,11 @@ public class PlcSINT extends PlcIECValue<Byte> {
return Byte.toString(value);
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
byte[] bytes = new byte[1];
bytes[0] = (byte) (value & 0xff);
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUBINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUBINT.java
index 772314204c..e688f9075c 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUBINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUBINT.java
@@ -266,6 +266,11 @@ public class PlcUBINT extends PlcIECValue<BigInteger> {
return value.toString();
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
byte[] tmp = value.toByteArray();
byte[] bytes = new byte[8];
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUDINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUDINT.java
index 861fd75a32..ede2fa8a03 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUDINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUDINT.java
@@ -259,6 +259,11 @@ public class PlcUDINT extends PlcIECValue<Long> {
return Long.toString(value);
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
byte[] bytes = new byte[4];
bytes[0] = (byte) ((value >> 24) & 0xff);
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUINT.java
index 8a82ea2359..ee7a66c4f5 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUINT.java
@@ -253,6 +253,11 @@ public class PlcUINT extends PlcIECValue<Integer> {
return Integer.toString(value);
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
return new byte[]{
(byte) ((value >> 8) & 0xff),
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcULINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcULINT.java
index 589f06114c..f9107815ae 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcULINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcULINT.java
@@ -267,6 +267,11 @@ public class PlcULINT extends PlcIECValue<BigInteger> {
return value.toString();
}
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
+
public byte[] getBytes() {
byte[] tmp = value.toByteArray();
byte[] bytes = new byte[8];
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUSINT.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUSINT.java
index 23346642c7..d6f16e8b2b 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUSINT.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcUSINT.java
@@ -255,6 +255,11 @@ public class PlcUSINT extends PlcIECValue<Short> {
public String toString() {
return Short.toString(value);
}
+
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
public byte[] getBytes() {
byte[] bytes = new byte[1];
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcWCHAR.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcWCHAR.java
index d72ac5b042..5e787a3db5 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcWCHAR.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcWCHAR.java
@@ -274,6 +274,11 @@ public class PlcWCHAR extends PlcIECValue<Integer> {
public String toString() {
return Character.toString((char) ((int) value));
}
+
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
public byte[] getBytes() {
return new byte[]{
diff --git
a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcWORD.java
b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcWORD.java
index a227dc41e0..9003db7753 100644
--- a/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcWORD.java
+++ b/plc4j/spi/src/main/java/org/apache/plc4x/java/spi/values/PlcWORD.java
@@ -262,6 +262,11 @@ public class PlcWORD extends PlcIECValue<Integer> {
public String toString() {
return Integer.toString(value);
}
+
+ @Override
+ public byte[] getRaw() {
+ return getBytes();
+ }
public byte[] getBytes() {
return new byte[]{
diff --git
a/protocols/knxnetip/src/main/generated/protocols/knxnetip/knx-master-data.mspec
b/protocols/knxnetip/src/main/generated/protocols/knxnetip/knx-master-data.mspec
index 7aed105661..f1f77e7829 100644
---
a/protocols/knxnetip/src/main/generated/protocols/knxnetip/knx-master-data.mspec
+++
b/protocols/knxnetip/src/main/generated/protocols/knxnetip/knx-master-data.mspec
@@ -1419,8 +1419,7 @@
['668' M_MUTLUSAN_ELECTRIC ['726', '"Mutlusan Electric"']]
['669' M_HANGZHOU_BROADLINK_TECHNOLOGY_CO__LTD_ ['727', '"Hangzhou
BroadLink Technology Co.,Ltd."']]
['670' M_ABB___RESERVED ['43954', '"ABB - reserved"']]
- ['671' M_BUSCH_JAEGER_ELEKTRO___RESERVED ['43959', '"Busch-Jaeger Elektro
- reserved"']]
-
+ ['671' M_BUSCH_JAEGER_ELEKTRO___RESERVED ['43959', '"Busch-Jaeger Elektro
- reserved"']]
]
[dataIo KnxDatapoint(KnxDatapointType datapointType)
