You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2026/08/18 10:37:57 UTC
(plc4x) branch develop updated (983ff4dd76 -> 5c316cce12)
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a change to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
from 983ff4dd76 fix(plc4go): widen length/position arithmetic to close the f009 uint16 wraps, and regenerate
new 6f2ac81e84 fix(plc4go/s7): use the generated ControllerType enum instead of a colliding local copy
new 80451d91b9 fix(plc4go/s7): port read-response decoding and the stubbed S7 static helpers
new de83bbff8f fix(plc4go/s7): write path honors string length, controller type and multi-element values
new 0815d3fc18 feat(plc4go/s7): parse S5TIME, var-length strings and alarm tag addresses
new 3aef13b873 fix(plc4go/s7): restore SZL controller-type detection broken since the s7h mspec refactor
new 97253814c1 feat(plc4go/s7): implement browse (static areas + DB block enumeration)
new 592473f776 feat(plc4go/s7): implement alarm + cyclic subscriptions, alarm query and push dispatch
new 203cfdd5ad feat(plc4go/s7): real round-trip ping
new 5c316cce12 test(plc4go/s7): re-enable the driver testsuite cases skipped since the s7h drift
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../testing/protocols/s7/DriverTestsuite.xml | 71 +-
plc4go/internal/s7/AlarmTag.go | 118 ++++
plc4go/internal/s7/Browser.go | 121 ++++
plc4go/internal/s7/Configuration.go | 24 +-
plc4go/internal/s7/Configuration_test.go | 75 +++
plc4go/internal/s7/Connection.go | 363 ++++++----
plc4go/internal/s7/DriverContext.go | 16 +-
plc4go/internal/s7/MessageCodec.go | 78 ++-
plc4go/internal/s7/{ControllerType.go => Query.go} | 23 +-
plc4go/internal/s7/Reader.go | 119 ++--
plc4go/internal/s7/Reader_test.go | 124 ++++
plc4go/internal/s7/Subscriber.go | 530 +++++++++++++++
plc4go/internal/s7/Subscriber_test.go | 220 +++++++
plc4go/internal/{cbus => s7}/SubscriptionEvent.go | 21 +-
plc4go/internal/{cbus => s7}/SubscriptionHandle.go | 48 +-
plc4go/internal/s7/Tag.go | 11 +
plc4go/internal/s7/TagHandler.go | 124 +++-
plc4go/internal/s7/TagHandler_test.go | 101 +++
plc4go/internal/s7/UserDataService.go | 727 +++++++++++++++++++++
plc4go/internal/s7/UserDataService_test.go | 153 +++++
plc4go/internal/s7/Writer.go | 83 ++-
plc4go/internal/s7/Writer_test.go | 74 +++
.../protocols/s7/readwrite/model/StaticHelper.go | 209 +++++-
plc4go/spi/model/DefaultPlcSubscriptionHandle.go | 6 +
.../spi/model/DefaultPlcUnsubscriptionRequest.go | 10 +-
plc4go/tests/drivers/tests/s7_driver_test.go | 5 -
.../resources/protocols/s7/DriverTestsuite.xml | 71 +-
27 files changed, 3086 insertions(+), 439 deletions(-)
create mode 100644 plc4go/internal/s7/AlarmTag.go
create mode 100644 plc4go/internal/s7/Browser.go
create mode 100644 plc4go/internal/s7/Configuration_test.go
rename plc4go/internal/s7/{ControllerType.go => Query.go} (70%)
create mode 100644 plc4go/internal/s7/Reader_test.go
create mode 100644 plc4go/internal/s7/Subscriber.go
create mode 100644 plc4go/internal/s7/Subscriber_test.go
copy plc4go/internal/{cbus => s7}/SubscriptionEvent.go (84%)
copy plc4go/internal/{cbus => s7}/SubscriptionHandle.go (55%)
create mode 100644 plc4go/internal/s7/TagHandler_test.go
create mode 100644 plc4go/internal/s7/UserDataService.go
create mode 100644 plc4go/internal/s7/UserDataService_test.go
create mode 100644 plc4go/internal/s7/Writer_test.go
(plc4x) 04/09: feat(plc4go/s7): parse S5TIME, var-length strings and alarm tag addresses
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 0815d3fc18ad90be08051e0fc0b5e60c1e2fa2f4
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 11:59:39 2026 +0200
feat(plc4go/s7): parse S5TIME, var-length strings and alarm tag addresses
- allow the S5 prefix in the data-type regex groups so %M0:S5TIME parses
- add var-length STRING/WSTRING patterns behaving as 254-char strings,
matching the Java driver semantics
- add the ALM / QUERY:ALARM_S / QUERY:ALARM_8 alarm tag addresses
- implement PlcSubscriptionTag on the tags so subscription request
builders accept them
- reject byte offsets that would truncate in the 16-bit wire address
instead of silently wrapping
---
plc4go/internal/s7/AlarmTag.go | 118 +++++++++++++++++++++++++++++++++
plc4go/internal/s7/Tag.go | 11 ++++
plc4go/internal/s7/TagHandler.go | 121 +++++++++++++++++++++++++++++-----
plc4go/internal/s7/TagHandler_test.go | 101 ++++++++++++++++++++++++++++
4 files changed, 335 insertions(+), 16 deletions(-)
diff --git a/plc4go/internal/s7/AlarmTag.go b/plc4go/internal/s7/AlarmTag.go
new file mode 100644
index 0000000000..3b041d37bf
--- /dev/null
+++ b/plc4go/internal/s7/AlarmTag.go
@@ -0,0 +1,118 @@
+/*
+ * 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 s7
+
+import (
+ "context"
+ "strings"
+ "time"
+
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi/utils"
+)
+
+const (
+ alarmTagAddressAlarm = "ALM"
+ alarmTagAddressQueryS = "QUERY:ALARM_S"
+ alarmTagAddressQuery8 = "QUERY:ALARM_8"
+)
+
+type AlarmTagKind uint8
+
+const (
+ AlarmTagPush AlarmTagKind = iota
+ AlarmTagQuery
+)
+
+// AlarmTag addresses the S7 alarm/event services instead of a memory location. "ALM" subscribes
+// to pushed alarm indications, "QUERY:ALARM_S"/"QUERY:ALARM_8" run a one-shot alarm query.
+type AlarmTag struct {
+ address string
+ kind AlarmTagKind
+ queryType readWriteModel.QueryType
+}
+
+func tryParseAlarmTag(tagAddress string) (*AlarmTag, bool) {
+ switch {
+ case strings.EqualFold(tagAddress, alarmTagAddressAlarm):
+ return &AlarmTag{address: alarmTagAddressAlarm, kind: AlarmTagPush, queryType: readWriteModel.QueryType_ALARM_S}, true
+ case strings.EqualFold(tagAddress, alarmTagAddressQueryS):
+ return &AlarmTag{address: alarmTagAddressQueryS, kind: AlarmTagQuery, queryType: readWriteModel.QueryType_ALARM_S}, true
+ case strings.EqualFold(tagAddress, alarmTagAddressQuery8):
+ return &AlarmTag{address: alarmTagAddressQuery8, kind: AlarmTagQuery, queryType: readWriteModel.QueryType_ALARM_8}, true
+ default:
+ return nil, false
+ }
+}
+
+func (m *AlarmTag) GetAddressString() string {
+ return m.address
+}
+
+func (m *AlarmTag) GetValueType() apiValues.PlcValueType {
+ return apiValues.Struct
+}
+
+func (m *AlarmTag) GetArrayInfo() []apiModel.ArrayInfo {
+ return []apiModel.ArrayInfo{}
+}
+
+func (m *AlarmTag) GetKind() AlarmTagKind {
+ return m.kind
+}
+
+func (m *AlarmTag) GetQueryType() readWriteModel.QueryType {
+ return m.queryType
+}
+
+func (m *AlarmTag) GetPlcSubscriptionType() apiModel.PlcSubscriptionType {
+ return apiModel.SubscriptionEvent
+}
+
+func (m *AlarmTag) GetDuration() time.Duration {
+ return 0
+}
+
+func (m *AlarmTag) Serialize() ([]byte, error) {
+ wb := utils.NewWriteBufferByteBased()
+ if err := m.SerializeWithWriteBuffer(context.Background(), wb); err != nil {
+ return nil, err
+ }
+ return wb.GetBytes(), nil
+}
+
+func (m *AlarmTag) SerializeWithWriteBuffer(ctx context.Context, wb utils.WriteBuffer) error {
+ if err := wb.PushContext("S7AlarmTag"); err != nil {
+ return err
+ }
+ if err := wb.WriteString("address", uint32(len(m.address)*8), m.address); err != nil {
+ return err
+ }
+ if err := wb.PopContext("S7AlarmTag"); err != nil {
+ return err
+ }
+ return nil
+}
+
+func (m *AlarmTag) String() string {
+ return m.address
+}
diff --git a/plc4go/internal/s7/Tag.go b/plc4go/internal/s7/Tag.go
index f2be4b13d8..de01f16622 100644
--- a/plc4go/internal/s7/Tag.go
+++ b/plc4go/internal/s7/Tag.go
@@ -23,6 +23,7 @@ import (
"context"
"encoding/binary"
"fmt"
+ "time"
apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
@@ -137,6 +138,16 @@ func (m plcTag) GetQuantity() uint16 {
return m.NumElements
}
+// GetPlcSubscriptionType makes address tags usable in subscription requests (cyclic services).
+func (m plcTag) GetPlcSubscriptionType() apiModel.PlcSubscriptionType {
+ return apiModel.SubscriptionCyclic
+}
+
+// GetDuration returns no default interval; the effective cycle time comes from the request.
+func (m plcTag) GetDuration() time.Duration {
+ return 0
+}
+
func (m plcTag) Serialize() ([]byte, error) {
wb := utils.NewWriteBufferByteBased(utils.WithByteOrderForByteBasedBuffer(binary.BigEndian))
if err := m.SerializeWithWriteBuffer(context.Background(), wb); err != nil {
diff --git a/plc4go/internal/s7/TagHandler.go b/plc4go/internal/s7/TagHandler.go
index e61826325f..29b09a1215 100644
--- a/plc4go/internal/s7/TagHandler.go
+++ b/plc4go/internal/s7/TagHandler.go
@@ -50,12 +50,14 @@ func (i TagType) GetName() string {
}
type TagHandler struct {
- addressPattern *regexp.Regexp
- dataBlockAddressPattern *regexp.Regexp
- dataBlockShortPattern *regexp.Regexp
- dataBlockStringAddressPattern *regexp.Regexp
- dataBlockStringShortPattern *regexp.Regexp
- plcProxyAddressPattern *regexp.Regexp
+ addressPattern *regexp.Regexp
+ dataBlockAddressPattern *regexp.Regexp
+ dataBlockShortPattern *regexp.Regexp
+ dataBlockStringAddressPattern *regexp.Regexp
+ dataBlockStringShortPattern *regexp.Regexp
+ dataBlockStringVarLengthAddressPattern *regexp.Regexp
+ dataBlockStringVarLengthShortPattern *regexp.Regexp
+ plcProxyAddressPattern *regexp.Regexp
passLogToModel bool
log zerolog.Logger
@@ -65,13 +67,16 @@ func NewTagHandler(_options ...options.WithOption) TagHandler {
passLoggerToModel, _ := options.ExtractPassLoggerToModel(_options...)
customLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...)
return TagHandler{
- addressPattern: regexp.MustCompile(`^%(?P<memoryArea>.)(?P<transferSizeCode>[XBWD]?)(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>[a-zA-Z_]+)(\[(?P<numElements>\d+)])?`),
+ // the (?:S5)? prefix is required because the character class doesn't allow digits (S5TIME)
+ addressPattern: regexp.MustCompile(`^%(?P<memoryArea>.)(?P<transferSizeCode>[XBWD]?)(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>(?:S5)?[a-zA-Z_]+)(\[(?P<numElements>\d+)])?`),
//blockNumber usually has its max hat around 64000 --> 5digits
- dataBlockAddressPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}).DB(?P<transferSizeCode>[XBWD]?)(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>[a-zA-Z_]+)(\[(?P<numElements>\d+)])?`),
- dataBlockShortPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}):(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>[a-zA-Z_]+)(\[(?P<numElements>\d+)])?`),
- dataBlockStringAddressPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}).DB(?P<transferSizeCode>[XBWD]?)(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>STRING|WSTRING)\((?P<stringLength>\d{1,3})\)(\[(?P<numElements>\d+)])?`),
- dataBlockStringShortPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}):(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>STRING|WSTRING)\((?P<stringLength>\d{1,3})\)(\[(?P<numElements>\d+)])?`),
- plcProxyAddressPattern: regexp.MustCompile(`[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}`),
+ dataBlockAddressPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}).DB(?P<transferSizeCode>[XBWD]?)(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>(?:S5)?[a-zA-Z_]+)(\[(?P<numElements>\d+)])?`),
+ dataBlockShortPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}):(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>(?:S5)?[a-zA-Z_]+)(\[(?P<numElements>\d+)])?`),
+ dataBlockStringAddressPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}).DB(?P<transferSizeCode>[XBWD]?)(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>STRING|WSTRING)\((?P<stringLength>\d{1,3})\)(\[(?P<numElements>\d+)])?`),
+ dataBlockStringShortPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}):(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>STRING|WSTRING)\((?P<stringLength>\d{1,3})\)(\[(?P<numElements>\d+)])?`),
+ dataBlockStringVarLengthAddressPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}).DB(?P<transferSizeCode>[XBWD]?)(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>STRING|WSTRING)(\[(?P<numElements>\d+)])?$`),
+ dataBlockStringVarLengthShortPattern: regexp.MustCompile(`^%DB(?P<blockNumber>\d{1,5}):(?P<byteOffset>\d{1,7})(.(?P<bitOffset>[0-7]))?:(?P<dataType>STRING|WSTRING)(\[(?P<numElements>\d+)])?$`),
+ plcProxyAddressPattern: regexp.MustCompile(`[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}-[0-9A-F]{2}`),
passLogToModel: passLoggerToModel,
log: customLogger,
@@ -91,6 +96,9 @@ const (
func (m TagHandler) ParseTag(tagAddress string) (apiModel.PlcTag, error) {
ctx := context.TODO()
+ if alarmTag, ok := tryParseAlarmTag(tagAddress); ok {
+ return alarmTag, nil
+ }
if match := utils.GetSubgroupMatches(m.dataBlockStringAddressPattern, tagAddress); match != nil {
dataType, ok := readWriteModel.TransportSizeByName(match[DATA_TYPE])
if !ok {
@@ -154,7 +162,7 @@ func (m TagHandler) ParseTag(tagAddress string) (apiModel.PlcTag, error) {
if err != nil {
return nil, errors.Wrap(err, "Error checking blocknumber")
}
- parsedByteOffset, err := strconv.ParseUint(match[BYTE_OFFSET], 10, 8)
+ parsedByteOffset, err := strconv.ParseUint(match[BYTE_OFFSET], 10, 64)
if err != nil {
return nil, errors.Wrap(err, "Error converting byteoffset")
}
@@ -173,6 +181,86 @@ func (m TagHandler) ParseTag(tagAddress string) (apiModel.PlcTag, error) {
}
return NewStringTag(memoryArea, blockNumber, byteOffset, bitOffset, numElements, stringLength, dataType), nil
+ } else if match := utils.GetSubgroupMatches(m.dataBlockStringVarLengthAddressPattern, tagAddress); match != nil {
+ dataType, ok := readWriteModel.TransportSizeByName(match[DATA_TYPE])
+ if !ok {
+ return nil, errors.Errorf("Unknown type %s", match[DATA_TYPE])
+ }
+ memoryArea := readWriteModel.MemoryArea_DATA_BLOCKS
+ transferSizeCode := getSizeCode(match[TRANSFER_SIZE_CODE])
+ parsedBlockNumber, err := strconv.ParseUint(match[BLOCK_NUMBER], 10, 16)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting blocknumber")
+ }
+ blockNumber, err := checkDatablockNumber(parsedBlockNumber)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error checking blocknumber")
+ }
+ parsedByteOffset, err := strconv.ParseUint(match[BYTE_OFFSET], 10, 64)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting byteoffset")
+ }
+ byteOffset, err := checkByteOffset(parsedByteOffset)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting byteoffset")
+ }
+ bitOffset := uint8(0)
+ if match[BIT_OFFSET] != "" {
+ parsedBitOffset, err := strconv.ParseUint(match[BIT_OFFSET], 10, 8)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting bitoffset")
+ }
+ bitOffset = uint8(parsedBitOffset)
+ }
+ numElements := uint16(1)
+ if match[NUM_ELEMENTS] != "" {
+ parsedNumElements, err := strconv.ParseUint(match[NUM_ELEMENTS], 10, 16)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting numelements")
+ }
+ numElements = uint16(parsedNumElements)
+ }
+
+ if (transferSizeCode != 0) && (dataType.ShortName() != transferSizeCode) {
+ return nil, errors.Errorf("Transfer size code '%d' doesn't match specified data type '%s'", transferSizeCode, dataType)
+ }
+
+ // Var-length strings behave like fixed-length strings with the maximum length of 254.
+ return NewStringTag(memoryArea, blockNumber, byteOffset, bitOffset, numElements, 254, dataType), nil
+ } else if match := utils.GetSubgroupMatches(m.dataBlockStringVarLengthShortPattern, tagAddress); match != nil {
+ dataType, ok := readWriteModel.TransportSizeByName(match[DATA_TYPE])
+ if !ok {
+ return nil, errors.Errorf("Unknown type %s", match[DATA_TYPE])
+ }
+ memoryArea := readWriteModel.MemoryArea_DATA_BLOCKS
+ parsedBlockNumber, err := strconv.ParseUint(match[BLOCK_NUMBER], 10, 16)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting blocknumber")
+ }
+ blockNumber, err := checkDatablockNumber(parsedBlockNumber)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error checking blocknumber")
+ }
+ parsedByteOffset, err := strconv.ParseUint(match[BYTE_OFFSET], 10, 64)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting byteoffset")
+ }
+ byteOffset, err := checkByteOffset(parsedByteOffset)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting byteoffset")
+ }
+ bitOffset := uint8(0)
+ numElements := uint16(1)
+ if match[NUM_ELEMENTS] != "" {
+ parsedNumElements, err := strconv.ParseUint(match[NUM_ELEMENTS], 10, 16)
+ if err != nil {
+ return nil, errors.Wrap(err, "Error converting numelements")
+ }
+ numElements = uint16(parsedNumElements)
+ }
+
+ // Var-length strings behave like fixed-length strings with the maximum length of 254.
+ return NewStringTag(memoryArea, blockNumber, byteOffset, bitOffset, numElements, 254, dataType), nil
} else if match := utils.GetSubgroupMatches(m.dataBlockAddressPattern, tagAddress); match != nil {
dataType, ok := readWriteModel.TransportSizeByName(match[DATA_TYPE])
if !ok {
@@ -347,9 +435,10 @@ func checkDatablockNumber(blockNumber uint64) (uint16, error) {
}
func checkByteOffset(byteOffset uint64) (uint16, error) {
- //ToDo check the value or add reference
- if byteOffset > 2097151 || byteOffset < 0 {
- return 0, errors.New("ByteOffset must be smaller than 2097151 and positive.")
+ // The generated S7AddressAny model limits the byte address to 16 bits, so anything
+ // larger would silently truncate on the wire.
+ if byteOffset > 0xFFFF {
+ return 0, errors.New("ByteOffset must fit into 16 bits (0..65535).")
}
return uint16(byteOffset), nil
}
diff --git a/plc4go/internal/s7/TagHandler_test.go b/plc4go/internal/s7/TagHandler_test.go
new file mode 100644
index 0000000000..3cfc2d7385
--- /dev/null
+++ b/plc4go/internal/s7/TagHandler_test.go
@@ -0,0 +1,101 @@
+/*
+ * 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 s7
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+)
+
+func TestTagHandlerParseTag(t *testing.T) {
+ handler := NewTagHandler()
+
+ t.Run("S5TIME address", func(t *testing.T) {
+ tag, err := handler.ParseTag("%M0:S5TIME")
+ require.NoError(t, err)
+ s7Tag := tag.(PlcTag)
+ assert.Equal(t, readWriteModel.TransportSize_S5TIME, s7Tag.GetDataType())
+ assert.Equal(t, readWriteModel.MemoryArea_FLAGS_MARKERS, s7Tag.GetMemoryArea())
+ })
+ t.Run("S5TIME in data block", func(t *testing.T) {
+ tag, err := handler.ParseTag("%DB10:4:S5TIME")
+ require.NoError(t, err)
+ s7Tag := tag.(PlcTag)
+ assert.Equal(t, readWriteModel.TransportSize_S5TIME, s7Tag.GetDataType())
+ assert.Equal(t, uint16(10), s7Tag.GetBlockNumber())
+ })
+ t.Run("fixed length string", func(t *testing.T) {
+ tag, err := handler.ParseTag("%DB69:68:STRING(20)")
+ require.NoError(t, err)
+ stringTag := tag.(PlcStringTag)
+ assert.Equal(t, uint16(20), stringTag.stringLength)
+ })
+ t.Run("var length string short form", func(t *testing.T) {
+ tag, err := handler.ParseTag("%DB69:68:STRING")
+ require.NoError(t, err)
+ stringTag := tag.(PlcStringTag)
+ assert.Equal(t, uint16(254), stringTag.stringLength)
+ assert.Equal(t, readWriteModel.TransportSize_STRING, stringTag.GetDataType())
+ })
+ t.Run("var length string long form with array", func(t *testing.T) {
+ tag, err := handler.ParseTag("%DB69.DBX68:WSTRING[3]")
+ require.NoError(t, err)
+ stringTag := tag.(PlcStringTag)
+ assert.Equal(t, uint16(254), stringTag.stringLength)
+ assert.Equal(t, uint16(3), stringTag.GetNumElements())
+ assert.Equal(t, readWriteModel.TransportSize_WSTRING, stringTag.GetDataType())
+ })
+ t.Run("alarm subscription tag", func(t *testing.T) {
+ tag, err := handler.ParseTag("ALM")
+ require.NoError(t, err)
+ alarmTag := tag.(*AlarmTag)
+ assert.Equal(t, AlarmTagPush, alarmTag.GetKind())
+ })
+ t.Run("alarm query tags", func(t *testing.T) {
+ tag, err := handler.ParseTag("QUERY:ALARM_S")
+ require.NoError(t, err)
+ alarmTag := tag.(*AlarmTag)
+ assert.Equal(t, AlarmTagQuery, alarmTag.GetKind())
+ assert.Equal(t, readWriteModel.QueryType_ALARM_S, alarmTag.GetQueryType())
+
+ tag, err = handler.ParseTag("query:alarm_8")
+ require.NoError(t, err)
+ alarmTag = tag.(*AlarmTag)
+ assert.Equal(t, readWriteModel.QueryType_ALARM_8, alarmTag.GetQueryType())
+ })
+ t.Run("existing forms still parse", func(t *testing.T) {
+ for _, address := range []string{"%Q0.0:BOOL", "%M100:INT[10]", "%DB1.DBX0.0:BOOL", "%DB1:0.0:BOOL", "%I0:BYTE"} {
+ _, err := handler.ParseTag(address)
+ assert.NoError(t, err, address)
+ }
+ })
+ t.Run("oversized byte offset is rejected", func(t *testing.T) {
+ _, err := handler.ParseTag("%DB1:70000:INT")
+ assert.Error(t, err)
+ })
+ t.Run("bogus string type suffix is rejected", func(t *testing.T) {
+ _, err := handler.ParseTag("%DB1:0:STRINGX")
+ assert.Error(t, err)
+ })
+}
(plc4x) 08/09: feat(plc4go/s7): real round-trip ping
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 203cfdd5ad5048b81b13f1fe125a5021e75c1121
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 12:17:25 2026 +0200
feat(plc4go/s7): real round-trip ping
The default ping only checked connection flags and couldn't detect a
half-open TCP connection or an unresponsive CPU. Port the plc4j approach:
a one-byte ReadVar of %M0 where any S7 answer counts as alive.
---
plc4go/internal/s7/Connection.go | 60 ++++++++++++++++++++++++++++++++++++++++
1 file changed, 60 insertions(+)
diff --git a/plc4go/internal/s7/Connection.go b/plc4go/internal/s7/Connection.go
index 7f3589bb5a..9e0d293623 100644
--- a/plc4go/internal/s7/Connection.go
+++ b/plc4go/internal/s7/Connection.go
@@ -415,6 +415,66 @@ func (c *Connection) GetMetadata() apiModel.PlcConnectionMetadata {
}
}
+// Ping does a real round trip: S7 has no dedicated ping, so a one-byte read of %M0 is
+// issued and any S7 answer (including a PLC error) counts as alive - only transport
+// failures and timeouts report an error.
+func (c *Connection) Ping(ctx context.Context) error {
+ if !c.IsConnected() {
+ return errors.New("not connected")
+ }
+ tpduId := c.tpduGenerator.getAndIncrement()
+ pingRequest := readWriteModel.NewTPKTPacket(readWriteModel.NewCOTPPacketData(
+ nil,
+ readWriteModel.NewS7MessageRequest(
+ tpduId,
+ readWriteModel.NewS7ParameterReadVarRequest([]readWriteModel.S7VarRequestParameterItem{
+ readWriteModel.NewS7VarRequestParameterItemAddress(
+ readWriteModel.NewS7AddressAny(readWriteModel.TransportSize_BYTE, 1, 0, readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0),
+ ),
+ }),
+ nil,
+ ),
+ true,
+ uint8(tpduId),
+ ))
+ successChan := make(chan struct{}, 1)
+ errChan := make(chan error, 1)
+ if err := c.messageCodec.SendRequest(ctx, "ping", pingRequest, func(message spi.Message) bool {
+ tpktPacket, ok := message.(readWriteModel.TPKTPacket)
+ if !ok {
+ return false
+ }
+ cotpPacketData, ok := tpktPacket.GetPayload().(readWriteModel.COTPPacketData)
+ if !ok {
+ return false
+ }
+ payload := cotpPacketData.GetPayload()
+ return payload != nil && payload.GetTpduReference() == tpduId
+ }, func(message spi.Message) error {
+ select {
+ case successChan <- struct{}{}:
+ default:
+ }
+ return nil
+ }, func(err error) error {
+ select {
+ case errChan <- err:
+ default:
+ }
+ return nil
+ }); err != nil {
+ return errors.Wrap(err, "error sending ping request")
+ }
+ select {
+ case <-successChan:
+ return nil
+ case err := <-errChan:
+ return errors.Wrap(err, "got error while waiting for ping response")
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+}
+
func (c *Connection) dispatchUnsolicitedUserData(channel <-chan readWriteModel.S7MessageUserData) {
for message := range channel {
c.subscriberMutex.Lock()
(plc4x) 06/09: feat(plc4go/s7): implement browse (static areas + DB block enumeration)
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 97253814c1489e342cd2edb7d1bc7d02b30f775a
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 12:05:57 2026 +0200
feat(plc4go/s7): implement browse (static areas + DB block enumeration)
Port of the plc4j browse behavior: every query returns the static memory
areas %M/%I/%Q plus one item per data block found via the 'list blocks of
type DB' block function, with a graceful fallback to the static areas
when the PLC doesn't honor block functions. Browse is gated on the
UserData capability detected at connect time; ParseQuery now returns a
named query instead of erroring out.
---
plc4go/internal/s7/Browser.go | 121 +++++++++++++++++++++++++++++
plc4go/internal/s7/Connection.go | 32 ++++++--
plc4go/internal/s7/Query.go | 34 ++++++++
plc4go/internal/s7/TagHandler.go | 3 +-
plc4go/internal/s7/UserDataService.go | 45 +++++++++++
plc4go/internal/s7/UserDataService_test.go | 35 +++++++++
6 files changed, 261 insertions(+), 9 deletions(-)
diff --git a/plc4go/internal/s7/Browser.go b/plc4go/internal/s7/Browser.go
new file mode 100644
index 0000000000..cd58572aea
--- /dev/null
+++ b/plc4go/internal/s7/Browser.go
@@ -0,0 +1,121 @@
+/*
+ * 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 s7
+
+import (
+ "context"
+ "fmt"
+
+ "github.com/rs/zerolog"
+
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi/default"
+ spiModel "github.com/apache/plc4x/plc4go/spi/model"
+ "github.com/apache/plc4x/plc4go/spi/options"
+)
+
+type Browser struct {
+ _default.DefaultBrowser
+
+ connection *Connection
+
+ log zerolog.Logger
+ _options []options.WithOption // Used to pass them downstream
+}
+
+func NewBrowser(connection *Connection, _options ...options.WithOption) *Browser {
+ customLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...)
+ browser := &Browser{
+ connection: connection,
+ log: customLogger,
+ _options: _options,
+ }
+ browser.DefaultBrowser = _default.NewDefaultBrowser(browser, _options...)
+ return browser
+}
+
+// BrowseQuery returns the static memory areas plus one item per data block reported by the
+// PLC's "list blocks of type DB" block function. Browse rides the UserData services, so
+// devices without them get UNSUPPORTED. A failing block-list request degrades to the static
+// areas only - some PLCs don't honor the block functions.
+func (m *Browser) BrowseQuery(ctx context.Context, interceptor func(result apiModel.PlcBrowseItem) bool, queryName string, query apiModel.PlcQuery) (apiModel.PlcResponseCode, []apiModel.PlcBrowseItem) {
+ if !m.connection.driverContext.UserDataServicesSupported {
+ return apiModel.PlcResponseCode_UNSUPPORTED, nil
+ }
+ subscribable := true
+ items := staticAreaBrowseItems(subscribable)
+
+ tpduId := m.connection.tpduGenerator.getAndIncrement()
+ response, err := m.connection.sendUserData(ctx, tpduId, buildListBlocksOfTypeRequest(tpduId, blockTypeDataBlock), "browse_list_blocks")
+ if err != nil {
+ m.log.Debug().Err(err).Msg("Block-list browse failed; returning static areas only")
+ } else if blockNumbers, err := parseListBlocksOfTypeResponse(response); err != nil {
+ m.log.Debug().Err(err).Msg("Block-list browse rejected; returning static areas only")
+ } else {
+ m.log.Debug().Int("count", len(blockNumbers)).Msg("Block-list browse found data blocks")
+ for _, blockNumber := range blockNumbers {
+ items = append(items, dataBlockBrowseItem(blockNumber, subscribable))
+ }
+ }
+
+ results := make([]apiModel.PlcBrowseItem, 0, len(items))
+ for _, item := range items {
+ if interceptor == nil || interceptor(item) {
+ results = append(results, item)
+ }
+ }
+ return apiModel.PlcResponseCode_OK, results
+}
+
+func staticAreaBrowseItems(subscribable bool) []apiModel.PlcBrowseItem {
+ return []apiModel.PlcBrowseItem{
+ staticAreaBrowseItem("%M", readWriteModel.MemoryArea_FLAGS_MARKERS, subscribable),
+ staticAreaBrowseItem("%I", readWriteModel.MemoryArea_INPUTS, subscribable),
+ staticAreaBrowseItem("%Q", readWriteModel.MemoryArea_OUTPUTS, subscribable),
+ }
+}
+
+func staticAreaBrowseItem(name string, area readWriteModel.MemoryArea, subscribable bool) apiModel.PlcBrowseItem {
+ // Placeholder tag for the area root, the caller composes the real offset / data type.
+ return spiModel.NewDefaultPlcBrowseItem(
+ NewTag(area, 0, 0, 0, 1, readWriteModel.TransportSize_BYTE),
+ name,
+ "",
+ true,
+ true,
+ subscribable,
+ nil,
+ nil,
+ )
+}
+
+func dataBlockBrowseItem(blockNumber uint16, subscribable bool) apiModel.PlcBrowseItem {
+ return spiModel.NewDefaultPlcBrowseItem(
+ NewTag(readWriteModel.MemoryArea_DATA_BLOCKS, blockNumber, 0, 0, 1, readWriteModel.TransportSize_BYTE),
+ fmt.Sprintf("DB%d", blockNumber),
+ "",
+ true,
+ true,
+ subscribable,
+ nil,
+ nil,
+ )
+}
diff --git a/plc4go/internal/s7/Connection.go b/plc4go/internal/s7/Connection.go
index b63c2c0d26..97673f11c0 100644
--- a/plc4go/internal/s7/Connection.go
+++ b/plc4go/internal/s7/Connection.go
@@ -299,9 +299,19 @@ func (c *Connection) identifyRemote(ctx context.Context) {
func (c *Connection) trySzlProbe(ctx context.Context, szlId readWriteModel.SzlId, szlIndex uint16) (string, readWriteModel.ControllerType, error) {
tpduId := c.tpduGenerator.getAndIncrement()
+ message, err := c.sendUserData(ctx, tpduId, buildSzlRequest(tpduId, szlId, szlIndex), "setup_connection_identify_remote_message")
+ if err != nil {
+ return "", 0, err
+ }
+ return parseSzlProbeResponse(message)
+}
+
+// sendUserData sends a single S7 UserData request and waits (blocking) for the response
+// with the matching tpdu reference.
+func (c *Connection) sendUserData(ctx context.Context, tpduId uint16, request readWriteModel.TPKTPacket, interactionInfo string) (readWriteModel.S7Message, error) {
resultChan := make(chan readWriteModel.S7Message, 1)
errChan := make(chan error, 1)
- if err := c.messageCodec.SendRequest(ctx, "setup_connection_identify_remote_message", buildSzlRequest(tpduId, szlId, szlIndex), func(message spi.Message) bool {
+ if err := c.messageCodec.SendRequest(ctx, interactionInfo, request, func(message spi.Message) bool {
tpktPacket, ok := message.(readWriteModel.TPKTPacket)
if !ok {
return false
@@ -324,15 +334,15 @@ func (c *Connection) trySzlProbe(ctx context.Context, szlId readWriteModel.SzlId
errChan <- err
return nil
}); err != nil {
- return "", 0, errors.Wrap(err, "error sending SZL identification request")
+ return nil, errors.Wrapf(err, "error sending %s request", interactionInfo)
}
select {
case message := <-resultChan:
- return parseSzlProbeResponse(message)
+ return message, nil
case err := <-errChan:
- return "", 0, err
+ return nil, err
case <-ctx.Done():
- return "", 0, ctx.Err()
+ return nil, ctx.Err()
}
}
@@ -384,11 +394,19 @@ func (c *Connection) GetMetadata() apiModel.PlcConnectionMetadata {
"max-amq-caller": strconv.Itoa(int(c.driverContext.MaxAmqCaller)),
"max-amq-callee": strconv.Itoa(int(c.driverContext.MaxAmqCallee)),
},
- ProvidesReading: true,
- ProvidesWriting: true,
+ ProvidesReading: true,
+ ProvidesWriting: true,
+ ProvidesBrowsing: c.driverContext.UserDataServicesSupported,
}
}
+func (c *Connection) BrowseRequestBuilder() apiModel.PlcBrowseRequestBuilder {
+ return spiModel.NewDefaultPlcBrowseRequestBuilder(
+ c.GetPlcTagHandler(),
+ NewBrowser(c, append(c._options, options.WithCustomLogger(c.log))...),
+ )
+}
+
func (c *Connection) ReadRequestBuilder() apiModel.PlcReadRequestBuilder {
return spiModel.NewDefaultPlcReadRequestBuilder(
c.GetPlcTagHandler(),
diff --git a/plc4go/internal/s7/Query.go b/plc4go/internal/s7/Query.go
new file mode 100644
index 0000000000..d640172f06
--- /dev/null
+++ b/plc4go/internal/s7/Query.go
@@ -0,0 +1,34 @@
+/*
+ * 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 s7
+
+// s7Query names a browse result bucket. Like the Java driver, S7 has no query language -
+// every query returns the full set of browsable items (static areas + data blocks).
+type s7Query struct {
+ queryString string
+}
+
+func (m s7Query) GetQueryString() string {
+ return m.queryString
+}
+
+func (m s7Query) String() string {
+ return m.queryString
+}
diff --git a/plc4go/internal/s7/TagHandler.go b/plc4go/internal/s7/TagHandler.go
index 29b09a1215..e2fca04d5b 100644
--- a/plc4go/internal/s7/TagHandler.go
+++ b/plc4go/internal/s7/TagHandler.go
@@ -22,7 +22,6 @@ package s7
import (
"context"
"encoding/hex"
- "fmt"
"regexp"
"strconv"
"strings"
@@ -423,7 +422,7 @@ func (m TagHandler) ParseTag(tagAddress string) (apiModel.PlcTag, error) {
}
func (m TagHandler) ParseQuery(query string) (apiModel.PlcQuery, error) {
- return nil, fmt.Errorf("queries not supported")
+ return s7Query{queryString: query}, nil
}
func checkDatablockNumber(blockNumber uint64) (uint16, error) {
diff --git a/plc4go/internal/s7/UserDataService.go b/plc4go/internal/s7/UserDataService.go
index 9045de5568..319424260e 100644
--- a/plc4go/internal/s7/UserDataService.go
+++ b/plc4go/internal/s7/UserDataService.go
@@ -78,6 +78,51 @@ func buildSzlRequest(tpduId uint16, szlId readWriteModel.SzlId, szlIndex uint16)
)
}
+// blockTypeDataBlock is the "list blocks of type" wire code for DBs: the two ASCII chars "0A".
+const blockTypeDataBlock = uint16(0x3041)
+
+func buildListBlocksOfTypeRequest(tpduId uint16, blockType uint16) readWriteModel.TPKTPacket {
+ return newUserDataMessage(
+ tpduId,
+ newCpuFunctionsRequestParameter(0x03, 0x02), // block functions / list blocks of type
+ readWriteModel.NewS7PayloadUserDataItemCpuFunctionListBlocksOfTypeRequest(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_OCTET_STRING,
+ 2, // dataLength: the packed block-type code
+ blockType,
+ ),
+ )
+}
+
+// parseListBlocksOfTypeResponse extracts the block numbers from a list-blocks-of-type
+// response. Each entry is 4 bytes: blockNumber (uint16 BE), flags, language.
+func parseListBlocksOfTypeResponse(message readWriteModel.S7Message) ([]uint16, error) {
+ messageUserData, ok := message.(readWriteModel.S7MessageUserData)
+ if !ok {
+ return nil, errors.Errorf("expected S7MessageUserData, got %T", message)
+ }
+ if err := checkUserDataErrorCode(messageUserData); err != nil {
+ return nil, err
+ }
+ payloadUserData, ok := messageUserData.GetPayload().(readWriteModel.S7PayloadUserData)
+ if !ok {
+ return nil, errors.Errorf("expected S7PayloadUserData, got %T", messageUserData.GetPayload())
+ }
+ for _, item := range payloadUserData.GetItems() {
+ response, ok := item.(readWriteModel.S7PayloadUserDataItemCpuFunctionListBlocksOfTypeResponse)
+ if !ok {
+ continue
+ }
+ data := response.GetItems()
+ blockNumbers := make([]uint16, 0, len(data)/4)
+ for i := 0; i+4 <= len(data); i += 4 {
+ blockNumbers = append(blockNumbers, uint16(data[i])<<8|uint16(data[i+1]))
+ }
+ return blockNumbers, nil
+ }
+ return nil, errors.New("no list-blocks-of-type response item in payload")
+}
+
// checkUserDataErrorCode returns an error if any CPUFunctions parameter item reports a non-zero error code.
func checkUserDataErrorCode(message readWriteModel.S7MessageUserData) error {
parameterUserData, ok := message.GetParameter().(readWriteModel.S7ParameterUserData)
diff --git a/plc4go/internal/s7/UserDataService_test.go b/plc4go/internal/s7/UserDataService_test.go
index a14140602f..cc2f32cd68 100644
--- a/plc4go/internal/s7/UserDataService_test.go
+++ b/plc4go/internal/s7/UserDataService_test.go
@@ -113,6 +113,41 @@ func TestSupportsUserDataServices(t *testing.T) {
assert.False(t, supportsUserDataServices(readWriteModel.ControllerType_ANY))
}
+func TestParseListBlocksOfTypeResponse(t *testing.T) {
+ makeResponse := func(items []byte, errorCode *uint16) readWriteModel.S7Message {
+ return readWriteModel.NewS7MessageUserData(
+ 12,
+ readWriteModel.NewS7ParameterUserData([]readWriteModel.S7ParameterUserDataItem{
+ readWriteModel.NewS7ParameterUserDataItemCPUFunctions(
+ 0x12, 0x08, 0x03, 0x02, 0x00,
+ ptr(uint8(0)), ptr(uint8(1)), errorCode,
+ ),
+ }),
+ readWriteModel.NewS7PayloadUserData([]readWriteModel.S7PayloadUserDataItem{
+ readWriteModel.NewS7PayloadUserDataItemCpuFunctionListBlocksOfTypeResponse(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_OCTET_STRING,
+ uint16(len(items)),
+ items,
+ ),
+ }),
+ )
+ }
+
+ t.Run("two data blocks", func(t *testing.T) {
+ blockNumbers, err := parseListBlocksOfTypeResponse(makeResponse([]byte{
+ 0x00, 0x01, 0x22, 0x01, // DB1
+ 0x00, 0x45, 0x22, 0x01, // DB69
+ }, ptr(uint16(0))))
+ require.NoError(t, err)
+ assert.Equal(t, []uint16{1, 69}, blockNumbers)
+ })
+ t.Run("rejected request", func(t *testing.T) {
+ _, err := parseListBlocksOfTypeResponse(makeResponse([]byte{}, ptr(uint16(0xD401))))
+ assert.Error(t, err)
+ })
+}
+
func ptr[T any](value T) *T {
return &value
}
(plc4x) 09/09: test(plc4go/s7): re-enable the driver testsuite cases skipped since the s7h drift
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 5c316cce12a042fa9c9924fe8667896a6f9df6de
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 12:30:34 2026 +0200
test(plc4go/s7): re-enable the driver testsuite cases skipped since the s7h drift
Two fixes were needed to get them green:
- the S7 MessageCodec never called FillBuffer, so transports that only
surface queued data through fills (like the test transport) starved
the receive worker forever - the setup handshake stalled after the
COTP connection response. Fill up to the 4-byte TPKT header first,
like the modbus codec does.
- the testsuite XML still described the pre-s7h SZL identification
exchange (structured SzlDataTreeItem response, fixed tpdu ids). It now
matches the ported driver: COMPONENT_IDENTIFICATION probe with
generator-assigned tpdu ids and a raw-bytes SZL response carrying the
MLFB.
The XML is updated in its canonical location (protocols/s7) and the
build-copied plc4go/assets duplicate is kept in sync.
---
.../testing/protocols/s7/DriverTestsuite.xml | 71 ++++++----------------
plc4go/internal/s7/MessageCodec.go | 22 ++++++-
plc4go/tests/drivers/tests/s7_driver_test.go | 5 --
.../resources/protocols/s7/DriverTestsuite.xml | 71 ++++++----------------
4 files changed, 55 insertions(+), 114 deletions(-)
diff --git a/plc4go/assets/testing/protocols/s7/DriverTestsuite.xml b/plc4go/assets/testing/protocols/s7/DriverTestsuite.xml
index 5e279c9d24..1984513b0b 100644
--- a/plc4go/assets/testing/protocols/s7/DriverTestsuite.xml
+++ b/plc4go/assets/testing/protocols/s7/DriverTestsuite.xml
@@ -221,7 +221,7 @@
<tpduCode dataType="uint" bitLength="8">240</tpduCode>
<COTPPacketData>
<eot dataType="bit" bitLength="1">true</eot>
- <tpduRef dataType="uint" bitLength="7">2</tpduRef>
+ <tpduRef dataType="uint" bitLength="7">10</tpduRef>
</COTPPacketData>
<parameters isList="true">
</parameters>
@@ -230,7 +230,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">7</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">1</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">10</tpduReference>
<parameterLength dataType="uint" bitLength="16">8</parameterLength>
<payloadLength dataType="uint" bitLength="16">8</payloadLength>
<S7MessageUserData>
@@ -280,12 +280,12 @@
<sublistExtract dataType="uint" bitLength="4">0</sublistExtract>
<sublistList>
<SzlSublist dataType="uint" bitLength="8"
- stringRepresentation="MODULE_IDENTIFICATION">17
+ stringRepresentation="COMPONENT_IDENTIFICATION">28
</SzlSublist>
</sublistList>
</SzlId>
</szlId>
- <szlIndex dataType="uint" bitLength="16">0</szlIndex>
+ <szlIndex dataType="uint" bitLength="16">1</szlIndex>
</S7PayloadUserDataItemCpuFunctionReadSzlRequest>
</S7PayloadUserDataItem>
</items>
@@ -303,7 +303,7 @@
<TPKTPacket>
<protocolId dataType="uint" bitLength="8">3</protocolId>
<reserved dataType="uint" bitLength="8">0</reserved>
- <len dataType="uint" bitLength="16">125</len>
+ <len dataType="uint" bitLength="16">69</len>
<payload>
<COTPPacket>
<headerLength dataType="uint" bitLength="8">2</headerLength>
@@ -319,9 +319,9 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">7</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">1</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">10</tpduReference>
<parameterLength dataType="uint" bitLength="16">12</parameterLength>
- <payloadLength dataType="uint" bitLength="16">96</payloadLength>
+ <payloadLength dataType="uint" bitLength="16">40</payloadLength>
<S7MessageUserData>
</S7MessageUserData>
<parameter>
@@ -361,48 +361,11 @@
<DataTransportSize dataType="uint" bitLength="8" stringRepresentation="OCTET_STRING">9
</DataTransportSize>
</transportSize>
- <dataLength dataType="uint" bitLength="16">92</dataLength>
+ <dataLength dataType="uint" bitLength="16">36</dataLength>
<S7PayloadUserDataItemCpuFunctionReadSzlResponse>
- <szlId>
- <SzlId>
- <typeClass>
- <SzlModuleTypeClass dataType="uint" bitLength="4" stringRepresentation="CPU">0
- </SzlModuleTypeClass>
- </typeClass>
- <sublistExtract dataType="uint" bitLength="4">0</sublistExtract>
- <sublistList>
- <SzlSublist dataType="uint" bitLength="8"
- stringRepresentation="MODULE_IDENTIFICATION">17
- </SzlSublist>
- </sublistList>
- </SzlId>
- </szlId>
- <szlIndex dataType="uint" bitLength="16">0</szlIndex>
- <szlItemLength dataType="uint" bitLength="16">28</szlItemLength>
- <szlItemCount dataType="uint" bitLength="16">3</szlItemCount>
- <items isList="true">
- <SzlDataTreeItem>
- <itemIndex dataType="uint" bitLength="16">1</itemIndex>
- <mlfb dataType="byte" bitLength="160">0x36455337203231322d31424433302d3058423020</mlfb>
- <moduleTypeId dataType="uint" bitLength="16">8224</moduleTypeId>
- <ausbg dataType="uint" bitLength="16">1</ausbg>
- <ausbe dataType="uint" bitLength="16">8224</ausbe>
- </SzlDataTreeItem>
- <SzlDataTreeItem>
- <itemIndex dataType="uint" bitLength="16">6</itemIndex>
- <mlfb dataType="byte" bitLength="160">0x36455337203231322d31424433302d3058423020</mlfb>
- <moduleTypeId dataType="uint" bitLength="16">8224</moduleTypeId>
- <ausbg dataType="uint" bitLength="16">1</ausbg>
- <ausbe dataType="uint" bitLength="16">8224</ausbe>
- </SzlDataTreeItem>
- <SzlDataTreeItem>
- <itemIndex dataType="uint" bitLength="16">7</itemIndex>
- <mlfb dataType="byte" bitLength="160">0x36455337203231322d31424433302d3058423020</mlfb>
- <moduleTypeId dataType="uint" bitLength="16">8224</moduleTypeId>
- <ausbg dataType="uint" bitLength="16">22018</ausbg>
- <ausbe dataType="uint" bitLength="16">2</ausbe>
- </SzlDataTreeItem>
- </items>
+ <!-- SZL header (id/index/itemLength/itemCount) + one 28-byte item
+ carrying the MLFB "6ES7 212-1BD30-0XB0 " -->
+ <items dataType="byte" bitLength="288">0x001c0001001c0001000136455337203231322d31424433302d3058423020202000012020</items>
</S7PayloadUserDataItemCpuFunctionReadSzlResponse>
</S7PayloadUserDataItem>
</items>
@@ -441,7 +404,7 @@
<tpduCode dataType="uint" bitLength="8">240</tpduCode>
<COTPPacketData>
<eot dataType="bit" bitLength="1">true</eot>
- <tpduRef dataType="uint" bitLength="7">10</tpduRef>
+ <tpduRef dataType="uint" bitLength="7">11</tpduRef>
</COTPPacketData>
<parameters isList="true">
</parameters>
@@ -450,7 +413,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">1</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">10</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">11</tpduReference>
<parameterLength dataType="uint" bitLength="16">14</parameterLength>
<payloadLength dataType="uint" bitLength="16">0</payloadLength>
<S7MessageRequest>
@@ -517,7 +480,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">3</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">10</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">11</tpduReference>
<parameterLength dataType="uint" bitLength="16">2</parameterLength>
<payloadLength dataType="uint" bitLength="16">6</payloadLength>
<S7MessageResponseData>
@@ -619,7 +582,7 @@
<tpduCode dataType="uint" bitLength="8">240</tpduCode>
<COTPPacketData>
<eot dataType="bit" bitLength="1">true</eot>
- <tpduRef dataType="uint" bitLength="7">10</tpduRef>
+ <tpduRef dataType="uint" bitLength="7">11</tpduRef>
</COTPPacketData>
<parameters isList="true">
</parameters>
@@ -628,7 +591,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">1</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">10</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">11</tpduReference>
<parameterLength dataType="uint" bitLength="16">14</parameterLength>
<payloadLength dataType="uint" bitLength="16">0</payloadLength>
<S7MessageRequest>
@@ -695,7 +658,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">2</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">10</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">11</tpduReference>
<parameterLength dataType="uint" bitLength="16">0</parameterLength>
<payloadLength dataType="uint" bitLength="16">0</payloadLength>
<S7MessageResponse>
diff --git a/plc4go/internal/s7/MessageCodec.go b/plc4go/internal/s7/MessageCodec.go
index 121d295aaa..afbc69d02f 100644
--- a/plc4go/internal/s7/MessageCodec.go
+++ b/plc4go/internal/s7/MessageCodec.go
@@ -128,7 +128,27 @@ func (m *MessageCodec) Send(ctx context.Context, interactionInfo string, message
}
func (m *MessageCodec) Receive(ctx context.Context) (spi.Message, error) {
- // We need at least 6 bytes in order to know how big the packet is in total
+ transportInstance := m.GetTransportInstance()
+ if !transportInstance.IsConnected() {
+ return nil, errors.New("Transport instance not connected")
+ }
+ // Pull data from the transport until at least the 4-byte TPKT header is buffered.
+ // Some transports (e.g. the test transport) only surface queued data through fills,
+ // so checking the buffer without filling first would starve the receive worker.
+ if err := transportInstance.FillBuffer(ctx, func(pos uint, currentByte byte, reader transports.ExtendedReader) bool {
+ numBytesAvailable, err := transportInstance.GetNumBytesAvailableInBuffer()
+ if err != nil {
+ return false
+ }
+ return numBytesAvailable < 4
+ }); err != nil {
+ if transportError, ok := transports.AsTransportError(err); ok && transportError.Kind() == transports.TransportErrorFatal {
+ return nil, err
+ }
+ // Fall through on non-fatal errors, we might have enough data buffered already.
+ m.log.Trace().Err(err).Msg("Error filling buffer, continuing with what's available")
+ }
+ // We need at least 4 bytes in order to know how big the packet is in total
if num, err := m.GetTransportInstance().GetNumBytesAvailableInBuffer(); (err == nil) && (num >= 4) {
m.log.Debug().Uint32("num", num).Msg("we got %d readable bytes")
data, err := m.GetTransportInstance().PeekReadableBytes(ctx, 4)
diff --git a/plc4go/tests/drivers/tests/s7_driver_test.go b/plc4go/tests/drivers/tests/s7_driver_test.go
index 31d88fc3ed..5428202c10 100644
--- a/plc4go/tests/drivers/tests/s7_driver_test.go
+++ b/plc4go/tests/drivers/tests/s7_driver_test.go
@@ -41,11 +41,6 @@ func TestS7Driver(t *testing.T) {
s7IO.S7XmlParserHelper{},
append(optionsForTesting,
testutils.WithRootTypeParser(parser),
- testutils.WithSkippedTestCases(
- // TODO: ignored due to carcia changes
- "Single element read request",
- "Single element read request with disabled PUT/GET",
- ),
)...,
)
}
diff --git a/protocols/s7/src/test/resources/protocols/s7/DriverTestsuite.xml b/protocols/s7/src/test/resources/protocols/s7/DriverTestsuite.xml
index 5e279c9d24..1984513b0b 100644
--- a/protocols/s7/src/test/resources/protocols/s7/DriverTestsuite.xml
+++ b/protocols/s7/src/test/resources/protocols/s7/DriverTestsuite.xml
@@ -221,7 +221,7 @@
<tpduCode dataType="uint" bitLength="8">240</tpduCode>
<COTPPacketData>
<eot dataType="bit" bitLength="1">true</eot>
- <tpduRef dataType="uint" bitLength="7">2</tpduRef>
+ <tpduRef dataType="uint" bitLength="7">10</tpduRef>
</COTPPacketData>
<parameters isList="true">
</parameters>
@@ -230,7 +230,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">7</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">1</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">10</tpduReference>
<parameterLength dataType="uint" bitLength="16">8</parameterLength>
<payloadLength dataType="uint" bitLength="16">8</payloadLength>
<S7MessageUserData>
@@ -280,12 +280,12 @@
<sublistExtract dataType="uint" bitLength="4">0</sublistExtract>
<sublistList>
<SzlSublist dataType="uint" bitLength="8"
- stringRepresentation="MODULE_IDENTIFICATION">17
+ stringRepresentation="COMPONENT_IDENTIFICATION">28
</SzlSublist>
</sublistList>
</SzlId>
</szlId>
- <szlIndex dataType="uint" bitLength="16">0</szlIndex>
+ <szlIndex dataType="uint" bitLength="16">1</szlIndex>
</S7PayloadUserDataItemCpuFunctionReadSzlRequest>
</S7PayloadUserDataItem>
</items>
@@ -303,7 +303,7 @@
<TPKTPacket>
<protocolId dataType="uint" bitLength="8">3</protocolId>
<reserved dataType="uint" bitLength="8">0</reserved>
- <len dataType="uint" bitLength="16">125</len>
+ <len dataType="uint" bitLength="16">69</len>
<payload>
<COTPPacket>
<headerLength dataType="uint" bitLength="8">2</headerLength>
@@ -319,9 +319,9 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">7</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">1</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">10</tpduReference>
<parameterLength dataType="uint" bitLength="16">12</parameterLength>
- <payloadLength dataType="uint" bitLength="16">96</payloadLength>
+ <payloadLength dataType="uint" bitLength="16">40</payloadLength>
<S7MessageUserData>
</S7MessageUserData>
<parameter>
@@ -361,48 +361,11 @@
<DataTransportSize dataType="uint" bitLength="8" stringRepresentation="OCTET_STRING">9
</DataTransportSize>
</transportSize>
- <dataLength dataType="uint" bitLength="16">92</dataLength>
+ <dataLength dataType="uint" bitLength="16">36</dataLength>
<S7PayloadUserDataItemCpuFunctionReadSzlResponse>
- <szlId>
- <SzlId>
- <typeClass>
- <SzlModuleTypeClass dataType="uint" bitLength="4" stringRepresentation="CPU">0
- </SzlModuleTypeClass>
- </typeClass>
- <sublistExtract dataType="uint" bitLength="4">0</sublistExtract>
- <sublistList>
- <SzlSublist dataType="uint" bitLength="8"
- stringRepresentation="MODULE_IDENTIFICATION">17
- </SzlSublist>
- </sublistList>
- </SzlId>
- </szlId>
- <szlIndex dataType="uint" bitLength="16">0</szlIndex>
- <szlItemLength dataType="uint" bitLength="16">28</szlItemLength>
- <szlItemCount dataType="uint" bitLength="16">3</szlItemCount>
- <items isList="true">
- <SzlDataTreeItem>
- <itemIndex dataType="uint" bitLength="16">1</itemIndex>
- <mlfb dataType="byte" bitLength="160">0x36455337203231322d31424433302d3058423020</mlfb>
- <moduleTypeId dataType="uint" bitLength="16">8224</moduleTypeId>
- <ausbg dataType="uint" bitLength="16">1</ausbg>
- <ausbe dataType="uint" bitLength="16">8224</ausbe>
- </SzlDataTreeItem>
- <SzlDataTreeItem>
- <itemIndex dataType="uint" bitLength="16">6</itemIndex>
- <mlfb dataType="byte" bitLength="160">0x36455337203231322d31424433302d3058423020</mlfb>
- <moduleTypeId dataType="uint" bitLength="16">8224</moduleTypeId>
- <ausbg dataType="uint" bitLength="16">1</ausbg>
- <ausbe dataType="uint" bitLength="16">8224</ausbe>
- </SzlDataTreeItem>
- <SzlDataTreeItem>
- <itemIndex dataType="uint" bitLength="16">7</itemIndex>
- <mlfb dataType="byte" bitLength="160">0x36455337203231322d31424433302d3058423020</mlfb>
- <moduleTypeId dataType="uint" bitLength="16">8224</moduleTypeId>
- <ausbg dataType="uint" bitLength="16">22018</ausbg>
- <ausbe dataType="uint" bitLength="16">2</ausbe>
- </SzlDataTreeItem>
- </items>
+ <!-- SZL header (id/index/itemLength/itemCount) + one 28-byte item
+ carrying the MLFB "6ES7 212-1BD30-0XB0 " -->
+ <items dataType="byte" bitLength="288">0x001c0001001c0001000136455337203231322d31424433302d3058423020202000012020</items>
</S7PayloadUserDataItemCpuFunctionReadSzlResponse>
</S7PayloadUserDataItem>
</items>
@@ -441,7 +404,7 @@
<tpduCode dataType="uint" bitLength="8">240</tpduCode>
<COTPPacketData>
<eot dataType="bit" bitLength="1">true</eot>
- <tpduRef dataType="uint" bitLength="7">10</tpduRef>
+ <tpduRef dataType="uint" bitLength="7">11</tpduRef>
</COTPPacketData>
<parameters isList="true">
</parameters>
@@ -450,7 +413,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">1</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">10</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">11</tpduReference>
<parameterLength dataType="uint" bitLength="16">14</parameterLength>
<payloadLength dataType="uint" bitLength="16">0</payloadLength>
<S7MessageRequest>
@@ -517,7 +480,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">3</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">10</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">11</tpduReference>
<parameterLength dataType="uint" bitLength="16">2</parameterLength>
<payloadLength dataType="uint" bitLength="16">6</payloadLength>
<S7MessageResponseData>
@@ -619,7 +582,7 @@
<tpduCode dataType="uint" bitLength="8">240</tpduCode>
<COTPPacketData>
<eot dataType="bit" bitLength="1">true</eot>
- <tpduRef dataType="uint" bitLength="7">10</tpduRef>
+ <tpduRef dataType="uint" bitLength="7">11</tpduRef>
</COTPPacketData>
<parameters isList="true">
</parameters>
@@ -628,7 +591,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">1</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">10</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">11</tpduReference>
<parameterLength dataType="uint" bitLength="16">14</parameterLength>
<payloadLength dataType="uint" bitLength="16">0</payloadLength>
<S7MessageRequest>
@@ -695,7 +658,7 @@
<protocolId dataType="uint" bitLength="8">50</protocolId>
<messageType dataType="uint" bitLength="8">2</messageType>
<reserved dataType="uint" bitLength="16">0</reserved>
- <tpduReference dataType="uint" bitLength="16">10</tpduReference>
+ <tpduReference dataType="uint" bitLength="16">11</tpduReference>
<parameterLength dataType="uint" bitLength="16">0</parameterLength>
<payloadLength dataType="uint" bitLength="16">0</payloadLength>
<S7MessageResponse>
(plc4x) 07/09: feat(plc4go/s7): implement alarm + cyclic subscriptions, alarm query and push dispatch
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 592473f77637cf748131bc81bec6e398cb139098
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 12:16:45 2026 +0200
feat(plc4go/s7): implement alarm + cyclic subscriptions, alarm query and push dispatch
Port of the plc4j subscription services:
- ALM alarm tags arm the ALARM_S/SQ (ALARM_8 on S7-400) push stream via
the MsgSubscription UserData service; indications are decoded into the
same PlcStruct shape as plc4j (plcTimestamp, eventId, state masks, ...)
- address tags subscribe to the cyclic services, grouped by their
TimeBase/factor cadence so equal intervals share one PLC job; pushed
values are decoded with the regular read decode path
- QUERY:ALARM_S / QUERY:ALARM_8 tags run a one-shot alarm query whose raw
result is delivered as an event when a consumer registers
- unsubscribing cancels cyclic jobs / the alarm stream once their last
handle is gone, and Close drains all PLC-side subscription state
Unsolicited UserData pushes are claimed by a custom message handler in
the codec (ahead of the request expectations) and dispatched by a
connection goroutine, so pushes don't idle in the receive worker.
The generic unsubscription request in spi/model asserted the concrete
*DefaultPlcSubscriptionHandle and panicked for driver handles embedding
it; it now uses an interface assertion that embedding satisfies and
errors instead of panicking on foreign handles.
---
plc4go/internal/s7/Connection.go | 63 ++-
plc4go/internal/s7/MessageCodec.go | 56 ++-
plc4go/internal/s7/Subscriber.go | 530 +++++++++++++++++++++
plc4go/internal/s7/Subscriber_test.go | 220 +++++++++
plc4go/internal/s7/SubscriptionEvent.go | 55 +++
plc4go/internal/s7/SubscriptionHandle.go | 64 +++
plc4go/internal/s7/UserDataService.go | 441 +++++++++++++++++
plc4go/spi/model/DefaultPlcSubscriptionHandle.go | 6 +
.../spi/model/DefaultPlcUnsubscriptionRequest.go | 10 +-
9 files changed, 1438 insertions(+), 7 deletions(-)
diff --git a/plc4go/internal/s7/Connection.go b/plc4go/internal/s7/Connection.go
index 97673f11c0..7f3589bb5a 100644
--- a/plc4go/internal/s7/Connection.go
+++ b/plc4go/internal/s7/Connection.go
@@ -24,6 +24,7 @@ import (
"fmt"
"strconv"
"sync"
+ "time"
"github.com/rs/zerolog"
@@ -66,6 +67,11 @@ type Connection struct {
driverContext DriverContext
tm transactions.RequestTransactionManager
+ // subscriber is created lazily and shared: the alarm subscription and cyclic jobs are
+ // PLC-side state bound to this connection.
+ subscriber *Subscriber
+ subscriberMutex sync.Mutex
+
connectionId string
tracer tracer.Tracer
@@ -129,6 +135,14 @@ func (c *Connection) Connect(ctx context.Context) error {
return errors.Wrap(err, "Error during message codec setup")
}
+ // Route unsolicited UserData pushes (cyclic data, alarm indications) to the subscriber.
+ // The channel is closed when the codec disconnects, which ends this goroutine.
+ if codec, ok := c.messageCodec.(*MessageCodec); ok {
+ c.wg.Go(func() {
+ c.dispatchUnsolicitedUserData(codec.GetUnsolicitedUserData())
+ })
+ }
+
// Only on active connections we do a connection
if c.driverContext.PassiveMode {
c.log.Info().Msg("S7 Driver running in PASSIVE mode.")
@@ -394,10 +408,53 @@ func (c *Connection) GetMetadata() apiModel.PlcConnectionMetadata {
"max-amq-caller": strconv.Itoa(int(c.driverContext.MaxAmqCaller)),
"max-amq-callee": strconv.Itoa(int(c.driverContext.MaxAmqCallee)),
},
- ProvidesReading: true,
- ProvidesWriting: true,
- ProvidesBrowsing: c.driverContext.UserDataServicesSupported,
+ ProvidesReading: true,
+ ProvidesWriting: true,
+ ProvidesSubscribing: c.driverContext.UserDataServicesSupported,
+ ProvidesBrowsing: c.driverContext.UserDataServicesSupported,
+ }
+}
+
+func (c *Connection) dispatchUnsolicitedUserData(channel <-chan readWriteModel.S7MessageUserData) {
+ for message := range channel {
+ c.subscriberMutex.Lock()
+ subscriber := c.subscriber
+ c.subscriberMutex.Unlock()
+ if subscriber == nil || !subscriber.handleUserDataPush(message) {
+ c.log.Debug().Stringer("message", message).Msg("Unsolicited user data message not handled by any subscription")
+ }
+ }
+}
+
+func (c *Connection) getSubscriber() *Subscriber {
+ c.subscriberMutex.Lock()
+ defer c.subscriberMutex.Unlock()
+ if c.subscriber == nil {
+ c.subscriber = NewSubscriber(c, append(c._options, options.WithCustomLogger(c.log))...)
+ }
+ return c.subscriber
+}
+
+func (c *Connection) SubscriptionRequestBuilder() apiModel.PlcSubscriptionRequestBuilder {
+ return spiModel.NewDefaultPlcSubscriptionRequestBuilder(
+ c.GetPlcTagHandler(), c.GetPlcValueHandler(), c.getSubscriber())
+}
+
+func (c *Connection) UnsubscriptionRequestBuilder() apiModel.PlcUnsubscriptionRequestBuilder {
+ return spiModel.NewDefaultPlcUnsubscriptionRequestBuilder()
+}
+
+func (c *Connection) Close() error {
+ // Cancel PLC-side subscription state before tearing the connection down (bounded).
+ c.subscriberMutex.Lock()
+ subscriber := c.subscriber
+ c.subscriberMutex.Unlock()
+ if subscriber != nil {
+ drainCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ subscriber.drain(drainCtx)
+ cancel()
}
+ return c.DefaultConnection.Close()
}
func (c *Connection) BrowseRequestBuilder() apiModel.PlcBrowseRequestBuilder {
diff --git a/plc4go/internal/s7/MessageCodec.go b/plc4go/internal/s7/MessageCodec.go
index e20b2d72c6..121d295aaa 100644
--- a/plc4go/internal/s7/MessageCodec.go
+++ b/plc4go/internal/s7/MessageCodec.go
@@ -21,6 +21,7 @@ package s7
import (
"context"
+ "sync"
"github.com/rs/zerolog"
@@ -36,6 +37,11 @@ import (
type MessageCodec struct {
_default.DefaultCodec
+ // unsolicitedUserData carries pushed UserData messages (cyclic data, alarm indications,
+ // mode transitions) that no request expectation claims. Closed on Disconnect.
+ unsolicitedUserData chan model.S7MessageUserData `ignore:"true"`
+ unsolicitedCloseOnce sync.Once `ignore:"true"`
+
passLogToModel bool
log zerolog.Logger
}
@@ -48,13 +54,57 @@ func NewMessageCodec(transportInstance transports.TransportInstance, _options ..
passLoggerToModel, _ := options.ExtractPassLoggerToModel(_options...)
extractCustomLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...)
codec := &MessageCodec{
- passLogToModel: passLoggerToModel,
- log: extractCustomLogger,
+ unsolicitedUserData: make(chan model.S7MessageUserData, 100),
+ passLogToModel: passLoggerToModel,
+ log: extractCustomLogger,
}
- codec.DefaultCodec = _default.NewDefaultCodec(codec, transportInstance, _options...)
+ codec.DefaultCodec = _default.NewDefaultCodec(codec, transportInstance,
+ append(_options, _default.WithCustomMessageHandler(extractUnsolicitedUserData(codec)))...)
return codec
}
+// extractUnsolicitedUserData claims pushed UserData messages (cpuFunctionType 0x00) before
+// the request expectations see them and forwards them to the subscription dispatch.
+func extractUnsolicitedUserData(codec *MessageCodec) _default.CustomMessageHandler {
+ return func(ctx context.Context, _ _default.DefaultCodecRequirements, message spi.Message) bool {
+ tpktPacket, ok := message.(model.TPKTPacket)
+ if !ok {
+ return false
+ }
+ cotpPacketData, ok := tpktPacket.GetPayload().(model.COTPPacketData)
+ if !ok {
+ return false
+ }
+ messageUserData, ok := cotpPacketData.GetPayload().(model.S7MessageUserData)
+ if !ok {
+ return false
+ }
+ _, functionType, _, ok := userDataPushKey(messageUserData)
+ if !ok || functionType != 0x00 {
+ return false
+ }
+ select {
+ case codec.unsolicitedUserData <- messageUserData:
+ default:
+ codec.log.Warn().Msg("Unsolicited user data message discarded, channel full")
+ }
+ return true
+ }
+}
+
+// GetUnsolicitedUserData exposes the push channel drained by the connection's dispatcher.
+func (m *MessageCodec) GetUnsolicitedUserData() <-chan model.S7MessageUserData {
+ return m.unsolicitedUserData
+}
+
+func (m *MessageCodec) Disconnect() error {
+ err := m.DefaultCodec.Disconnect()
+ m.unsolicitedCloseOnce.Do(func() {
+ close(m.unsolicitedUserData)
+ })
+ return err
+}
+
func (m *MessageCodec) GetCodec() spi.MessageCodec {
return m
}
diff --git a/plc4go/internal/s7/Subscriber.go b/plc4go/internal/s7/Subscriber.go
new file mode 100644
index 0000000000..c723778d50
--- /dev/null
+++ b/plc4go/internal/s7/Subscriber.go
@@ -0,0 +1,530 @@
+/*
+ * 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 s7
+
+import (
+ "context"
+ "runtime/debug"
+ "slices"
+ "sync"
+ "time"
+
+ "github.com/rs/zerolog"
+
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi/errors"
+ spiModel "github.com/apache/plc4x/plc4go/spi/model"
+ "github.com/apache/plc4x/plc4go/spi/options"
+ "github.com/apache/plc4x/plc4go/spi/utils"
+ spiValues "github.com/apache/plc4x/plc4go/spi/values"
+)
+
+// Subscriber implements the three S7 subscription flavours (matching plc4j):
+// - AlarmTag "ALM" + EVENT -> alarm indication push (UserData group 0x04)
+// - AlarmTag "QUERY:..." + EVENT -> one-shot alarm query, delivered on Register
+// - address tag + CYCLIC -> cyclic value push (UserData group 0x02)
+//
+// There is one Subscriber per connection: the alarm push subscription and the cyclic jobs
+// are PLC-side state bound to the TCP connection.
+type Subscriber struct {
+ connection *Connection
+
+ consumers map[*spiModel.DefaultPlcConsumerRegistration]apiModel.PlcSubscriptionEventConsumer
+ consumersMutex sync.RWMutex
+
+ alarmHandles []*SubscriptionHandle
+ alarmSubscriptionActive bool
+ alarmMutex sync.Mutex
+
+ cyclicHandlesByJob map[uint8][]*SubscriptionHandle
+ cyclicMutex sync.RWMutex
+
+ wg sync.WaitGroup // use to track spawned go routines
+
+ log zerolog.Logger
+ _options []options.WithOption // Used to pass them downstream
+}
+
+func NewSubscriber(connection *Connection, _options ...options.WithOption) *Subscriber {
+ customLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...)
+ return &Subscriber{
+ connection: connection,
+ consumers: map[*spiModel.DefaultPlcConsumerRegistration]apiModel.PlcSubscriptionEventConsumer{},
+ cyclicHandlesByJob: map[uint8][]*SubscriptionHandle{},
+ log: customLogger,
+ _options: _options,
+ }
+}
+
+func (s *Subscriber) Subscribe(ctx context.Context, subscriptionRequest apiModel.PlcSubscriptionRequest) <-chan apiModel.PlcSubscriptionRequestResult {
+ result := make(chan apiModel.PlcSubscriptionRequestResult, 1)
+ s.wg.Go(func() {
+ defer func() {
+ if err := recover(); err != nil {
+ utils.DeliverResult(s.log, result, spiModel.NewDefaultPlcSubscriptionRequestResult(subscriptionRequest, nil, errors.Errorf("panic-ed %v. Stack: %s", err, debug.Stack())))
+ }
+ }()
+ request := subscriptionRequest.(*spiModel.DefaultPlcSubscriptionRequest)
+ responseCodes := map[string]apiModel.PlcResponseCode{}
+ handles := map[string]apiModel.PlcSubscriptionHandle{}
+ deliver := func() {
+ utils.DeliverResult(s.log, result, spiModel.NewDefaultPlcSubscriptionRequestResult(
+ subscriptionRequest,
+ spiModel.NewDefaultPlcSubscriptionResponse(subscriptionRequest, responseCodes, handles, append(s._options, options.WithCustomLogger(s.log))...),
+ nil,
+ ))
+ }
+
+ if !s.connection.driverContext.UserDataServicesSupported {
+ for _, tagName := range request.GetTagNames() {
+ responseCodes[tagName] = apiModel.PlcResponseCode_UNSUPPORTED
+ }
+ deliver()
+ return
+ }
+
+ var alarmNames, queryNames, cyclicNames []string
+ alarmTags := map[string]*AlarmTag{}
+ queryTags := map[string]*AlarmTag{}
+ cyclicTags := map[string]PlcTag{}
+ for _, tagName := range request.GetTagNames() {
+ subscriptionType := request.GetType(tagName)
+ switch tag := s.resolveTag(request.GetTag(tagName)).(type) {
+ case *AlarmTag:
+ if subscriptionType != apiModel.SubscriptionEvent {
+ responseCodes[tagName] = apiModel.PlcResponseCode_INVALID_ADDRESS
+ continue
+ }
+ if tag.GetKind() == AlarmTagQuery {
+ queryNames = append(queryNames, tagName)
+ queryTags[tagName] = tag
+ } else {
+ alarmNames = append(alarmNames, tagName)
+ alarmTags[tagName] = tag
+ }
+ case PlcTag:
+ if subscriptionType != apiModel.SubscriptionCyclic {
+ responseCodes[tagName] = apiModel.PlcResponseCode_INVALID_ADDRESS
+ continue
+ }
+ cyclicNames = append(cyclicNames, tagName)
+ cyclicTags[tagName] = tag
+ default:
+ responseCodes[tagName] = apiModel.PlcResponseCode_INVALID_ADDRESS
+ }
+ }
+
+ s.subscribeAlarms(ctx, alarmNames, alarmTags, responseCodes, handles)
+ s.subscribeCyclic(ctx, request, cyclicNames, cyclicTags, responseCodes, handles)
+ s.runAlarmQueries(ctx, queryNames, queryTags, responseCodes, handles)
+ deliver()
+ })
+ return result
+}
+
+// resolveTag maps a subscription tag back to a driver tag. Tags added by address are already
+// driver types; tags wrapped in a DefaultPlcSubscriptionTag don't expose their inner tag, so
+// they are re-parsed from the address string.
+func (s *Subscriber) resolveTag(tag apiModel.PlcSubscriptionTag) apiModel.PlcTag {
+ switch typed := tag.(type) {
+ case *AlarmTag:
+ return typed
+ case PlcTag:
+ return typed
+ default:
+ parsed, err := NewTagHandler(s._options...).ParseTag(tag.GetAddressString())
+ if err != nil {
+ s.log.Debug().Err(err).Str("address", tag.GetAddressString()).Msg("Unable to re-parse subscription tag")
+ return nil
+ }
+ return parsed
+ }
+}
+
+func (s *Subscriber) subscribeAlarms(ctx context.Context, tagNames []string, tags map[string]*AlarmTag, responseCodes map[string]apiModel.PlcResponseCode, handles map[string]apiModel.PlcSubscriptionHandle) {
+ if len(tagNames) == 0 {
+ return
+ }
+ if err := s.ensureAlarmSubscriptionActive(ctx); err != nil {
+ s.log.Debug().Err(err).Msg("Alarm subscription failed")
+ for _, tagName := range tagNames {
+ responseCodes[tagName] = apiModel.PlcResponseCode_REMOTE_ERROR
+ }
+ return
+ }
+ s.alarmMutex.Lock()
+ defer s.alarmMutex.Unlock()
+ for _, tagName := range tagNames {
+ handle := NewSubscriptionHandle(s, tagName, tags[tagName], subscriptionKindAlarm)
+ s.alarmHandles = append(s.alarmHandles, handle)
+ handles[tagName] = handle
+ responseCodes[tagName] = apiModel.PlcResponseCode_OK
+ }
+}
+
+func (s *Subscriber) ensureAlarmSubscriptionActive(ctx context.Context) error {
+ s.alarmMutex.Lock()
+ alreadyActive := s.alarmSubscriptionActive
+ s.alarmMutex.Unlock()
+ if alreadyActive {
+ return nil
+ }
+ tpduId := s.connection.tpduGenerator.getAndIncrement()
+ alarmState := alarmStateForController(s.connection.driverContext.ControllerType, false)
+ response, err := s.connection.sendUserData(ctx, tpduId, buildMsgSubscriptionRequest(tpduId, alarmState), "alarm_subscribe")
+ if err != nil {
+ return errors.Wrap(err, "error sending alarm subscription request")
+ }
+ accepted, err := parseMsgSubscriptionResponse(response)
+ if err != nil {
+ return err
+ }
+ if !accepted {
+ return errors.New("alarm subscription rejected by PLC")
+ }
+ s.alarmMutex.Lock()
+ s.alarmSubscriptionActive = true
+ s.alarmMutex.Unlock()
+ return nil
+}
+
+func (s *Subscriber) subscribeCyclic(ctx context.Context, request *spiModel.DefaultPlcSubscriptionRequest, tagNames []string, tags map[string]PlcTag, responseCodes map[string]apiModel.PlcResponseCode, handles map[string]apiModel.PlcSubscriptionHandle) {
+ if len(tagNames) == 0 {
+ return
+ }
+ // Group tags by their effective cadence so callers requesting the same interval ride
+ // one PLC job instead of one per tag. Iterate names in request order for stable indices.
+ groups := map[cyclicInterval][]string{}
+ var groupOrder []cyclicInterval
+ for _, tagName := range tagNames {
+ interval := pickCyclicInterval(request.GetInterval(tagName))
+ if _, ok := groups[interval]; !ok {
+ groupOrder = append(groupOrder, interval)
+ }
+ groups[interval] = append(groups[interval], tagName)
+ }
+ for _, interval := range groupOrder {
+ groupNames := groups[interval]
+ groupTags := make([]PlcTag, len(groupNames))
+ for i, tagName := range groupNames {
+ groupTags[i] = tags[tagName]
+ }
+ tpduId := s.connection.tpduGenerator.getAndIncrement()
+ response, err := s.connection.sendUserData(ctx, tpduId, buildCyclicSubscribeRequest(tpduId, groupTags, interval), "cyclic_subscribe")
+ var jobId uint8
+ if err == nil {
+ jobId, err = parseCyclicSubscribeResponse(response)
+ }
+ if err != nil {
+ s.log.Debug().Err(err).Msg("Cyclic subscribe failed")
+ for _, tagName := range groupNames {
+ responseCodes[tagName] = apiModel.PlcResponseCode_REMOTE_ERROR
+ }
+ continue
+ }
+ jobHandles := make([]*SubscriptionHandle, 0, len(groupNames))
+ for i, tagName := range groupNames {
+ handle := NewSubscriptionHandle(s, tagName, groupTags[i], subscriptionKindCyclic)
+ handle.jobId = jobId
+ handle.itemIndex = i
+ jobHandles = append(jobHandles, handle)
+ handles[tagName] = handle
+ responseCodes[tagName] = apiModel.PlcResponseCode_OK
+ }
+ s.cyclicMutex.Lock()
+ s.cyclicHandlesByJob[jobId] = jobHandles
+ s.cyclicMutex.Unlock()
+ s.log.Debug().
+ Uint8("jobId", jobId).
+ Strs("tags", groupNames).
+ Dur("interval", interval.toDuration()).
+ Msg("Cyclic subscription active")
+ }
+}
+
+func (s *Subscriber) runAlarmQueries(ctx context.Context, tagNames []string, tags map[string]*AlarmTag, responseCodes map[string]apiModel.PlcResponseCode, handles map[string]apiModel.PlcSubscriptionHandle) {
+ for _, tagName := range tagNames {
+ tpduId := s.connection.tpduGenerator.getAndIncrement()
+ response, err := s.connection.sendUserData(ctx, tpduId, buildAlarmQueryRequest(tpduId, tags[tagName].GetQueryType()), "alarm_query")
+ var payload []byte
+ if err == nil {
+ payload, err = parseAlarmQueryResponse(response)
+ }
+ if err != nil {
+ s.log.Debug().Err(err).Str("tagName", tagName).Msg("Alarm query failed")
+ responseCodes[tagName] = apiModel.PlcResponseCode_REMOTE_ERROR
+ continue
+ }
+ handle := NewSubscriptionHandle(s, tagName, tags[tagName], subscriptionKindQuery)
+ handle.queryPayload = payload
+ handles[tagName] = handle
+ responseCodes[tagName] = apiModel.PlcResponseCode_OK
+ }
+}
+
+func (s *Subscriber) Unsubscribe(ctx context.Context, unsubscriptionRequest apiModel.PlcUnsubscriptionRequest) <-chan apiModel.PlcUnsubscriptionRequestResult {
+ result := make(chan apiModel.PlcUnsubscriptionRequestResult, 1)
+ s.wg.Go(func() {
+ defer func() {
+ if err := recover(); err != nil {
+ utils.DeliverResult(s.log, result, spiModel.NewDefaultPlcUnsubscriptionRequestResult(unsubscriptionRequest, nil, errors.Errorf("panic-ed %v. Stack: %s", err, debug.Stack())))
+ }
+ }()
+ request, ok := unsubscriptionRequest.(*spiModel.DefaultPlcUnsubscriptionRequest)
+ if !ok {
+ utils.DeliverResult(s.log, result, spiModel.NewDefaultPlcUnsubscriptionRequestResult(unsubscriptionRequest, nil, errors.Errorf("unsupported unsubscription request type %T", unsubscriptionRequest)))
+ return
+ }
+ err := s.removeHandles(ctx, request.GetSubscriptionHandles())
+ utils.DeliverResult(s.log, result, spiModel.NewDefaultPlcUnsubscriptionRequestResult(
+ unsubscriptionRequest,
+ spiModel.NewDefaultPlcUnsubscriptionResponse(unsubscriptionRequest),
+ err,
+ ))
+ })
+ return result
+}
+
+// removeHandles drops the given handles from the local bookkeeping and cancels the PLC-side
+// subscriptions that no longer have any handle attached.
+func (s *Subscriber) removeHandles(ctx context.Context, subscriptionHandles []apiModel.PlcSubscriptionHandle) error {
+ jobsToCancel := map[uint8]struct{}{}
+ cancelAlarms := false
+ for _, subscriptionHandle := range subscriptionHandles {
+ handle, ok := subscriptionHandle.(*SubscriptionHandle)
+ if !ok {
+ continue
+ }
+ switch handle.kind {
+ case subscriptionKindAlarm:
+ s.alarmMutex.Lock()
+ s.alarmHandles = slices.DeleteFunc(s.alarmHandles, func(other *SubscriptionHandle) bool { return other == handle })
+ if len(s.alarmHandles) == 0 && s.alarmSubscriptionActive {
+ cancelAlarms = true
+ }
+ s.alarmMutex.Unlock()
+ case subscriptionKindCyclic:
+ s.cyclicMutex.Lock()
+ remaining := slices.DeleteFunc(s.cyclicHandlesByJob[handle.jobId], func(other *SubscriptionHandle) bool { return other == handle })
+ if len(remaining) == 0 {
+ delete(s.cyclicHandlesByJob, handle.jobId)
+ jobsToCancel[handle.jobId] = struct{}{}
+ } else {
+ s.cyclicHandlesByJob[handle.jobId] = remaining
+ }
+ s.cyclicMutex.Unlock()
+ }
+ }
+ var firstErr error
+ for jobId := range jobsToCancel {
+ tpduId := s.connection.tpduGenerator.getAndIncrement()
+ if _, err := s.connection.sendUserData(ctx, tpduId, buildCyclicUnsubscribeRequest(tpduId, jobId), "cyclic_unsubscribe"); err != nil {
+ s.log.Debug().Err(err).Uint8("jobId", jobId).Msg("Cyclic cancel failed (ignoring)")
+ if firstErr == nil {
+ firstErr = err
+ }
+ }
+ }
+ if cancelAlarms {
+ tpduId := s.connection.tpduGenerator.getAndIncrement()
+ alarmState := alarmStateForController(s.connection.driverContext.ControllerType, true)
+ if _, err := s.connection.sendUserData(ctx, tpduId, buildMsgSubscriptionRequest(tpduId, alarmState), "alarm_unsubscribe"); err != nil {
+ s.log.Debug().Err(err).Msg("Alarm cancel failed (ignoring)")
+ if firstErr == nil {
+ firstErr = err
+ }
+ }
+ s.alarmMutex.Lock()
+ s.alarmSubscriptionActive = false
+ s.alarmMutex.Unlock()
+ }
+ return firstErr
+}
+
+// drain cancels every active PLC-side subscription; used on connection close.
+func (s *Subscriber) drain(ctx context.Context) {
+ s.alarmMutex.Lock()
+ handles := make([]apiModel.PlcSubscriptionHandle, 0, len(s.alarmHandles))
+ for _, handle := range s.alarmHandles {
+ handles = append(handles, handle)
+ }
+ s.alarmMutex.Unlock()
+ s.cyclicMutex.Lock()
+ for _, jobHandles := range s.cyclicHandlesByJob {
+ for _, handle := range jobHandles {
+ handles = append(handles, handle)
+ }
+ }
+ s.cyclicMutex.Unlock()
+ if len(handles) == 0 {
+ return
+ }
+ if err := s.removeHandles(ctx, handles); err != nil {
+ s.log.Debug().Err(err).Msg("Error draining subscriptions on close")
+ }
+}
+
+func (s *Subscriber) Register(consumer apiModel.PlcSubscriptionEventConsumer, handles []apiModel.PlcSubscriptionHandle) apiModel.PlcConsumerRegistration {
+ s.consumersMutex.Lock()
+ consumerRegistration := spiModel.NewDefaultPlcConsumerRegistration(s, consumer, handles...)
+ s.consumers[consumerRegistration.(*spiModel.DefaultPlcConsumerRegistration)] = consumer
+ s.consumersMutex.Unlock()
+
+ // Query handles carry their result inline (fetched at subscribe time). Deliver it as a
+ // single event now; the handle goes inert thereafter.
+ tags := map[string]apiModel.PlcTag{}
+ types := map[string]apiModel.PlcSubscriptionType{}
+ intervals := map[string]time.Duration{}
+ responseCodes := map[string]apiModel.PlcResponseCode{}
+ addresses := map[string]string{}
+ values := map[string]apiValues.PlcValue{}
+ for _, subscriptionHandle := range handles {
+ handle, ok := subscriptionHandle.(*SubscriptionHandle)
+ if !ok || handle.kind != subscriptionKindQuery {
+ continue
+ }
+ tags[handle.tagName] = handle.tag
+ types[handle.tagName] = apiModel.SubscriptionEvent
+ intervals[handle.tagName] = 0
+ responseCodes[handle.tagName] = apiModel.PlcResponseCode_OK
+ addresses[handle.tagName] = handle.tag.GetAddressString()
+ values[handle.tagName] = spiValues.NewPlcRawByteArray(handle.queryPayload)
+ }
+ if len(values) > 0 {
+ event := NewSubscriptionEvent(tags, types, intervals, responseCodes, addresses, values, append(s._options, options.WithCustomLogger(s.log))...)
+ consumer(&event)
+ }
+ return consumerRegistration
+}
+
+func (s *Subscriber) Unregister(registration apiModel.PlcConsumerRegistration) {
+ s.consumersMutex.Lock()
+ defer s.consumersMutex.Unlock()
+ delete(s.consumers, registration.(*spiModel.DefaultPlcConsumerRegistration))
+}
+
+// handleUserDataPush routes an unsolicited UserData push to the matching subscriptions.
+// Returns true if the message was consumed.
+func (s *Subscriber) handleUserDataPush(message readWriteModel.S7MessageUserData) bool {
+ group, functionType, subfunction, ok := userDataPushKey(message)
+ if !ok || functionType != 0x00 {
+ return false
+ }
+ switch {
+ case group == 0x02 && (subfunction == 0x01 || subfunction == 0x05):
+ return s.handleCyclicPush(message)
+ case group == 0x04:
+ if _, isAlarmIndication := alarmIndicationSubfunctions[subfunction]; isAlarmIndication {
+ return s.handleAlarmPush(message)
+ }
+ }
+ return false
+}
+
+func (s *Subscriber) handleCyclicPush(message readWriteModel.S7MessageUserData) bool {
+ jobId, ok := userDataSequenceNumber(message)
+ if !ok {
+ return false
+ }
+ s.cyclicMutex.RLock()
+ jobHandles := slices.Clone(s.cyclicHandlesByJob[jobId])
+ s.cyclicMutex.RUnlock()
+ if len(jobHandles) == 0 {
+ return false
+ }
+ items := extractCyclicPushItems(message)
+ if items == nil {
+ return false
+ }
+ tags := map[string]apiModel.PlcTag{}
+ types := map[string]apiModel.PlcSubscriptionType{}
+ intervals := map[string]time.Duration{}
+ responseCodes := map[string]apiModel.PlcResponseCode{}
+ addresses := map[string]string{}
+ values := map[string]apiValues.PlcValue{}
+ for _, handle := range jobHandles {
+ tags[handle.tagName] = handle.tag
+ types[handle.tagName] = apiModel.SubscriptionCyclic
+ intervals[handle.tagName] = 0
+ addresses[handle.tagName] = handle.tag.GetAddressString()
+ if handle.itemIndex >= len(items) {
+ responseCodes[handle.tagName] = apiModel.PlcResponseCode_NOT_FOUND
+ continue
+ }
+ value, err := parsePlcValue(context.Background(), handle.tag.(PlcTag), items[handle.itemIndex], s.connection.driverContext.ControllerType)
+ if err != nil {
+ s.log.Debug().Err(err).Str("tagName", handle.tagName).Msg("Cyclic push decode failed")
+ responseCodes[handle.tagName] = apiModel.PlcResponseCode_INTERNAL_ERROR
+ continue
+ }
+ responseCodes[handle.tagName] = apiModel.PlcResponseCode_OK
+ values[handle.tagName] = value
+ }
+ event := NewSubscriptionEvent(tags, types, intervals, responseCodes, addresses, values, append(s._options, options.WithCustomLogger(s.log))...)
+
+ s.consumersMutex.RLock()
+ defer s.consumersMutex.RUnlock()
+ for registration, consumer := range s.consumers {
+ for _, registeredHandle := range registration.GetSubscriptionHandles() {
+ if handle, isS7Handle := registeredHandle.(*SubscriptionHandle); isS7Handle && handle.kind == subscriptionKindCyclic && handle.jobId == jobId {
+ // One event per registration, even if multiple handles match.
+ consumer(&event)
+ break
+ }
+ }
+ }
+ return true
+}
+
+func (s *Subscriber) handleAlarmPush(message readWriteModel.S7MessageUserData) bool {
+ payload := parseAlarmIndication(message)
+ if payload == nil {
+ return false
+ }
+ s.alarmMutex.Lock()
+ anyHandles := len(s.alarmHandles) > 0
+ s.alarmMutex.Unlock()
+ if !anyHandles {
+ return false
+ }
+ s.consumersMutex.RLock()
+ defer s.consumersMutex.RUnlock()
+ for registration, consumer := range s.consumers {
+ for _, registeredHandle := range registration.GetSubscriptionHandles() {
+ handle, isS7Handle := registeredHandle.(*SubscriptionHandle)
+ if !isS7Handle || handle.kind != subscriptionKindAlarm {
+ continue
+ }
+ event := NewSubscriptionEvent(
+ map[string]apiModel.PlcTag{handle.tagName: handle.tag},
+ map[string]apiModel.PlcSubscriptionType{handle.tagName: apiModel.SubscriptionEvent},
+ map[string]time.Duration{handle.tagName: 0},
+ map[string]apiModel.PlcResponseCode{handle.tagName: apiModel.PlcResponseCode_OK},
+ map[string]string{handle.tagName: handle.tag.GetAddressString()},
+ map[string]apiValues.PlcValue{handle.tagName: payload},
+ append(s._options, options.WithCustomLogger(s.log))...,
+ )
+ consumer(&event)
+ }
+ }
+ return true
+}
diff --git a/plc4go/internal/s7/Subscriber_test.go b/plc4go/internal/s7/Subscriber_test.go
new file mode 100644
index 0000000000..1bf8ee962d
--- /dev/null
+++ b/plc4go/internal/s7/Subscriber_test.go
@@ -0,0 +1,220 @@
+/*
+ * 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 s7
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+)
+
+func TestPickCyclicInterval(t *testing.T) {
+ tests := []struct {
+ requested time.Duration
+ base readWriteModel.TimeBase
+ factor uint8
+ }{
+ {0, readWriteModel.TimeBase_B1SEC, 1},
+ {100 * time.Millisecond, readWriteModel.TimeBase_B01SEC, 1},
+ {500 * time.Millisecond, readWriteModel.TimeBase_B01SEC, 5},
+ {time.Second, readWriteModel.TimeBase_B1SEC, 1},
+ {2500 * time.Millisecond, readWriteModel.TimeBase_B1SEC, 3},
+ {30 * time.Second, readWriteModel.TimeBase_B10SEC, 3},
+ {255 * time.Second, readWriteModel.TimeBase_B1SEC, 255},
+ {2550 * time.Second, readWriteModel.TimeBase_B10SEC, 255},
+ }
+ for _, test := range tests {
+ t.Run(test.requested.String(), func(t *testing.T) {
+ interval := pickCyclicInterval(test.requested)
+ assert.Equal(t, test.base, interval.base, "base")
+ assert.Equal(t, test.factor, interval.factor, "factor")
+ })
+ }
+}
+
+// roundTrip serializes a request and parses it back, proving the declared lengths are
+// consistent with the actual wire representation.
+func roundTrip(t *testing.T, packet readWriteModel.TPKTPacket) readWriteModel.S7MessageUserData {
+ t.Helper()
+ data, err := packet.Serialize()
+ require.NoError(t, err)
+ parsed, err := readWriteModel.TPKTPacketParse(t.Context(), data)
+ require.NoError(t, err)
+ cotpPacketData := parsed.GetPayload().(readWriteModel.COTPPacketData)
+ return cotpPacketData.GetPayload().(readWriteModel.S7MessageUserData)
+}
+
+func TestBuildMsgSubscriptionRequest(t *testing.T) {
+ message := roundTrip(t, buildMsgSubscriptionRequest(42, readWriteModel.AlarmStateType_ALARM_S_INITIATE))
+ payload := message.GetPayload().(readWriteModel.S7PayloadUserData).GetItems()[0]
+ request := payload.(readWriteModel.S7PayloadUserDataItemCpuFunctionMsgSubscriptionRequest)
+ assert.Equal(t, uint8(0x80), request.GetSubscription())
+ assert.Equal(t, msgSubscriptionMagicKey, request.GetMagicKey())
+ require.NotNil(t, request.GetAlarmtype())
+ assert.Equal(t, readWriteModel.AlarmStateType_ALARM_S_INITIATE, *request.GetAlarmtype())
+}
+
+func TestBuildCyclicSubscribeRequest(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_DATA_BLOCKS, 5, 10, 0, 2, readWriteModel.TransportSize_INT).(plcTag)
+ message := roundTrip(t, buildCyclicSubscribeRequest(43, []PlcTag{tag}, cyclicInterval{readWriteModel.TimeBase_B1SEC, 2}))
+ payload := message.GetPayload().(readWriteModel.S7PayloadUserData).GetItems()[0]
+ request := payload.(readWriteModel.S7PayloadUserDataItemCyclicServicesSubscribeRequest)
+ assert.Equal(t, uint16(1), request.GetItemsCount())
+ assert.Equal(t, readWriteModel.TimeBase_B1SEC, request.GetTimeBase())
+ assert.Equal(t, uint8(2), request.GetTimeFactor())
+ item := request.GetItem()[0].(readWriteModel.CycServiceItemAnyType)
+ assert.Equal(t, readWriteModel.TransportSize_INT, item.GetTransportSize())
+ assert.Equal(t, uint16(2), item.GetLength())
+ assert.Equal(t, uint16(5), item.GetDbNumber())
+ assert.Equal(t, readWriteModel.MemoryArea_DATA_BLOCKS, item.GetMemoryArea())
+ assert.Equal(t, uint32(10<<3), item.GetAddress())
+}
+
+func TestBuildAlarmQueryRequest(t *testing.T) {
+ message := roundTrip(t, buildAlarmQueryRequest(44, readWriteModel.QueryType_ALARM_S))
+ payload := message.GetPayload().(readWriteModel.S7PayloadUserData).GetItems()[0]
+ request := payload.(readWriteModel.S7PayloadUserDataItemCpuFunctionAlarmQueryRequest)
+ assert.Equal(t, readWriteModel.SyntaxIdType_ALARM_QUERYREQSET, request.GetSyntaxId())
+ assert.Equal(t, readWriteModel.QueryType_ALARM_S, request.GetQueryType())
+ assert.Equal(t, readWriteModel.AlarmType_ALARM_S, request.GetAlarmType())
+}
+
+func TestParseCyclicSubscribeResponse(t *testing.T) {
+ makeResponse := func(payloadItem readWriteModel.S7PayloadUserDataItem, sequenceNumber uint8) readWriteModel.S7Message {
+ return readWriteModel.NewS7MessageUserData(
+ 45,
+ readWriteModel.NewS7ParameterUserData([]readWriteModel.S7ParameterUserDataItem{
+ readWriteModel.NewS7ParameterUserDataItemCPUFunctions(
+ 0x12, 0x08, 0x02, 0x01, sequenceNumber,
+ ptr(uint8(0)), ptr(uint8(1)), ptr(uint16(0)),
+ ),
+ }),
+ readWriteModel.NewS7PayloadUserData([]readWriteModel.S7PayloadUserDataItem{payloadItem}),
+ )
+ }
+
+ t.Run("empty response carries jobId", func(t *testing.T) {
+ jobId, err := parseCyclicSubscribeResponse(makeResponse(
+ readWriteModel.NewS7PayloadUserDataItemCyclicServicesSubscribeEmptyResponse(
+ readWriteModel.DataTransportErrorCode_OK, readWriteModel.DataTransportSize_NULL, 0),
+ 7,
+ ))
+ require.NoError(t, err)
+ assert.Equal(t, uint8(7), jobId)
+ })
+ t.Run("value-carrying response", func(t *testing.T) {
+ jobId, err := parseCyclicSubscribeResponse(makeResponse(
+ readWriteModel.NewS7PayloadUserDataItemCyclicServicesSubscribeResponse(
+ readWriteModel.DataTransportErrorCode_OK, readWriteModel.DataTransportSize_OCTET_STRING, 8,
+ 1,
+ []readWriteModel.AssociatedValueType{
+ readWriteModel.NewAssociatedValueType(readWriteModel.DataTransportErrorCode_OK, readWriteModel.DataTransportSize_BYTE_WORD_DWORD, 2, []uint8{0x00, 0x04}),
+ },
+ ),
+ 9,
+ ))
+ require.NoError(t, err)
+ assert.Equal(t, uint8(9), jobId)
+ })
+}
+
+func TestExtractCyclicPushItems(t *testing.T) {
+ message := readWriteModel.NewS7MessageUserData(
+ 0,
+ readWriteModel.NewS7ParameterUserData([]readWriteModel.S7ParameterUserDataItem{
+ readWriteModel.NewS7ParameterUserDataItemCPUFunctions(
+ 0x12, 0x00, 0x02, 0x01, 3,
+ ptr(uint8(0)), ptr(uint8(1)), nil,
+ ),
+ }),
+ readWriteModel.NewS7PayloadUserData([]readWriteModel.S7PayloadUserDataItem{
+ readWriteModel.NewS7PayloadUserDataItemCyclicServicesPush(
+ readWriteModel.DataTransportErrorCode_OK, readWriteModel.DataTransportSize_OCTET_STRING, 12,
+ 2,
+ []readWriteModel.AssociatedValueType{
+ readWriteModel.NewAssociatedValueType(readWriteModel.DataTransportErrorCode_OK, readWriteModel.DataTransportSize_BYTE_WORD_DWORD, 2, []uint8{0x00, 0x04}),
+ readWriteModel.NewAssociatedValueType(readWriteModel.DataTransportErrorCode_OK, readWriteModel.DataTransportSize_BYTE_WORD_DWORD, 1, []uint8{0xFF}),
+ },
+ ),
+ }),
+ )
+ items := extractCyclicPushItems(message)
+ require.Len(t, items, 2)
+ assert.Equal(t, []byte{0x00, 0x04}, items[0])
+ assert.Equal(t, []byte{0xFF}, items[1])
+
+ group, functionType, subfunction, ok := userDataPushKey(message)
+ require.True(t, ok)
+ assert.Equal(t, uint8(0x02), group)
+ assert.Equal(t, uint8(0x00), functionType)
+ assert.Equal(t, uint8(0x01), subfunction)
+ jobId, ok := userDataSequenceNumber(message)
+ require.True(t, ok)
+ assert.Equal(t, uint8(3), jobId)
+}
+
+func TestParseAlarmIndication(t *testing.T) {
+ state := func(mask uint8) readWriteModel.State {
+ return readWriteModel.NewState(
+ mask&0x80 != 0, mask&0x40 != 0, mask&0x20 != 0, mask&0x10 != 0,
+ mask&0x08 != 0, mask&0x04 != 0, mask&0x02 != 0, mask&0x01 != 0,
+ )
+ }
+ message := readWriteModel.NewS7MessageUserData(
+ 0,
+ readWriteModel.NewS7ParameterUserData([]readWriteModel.S7ParameterUserDataItem{
+ readWriteModel.NewS7ParameterUserDataItemCPUFunctions(
+ 0x12, 0x00, 0x04, 0x12, 0,
+ ptr(uint8(0)), ptr(uint8(1)), nil,
+ ),
+ }),
+ readWriteModel.NewS7PayloadUserData([]readWriteModel.S7PayloadUserDataItem{
+ readWriteModel.NewS7PayloadAlarmS(
+ readWriteModel.DataTransportErrorCode_OK, readWriteModel.DataTransportSize_OCTET_STRING, 0,
+ readWriteModel.NewAlarmMessagePushType(
+ readWriteModel.NewDateAndTime(24, 8, 18, 10, 30, 0, 123, 1),
+ 0x11,
+ 1,
+ []readWriteModel.AlarmMessageObjectPushType{
+ readWriteModel.NewAlarmMessageObjectPushType(
+ 0x04, readWriteModel.SyntaxIdType_ALARM_INDSET, 0,
+ 0x12345678,
+ state(0x01), state(0x00), state(0x00), state(0x01),
+ nil,
+ ),
+ },
+ ),
+ ),
+ }),
+ )
+ value := parseAlarmIndication(message)
+ require.NotNil(t, value)
+ require.True(t, value.IsStruct())
+ structValue := value.GetStruct()
+ assert.Equal(t, uint32(0x12345678), structValue["eventId"].GetUint32())
+ assert.Equal(t, uint32(0x01), structValue["eventState"].GetUint32())
+ assert.Equal(t, uint32(0x01), structValue["ackStateComing"].GetUint32())
+ assert.Equal(t, "2024-08-18T10:30:00.123", structValue["plcTimestamp"].GetString())
+ assert.Contains(t, structValue, "receivedAt")
+}
diff --git a/plc4go/internal/s7/SubscriptionEvent.go b/plc4go/internal/s7/SubscriptionEvent.go
new file mode 100644
index 0000000000..2b57cfe43b
--- /dev/null
+++ b/plc4go/internal/s7/SubscriptionEvent.go
@@ -0,0 +1,55 @@
+/*
+ * 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 s7
+
+import (
+ "time"
+
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
+ spiModel "github.com/apache/plc4x/plc4go/spi/model"
+ "github.com/apache/plc4x/plc4go/spi/options"
+)
+
+type SubscriptionEvent struct {
+ *spiModel.DefaultPlcSubscriptionEvent
+
+ addresses map[string]string
+}
+
+func NewSubscriptionEvent(
+ tags map[string]apiModel.PlcTag,
+ types map[string]apiModel.PlcSubscriptionType,
+ intervals map[string]time.Duration,
+ responseCodes map[string]apiModel.PlcResponseCode,
+ addresses map[string]string,
+ values map[string]apiValues.PlcValue,
+ _options ...options.WithOption,
+) SubscriptionEvent {
+ subscriptionEvent := SubscriptionEvent{addresses: addresses}
+ event := spiModel.NewDefaultPlcSubscriptionEvent(&subscriptionEvent, tags, types, intervals, responseCodes, values, _options...)
+ subscriptionEvent.DefaultPlcSubscriptionEvent = event.(*spiModel.DefaultPlcSubscriptionEvent)
+ return subscriptionEvent
+}
+
+// GetAddress returns the tag address of the requested tag.
+func (m SubscriptionEvent) GetAddress(name string) string {
+ return m.addresses[name]
+}
diff --git a/plc4go/internal/s7/SubscriptionHandle.go b/plc4go/internal/s7/SubscriptionHandle.go
new file mode 100644
index 0000000000..cd563ad36a
--- /dev/null
+++ b/plc4go/internal/s7/SubscriptionHandle.go
@@ -0,0 +1,64 @@
+/*
+ * 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 s7
+
+import (
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ spiModel "github.com/apache/plc4x/plc4go/spi/model"
+)
+
+type subscriptionKind uint8
+
+const (
+ subscriptionKindAlarm subscriptionKind = iota
+ subscriptionKindCyclic
+ subscriptionKindQuery
+)
+
+type SubscriptionHandle struct {
+ *spiModel.DefaultPlcSubscriptionHandle
+
+ tagName string
+ tag apiModel.PlcTag
+ kind subscriptionKind
+ // jobId is the PLC assigned cyclic job, itemIndex the tag's position within that job.
+ jobId uint8
+ itemIndex int
+ // queryPayload carries the one-shot alarm query result, delivered on Register.
+ queryPayload []byte
+}
+
+func NewSubscriptionHandle(subscriber *Subscriber, tagName string, tag apiModel.PlcTag, kind subscriptionKind) *SubscriptionHandle {
+ handle := &SubscriptionHandle{
+ tagName: tagName,
+ tag: tag,
+ kind: kind,
+ }
+ handle.DefaultPlcSubscriptionHandle = spiModel.NewDefaultPlcSubscriptionHandleWithHandleToRegister(subscriber, handle)
+ return handle
+}
+
+func (h *SubscriptionHandle) GetTagName() string {
+ return h.tagName
+}
+
+func (h *SubscriptionHandle) String() string {
+ return h.tagName
+}
diff --git a/plc4go/internal/s7/UserDataService.go b/plc4go/internal/s7/UserDataService.go
index 319424260e..1f2fd085c0 100644
--- a/plc4go/internal/s7/UserDataService.go
+++ b/plc4go/internal/s7/UserDataService.go
@@ -20,10 +20,14 @@
package s7
import (
+ "fmt"
"strings"
+ "time"
+ apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
"github.com/apache/plc4x/plc4go/spi/errors"
+ spiValues "github.com/apache/plc4x/plc4go/spi/values"
)
// Helpers for the S7Comm UserData services (SZL identification, block functions,
@@ -271,6 +275,443 @@ func decodeControllerType(descriptor string) readWriteModel.ControllerType {
return readWriteModel.ControllerType_ANY
}
+// msgSubscriptionMagicKey is the magic key the PLC echoes back in subscription confirmations.
+// The legacy driver used "HmiRtm " (HMI runtime, padded to 8 ASCII bytes); keep it for
+// compatibility with PLCs matching the key pattern in some firmware paths.
+const msgSubscriptionMagicKey = "HmiRtm "
+
+// alarmStateForController picks the alarm stream to arm: S7-400 uses the block-of-8 ALARM_8
+// family, everything else the ALARM_S/SQ family (SFC17/18).
+func alarmStateForController(controllerType readWriteModel.ControllerType, abort bool) readWriteModel.AlarmStateType {
+ if controllerType == readWriteModel.ControllerType_S7_400 {
+ if abort {
+ return readWriteModel.AlarmStateType_ALARM_ABORT
+ }
+ return readWriteModel.AlarmStateType_ALARM_INITIATE
+ }
+ if abort {
+ return readWriteModel.AlarmStateType_ALARM_S_ABORT
+ }
+ return readWriteModel.AlarmStateType_ALARM_S_INITIATE
+}
+
+// buildMsgSubscriptionRequest arms (or with an abort alarm state: cancels) the ALARM_S/SQ
+// push stream on this connection (UserData group 0x04, subfunction 0x02, 12-byte variant).
+func buildMsgSubscriptionRequest(tpduId uint16, alarmState readWriteModel.AlarmStateType) readWriteModel.TPKTPacket {
+ reserve := uint8(0)
+ return newUserDataMessage(
+ tpduId,
+ newCpuFunctionsRequestParameter(0x04, 0x02), // CPU functions / MsgSubscription
+ readWriteModel.NewS7PayloadUserDataItemCpuFunctionMsgSubscriptionRequest(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_OCTET_STRING,
+ 12, // dataLength: the ALARM_S/SQ variant with appended alarm state
+ 0x80, // subscription bitmask: ALM
+ msgSubscriptionMagicKey,
+ &alarmState,
+ &reserve,
+ ),
+ )
+}
+
+func parseMsgSubscriptionResponse(message readWriteModel.S7Message) (bool, error) {
+ messageUserData, ok := message.(readWriteModel.S7MessageUserData)
+ if !ok {
+ return false, errors.Errorf("expected S7MessageUserData, got %T", message)
+ }
+ if err := checkUserDataErrorCode(messageUserData); err != nil {
+ return false, err
+ }
+ payloadUserData, ok := messageUserData.GetPayload().(readWriteModel.S7PayloadUserData)
+ if !ok {
+ return false, errors.Errorf("expected S7PayloadUserData, got %T", messageUserData.GetPayload())
+ }
+ for _, item := range payloadUserData.GetItems() {
+ switch response := item.(type) {
+ case readWriteModel.S7PayloadUserDataItemCpuFunctionMsgSubscriptionAlarmResponse:
+ result := response.GetResult()
+ return result == 0x00 || result == 0x02, nil
+ case readWriteModel.S7PayloadUserDataItemCpuFunctionMsgSubscriptionResponse:
+ // Plain "ok" response with no body.
+ return true, nil
+ }
+ }
+ return false, nil
+}
+
+// cyclicInterval is the TimeBase + factor pair approximating a requested cycle time.
+type cyclicInterval struct {
+ base readWriteModel.TimeBase
+ factor uint8
+}
+
+func (c cyclicInterval) toDuration() time.Duration {
+ base := 100 * time.Millisecond
+ switch c.base {
+ case readWriteModel.TimeBase_B1SEC:
+ base = time.Second
+ case readWriteModel.TimeBase_B10SEC:
+ base = 10 * time.Second
+ }
+ return base * time.Duration(c.factor)
+}
+
+// pickCyclicInterval prefers the coarsest base whose factor still fits into a byte -
+// empirically required for S7-300 firmware (>= V3.2), which rejects the 100ms base with
+// errCode 0xD804 and only honors B1SEC/B10SEC. Range: 100ms ... 2550s; non-positive
+// durations default to 1s.
+func pickCyclicInterval(requested time.Duration) cyclicInterval {
+ ms := int64(1000)
+ if requested > 0 {
+ ms = requested.Milliseconds()
+ if ms < 100 {
+ ms = 100
+ }
+ }
+ if ms >= 10_000 && ms <= 2_550_000 && ms%10_000 == 0 {
+ return cyclicInterval{readWriteModel.TimeBase_B10SEC, uint8(ms / 10_000)}
+ }
+ if ms >= 1_000 && ms <= 255_000 && ms%1_000 == 0 {
+ return cyclicInterval{readWriteModel.TimeBase_B1SEC, uint8(ms / 1_000)}
+ }
+ // Round near-second cadences to full seconds rather than using the often-rejected
+ // 100ms base when the cadence changes by less than ~10%.
+ if ms >= 900 && ms <= 255_000 {
+ factor := (ms + 500) / 1_000
+ if factor < 1 {
+ factor = 1
+ }
+ return cyclicInterval{readWriteModel.TimeBase_B1SEC, uint8(factor)}
+ }
+ if ms >= 9_000 && ms <= 2_550_000 {
+ factor := (ms + 5_000) / 10_000
+ if factor < 1 {
+ factor = 1
+ }
+ return cyclicInterval{readWriteModel.TimeBase_B10SEC, uint8(factor)}
+ }
+ // Genuine sub-second request - only the 100ms base can express it; some firmwares
+ // reject this with 0xD804, callers should treat that as "not supported".
+ factor := (ms + 50) / 100
+ if factor < 1 {
+ factor = 1
+ }
+ if factor > 255 {
+ factor = 255
+ }
+ return cyclicInterval{readWriteModel.TimeBase_B01SEC, uint8(factor)}
+}
+
+func buildCyclicSubscribeRequest(tpduId uint16, tags []PlcTag, interval cyclicInterval) readWriteModel.TPKTPacket {
+ items := make([]readWriteModel.CycServiceItemType, len(tags))
+ for i, tag := range tags {
+ // The 24-bit address encodes bit-addressed memory: byteOffset shifted left by 3
+ // plus the bit number - same as in a regular S7AddressAny.
+ address := uint32(tag.GetByteOffset())<<3 | uint32(tag.GetBitOffset()&0x07)
+ items[i] = readWriteModel.NewCycServiceItemAnyType(
+ 0x0a, // byteLength: 10 bytes after the byteLength field
+ 0x10, // syntaxId: ANY
+ tag.GetDataType(),
+ tag.GetNumElements(),
+ tag.GetBlockNumber(),
+ tag.GetMemoryArea(),
+ address,
+ )
+ }
+ // dataLength: itemsCount(2) + timeBase(1) + timeFactor(1) + 12 bytes per ANY item.
+ payloadLength := uint16(4 + len(items)*12)
+ return newUserDataMessage(
+ tpduId,
+ newCpuFunctionsRequestParameter(0x02, 0x01), // cyclic services / subscribe
+ readWriteModel.NewS7PayloadUserDataItemCyclicServicesSubscribeRequest(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_OCTET_STRING,
+ payloadLength,
+ uint16(len(items)),
+ interval.base,
+ interval.factor,
+ items,
+ ),
+ )
+}
+
+func buildCyclicUnsubscribeRequest(tpduId uint16, jobId uint8) readWriteModel.TPKTPacket {
+ return newUserDataMessage(
+ tpduId,
+ newCpuFunctionsRequestParameter(0x02, 0x04), // cyclic services / unsubscribe
+ readWriteModel.NewS7PayloadUserDataItemCyclicServicesUnsubscribeRequest(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_OCTET_STRING,
+ 2,
+ 0x05, // function: unsubscribe single job
+ jobId,
+ ),
+ )
+}
+
+// parseCyclicSubscribeResponse returns the jobId the PLC assigned to the subscription
+// (carried as the parameter's sequence number).
+func parseCyclicSubscribeResponse(message readWriteModel.S7Message) (uint8, error) {
+ messageUserData, ok := message.(readWriteModel.S7MessageUserData)
+ if !ok {
+ return 0, errors.Errorf("expected S7MessageUserData, got %T", message)
+ }
+ if err := checkUserDataErrorCode(messageUserData); err != nil {
+ return 0, err
+ }
+ payloadUserData, ok := messageUserData.GetPayload().(readWriteModel.S7PayloadUserData)
+ if !ok {
+ return 0, errors.Errorf("expected S7PayloadUserData, got %T", messageUserData.GetPayload())
+ }
+ for _, item := range payloadUserData.GetItems() {
+ switch item.(type) {
+ // Two success shapes: a response carrying the initial values, or an empty response
+ // when the PLC doesn't bundle them - pushes start arriving on the cycle anyway.
+ case readWriteModel.S7PayloadUserDataItemCyclicServicesSubscribeResponse,
+ readWriteModel.S7PayloadUserDataItemCyclicServicesSubscribeEmptyResponse:
+ jobId, ok := userDataSequenceNumber(messageUserData)
+ if !ok {
+ return 0, errors.New("cyclic subscribe response carries no sequence number")
+ }
+ return jobId, nil
+ case readWriteModel.S7PayloadUserDataItemCyclicServicesErrorResponse:
+ return 0, errors.New("cyclic subscribe rejected by PLC")
+ }
+ }
+ return 0, errors.New("no cyclic subscribe response item in payload")
+}
+
+func userDataSequenceNumber(message readWriteModel.S7MessageUserData) (uint8, bool) {
+ parameterUserData, ok := message.GetParameter().(readWriteModel.S7ParameterUserData)
+ if !ok {
+ return 0, false
+ }
+ for _, item := range parameterUserData.GetItems() {
+ if cpuFunctions, ok := item.(readWriteModel.S7ParameterUserDataItemCPUFunctions); ok {
+ return cpuFunctions.GetSequenceNumber(), true
+ }
+ }
+ return 0, false
+}
+
+// userDataPushKey extracts the (functionGroup, functionType, subfunction) routing triple.
+func userDataPushKey(message readWriteModel.S7MessageUserData) (group uint8, functionType uint8, subfunction uint8, ok bool) {
+ parameterUserData, isUserData := message.GetParameter().(readWriteModel.S7ParameterUserData)
+ if !isUserData {
+ return 0, 0, 0, false
+ }
+ for _, item := range parameterUserData.GetItems() {
+ if cpuFunctions, isCpuFunctions := item.(readWriteModel.S7ParameterUserDataItemCPUFunctions); isCpuFunctions {
+ return cpuFunctions.GetCpuFunctionGroup(), cpuFunctions.GetCpuFunctionType(), cpuFunctions.GetCpuSubfunction(), true
+ }
+ }
+ return 0, 0, 0, false
+}
+
+// extractCyclicPushItems returns the per-item raw payloads of a cyclic services push.
+func extractCyclicPushItems(message readWriteModel.S7MessageUserData) [][]byte {
+ payloadUserData, ok := message.GetPayload().(readWriteModel.S7PayloadUserData)
+ if !ok {
+ return nil
+ }
+ toByteSlices := func(values []readWriteModel.AssociatedValueType) [][]byte {
+ result := make([][]byte, len(values))
+ for i, value := range values {
+ result[i] = value.GetData()
+ }
+ return result
+ }
+ toByteSlicesQuery := func(values []readWriteModel.AssociatedQueryValueType) [][]byte {
+ result := make([][]byte, len(values))
+ for i, value := range values {
+ result[i] = value.GetData()
+ }
+ return result
+ }
+ for _, item := range payloadUserData.GetItems() {
+ switch push := item.(type) {
+ case readWriteModel.S7PayloadUserDataItemCyclicServicesPush:
+ return toByteSlices(push.GetItems())
+ case readWriteModel.S7PayloadUserDataItemCyclicServicesChangeDrivenPush:
+ return toByteSlicesQuery(push.GetItems())
+ }
+ }
+ return nil
+}
+
+func buildAlarmQueryRequest(tpduId uint16, queryType readWriteModel.QueryType) readWriteModel.TPKTPacket {
+ alarmType := readWriteModel.AlarmType_ALARM_S
+ if queryType == readWriteModel.QueryType_ALARM_8 {
+ alarmType = readWriteModel.AlarmType_ALARM_8
+ }
+ return newUserDataMessage(
+ tpduId,
+ newCpuFunctionsRequestParameter(0x04, 0x13), // CPU functions / AlarmQuery
+ readWriteModel.NewS7PayloadUserDataItemCpuFunctionAlarmQueryRequest(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_OCTET_STRING,
+ // 4 const bytes + 5 variable bytes = 9; the legacy driver used 12 and real
+ // S7-300s absorb the extra framing, so keep the proven value.
+ 12,
+ readWriteModel.SyntaxIdType_ALARM_QUERYREQSET,
+ queryType,
+ alarmType,
+ ),
+ )
+}
+
+// parseAlarmQueryResponse extracts the raw byte payload of an AlarmQuery response.
+func parseAlarmQueryResponse(message readWriteModel.S7Message) ([]byte, error) {
+ messageUserData, ok := message.(readWriteModel.S7MessageUserData)
+ if !ok {
+ return nil, errors.Errorf("expected S7MessageUserData, got %T", message)
+ }
+ if err := checkUserDataErrorCode(messageUserData); err != nil {
+ return nil, err
+ }
+ payloadUserData, ok := messageUserData.GetPayload().(readWriteModel.S7PayloadUserData)
+ if !ok {
+ return nil, errors.Errorf("expected S7PayloadUserData, got %T", messageUserData.GetPayload())
+ }
+ for _, item := range payloadUserData.GetItems() {
+ if response, ok := item.(readWriteModel.S7PayloadUserDataItemCpuFunctionAlarmQueryResponse); ok {
+ return response.GetItems(), nil
+ }
+ }
+ return nil, errors.New("no alarm query response item in payload")
+}
+
+// alarmIndicationSubfunctions are the push subfunctions carrying alarm indication payloads.
+var alarmIndicationSubfunctions = map[uint8]struct{}{
+ 0x05: {}, // ALARM8
+ 0x06: {}, // NOTIFY
+ 0x0c: {}, // ALARM_ACK_IND
+ 0x11: {}, // ALARM_SQ
+ 0x12: {}, // ALARM_S
+ 0x13: {}, // ALARM_SC
+ 0x16: {}, // NOTIFY8
+}
+
+// parseAlarmIndication turns a pushed alarm message into a PlcStruct. The struct keys match
+// the plc4j driver: plcTimestamp, functionId, numberOfObjects, eventId, eventState,
+// localState, ackStateGoing, ackStateComing, receivedAt.
+func parseAlarmIndication(message readWriteModel.S7MessageUserData) apiValues.PlcValue {
+ payloadUserData, ok := message.GetPayload().(readWriteModel.S7PayloadUserData)
+ if !ok {
+ return nil
+ }
+ for _, item := range payloadUserData.GetItems() {
+ switch payload := item.(type) {
+ case readWriteModel.S7PayloadAlarmS:
+ return alarmStructFromPush(payload.GetAlarmMessage())
+ case readWriteModel.S7PayloadAlarmSQ:
+ return alarmStructFromPush(payload.GetAlarmMessage())
+ case readWriteModel.S7PayloadAlarmSC:
+ return alarmStructFromPush(payload.GetAlarmMessage())
+ case readWriteModel.S7PayloadAlarm8:
+ return alarmStructFromPush(payload.GetAlarmMessage())
+ case readWriteModel.S7PayloadNotify:
+ return alarmStructFromPush(payload.GetAlarmMessage())
+ case readWriteModel.S7PayloadNotify8:
+ return alarmStructFromPush(payload.GetAlarmMessage())
+ case readWriteModel.S7PayloadAlarmAckInd:
+ return alarmStructFromAckPush(payload.GetAlarmMessage())
+ }
+ }
+ return nil
+}
+
+func alarmStructFromPush(push readWriteModel.AlarmMessagePushType) apiValues.PlcValue {
+ if push == nil {
+ return nil
+ }
+ children := map[string]apiValues.PlcValue{}
+ addAlarmHeader(children, push.GetTimeStamp(), push.GetFunctionId(), push.GetNumberOfObjects())
+ // Surface the first object directly - the common case is exactly one object per
+ // indication; multi-object indications can be inspected via numberOfObjects.
+ if objects := push.GetMessageObjects(); len(objects) > 0 {
+ first := objects[0]
+ children["eventId"] = spiValues.NewPlcUDINT(first.GetEventId())
+ children["eventState"] = spiValues.NewPlcUDINT(stateToMask(first.GetEventState()))
+ children["localState"] = spiValues.NewPlcUDINT(stateToMask(first.GetLocalState()))
+ children["ackStateGoing"] = spiValues.NewPlcUDINT(stateToMask(first.GetAckStateGoing()))
+ children["ackStateComing"] = spiValues.NewPlcUDINT(stateToMask(first.GetAckStateComing()))
+ }
+ children["receivedAt"] = spiValues.NewPlcLINT(time.Now().UnixMilli())
+ return spiValues.NewPlcStruct(children)
+}
+
+func alarmStructFromAckPush(push readWriteModel.AlarmMessageAckPushType) apiValues.PlcValue {
+ if push == nil {
+ return nil
+ }
+ children := map[string]apiValues.PlcValue{}
+ addAlarmHeader(children, push.GetTimeStamp(), push.GetFunctionId(), push.GetNumberOfObjects())
+ if objects := push.GetMessageObjects(); len(objects) > 0 {
+ first := objects[0]
+ children["eventId"] = spiValues.NewPlcUDINT(first.GetEventId())
+ children["ackStateGoing"] = spiValues.NewPlcUDINT(stateToMask(first.GetAckStateGoing()))
+ children["ackStateComing"] = spiValues.NewPlcUDINT(stateToMask(first.GetAckStateComing()))
+ }
+ children["receivedAt"] = spiValues.NewPlcLINT(time.Now().UnixMilli())
+ return spiValues.NewPlcStruct(children)
+}
+
+func addAlarmHeader(children map[string]apiValues.PlcValue, timestamp readWriteModel.DateAndTime, functionId uint8, numberOfObjects uint8) {
+ children["plcTimestamp"] = spiValues.NewPlcSTRING(formatAlarmDateAndTime(timestamp))
+ children["functionId"] = spiValues.NewPlcUDINT(uint32(functionId))
+ children["numberOfObjects"] = spiValues.NewPlcUDINT(uint32(numberOfObjects))
+}
+
+func stateToMask(state readWriteModel.State) uint32 {
+ if state == nil {
+ return 0
+ }
+ var mask uint32
+ if state.GetSIG_1() {
+ mask |= 0x01
+ }
+ if state.GetSIG_2() {
+ mask |= 0x02
+ }
+ if state.GetSIG_3() {
+ mask |= 0x04
+ }
+ if state.GetSIG_4() {
+ mask |= 0x08
+ }
+ if state.GetSIG_5() {
+ mask |= 0x10
+ }
+ if state.GetSIG_6() {
+ mask |= 0x20
+ }
+ if state.GetSIG_7() {
+ mask |= 0x40
+ }
+ if state.GetSIG_8() {
+ mask |= 0x80
+ }
+ return mask
+}
+
+func formatAlarmDateAndTime(dateAndTime readWriteModel.DateAndTime) string {
+ if dateAndTime == nil {
+ return ""
+ }
+ // The BCD year is 2-digit; S7 convention: < 90 -> 2000s, >= 90 -> 1900s.
+ year := uint16(dateAndTime.GetYear())
+ if year < 90 {
+ year += 2000
+ } else {
+ year += 1900
+ }
+ return fmt.Sprintf("%04d-%02d-%02dT%02d:%02d:%02d.%03d",
+ year, dateAndTime.GetMonth(), dateAndTime.GetDay(),
+ dateAndTime.GetHour(), dateAndTime.GetMinutes(), dateAndTime.GetSeconds(),
+ dateAndTime.GetMsec())
+}
+
// supportsUserDataServices reports whether the controller family supports the S7Comm
// UserData services (browse, alarms, cyclic subscriptions).
func supportsUserDataServices(controllerType readWriteModel.ControllerType) bool {
diff --git a/plc4go/spi/model/DefaultPlcSubscriptionHandle.go b/plc4go/spi/model/DefaultPlcSubscriptionHandle.go
index 6dd1185a71..4436ba079b 100644
--- a/plc4go/spi/model/DefaultPlcSubscriptionHandle.go
+++ b/plc4go/spi/model/DefaultPlcSubscriptionHandle.go
@@ -56,3 +56,9 @@ func NewDefaultPlcSubscriptionHandleWithHandleToRegister(plcSubscriber spi.PlcSu
func (d *DefaultPlcSubscriptionHandle) Register(consumer apiModel.PlcSubscriptionEventConsumer) apiModel.PlcConsumerRegistration {
return d.plcSubscriber.Register(consumer, []apiModel.PlcSubscriptionHandle{d.handleToRegister})
}
+
+// getPlcSubscriber exposes the subscriber to the unsubscription flow; through embedding it
+// also makes driver-specific handle types unsubscribable.
+func (d *DefaultPlcSubscriptionHandle) getPlcSubscriber() spi.PlcSubscriber {
+ return d.plcSubscriber
+}
diff --git a/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go b/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go
index af66276988..1aa05664b2 100644
--- a/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go
+++ b/plc4go/spi/model/DefaultPlcUnsubscriptionRequest.go
@@ -24,6 +24,7 @@ import (
"sync"
apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ "github.com/apache/plc4x/plc4go/spi"
"github.com/apache/plc4x/plc4go/spi/errors"
)
@@ -67,8 +68,15 @@ func (d *DefaultPlcUnsubscriptionRequest) Execute(ctx context.Context) <-chan ap
d.wg.Go(func() {
var collectedErrors []error
for _, handle := range d.subscriptionHandles {
+ // Driver-specific handles embed *DefaultPlcSubscriptionHandle, so the unexported
+ // interface assertion covers them too (a plain type assertion would panic).
+ subscriberProvider, ok := handle.(interface{ getPlcSubscriber() spi.PlcSubscriber })
+ if !ok {
+ collectedErrors = append(collectedErrors, errors.Errorf("%T is not a supported subscription handle", handle))
+ continue
+ }
select {
- case unsubscribe := <-handle.(*DefaultPlcSubscriptionHandle).plcSubscriber.Unsubscribe(ctx, d):
+ case unsubscribe := <-subscriberProvider.getPlcSubscriber().Unsubscribe(ctx, d):
if err := unsubscribe.GetErr(); err != nil {
collectedErrors = append(collectedErrors, err)
continue
(plc4x) 03/09: fix(plc4go/s7): write path honors string length, controller type and multi-element values
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit de83bbff8fb3b2bc9bfbbfe77c77c9b3af1e9b4c
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 11:56:17 2026 +0200
fix(plc4go/s7): write path honors string length, controller type and multi-element values
The *PlcStringTag pointer assertion never matched the value-typed tags
returned by the TagHandler, so every string write silently used the
254-char default. Wire the DriverContext into the Writer so the detected
controller type reaches DataItemSerialize, port the Java multi-element
handling (raw BYTE arrays, bit-packed BOOL arrays, per-element
serialization) and share the header error-class mapping with the Reader.
---
plc4go/internal/s7/Connection.go | 8 +++-
plc4go/internal/s7/Writer.go | 83 ++++++++++++++++++++++++++-------------
plc4go/internal/s7/Writer_test.go | 74 ++++++++++++++++++++++++++++++++++
3 files changed, 136 insertions(+), 29 deletions(-)
diff --git a/plc4go/internal/s7/Connection.go b/plc4go/internal/s7/Connection.go
index 5627970f0e..e8e510c8d6 100644
--- a/plc4go/internal/s7/Connection.go
+++ b/plc4go/internal/s7/Connection.go
@@ -453,7 +453,13 @@ func (c *Connection) ReadRequestBuilder() apiModel.PlcReadRequestBuilder {
func (c *Connection) WriteRequestBuilder() apiModel.PlcWriteRequestBuilder {
return spiModel.NewDefaultPlcWriteRequestBuilder(
- c.GetPlcTagHandler(), c.GetPlcValueHandler(), NewWriter(&c.tpduGenerator, c.messageCodec, c.tm))
+ c.GetPlcTagHandler(), c.GetPlcValueHandler(), NewWriter(
+ &c.tpduGenerator,
+ c.messageCodec,
+ c.tm,
+ &c.driverContext,
+ append(c._options, options.WithCustomLogger(c.log))...,
+ ))
}
func (c *Connection) String() string {
diff --git a/plc4go/internal/s7/Writer.go b/plc4go/internal/s7/Writer.go
index 8387fff491..c78b146ddb 100644
--- a/plc4go/internal/s7/Writer.go
+++ b/plc4go/internal/s7/Writer.go
@@ -42,18 +42,20 @@ type Writer struct {
tpduGenerator *TpduGenerator
messageCodec spi.MessageCodec
tm transactions.RequestTransactionManager
+ driverContext *DriverContext
wg sync.WaitGroup // use to track spawned go routines
log zerolog.Logger
}
-func NewWriter(tpduGenerator *TpduGenerator, messageCodec spi.MessageCodec, tm transactions.RequestTransactionManager, _options ...options.WithOption) *Writer {
+func NewWriter(tpduGenerator *TpduGenerator, messageCodec spi.MessageCodec, tm transactions.RequestTransactionManager, driverContext *DriverContext, _options ...options.WithOption) *Writer {
customLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...)
return &Writer{
tpduGenerator: tpduGenerator,
messageCodec: messageCodec,
tm: tm,
+ driverContext: driverContext,
log: customLogger,
}
}
@@ -78,7 +80,7 @@ func (m *Writer) Write(ctx context.Context, writeRequest apiModel.PlcWriteReques
return
}
parameterItems[i] = readWriteModel.NewS7VarRequestParameterItemAddress(s7Address)
- value, err := serializePlcValue(tag, plcValue)
+ value, err := serializePlcValue(tag, plcValue, m.driverContext.ControllerType)
if err != nil {
utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcWriteRequestResult(writeRequest, nil, errors.Wrapf(err, "Error encoding value for tag %s", tagName)))
return
@@ -173,28 +175,12 @@ func (m *Writer) ToPlc4xWriteResponse(response readWriteModel.S7Message, writeRe
// If the result contains any form of non-null error code, handle this instead.
if (errorClass != 0) || (errorCode != 0) {
- // This is usually the case if PUT/GET wasn't enabled on the PLC
- if (errorClass == 129) && (errorCode == 4) {
- m.log.Warn().Msg("Got an error response from the PLC. This particular response code usually indicates " +
- "that PUT/GET is not enabled on the PLC.")
- for _, tagName := range writeRequest.GetTagNames() {
- responseCodes[tagName] = apiModel.PlcResponseCode_ACCESS_DENIED
- }
- m.log.Trace().Msg("Returning the response")
- return spiModel.NewDefaultPlcWriteResponse(writeRequest, responseCodes), nil
- } else {
- m.log.Warn().
- Uint8("errorClass", errorClass).
- Uint8("errorCode", errorCode).
- Msg("Got an unknown error response from the PLC. Error Class: %d, Error Code %d. " +
- "We probably need to implement explicit handling for this, so please file a bug-report " +
- "on https://github.com/apache/plc4x/issues and ideally attach a WireShark dump " +
- "containing a capture of the communication.")
- for _, tagName := range writeRequest.GetTagNames() {
- responseCodes[tagName] = apiModel.PlcResponseCode_INTERNAL_ERROR
- }
- return spiModel.NewDefaultPlcWriteResponse(writeRequest, responseCodes), nil
+ responseCode := mapPlcErrorCode(m.log, errorClass, errorCode)
+ for _, tagName := range writeRequest.GetTagNames() {
+ responseCodes[tagName] = responseCode
}
+ m.log.Trace().Msg("Returning the response")
+ return spiModel.NewDefaultPlcWriteResponse(writeRequest, responseCodes), nil
}
// In all other cases all went well.
@@ -222,19 +208,60 @@ func (m *Writer) ToPlc4xWriteResponse(response readWriteModel.S7Message, writeRe
return spiModel.NewDefaultPlcWriteResponse(writeRequest, responseCodes), nil
}
-func serializePlcValue(tag apiModel.PlcTag, plcValue apiValues.PlcValue) (readWriteModel.S7VarPayloadDataItem, error) {
+func serializePlcValue(tag apiModel.PlcTag, plcValue apiValues.PlcValue, controllerType readWriteModel.ControllerType) (readWriteModel.S7VarPayloadDataItem, error) {
s7Tag, ok := tag.(PlcTag)
if !ok {
return nil, errors.Errorf("Unsupported address type %t", tag)
}
transportSize := s7Tag.GetDataType().DataTransportSize()
stringLength := uint16(254)
- if s7StringTag, ok := tag.(*PlcStringTag); ok {
+ if s7StringTag, ok := tag.(PlcStringTag); ok {
stringLength = s7StringTag.stringLength
}
- data, err := readWriteModel.DataItemSerialize(plcValue, s7Tag.GetDataType().DataProtocolId(), 0 /*TODO: port s7DriverContext.getControllerType()*/, int32(stringLength))
- if err != nil {
- return nil, errors.Wrapf(err, "Error serializing tag item of type: '%v'", s7Tag.GetDataType())
+ numElements := s7Tag.GetNumElements()
+ dataProtocolId := s7Tag.GetDataType().DataProtocolId()
+
+ if s7Tag.GetDataType() == readWriteModel.TransportSize_BYTE && numElements > 1 {
+ return readWriteModel.NewS7VarPayloadDataItem(
+ readWriteModel.DataTransportErrorCode_OK,
+ transportSize, plcValue.GetRaw(),
+ ), nil
+ }
+ if s7Tag.GetDataType() == readWriteModel.TransportSize_BOOL && numElements > 1 {
+ if !plcValue.IsList() {
+ return nil, errors.New("expected a list value for a multi-element BOOL tag")
+ }
+ list := plcValue.GetList()
+ if len(list) < int(numElements) {
+ return nil, errors.Errorf("expected %d list elements, got %d", numElements, len(list))
+ }
+ data := make([]byte, (numElements+7)/8)
+ for i := uint16(0); i < numElements; i++ {
+ if list[i].GetBool() {
+ data[i/8] |= 1 << (i % 8)
+ }
+ }
+ return readWriteModel.NewS7VarPayloadDataItem(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_BYTE_WORD_DWORD, data,
+ ), nil
+ }
+
+ var data []byte
+ if numElements == 1 {
+ serialized, err := readWriteModel.DataItemSerialize(plcValue, dataProtocolId, controllerType, int32(stringLength))
+ if err != nil {
+ return nil, errors.Wrapf(err, "Error serializing tag item of type: '%v'", s7Tag.GetDataType())
+ }
+ data = serialized
+ } else {
+ for i := uint16(0); i < numElements; i++ {
+ serialized, err := readWriteModel.DataItemSerialize(plcValue.GetIndex(uint32(i)), dataProtocolId, controllerType, int32(stringLength))
+ if err != nil {
+ return nil, errors.Wrapf(err, "Error serializing element %d of type: '%v'", i, s7Tag.GetDataType())
+ }
+ data = append(data, serialized...)
+ }
}
return readWriteModel.NewS7VarPayloadDataItem(
readWriteModel.DataTransportErrorCode_OK,
diff --git a/plc4go/internal/s7/Writer_test.go b/plc4go/internal/s7/Writer_test.go
new file mode 100644
index 0000000000..f41033ddc1
--- /dev/null
+++ b/plc4go/internal/s7/Writer_test.go
@@ -0,0 +1,74 @@
+/*
+ * 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 s7
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+ spiValues "github.com/apache/plc4x/plc4go/spi/values"
+)
+
+func TestSerializePlcValue(t *testing.T) {
+ anyController := readWriteModel.ControllerType_ANY
+
+ t.Run("STRING honors the tag string length", func(t *testing.T) {
+ tag := NewStringTag(readWriteModel.MemoryArea_DATA_BLOCKS, 1, 0, 0, 1, 10, readWriteModel.TransportSize_STRING)
+ item, err := serializePlcValue(tag, spiValues.NewPlcSTRING("ABC"), anyController)
+ require.NoError(t, err)
+ data := item.GetData()
+ // maxLen(1) + curLen(1) + 10 padded chars = 12 bytes, NOT the 256 of the old
+ // silently-default-254 behavior.
+ require.Len(t, data, 12)
+ assert.Equal(t, uint8(10), data[0])
+ assert.Equal(t, uint8(3), data[1])
+ assert.Equal(t, []byte("ABC"), data[2:5])
+ })
+ t.Run("BOOL array is bit packed", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 10, readWriteModel.TransportSize_BOOL)
+ list := make([]apiValues.PlcValue, 10)
+ for i := range list {
+ list[i] = spiValues.NewPlcBOOL(i == 0 || i == 1 || i == 9)
+ }
+ item, err := serializePlcValue(tag, spiValues.NewPlcList(list), anyController)
+ require.NoError(t, err)
+ assert.Equal(t, []byte{0b00000011, 0b00000010}, item.GetData())
+ assert.Equal(t, readWriteModel.DataTransportSize_BYTE_WORD_DWORD, item.GetTransportSize())
+ })
+ t.Run("S5TIME serializes to two bytes", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 1, readWriteModel.TransportSize_S5TIME)
+ item, err := serializePlcValue(tag, spiValues.NewPlcTIME(2*time.Second), anyController)
+ require.NoError(t, err)
+ // 2000ms -> finest fitting time base 10ms (0x0), counter 200 BCD
+ assert.Equal(t, []byte{0x02, 0x00}, item.GetData())
+ })
+ t.Run("INT array concatenates elements", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 2, readWriteModel.TransportSize_INT)
+ list := []apiValues.PlcValue{spiValues.NewPlcINT(1), spiValues.NewPlcINT(2)}
+ item, err := serializePlcValue(tag, spiValues.NewPlcList(list), anyController)
+ require.NoError(t, err)
+ assert.Equal(t, []byte{0x00, 0x01, 0x00, 0x02}, item.GetData())
+ })
+}
(plc4x) 02/09: fix(plc4go/s7): port read-response decoding and the stubbed S7 static helpers
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 80451d91b98cbe9e155d06df8830f91757e786a8
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 11:54:32 2026 +0200
fix(plc4go/s7): port read-response decoding and the stubbed S7 static helpers
parsePlcValue was an unported stub that panic-ed in the codec receive
worker, so every well-formed read timed out. Port the Java decode path
(single/multi element, raw BYTE arrays, bit-unpacked BOOL arrays) and
align the header error-class mapping (0x81/4, 0x83/4, 0x85 -> ACCESS_DENIED)
plus the DataTransportErrorCode mapping with plc4j.
encodeS7Address now converts every transport size without an own wire
code (temporal and 64-bit types, S5TIME) to plain byte-array requests
and packs BOOL arrays, like the Java driver.
The hand-written model StaticHelper was mostly stubs: S7 STRING/WSTRING
ignored the max/cur-length header, S5TIME and TIA DATE parsed to zero,
and RightShift3/LeftShift3/EventItemLength returned constants, breaking
AssociatedValueType (cyclic/alarm payloads). Ported all of them from the
Java StaticHelper.
---
plc4go/internal/s7/Connection.go | 1 +
plc4go/internal/s7/Reader.go | 119 ++++++++----
plc4go/internal/s7/Reader_test.go | 124 ++++++++++++
.../protocols/s7/readwrite/model/StaticHelper.go | 209 +++++++++++++++++----
4 files changed, 378 insertions(+), 75 deletions(-)
diff --git a/plc4go/internal/s7/Connection.go b/plc4go/internal/s7/Connection.go
index 98e764db38..5627970f0e 100644
--- a/plc4go/internal/s7/Connection.go
+++ b/plc4go/internal/s7/Connection.go
@@ -445,6 +445,7 @@ func (c *Connection) ReadRequestBuilder() apiModel.PlcReadRequestBuilder {
&c.tpduGenerator,
c.messageCodec,
c.tm,
+ &c.driverContext,
append(c._options, options.WithCustomLogger(c.log))...,
),
)
diff --git a/plc4go/internal/s7/Reader.go b/plc4go/internal/s7/Reader.go
index 7ddf5b158d..e423bb8c2f 100644
--- a/plc4go/internal/s7/Reader.go
+++ b/plc4go/internal/s7/Reader.go
@@ -43,6 +43,7 @@ type Reader struct {
tpduGenerator *TpduGenerator
messageCodec spi.MessageCodec
tm transactions.RequestTransactionManager
+ driverContext *DriverContext
wg sync.WaitGroup // use to track spawned go routines
@@ -50,13 +51,14 @@ type Reader struct {
log zerolog.Logger
}
-func NewReader(tpduGenerator *TpduGenerator, messageCodec spi.MessageCodec, tm transactions.RequestTransactionManager, _options ...options.WithOption) *Reader {
+func NewReader(tpduGenerator *TpduGenerator, messageCodec spi.MessageCodec, tm transactions.RequestTransactionManager, driverContext *DriverContext, _options ...options.WithOption) *Reader {
passLoggerToModel, _ := options.ExtractPassLoggerToModel(_options...)
customLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...)
return &Reader{
tpduGenerator: tpduGenerator,
messageCodec: messageCodec,
tm: tm,
+ driverContext: driverContext,
passLogToModel: passLoggerToModel,
log: customLogger,
}
@@ -200,30 +202,13 @@ func (m *Reader) ToPlc4xReadResponse(response readWriteModel.S7Message, readRequ
// If the result contains any form of non-null error code, handle this instead.
if (errorClass != 0) || (errorCode != 0) {
- // This is usually the case if PUT/GET wasn't enabled on the PLC
- if (errorClass == 129) && (errorCode == 4) {
- m.log.Warn().Msg("Got an error response from the PLC. This particular response code usually indicates " +
- "that PUT/GET is not enabled on the PLC.")
- for _, tagName := range readRequest.GetTagNames() {
- responseCodes[tagName] = apiModel.PlcResponseCode_ACCESS_DENIED
- plcValues[tagName] = spiValues.NewPlcNULL()
- }
- m.log.Trace().Msg("Returning the response")
- return spiModel.NewDefaultPlcReadResponse(readRequest, responseCodes, plcValues), nil
- } else {
- m.log.Warn().
- Uint8("errorClass", errorClass).
- Uint8("errorCode", errorCode).
- Msg("Got an unknown error response from the PLC. Error Class: %d, Error Code %d. " +
- "We probably need to implement explicit handling for this, so please file a bug-report " +
- "on https://github.com/apache/plc4x/issues and ideally attach a WireShark dump " +
- "containing a capture of the communication.")
- for _, tagName := range readRequest.GetTagNames() {
- responseCodes[tagName] = apiModel.PlcResponseCode_INTERNAL_ERROR
- plcValues[tagName] = spiValues.NewPlcNULL()
- }
- return spiModel.NewDefaultPlcReadResponse(readRequest, responseCodes, plcValues), nil
+ responseCode := mapPlcErrorCode(m.log, errorClass, errorCode)
+ for _, tagName := range readRequest.GetTagNames() {
+ responseCodes[tagName] = responseCode
+ plcValues[tagName] = spiValues.NewPlcNULL()
}
+ m.log.Trace().Msg("Returning the response")
+ return spiModel.NewDefaultPlcReadResponse(readRequest, responseCodes, plcValues), nil
}
// In all other cases all went well.
@@ -247,7 +232,7 @@ func (m *Reader) ToPlc4xReadResponse(response readWriteModel.S7Message, readRequ
responseCodes[tagName] = responseCode
if responseCode == apiModel.PlcResponseCode_OK {
ctxForModel := options.GetLoggerContextForModel(ctx, m.log, options.WithPassLoggerToModel(m.passLogToModel))
- plcValue, err := parsePlcValue(ctxForModel, tag, payloadItem.GetData())
+ plcValue, err := parsePlcValue(ctxForModel, tag, payloadItem.GetData(), m.driverContext.ControllerType)
if err != nil {
return nil, errors.Wrap(err, "Error parsing data item")
}
@@ -269,14 +254,6 @@ func encodeS7Address(tag apiModel.PlcTag) (readWriteModel.S7Address, error) {
}
transportSize := s7Tag.GetDataType()
numElements := s7Tag.GetNumElements()
- // For these date-types we have to convert the requests to simple byte-array requests
- // As otherwise the S7 will deny them with "Data type not supported" replies.
- if (transportSize == readWriteModel.TransportSize_TIME) /*|| (transportSize == TransportSize.S7_S5TIME)*/ ||
- (transportSize == readWriteModel.TransportSize_LTIME) || (transportSize == readWriteModel.TransportSize_DATE) ||
- (transportSize == readWriteModel.TransportSize_TIME_OF_DAY) || (transportSize == readWriteModel.TransportSize_DATE_AND_TIME) {
- numElements = numElements * uint16(transportSize.SizeInBytes())
- transportSize = readWriteModel.TransportSize_BYTE
- }
if transportSize == readWriteModel.TransportSize_STRING {
transportSize = readWriteModel.TransportSize_CHAR
stringLength := uint16(254)
@@ -291,6 +268,16 @@ func encodeS7Address(tag apiModel.PlcTag) (readWriteModel.S7Address, error) {
stringLength = s7StringTag.stringLength
}
numElements = numElements * (stringLength + 2) * 2
+ } else if transportSize == readWriteModel.TransportSize_BOOL && numElements > 1 {
+ // BOOL arrays are read as packed bytes.
+ numElements = (numElements + 7) / 8
+ transportSize = readWriteModel.TransportSize_BYTE
+ }
+ // Types without an own transport-size code (temporal types, 64-bit types, ...) have to be
+ // requested as plain byte arrays, the S7 denies them with "Data type not supported" otherwise.
+ if transportSize.Code() == 0x00 {
+ numElements = numElements * uint16(transportSize.SizeInBytes())
+ transportSize = readWriteModel.TransportSize_BYTE
}
return readWriteModel.NewS7AddressAny(
transportSize,
@@ -302,9 +289,63 @@ func encodeS7Address(tag apiModel.PlcTag) (readWriteModel.S7Address, error) {
), nil
}
-func parsePlcValue(ctx context.Context, tag PlcTag, data []byte) (apiValues.PlcValue, error) {
- // TODO: port over
- panic("not implemented yet")
+func parsePlcValue(ctx context.Context, tag PlcTag, data []byte, controllerType readWriteModel.ControllerType) (apiValues.PlcValue, error) {
+ stringLength := int32(254)
+ if s7StringTag, ok := tag.(PlcStringTag); ok {
+ stringLength = int32(s7StringTag.stringLength)
+ }
+ numElements := tag.GetNumElements()
+ dataProtocolId := tag.GetDataType().DataProtocolId()
+ if numElements == 1 {
+ return readWriteModel.DataItemParse(ctx, data, dataProtocolId, controllerType, stringLength)
+ }
+ switch tag.GetDataType() {
+ case readWriteModel.TransportSize_BYTE:
+ return spiValues.NewPlcRawByteArray(data), nil
+ case readWriteModel.TransportSize_BOOL:
+ // BOOL arrays arrive as packed bytes (LSB first), matching the write-side packing.
+ items := make([]apiValues.PlcValue, 0, numElements)
+ for i := uint16(0); i < numElements; i++ {
+ if int(i/8) >= len(data) {
+ return nil, errors.Errorf("BOOL array response too short: got %d bytes for %d elements", len(data), numElements)
+ }
+ items = append(items, spiValues.NewPlcBOOL((data[i/8]>>(i%8))&0x01 != 0))
+ }
+ return spiValues.NewPlcList(items), nil
+ default:
+ readBuffer := utils.NewReadBufferByteBased(data)
+ items := make([]apiValues.PlcValue, 0, numElements)
+ for i := uint16(0); i < numElements; i++ {
+ item, err := readWriteModel.DataItemParseWithBuffer(ctx, readBuffer, dataProtocolId, controllerType, stringLength)
+ if err != nil {
+ return nil, errors.Wrapf(err, "error parsing element %d of type %s", i, tag.GetDataType())
+ }
+ items = append(items, item)
+ }
+ return spiValues.NewPlcList(items), nil
+ }
+}
+
+// mapPlcErrorCode maps S7 header error class/code pairs to PLC4X response codes (ported from plc4j).
+func mapPlcErrorCode(localLog zerolog.Logger, errorClass uint8, errorCode uint8) apiModel.PlcResponseCode {
+ // 0x81/4 means PUT/GET is disabled. An S7-300 reports the same refusal as 0x83/4 (GH-599),
+ // and error class 0x85 covers further access-denied situations.
+ if (errorClass == 0x81 && errorCode == 4) || (errorClass == 0x83 && errorCode == 4) {
+ localLog.Warn().Msg("Got an error response from the PLC. This particular response code usually indicates " +
+ "that PUT/GET is not enabled on the PLC.")
+ return apiModel.PlcResponseCode_ACCESS_DENIED
+ }
+ if errorClass == 0x85 {
+ return apiModel.PlcResponseCode_ACCESS_DENIED
+ }
+ localLog.Warn().
+ Uint8("errorClass", errorClass).
+ Uint8("errorCode", errorCode).
+ Msg("Got an unknown error response from the PLC. " +
+ "We probably need to implement explicit handling for this, so please file a bug-report " +
+ "on https://github.com/apache/plc4x/issues and ideally attach a WireShark dump " +
+ "containing a capture of the communication.")
+ return apiModel.PlcResponseCode_INTERNAL_ERROR
}
// Helper to convert the return codes returned from the S7 into one of our standard
@@ -312,12 +353,12 @@ func decodeResponseCode(dataTransportErrorCode readWriteModel.DataTransportError
switch dataTransportErrorCode {
case readWriteModel.DataTransportErrorCode_OK:
return apiModel.PlcResponseCode_OK
- case readWriteModel.DataTransportErrorCode_NOT_FOUND:
- return apiModel.PlcResponseCode_NOT_FOUND
- case readWriteModel.DataTransportErrorCode_INVALID_ADDRESS:
+ case readWriteModel.DataTransportErrorCode_NOT_FOUND, readWriteModel.DataTransportErrorCode_INVALID_ADDRESS:
return apiModel.PlcResponseCode_INVALID_ADDRESS
case readWriteModel.DataTransportErrorCode_DATA_TYPE_NOT_SUPPORTED:
return apiModel.PlcResponseCode_INVALID_DATATYPE
+ case readWriteModel.DataTransportErrorCode_ACCESS_DENIED:
+ return apiModel.PlcResponseCode_ACCESS_DENIED
default:
return apiModel.PlcResponseCode_INTERNAL_ERROR
}
diff --git a/plc4go/internal/s7/Reader_test.go b/plc4go/internal/s7/Reader_test.go
new file mode 100644
index 0000000000..f327d8dc17
--- /dev/null
+++ b/plc4go/internal/s7/Reader_test.go
@@ -0,0 +1,124 @@
+/*
+ * 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 s7
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+)
+
+func TestParsePlcValue(t *testing.T) {
+ ctx := t.Context()
+ anyController := readWriteModel.ControllerType_ANY
+
+ t.Run("single BOOL", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 1, readWriteModel.TransportSize_BOOL)
+ value, err := parsePlcValue(ctx, tag, []byte{0x01}, anyController)
+ require.NoError(t, err)
+ assert.True(t, value.GetBool())
+ })
+ t.Run("single INT", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 1, readWriteModel.TransportSize_INT)
+ value, err := parsePlcValue(ctx, tag, []byte{0x00, 0x04}, anyController)
+ require.NoError(t, err)
+ assert.Equal(t, int16(4), value.GetInt16())
+ })
+ t.Run("single REAL", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 1, readWriteModel.TransportSize_REAL)
+ value, err := parsePlcValue(ctx, tag, []byte{0x41, 0x28, 0x00, 0x00}, anyController)
+ require.NoError(t, err)
+ assert.Equal(t, float32(10.5), value.GetFloat32())
+ })
+ t.Run("single STRING with fixed length", func(t *testing.T) {
+ tag := NewStringTag(readWriteModel.MemoryArea_DATA_BLOCKS, 1, 0, 0, 1, 5, readWriteModel.TransportSize_STRING)
+ // maxLen=5, curLen=3, "ABC" + 2 bytes padding
+ value, err := parsePlcValue(ctx, tag, []byte{0x05, 0x03, 'A', 'B', 'C', 0x00, 0x00}, anyController)
+ require.NoError(t, err)
+ assert.Equal(t, "ABC", value.GetString())
+ })
+ t.Run("INT array", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 2, readWriteModel.TransportSize_INT)
+ value, err := parsePlcValue(ctx, tag, []byte{0x00, 0x01, 0x00, 0x02}, anyController)
+ require.NoError(t, err)
+ require.True(t, value.IsList())
+ list := value.GetList()
+ require.Len(t, list, 2)
+ assert.Equal(t, int16(1), list[0].GetInt16())
+ assert.Equal(t, int16(2), list[1].GetInt16())
+ })
+ t.Run("BYTE array stays raw", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 4, readWriteModel.TransportSize_BYTE)
+ value, err := parsePlcValue(ctx, tag, []byte{0xDE, 0xAD, 0xBE, 0xEF}, anyController)
+ require.NoError(t, err)
+ assert.Equal(t, []byte{0xDE, 0xAD, 0xBE, 0xEF}, value.GetRaw())
+ })
+ t.Run("BOOL array unpacks bits LSB first", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 10, readWriteModel.TransportSize_BOOL)
+ value, err := parsePlcValue(ctx, tag, []byte{0b00000011, 0b00000010}, anyController)
+ require.NoError(t, err)
+ require.True(t, value.IsList())
+ list := value.GetList()
+ require.Len(t, list, 10)
+ assert.True(t, list[0].GetBool())
+ assert.True(t, list[1].GetBool())
+ assert.False(t, list[2].GetBool())
+ assert.False(t, list[8].GetBool())
+ assert.True(t, list[9].GetBool())
+ })
+}
+
+func TestEncodeS7Address(t *testing.T) {
+ t.Run("S5TIME is requested as byte array", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 1, readWriteModel.TransportSize_S5TIME)
+ address, err := encodeS7Address(tag)
+ require.NoError(t, err)
+ addressAny := address.(readWriteModel.S7AddressAny)
+ assert.Equal(t, readWriteModel.TransportSize_BYTE, addressAny.GetTransportSize())
+ assert.Equal(t, uint16(2), addressAny.GetNumberOfElements())
+ })
+ t.Run("TIME is requested as byte array", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 1, readWriteModel.TransportSize_TIME)
+ address, err := encodeS7Address(tag)
+ require.NoError(t, err)
+ addressAny := address.(readWriteModel.S7AddressAny)
+ assert.Equal(t, readWriteModel.TransportSize_BYTE, addressAny.GetTransportSize())
+ assert.Equal(t, uint16(4), addressAny.GetNumberOfElements())
+ })
+ t.Run("BOOL array is requested packed", func(t *testing.T) {
+ tag := NewTag(readWriteModel.MemoryArea_FLAGS_MARKERS, 0, 0, 0, 10, readWriteModel.TransportSize_BOOL)
+ address, err := encodeS7Address(tag)
+ require.NoError(t, err)
+ addressAny := address.(readWriteModel.S7AddressAny)
+ assert.Equal(t, readWriteModel.TransportSize_BYTE, addressAny.GetTransportSize())
+ assert.Equal(t, uint16(2), addressAny.GetNumberOfElements())
+ })
+ t.Run("STRING converts to CHAR elements", func(t *testing.T) {
+ tag := NewStringTag(readWriteModel.MemoryArea_DATA_BLOCKS, 1, 0, 0, 1, 20, readWriteModel.TransportSize_STRING)
+ address, err := encodeS7Address(tag)
+ require.NoError(t, err)
+ addressAny := address.(readWriteModel.S7AddressAny)
+ assert.Equal(t, readWriteModel.TransportSize_CHAR, addressAny.GetTransportSize())
+ assert.Equal(t, uint16(22), addressAny.GetNumberOfElements())
+ })
+}
diff --git a/plc4go/protocols/s7/readwrite/model/StaticHelper.go b/plc4go/protocols/s7/readwrite/model/StaticHelper.go
index 06157442ca..165dd8982b 100644
--- a/plc4go/protocols/s7/readwrite/model/StaticHelper.go
+++ b/plc4go/protocols/s7/readwrite/model/StaticHelper.go
@@ -21,11 +21,14 @@ package model
import (
"context"
+ "strings"
"time"
+ "unicode/utf16"
"github.com/apache/plc4x/plc4go/pkg/api/values"
"github.com/apache/plc4x/plc4go/spi/errors"
"github.com/apache/plc4x/plc4go/spi/utils"
+ spiValues "github.com/apache/plc4x/plc4go/spi/values"
)
func ParseTiaTime(ctx context.Context, io utils.ReadBuffer) (uint32, error) {
@@ -44,20 +47,45 @@ func SerializeTiaTime(ctx context.Context, io utils.WriteBuffer, value values.Pl
return nil
}
+// ParseS5Time decodes the BCD encoded S5TIME format: the high nibble selects the time base
+// (10ms/100ms/1s/10s), the lower three nibbles are a three digit BCD counter.
func ParseS5Time(ctx context.Context, io utils.ReadBuffer) (uint32, error) {
- /*try {
- int stuff = io.readInt(16);
- // TODO: Implement this correctly.
- throw new NotImplementedException("S5TIME not implemented");
- } catch (ParseException e) {
- return null;
- }*/
- return 0, nil
+ s5time, err := io.ReadUint16("s5time", 16)
+ if err != nil {
+ return 0, errors.Wrap(err, "Error parsing s5time")
+ }
+ timeValue := uint32(s5time&0x000F) + uint32((s5time&0x00F0)>>4)*10 + uint32((s5time&0x0F00)>>8)*100
+ timeBase := uint32(10)
+ for i := uint16(0); i < (s5time&0xF000)>>12; i++ {
+ timeBase *= 10
+ }
+ totalMs := timeValue * timeBase
+ if totalMs > 9990000 {
+ totalMs = 9990000
+ }
+ return totalMs, nil
}
func SerializeS5Time(ctx context.Context, io utils.WriteBuffer, value values.PlcValue) error {
- //throw new NotImplementedException("Serializing S5TIME not implemented");
- return nil
+ totalMs := value.GetDuration().Milliseconds()
+ var timeBase, timeValue int64
+ switch {
+ case totalMs < 0 || totalMs > 9990000:
+ // Out of the representable range, S5TIME stays 0.
+ case totalMs <= 9990:
+ timeBase, timeValue = 0, totalMs/10
+ case totalMs <= 99900:
+ timeBase, timeValue = 1, totalMs/100
+ case totalMs <= 999000:
+ timeBase, timeValue = 2, totalMs/1000
+ default:
+ timeBase, timeValue = 3, totalMs/10000
+ }
+ units := timeValue % 10
+ tens := (timeValue / 10) % 10
+ hundreds := (timeValue / 100) % 10
+ s5time := uint16(timeBase<<12 | hundreds<<8 | tens<<4 | units)
+ return io.WriteUint16("s5time", 16, s5time)
}
func ParseTiaLTime(ctx context.Context, io utils.ReadBuffer) (uint32, error) {
@@ -86,19 +114,29 @@ func SerializeTiaTimeOfDay(ctx context.Context, io utils.WriteBuffer, value valu
return nil
}
+// daysBetweenUnixAndSiemensEpoch is the offset between 1970-01-01 and the Siemens epoch 1990-01-01.
+const daysBetweenUnixAndSiemensEpoch = 7305
+
func ParseTiaDate(ctx context.Context, io utils.ReadBuffer) (uint16, error) {
- /*try {
- int daysSince1990 = io.readUnsignedInt(16);
- return LocalDate.now().withYear(1990).withDayOfMonth(1).withMonth(1).plus(daysSince1990, ChronoUnit.DAYS);
- } catch (ParseException e) {
- return null;
- }*/
- return 0, nil
+ daysSinceSiemensEpoch, err := io.ReadUint16("daysSinceSiemensEpoch", 16)
+ if err != nil {
+ return 0, errors.Wrap(err, "Error parsing daysSinceSiemensEpoch")
+ }
+ if daysSinceSiemensEpoch > 0xFFFF-daysBetweenUnixAndSiemensEpoch {
+ return 0xFFFF, nil
+ }
+ return daysSinceSiemensEpoch + daysBetweenUnixAndSiemensEpoch, nil
}
func SerializeTiaDate(ctx context.Context, io utils.WriteBuffer, value values.PlcValue) error {
- //throw new NotImplementedException("Serializing DATE not implemented");
- return nil
+ var daysSinceSiemensEpoch uint16
+ if date, ok := value.(spiValues.PlcDATE); ok {
+ daysSinceSiemensEpoch = date.GetDaysSinceSiemensEpoch()
+ } else {
+ siemensEpoch := time.Date(1990, time.January, 1, 0, 0, 0, 0, time.UTC)
+ daysSinceSiemensEpoch = uint16(value.GetDate().Sub(siemensEpoch).Hours() / 24)
+ }
+ return io.WriteUint16("daysSinceSiemensEpoch", 16, daysSinceSiemensEpoch)
}
func ParseTiaDateTime(ctx context.Context, io utils.ReadBuffer) (time.Time, error) {
@@ -133,26 +171,103 @@ func serializeTiaDate(ctx context.Context, io utils.WriteBuffer, value values.Pl
return nil
}
+// ParseS7String reads an S7 STRING/WSTRING: a max-length prefix, a current-length prefix and
+// max-length characters of which only the first current-length carry data. Up to stringLength
+// characters are consumed from the buffer so trailing padding is gobbled up as well.
func ParseS7String(ctx context.Context, io utils.ReadBuffer, stringLength int32, encoding string) (string, error) {
- var multiplier int32
- switch encoding {
- case "UTF8":
- multiplier = 8
- case "UTF16":
- multiplier = 16
+ switch {
+ case strings.EqualFold(encoding, "UTF8"):
+ if _, err := io.ReadUint8("maxLength", 8); err != nil {
+ return "", errors.Wrap(err, "Error parsing max length")
+ }
+ totalStringLength, err := io.ReadUint8("totalStringLength", 8)
+ if err != nil {
+ return "", errors.Wrap(err, "Error parsing total string length")
+ }
+ data := make([]byte, 0, totalStringLength)
+ for i := int32(0); i < stringLength && io.HasMore(8); i++ {
+ curByte, err := io.ReadUint8("", 8)
+ if err != nil {
+ return "", errors.Wrap(err, "Error parsing character")
+ }
+ if i < int32(totalStringLength) {
+ data = append(data, curByte)
+ }
+ }
+ return string(data), nil
+ case strings.EqualFold(encoding, "UTF16") || strings.EqualFold(encoding, "UTF16BE"):
+ if _, err := io.ReadUint16("maxLength", 16); err != nil {
+ return "", errors.Wrap(err, "Error parsing max length")
+ }
+ totalStringLength, err := io.ReadUint16("totalStringLength", 16)
+ if err != nil {
+ return "", errors.Wrap(err, "Error parsing total string length")
+ }
+ units := make([]uint16, 0, totalStringLength)
+ for i := int32(0); i < stringLength && io.HasMore(16); i++ {
+ curUnit, err := io.ReadUint16("", 16)
+ if err != nil {
+ return "", errors.Wrap(err, "Error parsing character")
+ }
+ if i < int32(totalStringLength) {
+ units = append(units, curUnit)
+ }
+ }
+ return string(utf16.Decode(units)), nil
+ default:
+ return "", errors.Errorf("unsupported string encoding %s", encoding)
}
- return io.ReadString("", uint32(stringLength*multiplier), utils.WithEncoding(encoding))
}
+// SerializeS7String writes an S7 STRING/WSTRING: [maxLen][curLen][maxLen chars, zero padded].
func SerializeS7String(ctx context.Context, io utils.WriteBuffer, value values.PlcValue, stringLength int32, encoding string) error {
- var multiplier int32
- switch encoding {
- case "UTF8":
- multiplier = 8
- case "UTF16":
- multiplier = 16
+ switch {
+ case strings.EqualFold(encoding, "UTF8"):
+ maxStringLength := int(stringLength)
+ if maxStringLength > 254 {
+ maxStringLength = 254
+ }
+ data := []byte(value.GetString())
+ if len(data) > maxStringLength {
+ data = data[:maxStringLength]
+ }
+ if err := io.WriteUint8("maxLength", 8, uint8(maxStringLength)); err != nil {
+ return errors.Wrap(err, "Error serializing max length")
+ }
+ if err := io.WriteUint8("totalStringLength", 8, uint8(len(data))); err != nil {
+ return errors.Wrap(err, "Error serializing total string length")
+ }
+ padded := make([]byte, maxStringLength)
+ copy(padded, data)
+ return io.WriteByteArray("chars", padded)
+ case strings.EqualFold(encoding, "UTF16") || strings.EqualFold(encoding, "UTF16BE"):
+ maxStringLength := int(stringLength)
+ if maxStringLength > 16382 {
+ maxStringLength = 16382
+ }
+ units := utf16.Encode([]rune(value.GetString()))
+ if len(units) > maxStringLength {
+ units = units[:maxStringLength]
+ }
+ if err := io.WriteUint16("maxLength", 16, uint16(maxStringLength)); err != nil {
+ return errors.Wrap(err, "Error serializing max length")
+ }
+ if err := io.WriteUint16("totalStringLength", 16, uint16(len(units))); err != nil {
+ return errors.Wrap(err, "Error serializing total string length")
+ }
+ for i := 0; i < maxStringLength; i++ {
+ var unit uint16
+ if i < len(units) {
+ unit = units[i]
+ }
+ if err := io.WriteUint16("", 16, unit); err != nil {
+ return errors.Wrap(err, "Error serializing character")
+ }
+ }
+ return nil
+ default:
+ return errors.Errorf("unsupported string encoding %s", encoding)
}
- return io.WriteString("", uint32(stringLength*multiplier), value.GetString(), utils.WithEncoding(encoding))
}
func ParseS7Char(ctx context.Context, io utils.ReadBuffer, encoding string) (uint8, error) {
@@ -163,16 +278,38 @@ func SerializeS7Char(ctx context.Context, io utils.WriteBuffer, value values.Plc
return io.WriteUint8("", 8, value.GetUint8())
}
+// RightShift3 reads a 16 bit length field which is expressed in bits for the numeric transport
+// sizes and in bytes for the octet-string like ones.
func RightShift3(ctx context.Context, readBuffer utils.ReadBuffer, dataTransportSize DataTransportSize) (any, error) {
- return uint16(0), nil
+ value, err := readBuffer.ReadUint16("valueLength", 16)
+ if err != nil {
+ return uint16(0), errors.Wrap(err, "Error parsing value length")
+ }
+ if dataTransportSize == DataTransportSize_OCTET_STRING ||
+ dataTransportSize == DataTransportSize_REAL ||
+ dataTransportSize == DataTransportSize_BIT {
+ return value, nil
+ }
+ return value >> 3, nil
}
func LeftShift3(ctx context.Context, writeBuffer utils.WriteBuffer, valueLength uint16) error {
- return nil
+ return writeBuffer.WriteUint16("valueLength", 16, valueLength<<3)
}
+// EventItemLength accounts for the pad byte after odd-length event payload items (as long as
+// the buffer actually still contains it).
func EventItemLength(ctx context.Context, readBuffer utils.ReadBuffer, valueLength uint16) uint16 {
- return 0
+ if valueLength%2 == 0 {
+ return valueLength
+ }
+ if rb, ok := readBuffer.(utils.ReadBufferByteBased); ok {
+ remainingBytes := rb.GetTotalBytes() - uint64(rb.GetPos())
+ if remainingBytes < uint64(valueLength)+1 {
+ return valueLength
+ }
+ }
+ return valueLength + 1
}
func BcdToInt(ctx context.Context, readBuffer utils.ReadBuffer) (any, error) {
(plc4x) 05/09: fix(plc4go/s7): restore SZL controller-type detection broken since the s7h mspec refactor
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 3aef13b873b876c186c6a93a37ad6e0b5ac962f1
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 12:03:11 2026 +0200
fix(plc4go/s7): restore SZL controller-type detection broken since the s7h mspec refactor
The detection code was stubbed with an 'articleNumber := "broken at the
moment"' placeholder after the mspec changed the ReadSzl response to a
raw byte blob. Port the plc4j probe flow: try the COMPONENT_IDENTIFICATION
SZL first (S7-1200/1500), fall back to MODULE_IDENTIFICATION (S7-300/400),
scan the raw bytes for the order number or CPU model name, and derive the
controller family from it. A pinned controller-type skips the probe.
Probe failures no longer fail the connection (LOGO and S7-200 devices
don't answer SZL requests) - they just disable the UserData services via
a new DriverContext.UserDataServicesSupported flag, which will gate
browse and subscriptions. The connection now also exposes article number,
controller type, PDU size and AMQ values as connection attributes.
---
plc4go/internal/s7/Connection.go | 217 ++++++++++----------------
plc4go/internal/s7/DriverContext.go | 9 +-
plc4go/internal/s7/UserDataService.go | 241 +++++++++++++++++++++++++++++
plc4go/internal/s7/UserDataService_test.go | 118 ++++++++++++++
4 files changed, 450 insertions(+), 135 deletions(-)
diff --git a/plc4go/internal/s7/Connection.go b/plc4go/internal/s7/Connection.go
index e8e510c8d6..b63c2c0d26 100644
--- a/plc4go/internal/s7/Connection.go
+++ b/plc4go/internal/s7/Connection.go
@@ -22,7 +22,7 @@ package s7
import (
"context"
"fmt"
- "strings"
+ "strconv"
"sync"
"github.com/rs/zerolog"
@@ -246,150 +246,94 @@ func (c *Connection) setupConnection(ctx context.Context) error {
// requests.
c.tm.SetNumberOfConcurrentRequests(int(c.driverContext.MaxAmqCallee))
- // If the controller type is explicitly set, were finished with the login
- // process. If it's set to ANY, we have to query the serial number information
- // in order to detect the type of PLC.
- if c.driverContext.ControllerType != readWriteModel.ControllerType_ANY {
- // Send an event that connection setup is complete.
- c.SetConnected(true)
- return nil
- }
+ // Detect the controller type (unless it was pinned by the user) and derive
+ // whether the device supports the S7Comm UserData services. Probe failures are
+ // not connection failures - plain read/write stays usable (e.g. LOGO devices).
+ c.identifyRemote(ctx)
- // Prepare a message to request the remote to identify itself.
- c.log.Debug().Msg("Sending S7 Identification Request")
- s7IdentificationResult := make(chan readWriteModel.S7PayloadUserData, 1)
- s7IdentificationErrorChan := make(chan error, 1)
- if err := c.messageCodec.SendRequest(ctx, "setup_connection_identify_remote_message", c.createIdentifyRemoteMessage(), func(message spi.Message) bool {
- tpktPacket, ok := message.(readWriteModel.TPKTPacket)
- if !ok {
- return false
- }
- cotpPacketData, ok := tpktPacket.GetPayload().(readWriteModel.COTPPacketData)
- if !ok {
- return false
- }
- messageUserData, ok := cotpPacketData.GetPayload().(readWriteModel.S7MessageUserData)
- if !ok {
- return false
- }
- _, ok = messageUserData.GetPayload().(readWriteModel.S7PayloadUserData)
- return ok
- }, func(message spi.Message) error {
- tpktPacket := message.(readWriteModel.TPKTPacket)
- cotpPacketData := tpktPacket.GetPayload().(readWriteModel.COTPPacketData)
- messageUserData := cotpPacketData.GetPayload().(readWriteModel.S7MessageUserData)
- s7IdentificationResult <- messageUserData.GetPayload().(readWriteModel.S7PayloadUserData)
- return nil
- }, func(err error) error {
- // If this is a timeout, do a check if the connection requires a reconnection
- var timeoutError utils.TimeoutError
- if errors.As(err, &timeoutError) {
- c.log.Warn().Msg("Timeout during Connection establishing, closing channel...")
- c.Close()
- }
- s7IdentificationErrorChan <- errors.Wrap(err, "got error processing request")
- return nil
- }); err != nil {
- return errors.Wrap(err, "Error during sending of identify remote Request")
- }
- select {
- case payloadUserData := <-s7IdentificationResult:
- c.log.Debug().Msg("Got S7 Identification Response")
- if err := c.extractControllerTypeAndFireConnected(payloadUserData); err != nil {
- return errors.Wrap(err, "Error extracting controller type")
- }
- case err := <-s7IdentificationErrorChan:
- return errors.Wrap(err, "Error receiving identify remote Request")
- }
+ // Send an event that connection setup is complete.
+ c.SetConnected(true)
+ return nil
case err := <-s7ConnectionErrorChan:
return errors.Wrap(err, "Error receiving S7 Connection Request")
}
case err := <-cotpConnectionErrorChan:
return errors.Wrap(err, "Error receiving of COTP Connection Request")
}
- return nil
}
-func (c *Connection) extractControllerTypeAndFireConnected(payloadUserData readWriteModel.S7PayloadUserData) error {
- // TODO: how do we handle the case if there no items at all? Should we assume it a successful or failure...
- // TODO ... opposed to the java implementation we treat it as a failure
- for _, item := range payloadUserData.GetItems() {
- switch readSzlResponseItem := item.(type) {
- case readWriteModel.S7PayloadUserDataItemCpuFunctionReadSzlResponse:
- for _, readSzlResponseItemItem := range readSzlResponseItem.GetItems() {
- _ = readSzlResponseItemItem
- /* TODO: broken by mspec changes from carcia
- if readSzlResponseItemItem.GetItemIndex() != 0x0001 {
- continue
- }
- articleNumber := string(readSzlResponseItemItem.GetMlfb())
- */
- articleNumber := "broken at the moment"
- var controllerType readWriteModel.ControllerType
- if !strings.HasPrefix(articleNumber, "6ES7 ") {
- controllerType = readWriteModel.ControllerType_ANY
- }
- blankIndex := strings.Index(articleNumber, " ")
- model := articleNumber[blankIndex+1 : blankIndex+2]
- switch model {
- case "2":
- controllerType = readWriteModel.ControllerType_S7_1200
- case "5":
- controllerType = readWriteModel.ControllerType_S7_1500
- case "3":
- controllerType = readWriteModel.ControllerType_S7_300
- case "4":
- controllerType = readWriteModel.ControllerType_S7_400
- default:
- c.log.Info().Str("articleNumber", articleNumber).Msg("Looking up unknown article number")
- controllerType = readWriteModel.ControllerType_ANY
- }
- c.driverContext.ControllerType = controllerType
-
- // Send an event that connection setup is complete.
- c.SetConnected(true)
- return nil
- }
- }
+// identifyRemote determines the controller type and UserData-service capability. If the user
+// pinned the controller type it is trusted and no SZL probe runs; otherwise the modern
+// COMPONENT_IDENTIFICATION SZL is tried first with the legacy MODULE_IDENTIFICATION as
+// fallback. Failure of both just disables the UserData services (browse/alarms/cyclic).
+func (c *Connection) identifyRemote(ctx context.Context) {
+ if c.driverContext.ControllerType != readWriteModel.ControllerType_ANY {
+ c.driverContext.UserDataServicesSupported = supportsUserDataServices(c.driverContext.ControllerType)
+ c.log.Info().
+ Stringer("controllerType", c.driverContext.ControllerType).
+ Bool("userDataServices", c.driverContext.UserDataServicesSupported).
+ Msg("Controller type pinned, skipping SZL probe")
+ return
+ }
+ c.log.Debug().Msg("Sending S7 Identification Request")
+ articleNumber, controllerType, err := c.trySzlProbe(ctx, szlIdComponentIdentification(), 0x0001)
+ if err != nil {
+ c.log.Debug().Err(err).Msg("COMPONENT_IDENTIFICATION SZL probe failed, trying MODULE_IDENTIFICATION")
+ articleNumber, controllerType, err = c.trySzlProbe(ctx, szlIdModuleIdentification(), 0x0000)
+ }
+ if err != nil {
+ // Common on LOGO and other non-SZL devices.
+ c.log.Info().Err(err).Msg("SZL probe yielded no usable identification; " +
+ "S7Comm UserData services disabled for this device")
+ c.driverContext.UserDataServicesSupported = false
+ return
}
- return errors.New("Coudln't find the required information")
+ c.driverContext.ArticleNumber = articleNumber
+ c.driverContext.ControllerType = controllerType
+ c.driverContext.UserDataServicesSupported = true
+ c.log.Info().
+ Str("articleNumber", articleNumber).
+ Stringer("controllerType", controllerType).
+ Msg("SZL probe ok")
}
-func (c *Connection) createIdentifyRemoteMessage() readWriteModel.TPKTPacket {
- identifyRemoteMessage := readWriteModel.NewS7MessageUserData(
- 1,
- readWriteModel.NewS7ParameterUserData(
- []readWriteModel.S7ParameterUserDataItem{
- readWriteModel.NewS7ParameterUserDataItemCPUFunctions(
- 0x11,
- 0x4,
- 0x4,
- 0x01,
- 0x00,
- nil,
- nil,
- nil,
- ),
- },
- ),
- readWriteModel.NewS7PayloadUserData(
- []readWriteModel.S7PayloadUserDataItem{
- readWriteModel.NewS7PayloadUserDataItemCpuFunctionReadSzlRequest(
- readWriteModel.DataTransportErrorCode_OK,
- readWriteModel.DataTransportSize_OCTET_STRING,
- 4,
- readWriteModel.NewSzlId(
- readWriteModel.SzlModuleTypeClass_CPU,
- 0x00,
- readWriteModel.SzlSublist_MODULE_IDENTIFICATION,
- ),
- 0x0000,
- ),
- },
- ),
- )
- cotpPacketData := readWriteModel.NewCOTPPacketData(nil, identifyRemoteMessage, true, 2)
- return readWriteModel.NewTPKTPacket(cotpPacketData)
+func (c *Connection) trySzlProbe(ctx context.Context, szlId readWriteModel.SzlId, szlIndex uint16) (string, readWriteModel.ControllerType, error) {
+ tpduId := c.tpduGenerator.getAndIncrement()
+ resultChan := make(chan readWriteModel.S7Message, 1)
+ errChan := make(chan error, 1)
+ if err := c.messageCodec.SendRequest(ctx, "setup_connection_identify_remote_message", buildSzlRequest(tpduId, szlId, szlIndex), func(message spi.Message) bool {
+ tpktPacket, ok := message.(readWriteModel.TPKTPacket)
+ if !ok {
+ return false
+ }
+ cotpPacketData, ok := tpktPacket.GetPayload().(readWriteModel.COTPPacketData)
+ if !ok {
+ return false
+ }
+ messageUserData, ok := cotpPacketData.GetPayload().(readWriteModel.S7MessageUserData)
+ if !ok {
+ return false
+ }
+ return messageUserData.GetTpduReference() == tpduId
+ }, func(message spi.Message) error {
+ tpktPacket := message.(readWriteModel.TPKTPacket)
+ cotpPacketData := tpktPacket.GetPayload().(readWriteModel.COTPPacketData)
+ resultChan <- cotpPacketData.GetPayload()
+ return nil
+ }, func(err error) error {
+ errChan <- err
+ return nil
+ }); err != nil {
+ return "", 0, errors.Wrap(err, "error sending SZL identification request")
+ }
+ select {
+ case message := <-resultChan:
+ return parseSzlProbeResponse(message)
+ case err := <-errChan:
+ return "", 0, err
+ case <-ctx.Done():
+ return "", 0, ctx.Err()
+ }
}
func (c *Connection) createS7ConnectionRequest(cotpPacketConnectionResponse readWriteModel.COTPPacketConnectionResponse) readWriteModel.TPKTPacket {
@@ -433,6 +377,13 @@ func (c *Connection) createCOTPConnectionRequest() readWriteModel.COTPPacket {
func (c *Connection) GetMetadata() apiModel.PlcConnectionMetadata {
return &_default.DefaultConnectionMetadata{
+ ConnectionAttributes: map[string]string{
+ "article-number": c.driverContext.ArticleNumber,
+ "controller-type": c.driverContext.ControllerType.String(),
+ "pdu-size": strconv.Itoa(int(c.driverContext.PduSize)),
+ "max-amq-caller": strconv.Itoa(int(c.driverContext.MaxAmqCaller)),
+ "max-amq-callee": strconv.Itoa(int(c.driverContext.MaxAmqCallee)),
+ },
ProvidesReading: true,
ProvidesWriting: true,
}
diff --git a/plc4go/internal/s7/DriverContext.go b/plc4go/internal/s7/DriverContext.go
index 919d0d45df..5d99899f34 100644
--- a/plc4go/internal/s7/DriverContext.go
+++ b/plc4go/internal/s7/DriverContext.go
@@ -32,8 +32,13 @@ type DriverContext struct {
MaxAmqCaller uint16
MaxAmqCallee uint16
ControllerType model.ControllerType
- awaitSetupComplete bool
- awaitDisconnectComplete bool
+ ArticleNumber string
+ // UserDataServicesSupported reports whether the device speaks the S7Comm UserData
+ // services (browse, alarms, cyclic subscriptions). Derived from the SZL probe at
+ // connect time or from a pinned controller type.
+ UserDataServicesSupported bool
+ awaitSetupComplete bool
+ awaitDisconnectComplete bool
}
func NewDriverContext(configuration Configuration) (DriverContext, error) {
diff --git a/plc4go/internal/s7/UserDataService.go b/plc4go/internal/s7/UserDataService.go
new file mode 100644
index 0000000000..9045de5568
--- /dev/null
+++ b/plc4go/internal/s7/UserDataService.go
@@ -0,0 +1,241 @@
+/*
+ * 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 s7
+
+import (
+ "strings"
+
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi/errors"
+)
+
+// Helpers for the S7Comm UserData services (SZL identification, block functions,
+// message subscriptions, cyclic services). Ported from the plc4j userdata services.
+
+// newCpuFunctionsRequestParameter builds the CPUFunctions parameter for a UserData request.
+func newCpuFunctionsRequestParameter(cpuFunctionGroup uint8, cpuSubfunction uint8) readWriteModel.S7ParameterUserDataItem {
+ return readWriteModel.NewS7ParameterUserDataItemCPUFunctions(
+ 0x11, // method: request
+ 0x04, // cpuFunctionType: request
+ cpuFunctionGroup,
+ cpuSubfunction,
+ 0x00, // sequenceNumber
+ nil,
+ nil,
+ nil,
+ )
+}
+
+// newUserDataMessage wraps a single parameter/payload item pair into an S7MessageUserData TPKT packet.
+func newUserDataMessage(tpduId uint16, parameter readWriteModel.S7ParameterUserDataItem, payload readWriteModel.S7PayloadUserDataItem) readWriteModel.TPKTPacket {
+ message := readWriteModel.NewS7MessageUserData(
+ tpduId,
+ readWriteModel.NewS7ParameterUserData([]readWriteModel.S7ParameterUserDataItem{parameter}),
+ readWriteModel.NewS7PayloadUserData([]readWriteModel.S7PayloadUserDataItem{payload}),
+ )
+ return readWriteModel.NewTPKTPacket(readWriteModel.NewCOTPPacketData(nil, message, true, uint8(tpduId)))
+}
+
+// szlIdModuleIdentification works on S7-300/400.
+func szlIdModuleIdentification() readWriteModel.SzlId {
+ return readWriteModel.NewSzlId(readWriteModel.SzlModuleTypeClass_CPU, 0x00, readWriteModel.SzlSublist_MODULE_IDENTIFICATION)
+}
+
+// szlIdComponentIdentification is the standard for S7-1200/1500; with index 0x0001 the
+// response carries the CPU order number (MLFB) in ASCII.
+func szlIdComponentIdentification() readWriteModel.SzlId {
+ return readWriteModel.NewSzlId(readWriteModel.SzlModuleTypeClass_CPU, 0x00, readWriteModel.SzlSublist_COMPONENT_IDENTIFICATION)
+}
+
+func buildSzlRequest(tpduId uint16, szlId readWriteModel.SzlId, szlIndex uint16) readWriteModel.TPKTPacket {
+ return newUserDataMessage(
+ tpduId,
+ newCpuFunctionsRequestParameter(0x04, 0x01), // CPU functions / ReadSZL
+ readWriteModel.NewS7PayloadUserDataItemCpuFunctionReadSzlRequest(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_OCTET_STRING,
+ 4, // dataLength: SzlId(2) + SzlIndex(2)
+ szlId,
+ szlIndex,
+ ),
+ )
+}
+
+// checkUserDataErrorCode returns an error if any CPUFunctions parameter item reports a non-zero error code.
+func checkUserDataErrorCode(message readWriteModel.S7MessageUserData) error {
+ parameterUserData, ok := message.GetParameter().(readWriteModel.S7ParameterUserData)
+ if !ok {
+ return errors.Errorf("expected S7ParameterUserData, got %T", message.GetParameter())
+ }
+ for _, item := range parameterUserData.GetItems() {
+ if cpuFunctions, ok := item.(readWriteModel.S7ParameterUserDataItemCPUFunctions); ok {
+ if errorCode := cpuFunctions.GetErrorCode(); errorCode != nil && *errorCode != 0 {
+ return errors.Errorf("request rejected by PLC, errorCode=0x%x", *errorCode)
+ }
+ }
+ }
+ return nil
+}
+
+// parseSzlProbeResponse extracts the device descriptor (article number or CPU model name)
+// from an SZL identification response and derives the controller type from it.
+func parseSzlProbeResponse(message readWriteModel.S7Message) (string, readWriteModel.ControllerType, error) {
+ messageUserData, ok := message.(readWriteModel.S7MessageUserData)
+ if !ok {
+ return "", 0, errors.Errorf("expected S7MessageUserData, got %T", message)
+ }
+ if err := checkUserDataErrorCode(messageUserData); err != nil {
+ return "", 0, err
+ }
+ payloadUserData, ok := messageUserData.GetPayload().(readWriteModel.S7PayloadUserData)
+ if !ok {
+ return "", 0, errors.Errorf("expected S7PayloadUserData, got %T", messageUserData.GetPayload())
+ }
+ for _, item := range payloadUserData.GetItems() {
+ szlResponse, ok := item.(readWriteModel.S7PayloadUserDataItemCpuFunctionReadSzlResponse)
+ if !ok {
+ continue
+ }
+ data := szlResponse.GetItems()
+ if len(data) < 4 {
+ return "", 0, errors.Errorf("SZL response too short: %d bytes", len(data))
+ }
+ // Best-effort identification: the order-number prefix first, then the model-name
+ // pattern (e.g. "CPU 315-2 PN/DP" on S7-300). A structurally valid response without
+ // either still proves the device speaks UserData services.
+ descriptor := findArticleNumber(data)
+ if descriptor == "" {
+ descriptor = findCpuModelName(data)
+ }
+ return descriptor, decodeControllerType(descriptor), nil
+ }
+ return "", 0, errors.New("no SZL response item in payload")
+}
+
+// findArticleNumber scans for a printable ASCII run starting with a Siemens order-number
+// prefix ("6ES7"/"6GK") anywhere in the raw SZL item bytes.
+func findArticleNumber(data []byte) string {
+ for start := 0; start+12 <= len(data); start++ {
+ if data[start] != '6' {
+ continue
+ }
+ end := start
+ maxEnd := start + 32
+ if maxEnd > len(data) {
+ maxEnd = len(data)
+ }
+ for end < maxEnd && data[end] >= 0x20 && data[end] < 0x7F {
+ end++
+ }
+ if end-start < 12 {
+ continue
+ }
+ candidate := strings.TrimSpace(string(data[start:end]))
+ if strings.HasPrefix(candidate, "6ES7") || strings.HasPrefix(candidate, "6GK") {
+ return candidate
+ }
+ }
+ return ""
+}
+
+// findCpuModelName scans for a "CPU ..." model-name run in the raw SZL item bytes.
+func findCpuModelName(data []byte) string {
+ for start := 0; start+6 <= len(data); start++ {
+ if data[start] != 'C' || data[start+1] != 'P' || data[start+2] != 'U' || data[start+3] != ' ' {
+ continue
+ }
+ end := start
+ maxEnd := start + 32
+ if maxEnd > len(data) {
+ maxEnd = len(data)
+ }
+ for end < maxEnd && data[end] >= 0x20 && data[end] < 0x7F {
+ end++
+ }
+ candidate := strings.TrimSpace(string(data[start:end]))
+ if len(candidate) > 4 {
+ return candidate
+ }
+ }
+ return ""
+}
+
+// decodeControllerType derives the controller family from an order number
+// (e.g. "6ES7 212-1AE40-0XB0") or a model name (e.g. "CPU 315-2 PN/DP").
+func decodeControllerType(descriptor string) readWriteModel.ControllerType {
+ if strings.HasPrefix(descriptor, "6ES7") {
+ rest := strings.TrimLeft(descriptor[4:], " ")
+ if rest == "" {
+ return readWriteModel.ControllerType_ANY
+ }
+ switch rest[0] {
+ case '2':
+ return readWriteModel.ControllerType_S7_1200
+ case '5':
+ return readWriteModel.ControllerType_S7_1500
+ case '3':
+ return readWriteModel.ControllerType_S7_300
+ case '4':
+ return readWriteModel.ControllerType_S7_400
+ default:
+ return readWriteModel.ControllerType_ANY
+ }
+ }
+ if strings.HasPrefix(descriptor, "CPU ") {
+ rest := strings.TrimLeft(descriptor[4:], " ")
+ digits := 0
+ number := 0
+ for digits < len(rest) && digits < 4 && rest[digits] >= '0' && rest[digits] <= '9' {
+ number = number*10 + int(rest[digits]-'0')
+ digits++
+ }
+ if digits == 3 {
+ switch rest[0] {
+ case '3':
+ return readWriteModel.ControllerType_S7_300
+ case '4':
+ return readWriteModel.ControllerType_S7_400
+ }
+ }
+ if digits == 4 {
+ if number >= 1200 && number <= 1299 {
+ return readWriteModel.ControllerType_S7_1200
+ }
+ if number >= 1500 && number <= 1599 {
+ return readWriteModel.ControllerType_S7_1500
+ }
+ }
+ }
+ return readWriteModel.ControllerType_ANY
+}
+
+// supportsUserDataServices reports whether the controller family supports the S7Comm
+// UserData services (browse, alarms, cyclic subscriptions).
+func supportsUserDataServices(controllerType readWriteModel.ControllerType) bool {
+ switch controllerType {
+ case readWriteModel.ControllerType_S7_300,
+ readWriteModel.ControllerType_S7_400,
+ readWriteModel.ControllerType_S7_1200,
+ readWriteModel.ControllerType_S7_1500:
+ return true
+ default: // S7_200, LOGO, ANY
+ return false
+ }
+}
diff --git a/plc4go/internal/s7/UserDataService_test.go b/plc4go/internal/s7/UserDataService_test.go
new file mode 100644
index 0000000000..a14140602f
--- /dev/null
+++ b/plc4go/internal/s7/UserDataService_test.go
@@ -0,0 +1,118 @@
+/*
+ * 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 s7
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+)
+
+func TestDecodeControllerType(t *testing.T) {
+ tests := []struct {
+ descriptor string
+ expected readWriteModel.ControllerType
+ }{
+ {"6ES7 212-1AE40-0XB0", readWriteModel.ControllerType_S7_1200},
+ {"6ES7 512-1DK01-0AB0", readWriteModel.ControllerType_S7_1500},
+ {"6ES7 315-2EH14-0AB0", readWriteModel.ControllerType_S7_300},
+ {"6ES7 412-2XJ05-0AB0", readWriteModel.ControllerType_S7_400},
+ {"CPU 315-2 PN/DP", readWriteModel.ControllerType_S7_300},
+ {"CPU 412-2", readWriteModel.ControllerType_S7_400},
+ {"CPU 1212C", readWriteModel.ControllerType_S7_1200},
+ {"CPU 1516-3", readWriteModel.ControllerType_S7_1500},
+ {"", readWriteModel.ControllerType_ANY},
+ {"garbage", readWriteModel.ControllerType_ANY},
+ {"6ES7", readWriteModel.ControllerType_ANY},
+ }
+ for _, test := range tests {
+ t.Run(test.descriptor, func(t *testing.T) {
+ assert.Equal(t, test.expected, decodeControllerType(test.descriptor))
+ })
+ }
+}
+
+func TestFindArticleNumber(t *testing.T) {
+ t.Run("embedded order number", func(t *testing.T) {
+ data := append(append([]byte{0x00, 0x01, 0xFF}, []byte("6ES7 212-1AE40-0XB0")...), 0x00, 0x00)
+ assert.Equal(t, "6ES7 212-1AE40-0XB0", findArticleNumber(data))
+ })
+ t.Run("no order number", func(t *testing.T) {
+ assert.Equal(t, "", findArticleNumber([]byte("hello world, no siemens here")))
+ })
+ t.Run("model name fallback", func(t *testing.T) {
+ data := append(append([]byte{0x00, 0x01}, []byte("CPU 315-2 PN/DP")...), 0x00)
+ assert.Equal(t, "CPU 315-2 PN/DP", findCpuModelName(data))
+ })
+}
+
+func TestParseSzlProbeResponse(t *testing.T) {
+ makeResponse := func(items []byte, errorCode *uint16) readWriteModel.S7Message {
+ return readWriteModel.NewS7MessageUserData(
+ 11,
+ readWriteModel.NewS7ParameterUserData([]readWriteModel.S7ParameterUserDataItem{
+ readWriteModel.NewS7ParameterUserDataItemCPUFunctions(
+ 0x12, 0x08, 0x04, 0x01, 0x00,
+ ptr(uint8(0)), ptr(uint8(1)), errorCode,
+ ),
+ }),
+ readWriteModel.NewS7PayloadUserData([]readWriteModel.S7PayloadUserDataItem{
+ readWriteModel.NewS7PayloadUserDataItemCpuFunctionReadSzlResponse(
+ readWriteModel.DataTransportErrorCode_OK,
+ readWriteModel.DataTransportSize_OCTET_STRING,
+ uint16(len(items)),
+ items,
+ ),
+ }),
+ )
+ }
+
+ t.Run("successful identification", func(t *testing.T) {
+ // SZL header (8 bytes) followed by an item starting with a 2 byte index and the MLFB
+ items := append([]byte{0x00, 0x11, 0x00, 0x01, 0x00, 0x1C, 0x00, 0x01, 0x00, 0x01}, []byte("6ES7 512-1DK01-0AB0 ")...)
+ article, controllerType, err := parseSzlProbeResponse(makeResponse(items, ptr(uint16(0))))
+ require.NoError(t, err)
+ assert.Equal(t, "6ES7 512-1DK01-0AB0", article)
+ assert.Equal(t, readWriteModel.ControllerType_S7_1500, controllerType)
+ })
+ t.Run("plc rejects the szl id", func(t *testing.T) {
+ _, _, err := parseSzlProbeResponse(makeResponse([]byte{0x00, 0x00, 0x00, 0x00}, ptr(uint16(0xD401))))
+ assert.Error(t, err)
+ })
+ t.Run("too short payload", func(t *testing.T) {
+ _, _, err := parseSzlProbeResponse(makeResponse([]byte{0x00}, ptr(uint16(0))))
+ assert.Error(t, err)
+ })
+}
+
+func TestSupportsUserDataServices(t *testing.T) {
+ assert.True(t, supportsUserDataServices(readWriteModel.ControllerType_S7_300))
+ assert.True(t, supportsUserDataServices(readWriteModel.ControllerType_S7_1500))
+ assert.False(t, supportsUserDataServices(readWriteModel.ControllerType_LOGO))
+ assert.False(t, supportsUserDataServices(readWriteModel.ControllerType_S7_200))
+ assert.False(t, supportsUserDataServices(readWriteModel.ControllerType_ANY))
+}
+
+func ptr[T any](value T) *T {
+ return &value
+}
(plc4x) 01/09: fix(plc4go/s7): use the generated ControllerType enum instead of a colliding local copy
Posted by sr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
sruehl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
commit 6f2ac81e8491c53c6b80a6eb880e095b3de7fc0d
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Tue Aug 18 11:47:39 2026 +0200
fix(plc4go/s7): use the generated ControllerType enum instead of a colliding local copy
The hand-written enum had S7_200 and LOGO both at 0x5 and diverged from
the generated model values, so a user pinning controller-type=S7_200 got
the LOGO PDU-size clamp. Parse controller-type via ControllerTypeByName
and carry readWriteModel.ControllerType through Configuration and
DriverContext.
---
plc4go/internal/s7/Configuration.go | 24 +++-------
plc4go/internal/s7/Configuration_test.go | 75 ++++++++++++++++++++++++++++++++
plc4go/internal/s7/Connection.go | 16 +++----
plc4go/internal/s7/ControllerType.go | 33 --------------
plc4go/internal/s7/DriverContext.go | 7 +--
5 files changed, 91 insertions(+), 64 deletions(-)
diff --git a/plc4go/internal/s7/Configuration.go b/plc4go/internal/s7/Configuration.go
index 6e0cb3ad8b..25760b2698 100644
--- a/plc4go/internal/s7/Configuration.go
+++ b/plc4go/internal/s7/Configuration.go
@@ -24,6 +24,7 @@ import (
"github.com/rs/zerolog"
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
"github.com/apache/plc4x/plc4go/spi/errors"
)
@@ -35,7 +36,7 @@ type Configuration struct {
pduSize uint16
maxAmqCaller uint16
maxAmqCallee uint16
- controllerType ControllerType
+ controllerType readWriteModel.ControllerType
}
func ParseFromOptions(localLog zerolog.Logger, options map[string][]string) (Configuration, error) {
@@ -47,7 +48,7 @@ func ParseFromOptions(localLog zerolog.Logger, options map[string][]string) (Con
pduSize: 1024,
maxAmqCaller: 8,
maxAmqCallee: 8,
- controllerType: ControllerType_UNKNOWN,
+ controllerType: readWriteModel.ControllerType_ANY,
}
if localRackString := getFromOptions(localLog, options, "local-rack"); localRackString != "" {
parsedInt, err := strconv.ParseInt(localRackString, 10, 32)
@@ -78,24 +79,11 @@ func ParseFromOptions(localLog zerolog.Logger, options map[string][]string) (Con
configuration.remoteSlot = int32(parsedInt)
}
if controllerTypeString := getFromOptions(localLog, options, "controller-type"); controllerTypeString != "" {
- switch controllerTypeString {
- case "ANY":
- configuration.controllerType = ControllerType_ANY
- case "S7_200":
- configuration.controllerType = ControllerType_S7_200
- case "S7_300":
- configuration.controllerType = ControllerType_S7_300
- case "S7_400":
- configuration.controllerType = ControllerType_S7_400
- case "S7_1200":
- configuration.controllerType = ControllerType_S7_1200
- case "S7_1500":
- configuration.controllerType = ControllerType_S7_1500
- case "LOGO":
- configuration.controllerType = ControllerType_LOGO
- default:
+ controllerType, ok := readWriteModel.ControllerTypeByName(controllerTypeString)
+ if !ok {
return Configuration{}, errors.Errorf("Unknown controller type %s", controllerTypeString)
}
+ configuration.controllerType = controllerType
}
pduSizeString := getFromOptions(localLog, options, "pdu-size")
diff --git a/plc4go/internal/s7/Configuration_test.go b/plc4go/internal/s7/Configuration_test.go
new file mode 100644
index 0000000000..216bc01265
--- /dev/null
+++ b/plc4go/internal/s7/Configuration_test.go
@@ -0,0 +1,75 @@
+/*
+ * 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 s7
+
+import (
+ "testing"
+
+ "github.com/rs/zerolog"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ readWriteModel "github.com/apache/plc4x/plc4go/protocols/s7/readwrite/model"
+)
+
+func TestParseFromOptions_controllerType(t *testing.T) {
+ t.Run("default is ANY", func(t *testing.T) {
+ configuration, err := ParseFromOptions(zerolog.Nop(), map[string][]string{})
+ require.NoError(t, err)
+ assert.Equal(t, readWriteModel.ControllerType_ANY, configuration.controllerType)
+ })
+ t.Run("explicit type is parsed", func(t *testing.T) {
+ configuration, err := ParseFromOptions(zerolog.Nop(), map[string][]string{
+ "controller-type": {"S7_1200"},
+ })
+ require.NoError(t, err)
+ assert.Equal(t, readWriteModel.ControllerType_S7_1200, configuration.controllerType)
+ })
+ t.Run("LOGO is distinct from S7_200", func(t *testing.T) {
+ logoConfiguration, err := ParseFromOptions(zerolog.Nop(), map[string][]string{
+ "controller-type": {"LOGO"},
+ })
+ require.NoError(t, err)
+ s7200Configuration, err := ParseFromOptions(zerolog.Nop(), map[string][]string{
+ "controller-type": {"S7_200"},
+ })
+ require.NoError(t, err)
+ assert.NotEqual(t, logoConfiguration.controllerType, s7200Configuration.controllerType)
+ })
+ t.Run("unknown type errors", func(t *testing.T) {
+ _, err := ParseFromOptions(zerolog.Nop(), map[string][]string{
+ "controller-type": {"S7_9000"},
+ })
+ assert.Error(t, err)
+ })
+}
+
+func TestNewDriverContext_logoPduSizeOverride(t *testing.T) {
+ configuration, err := ParseFromOptions(zerolog.Nop(), map[string][]string{
+ "controller-type": {"LOGO"},
+ })
+ require.NoError(t, err)
+ driverContext, err := NewDriverContext(configuration)
+ require.NoError(t, err)
+ // LOGO devices only support small PDUs; the default of 1024 gets clamped to 480,
+ // which then snaps to the nearest COTP TPDU size minus header overhead.
+ assert.LessOrEqual(t, driverContext.PduSize, uint16(496))
+ assert.Equal(t, readWriteModel.ControllerType_LOGO, driverContext.ControllerType)
+}
diff --git a/plc4go/internal/s7/Connection.go b/plc4go/internal/s7/Connection.go
index 9d1d215fc7..98e764db38 100644
--- a/plc4go/internal/s7/Connection.go
+++ b/plc4go/internal/s7/Connection.go
@@ -249,7 +249,7 @@ func (c *Connection) setupConnection(ctx context.Context) error {
// If the controller type is explicitly set, were finished with the login
// process. If it's set to ANY, we have to query the serial number information
// in order to detect the type of PLC.
- if c.driverContext.ControllerType != ControllerType_ANY {
+ if c.driverContext.ControllerType != readWriteModel.ControllerType_ANY {
// Send an event that connection setup is complete.
c.SetConnected(true)
return nil
@@ -325,24 +325,24 @@ func (c *Connection) extractControllerTypeAndFireConnected(payloadUserData readW
articleNumber := string(readSzlResponseItemItem.GetMlfb())
*/
articleNumber := "broken at the moment"
- var controllerType ControllerType
+ var controllerType readWriteModel.ControllerType
if !strings.HasPrefix(articleNumber, "6ES7 ") {
- controllerType = ControllerType_ANY
+ controllerType = readWriteModel.ControllerType_ANY
}
blankIndex := strings.Index(articleNumber, " ")
model := articleNumber[blankIndex+1 : blankIndex+2]
switch model {
case "2":
- controllerType = ControllerType_S7_1200
+ controllerType = readWriteModel.ControllerType_S7_1200
case "5":
- controllerType = ControllerType_S7_1500
+ controllerType = readWriteModel.ControllerType_S7_1500
case "3":
- controllerType = ControllerType_S7_300
+ controllerType = readWriteModel.ControllerType_S7_300
case "4":
- controllerType = ControllerType_S7_400
+ controllerType = readWriteModel.ControllerType_S7_400
default:
c.log.Info().Str("articleNumber", articleNumber).Msg("Looking up unknown article number")
- controllerType = ControllerType_ANY
+ controllerType = readWriteModel.ControllerType_ANY
}
c.driverContext.ControllerType = controllerType
diff --git a/plc4go/internal/s7/ControllerType.go b/plc4go/internal/s7/ControllerType.go
deleted file mode 100644
index c5d39d4598..0000000000
--- a/plc4go/internal/s7/ControllerType.go
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * 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 s7
-
-type ControllerType int
-
-const (
- ControllerType_UNKNOWN = -0x1
- ControllerType_ANY = 0x0
- ControllerType_S7_200 = 0x5
- ControllerType_S7_300 = 0x1
- ControllerType_S7_400 = 0x2
- ControllerType_S7_1200 = 0x3
- ControllerType_S7_1500 = 0x4
- ControllerType_LOGO = 0x5
-)
diff --git a/plc4go/internal/s7/DriverContext.go b/plc4go/internal/s7/DriverContext.go
index 752dc5ffd0..919d0d45df 100644
--- a/plc4go/internal/s7/DriverContext.go
+++ b/plc4go/internal/s7/DriverContext.go
@@ -31,7 +31,7 @@ type DriverContext struct {
PduSize uint16
MaxAmqCaller uint16
MaxAmqCallee uint16
- ControllerType ControllerType
+ ControllerType model.ControllerType
awaitSetupComplete bool
awaitDisconnectComplete bool
}
@@ -41,14 +41,11 @@ func NewDriverContext(configuration Configuration) (DriverContext, error) {
calledTsapId := encodeS7TsapId(model.DeviceGroup_PG_OR_PC, configuration.remoteRack, configuration.remoteSlot)
controllerType := configuration.controllerType
- if controllerType == ControllerType_UNKNOWN {
- controllerType = ControllerType_ANY
- }
pduSize := configuration.pduSize
// The Siemens LOGO device seems to only work with very limited settings,
// so we're overriding some of the defaults.
- if controllerType == ControllerType_LOGO && pduSize == 1024 {
+ if controllerType == model.ControllerType_LOGO && pduSize == 1024 {
pduSize = 480
}