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/09/11 15:10:17 UTC
(plc4x) 05/05: feat(plc4go/modbus): merge reads into block requests and honour the per-request ceilings
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 310b85450ead116bc2df58dfff1a307b2a94070e
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Fri Sep 11 16:57:40 2026 +0200
feat(plc4go/modbus): merge reads into block requests and honour the per-request ceilings
The driver answered a multi-tag read by splitting it into one request per tag
and dispatching them all at once, spaced only by a hardcoded 4ms sleep with no
in-flight cap. A direct-TCP device shrugs that off; a TCP<->RTU gateway, which
serialises each transaction onto a slow serial line, resets the connection
instead -- and the reset takes every in-flight request with it, so the read comes
back with nothing rather than with less. A profile of a few dozen points was
enough to do it.
Reads now go through the optimizer added in 1033e8b897: tags one request can
answer are merged into a block read, blocks are issued one at a time, and the
single response is split back into per-tag values. plc4j has behaved this way
since it gained ModbusReadOptimizer, and reaches the serialization from the
other side by capping its connections at one concurrent request.
Writes keep the per-tag interceptor. Modbus genuinely is one operation per PDU
for writes, and plc4j keeps them per-tag for the same reason.
max-coils-per-request and max-registers-per-request are now read, defaulting to
the spec maxima of 2000 and 125 exactly as plc4j does. These have been declared,
defaulted and documented on the plc4j side since 0.13.0 and on the website's
modbus pages, while plc4go silently ignored them -- so a connection string
written against the documentation did nothing here, with no warning, because
unknown options pass quietly. Values that are unusable but parseable are clamped
with a warning rather than refused: plc4j validates neither, and a connection
string that opens a connection there has to open one here.
BEHAVIOUR CHANGE, coil reads return different values. The optimizer's bit
extraction and the old ParseRegisters path disagree on coil bit order, and the
old one was wrong: Modbus packs the first coil in the LSB of the first data
byte, while ParseRegisters read MSB-first, and for a single coil read a bit the
response does not even contain. Anyone who compensated for that will see values
move.
The driver testsuite is updated accordingly: a two-tag read of adjacent REALs is
now one request for four registers rather than two requests for two. That asset
is only consumed by plc4go despite living under protocols/ -- no plc4j test
loads it, which is part of why the divergence went unnoticed.
The single-tag path is pinned by golden PDU bytes captured from the
implementation as it stood before this change, rather than by rebuilding the
expectation from the code under test, which would move with it.
---
.../protocols/modbus/tcp/DriverTestsuite.xml | 60 +--
plc4go/internal/modbus/Configuration.go | 70 ++++
plc4go/internal/modbus/Configuration_test.go | 58 +++
plc4go/internal/modbus/Connection.go | 43 +-
plc4go/internal/modbus/ReadOptimizer.go | 12 +-
plc4go/internal/modbus/ReadRequestBuilder.go | 106 +++++
plc4go/internal/modbus/ReadRequestBuilder_test.go | 110 ++++++
plc4go/internal/modbus/Reader.go | 436 +++++++++++++++------
plc4go/internal/modbus/Reader_test.go | 302 ++++++++++++++
plc4go/tests/configparity/parameter_names_test.go | 2 +
.../protocols/modbus/tcp/DriverTestsuite.xml | 60 +--
11 files changed, 1002 insertions(+), 257 deletions(-)
diff --git a/plc4go/assets/testing/protocols/modbus/tcp/DriverTestsuite.xml b/plc4go/assets/testing/protocols/modbus/tcp/DriverTestsuite.xml
index cc6976df48..2a175e94ec 100644
--- a/plc4go/assets/testing/protocols/modbus/tcp/DriverTestsuite.xml
+++ b/plc4go/assets/testing/protocols/modbus/tcp/DriverTestsuite.xml
@@ -235,7 +235,7 @@
</tags>
</TestReadRequest>
</api-request>
- <outgoing-plc-message name="Send First Item Modbus Input-Register Read Request">
+ <outgoing-plc-message name="Send Read Request for optimized chunk of data">
<parser-arguments>
<driverType>MODBUS_TCP</driverType>
<response>false</response>
@@ -252,14 +252,14 @@
<functionFlag dataType="uint" bitLength="7">3</functionFlag>
<ModbusPDUReadHoldingRegistersRequest>
<startingAddress dataType="uint" bitLength="16">0</startingAddress>
- <quantity dataType="uint" bitLength="16">2</quantity>
+ <quantity dataType="uint" bitLength="16">4</quantity>
</ModbusPDUReadHoldingRegistersRequest>
</ModbusPDU>
</pdu>
</ModbusTcpADU>
</ModbusADU>
</outgoing-plc-message>
- <incoming-plc-message name="Receive First Item Modbus Input-Register Read Response">
+ <incoming-plc-message name="Send Read Response for optimized chunk of data">
<parser-arguments>
<driverType>MODBUS_TCP</driverType>
<response>true</response>
@@ -268,63 +268,15 @@
<ModbusTcpADU>
<transactionIdentifier dataType="uint" bitLength="16">1</transactionIdentifier>
<protocolIdentifier dataType="uint" bitLength="16">0</protocolIdentifier>
- <length dataType="uint" bitLength="16">7</length>
- <unitIdentifier dataType="uint" bitLength="8">1</unitIdentifier>
- <pdu>
- <ModbusPDU>
- <errorFlag dataType="bit" bitLength="1">false</errorFlag>
- <functionFlag dataType="uint" bitLength="7">3</functionFlag>
- <ModbusPDUReadHoldingRegistersResponse>
- <byteCount dataType="uint" bitLength="8">4</byteCount>
- <value dataType="byte" bitLength="32">0x40490fdb</value>
- </ModbusPDUReadHoldingRegistersResponse>
- </ModbusPDU>
- </pdu>
- </ModbusTcpADU>
- </ModbusADU>
- </incoming-plc-message>
- <outgoing-plc-message name="Send Second Item Modbus Input-Register Read Request">
- <parser-arguments>
- <driverType>MODBUS_TCP</driverType>
- <response>false</response>
- </parser-arguments>
- <ModbusADU>
- <ModbusTcpADU>
- <transactionIdentifier dataType="uint" bitLength="16">2</transactionIdentifier>
- <protocolIdentifier dataType="uint" bitLength="16">0</protocolIdentifier>
- <length dataType="uint" bitLength="16">6</length>
- <unitIdentifier dataType="uint" bitLength="8">1</unitIdentifier>
- <pdu>
- <ModbusPDU>
- <errorFlag dataType="bit" bitLength="1">false</errorFlag>
- <functionFlag dataType="uint" bitLength="7">3</functionFlag>
- <ModbusPDUReadHoldingRegistersRequest>
- <startingAddress dataType="uint" bitLength="16">2</startingAddress>
- <quantity dataType="uint" bitLength="16">2</quantity>
- </ModbusPDUReadHoldingRegistersRequest>
- </ModbusPDU>
- </pdu>
- </ModbusTcpADU>
- </ModbusADU>
- </outgoing-plc-message>
- <incoming-plc-message name="Receive Second Item Modbus Input-Register Read Response">
- <parser-arguments>
- <driverType>MODBUS_TCP</driverType>
- <response>true</response>
- </parser-arguments>
- <ModbusADU>
- <ModbusTcpADU>
- <transactionIdentifier dataType="uint" bitLength="16">2</transactionIdentifier>
- <protocolIdentifier dataType="uint" bitLength="16">0</protocolIdentifier>
- <length dataType="uint" bitLength="16">7</length>
+ <length dataType="uint" bitLength="16">11</length>
<unitIdentifier dataType="uint" bitLength="8">1</unitIdentifier>
<pdu>
<ModbusPDU>
<errorFlag dataType="bit" bitLength="1">false</errorFlag>
<functionFlag dataType="uint" bitLength="7">3</functionFlag>
<ModbusPDUReadHoldingRegistersResponse>
- <byteCount dataType="uint" bitLength="8">4</byteCount>
- <value dataType="byte" bitLength="32">0x40490fdb</value>
+ <byteCount dataType="uint" bitLength="8">8</byteCount>
+ <value dataType="byte" bitLength="64">0x40490fdb40490fdb</value>
</ModbusPDUReadHoldingRegistersResponse>
</ModbusPDU>
</pdu>
diff --git a/plc4go/internal/modbus/Configuration.go b/plc4go/internal/modbus/Configuration.go
index 57d587d72b..3e39c55b67 100644
--- a/plc4go/internal/modbus/Configuration.go
+++ b/plc4go/internal/modbus/Configuration.go
@@ -43,6 +43,13 @@ type Configuration struct {
pingAddress string
// requestTimeout bounds how long a single request waits for its response.
requestTimeout time.Duration
+ // maxCoilsPerRequest is the widest run of coils or discrete inputs the read optimizer merges
+ // into one request.
+ maxCoilsPerRequest uint16
+ // maxRegistersPerRequest is the widest run of registers the read optimizer merges into one
+ // request. It applies to all three register areas; the extended registers are cut down further
+ // where FC 0x14's framing demands it (see maxPerRequestFor).
+ maxRegistersPerRequest uint16
// flavor is the framing this connection speaks. It is not something the connection string can
// say - each driver speaks exactly one flavor and sets it on the configuration it hands to the
// connection. The zero value is flavorTcp, so a configuration nobody told otherwise behaves
@@ -67,6 +74,12 @@ const (
defaultPingAddress = "4x00001:BOOL"
// defaultRequestTimeout is plc4j's request-timeout-ms default of five seconds.
defaultRequestTimeout = 5 * time.Second
+ // defaultMaxCoilsPerRequest and defaultMaxRegistersPerRequest are plc4j's
+ // max-coils-per-request and max-registers-per-request defaults, which are the ceilings the
+ // modbus specification itself imposes - merging up to them asks for the largest request that
+ // can be answered at all.
+ defaultMaxCoilsPerRequest = uint16(maxCoilQuantity)
+ defaultMaxRegistersPerRequest = uint16(maxRegisterQuantity)
)
// DefaultConfiguration is a connection without any options set.
@@ -76,6 +89,8 @@ func DefaultConfiguration() Configuration {
defaultPayloadByteOrder: BigEndianOrder,
pingAddress: defaultPingAddress,
requestTimeout: defaultRequestTimeout,
+ maxCoilsPerRequest: defaultMaxCoilsPerRequest,
+ maxRegistersPerRequest: defaultMaxRegistersPerRequest,
}
}
@@ -130,9 +145,64 @@ func ParseFromOptions(localLog zerolog.Logger, connectionOptions map[string][]st
configuration.requestTimeout = time.Duration(parsedUint) * time.Millisecond
}
+ // The two ceilings the read optimizer merges within. plc4j declares them on all three modbus
+ // configurations - ModbusTcpConfiguration and its RTU and ASCII twins - and the website
+ // documents them (modbus-tcp.adoc and friends), so a connection string that sets them is one
+ // users have been told to write. This driver never read them, which made merging silently
+ // ignore a device that can't answer a full-width request.
+ maxCoilsPerRequest, err := parseMaxPerRequest(localLog, reader, "max-coils-per-request",
+ configuration.maxCoilsPerRequest, maxCoilQuantity)
+ if err != nil {
+ return Configuration{}, err
+ }
+ configuration.maxCoilsPerRequest = maxCoilsPerRequest
+
+ maxRegistersPerRequest, err := parseMaxPerRequest(localLog, reader, "max-registers-per-request",
+ configuration.maxRegistersPerRequest, maxRegisterQuantity)
+ if err != nil {
+ return Configuration{}, err
+ }
+ configuration.maxRegistersPerRequest = maxRegistersPerRequest
+
return configuration, nil
}
+// parseMaxPerRequest reads one of the two per-request ceilings, falling back to the given default
+// when the connection string doesn't name it.
+//
+// Both bounds are refused rather than clamped. Zero would ask the device for nothing, which it
+// answers happily, so a read would report success and no data; anything above what the
+// specification allows produces a response that doesn't fit into a modbus PDU, which no device can
+// send. Either way a request built from such a number cannot work, and saying so beats sending it.
+func parseMaxPerRequest(localLog zerolog.Logger, reader *spiOptions.OptionReader, name string, defaultValue uint16, ceiling uint16) (uint16, error) {
+ value := reader.Get(name)
+ if value == "" {
+ return defaultValue, nil
+ }
+ parsedUint, err := strconv.ParseUint(value, 10, 16)
+ if err != nil {
+ return 0, errors.Wrapf(err, "Error parsing %s %s", name, value)
+ }
+ // Clamp rather than reject. plc4j declares these as plain ints with no validation at all
+ // (ModbusTcpConfiguration and its RTU/ASCII siblings), so a connection string that opens a
+ // connection there must open one here too -- refusing it would make the option LESS portable
+ // than it was when plc4go ignored it entirely, which is the opposite of the point.
+ //
+ // Zero would mean "no tag may ever be read" and the ceiling is what one request can physically
+ // carry, so both ends are pinned to something usable and the reason is logged.
+ if parsedUint == 0 {
+ localLog.Warn().Str("option", name).Str("value", value).Uint16("using", defaultValue).
+ Msg("a per-request ceiling of zero would read nothing, so the default is used")
+ return defaultValue, nil
+ }
+ if parsedUint > uint64(ceiling) {
+ localLog.Warn().Str("option", name).Str("value", value).Uint16("using", ceiling).
+ Msg("a per-request ceiling beyond what one modbus request can carry is clamped")
+ return ceiling, nil
+ }
+ return uint16(parsedUint), nil
+}
+
// withRequestTimeout bounds a single request. The codec turns the deadline of the context it is
// handed into the lifetime of its expectation (spi/default.defaultCodec.expect), so a deadline is
// all it takes to time a request out; a caller who brought a deadline of their own keeps it.
diff --git a/plc4go/internal/modbus/Configuration_test.go b/plc4go/internal/modbus/Configuration_test.go
index 79811e0285..881e625a2d 100644
--- a/plc4go/internal/modbus/Configuration_test.go
+++ b/plc4go/internal/modbus/Configuration_test.go
@@ -44,6 +44,8 @@ func TestParseFromOptions_defaults(t *testing.T) {
assert.Equal(t, BigEndianOrder, configuration.defaultPayloadByteOrder)
assert.Equal(t, "4x00001:BOOL", configuration.pingAddress)
assert.Equal(t, 5*time.Second, configuration.requestTimeout)
+ assert.Equal(t, uint16(2000), configuration.maxCoilsPerRequest)
+ assert.Equal(t, uint16(125), configuration.maxRegistersPerRequest)
assert.Equal(t, DefaultConfiguration(), configuration)
}
@@ -82,8 +84,61 @@ func TestParseFromOptions_requestTimeout(t *testing.T) {
assert.Equal(t, 250*time.Millisecond, configuration.requestTimeout)
}
+// The two ceilings the read optimizer merges within. plc4j has declared and documented them since
+// 0.13.0; this driver never read them, so a connection string that named them was ignored here.
+func TestParseFromOptions_maxPerRequest(t *testing.T) {
+ configuration := parseConfiguration(t, map[string][]string{
+ "max-coils-per-request": {"64"},
+ "max-registers-per-request": {"20"},
+ })
+ assert.Equal(t, uint16(64), configuration.maxCoilsPerRequest)
+ assert.Equal(t, uint16(20), configuration.maxRegistersPerRequest)
+
+ // Each stands on its own; naming one leaves the other at its default.
+ assert.Equal(t, defaultMaxRegistersPerRequest,
+ parseConfiguration(t, map[string][]string{"max-coils-per-request": {"64"}}).maxRegistersPerRequest)
+ assert.Equal(t, defaultMaxCoilsPerRequest,
+ parseConfiguration(t, map[string][]string{"max-registers-per-request": {"20"}}).maxCoilsPerRequest)
+}
+
// An option the driver can't make sense of is an error - silently falling back to a default would
// leave the caller talking to the device in a way they didn't ask for.
+// TestParseFromOptions_clampsPerRequestCeilings covers a value that is usable but out of range.
+//
+// These are CLAMPED rather than rejected, deliberately. plc4j declares both options as plain ints
+// with no validation, so a connection string that opens a connection there has to open one here -
+// refusing it would make the option less portable than it was while plc4go ignored it entirely,
+// which is the opposite of the point of supporting it. It also cannot break an existing
+// deployment: a connection string carrying a nonsense ceiling works today (silently ignored) and
+// must keep working. The value is pinned to something usable and the reason is logged.
+//
+// Values that are not numbers at all are still an error - see TestParseFromOptions_rejectsBadValues.
+func TestParseFromOptions_clampsPerRequestCeilings(t *testing.T) {
+ for _, test := range []struct {
+ name string
+ options map[string][]string
+ wantCoils uint16
+ wantRegister uint16
+ }{
+ {"zero reads nothing, so the default stands", map[string][]string{
+ "max-coils-per-request": {"0"}, "max-registers-per-request": {"0"}},
+ maxCoilQuantity, maxRegisterQuantity},
+ {"beyond what one request can carry is clamped", map[string][]string{
+ "max-coils-per-request": {"2001"}, "max-registers-per-request": {"126"}},
+ maxCoilQuantity, maxRegisterQuantity},
+ {"a usable value is taken as given", map[string][]string{
+ "max-coils-per-request": {"8"}, "max-registers-per-request": {"1"}},
+ 8, 1},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ configuration, err := ParseFromOptions(zerolog.Nop(), test.options)
+ require.NoError(t, err, "a usable-but-out-of-range ceiling must not stop the connection opening")
+ assert.Equal(t, test.wantCoils, configuration.maxCoilsPerRequest)
+ assert.Equal(t, test.wantRegister, configuration.maxRegistersPerRequest)
+ })
+ }
+}
+
func TestParseFromOptions_rejectsBadValues(t *testing.T) {
for _, test := range []struct {
name string
@@ -95,6 +150,9 @@ func TestParseFromOptions_rejectsBadValues(t *testing.T) {
{"unparsable ping address", map[string][]string{"ping-address": {"this is not an address"}}},
{"request timeout that isn't a number", map[string][]string{"request-timeout-ms": {"soon"}}},
{"request timeout of zero", map[string][]string{"request-timeout-ms": {"0"}}},
+ {"coils per request that isn't a number", map[string][]string{"max-coils-per-request": {"lots"}}},
+ {"registers per request that isn't a number", map[string][]string{"max-registers-per-request": {"lots"}}},
+ {"coils per request beyond a uint16", map[string][]string{"max-coils-per-request": {"65536"}}},
} {
t.Run(test.name, func(t *testing.T) {
_, err := ParseFromOptions(zerolog.Nop(), test.connectionOptions)
diff --git a/plc4go/internal/modbus/Connection.go b/plc4go/internal/modbus/Connection.go
index 38dc4a25c2..3fac7f684f 100644
--- a/plc4go/internal/modbus/Connection.go
+++ b/plc4go/internal/modbus/Connection.go
@@ -43,10 +43,13 @@ import (
type Connection struct {
_default.DefaultConnection
- configuration Configuration
- messageCodec spi.MessageCodec
- options map[string][]string
- requestInterceptor interceptors.RequestInterceptor
+ configuration Configuration
+ messageCodec spi.MessageCodec
+ options map[string][]string
+ // writeRequestInterceptor splits a write request into one request per tag. Reads have no such
+ // interceptor: they are merged into block requests by the read optimizer instead, which is the
+ // whole point of merging them.
+ writeRequestInterceptor interceptors.WriteRequestInterceptor
// transactionIdentifier numbers the requests Ping sends; reads and writes have counters of
// their own in Reader and Writer.
@@ -71,7 +74,10 @@ func NewConnection(configuration Configuration, messageCodec spi.MessageCodec, c
configuration: configuration,
messageCodec: messageCodec,
options: connectionOptions,
- requestInterceptor: interceptors.NewSingleItemRequestInterceptor(
+ // Modbus writes one operation per PDU - there is no function code that writes two
+ // unrelated addresses - so a write request is still cut into one request per tag. plc4j
+ // keeps per-tag writes for the same reason (ModbusTcpConnection.onWrite).
+ writeRequestInterceptor: interceptors.NewSingleItemRequestInterceptor(
spiModel.NewDefaultPlcReadRequest,
spiModel.NewDefaultPlcWriteRequest,
spiModel.NewDefaultPlcReadResponse,
@@ -222,15 +228,26 @@ func (c *Connection) GetMetadata() apiModel.PlcConnectionMetadata {
}
}
+// ReadRequestBuilder builds a read request that reaches the reader whole. Reads used to be split
+// into one request per tag by a SingleItemRequestInterceptor, which cost a round trip per tag; the
+// reader merges them into block reads instead (see ReadOptimizer.go), and it can only do that if
+// it is handed every tag at once.
+//
+// The builder is this driver's own so that an address it can't parse costs only its own tag rather
+// than the entire request (see ReadRequestBuilder.go).
func (c *Connection) ReadRequestBuilder() apiModel.PlcReadRequestBuilder {
- return spiModel.NewDefaultPlcReadRequestBuilderWithInterceptor(
- c.GetPlcTagHandler(),
- NewReader(
- c.configuration,
- c.messageCodec,
- append(c._options, options.WithCustomLogger(c.log))...,
+ tagHandler := c.GetPlcTagHandler()
+ return newReadRequestBuilder(
+ spiModel.NewDefaultPlcReadRequestBuilder(
+ tagHandler,
+ NewReader(
+ c.configuration,
+ c.messageCodec,
+ append(c._options, options.WithCustomLogger(c.log))...,
+ ),
),
- c.requestInterceptor,
+ tagHandler,
+ c.log,
)
}
@@ -243,7 +260,7 @@ func (c *Connection) WriteRequestBuilder() apiModel.PlcWriteRequestBuilder {
c.messageCodec,
append(c._options, options.WithCustomLogger(c.log))...,
),
- c.requestInterceptor,
+ c.writeRequestInterceptor,
)
}
diff --git a/plc4go/internal/modbus/ReadOptimizer.go b/plc4go/internal/modbus/ReadOptimizer.go
index 49f2d555a9..c05e7689c0 100644
--- a/plc4go/internal/modbus/ReadOptimizer.go
+++ b/plc4go/internal/modbus/ReadOptimizer.go
@@ -40,13 +40,10 @@ import (
// the single response is split apart again afterwards.
//
// Nothing in this file talks to a connection. optimizeReads decides what to ask for and
-// splitBlockResponse decodes what came back; issuing the requests is the caller's business.
+// splitBlockResponse decodes what came back; issuing the requests is Reader.read's business.
//
-// NOTE: nothing calls this yet. The driver still issues one request per tag (see Reader.Read,
-// which rejects multi-tag requests outright, and the SingleItemRequestInterceptor the connection
-// installs). Wiring it in changes per-tag failure semantics - a failed block fails every tag in
-// it - and so is a separate, separately reviewable step. Until then this file changes no
-// behaviour.
+// How wide a block may get is the connection's max-coils-per-request and max-registers-per-request
+// (see Configuration), which default to what the specification itself allows.
// namedTag pairs a tag with the name the read request knows it by. Tags travel by value here, the
// way they do everywhere else in this package - castToModbusTagFromPlcTag hands out a value too.
@@ -112,7 +109,8 @@ type unitGroupKey struct {
}
// optimizeReadsWithSpecLimits merges tags using the largest request the modbus specification
-// allows, which is also what plc4j defaults to.
+// allows, which is what plc4j defaults to and what a connection nobody configured otherwise uses
+// (see defaultMaxCoilsPerRequest and defaultMaxRegistersPerRequest).
func optimizeReadsWithSpecLimits(tags []namedTag) []readBlock {
return optimizeReads(tags, maxCoilQuantity, maxRegisterQuantity)
}
diff --git a/plc4go/internal/modbus/ReadRequestBuilder.go b/plc4go/internal/modbus/ReadRequestBuilder.go
new file mode 100644
index 0000000000..71261c7199
--- /dev/null
+++ b/plc4go/internal/modbus/ReadRequestBuilder.go
@@ -0,0 +1,106 @@
+/*
+ * 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 modbus
+
+import (
+ "fmt"
+
+ "github.com/rs/zerolog"
+
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
+ "github.com/apache/plc4x/plc4go/spi"
+)
+
+// readRequestBuilder assembles a read request in which one rejected address costs only its own
+// tag. The shared spiModel.DefaultPlcReadRequestBuilder parses the addresses in Build and returns
+// nothing at all as soon as one of them doesn't parse, so a single bad address takes down the
+// whole scan - and modbus addresses regularly come from somewhere other than the caller (a device
+// profile handed down from a server, say), where one typo among hundreds of good points is
+// ordinary rather than exceptional.
+//
+// plc4j does the same thing one layer up, in the shared builder: DefaultPlcReadRequest.Builder
+// keeps the tag in the request carrying INVALID_ADDRESS and a null tag, and the driver reads that
+// back with getTagResponseCode. The equivalent move here would change the builder every driver
+// uses, so the rejection is carried by a tag type of this driver's own instead: the tag takes part
+// in the request, the reader recognizes it (see modbusTagsOf) and reports it as INVALID_ADDRESS,
+// and nothing of it ever reaches the wire.
+type readRequestBuilder struct {
+ // The default builder does the rest of the work, and it only ever sees tags that are already
+ // parsed - which is why its Build can no longer fail.
+ apiModel.PlcReadRequestBuilder
+
+ tagHandler spi.PlcTagHandler
+ log zerolog.Logger
+}
+
+// newReadRequestBuilder wraps the default builder, which is handed the reader and assembles the
+// request itself.
+func newReadRequestBuilder(delegate apiModel.PlcReadRequestBuilder, tagHandler spi.PlcTagHandler, log zerolog.Logger) apiModel.PlcReadRequestBuilder {
+ return &readRequestBuilder{
+ PlcReadRequestBuilder: delegate,
+ tagHandler: tagHandler,
+ log: log,
+ }
+}
+
+// AddTagAddress parses the address right away, so that the tag the request ends up with is either
+// a modbus tag or the record of why there isn't one. The default builder is only ever handed
+// already parsed tags, so it never parses anything of its own.
+func (b *readRequestBuilder) AddTagAddress(tagName string, tagAddress string) apiModel.PlcReadRequestBuilder {
+ tag, err := b.tagHandler.ParseTag(tagAddress)
+ if err != nil {
+ // The response carries only the code, so this is the one place the reason is known.
+ b.log.Warn().Err(err).Str("tagName", tagName).Str("tagAddress", tagAddress).
+ Msg("Couldn't parse an address, so this one tag is answered with INVALID_ADDRESS")
+ b.PlcReadRequestBuilder.AddTag(tagName, unparsedTag{address: tagAddress, err: err})
+ return b
+ }
+ b.PlcReadRequestBuilder.AddTag(tagName, tag)
+ return b
+}
+
+// AddTag keeps the chain on this builder; a tag that is handed over already parsed needs nothing
+// else done to it.
+func (b *readRequestBuilder) AddTag(tagName string, tag apiModel.PlcTag) apiModel.PlcReadRequestBuilder {
+ b.PlcReadRequestBuilder.AddTag(tagName, tag)
+ return b
+}
+
+var _ apiModel.PlcTag = unparsedTag{}
+
+// unparsedTag is a tag whose address this driver couldn't make sense of. It is a tag only so far
+// as it can take part in a request and be reported back; it carries no address the driver could
+// read, which is the whole reason it exists.
+type unparsedTag struct {
+ address string
+ err error
+}
+
+func (u unparsedTag) GetAddressString() string { return u.address }
+
+// GetValueType is NULL, the value a tag that can't be read comes back with (see valueOrNull).
+func (u unparsedTag) GetValueType() apiValues.PlcValueType { return apiValues.NULL }
+
+func (u unparsedTag) GetArrayInfo() []apiModel.ArrayInfo { return nil }
+
+func (u unparsedTag) String() string {
+ return fmt.Sprintf("unparsedTag{address: %s, err: %v}", u.address, u.err)
+}
diff --git a/plc4go/internal/modbus/ReadRequestBuilder_test.go b/plc4go/internal/modbus/ReadRequestBuilder_test.go
new file mode 100644
index 0000000000..a49a10fbf4
--- /dev/null
+++ b/plc4go/internal/modbus/ReadRequestBuilder_test.go
@@ -0,0 +1,110 @@
+/*
+ * 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 modbus
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ 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/modbus/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+)
+
+// One address the driver can't parse costs its own tag and nothing else. Addresses regularly come
+// from somewhere other than the caller - a device profile handed down from a server - so one typo
+// among hundreds of good points is ordinary; failing the Build would turn it into an entire scan
+// that reads nothing. plc4j answers the same request the same way, by keeping the rejected tag in
+// the request with INVALID_ADDRESS (DefaultPlcReadRequest.Builder).
+func TestReadRequestBuilder_anUnparseableAddressCostsOnlyItsOwnTag(t *testing.T) {
+ codec := newCaptureCodec(nil)
+ connection := testConnection(DefaultConfiguration(), codec)
+
+ request, err := connection.ReadRequestBuilder().
+ AddTagAddress("good1", "holding-register:1:REAL").
+ AddTagAddress("bad", "holding-register:nonsense:REAL").
+ AddTagAddress("good2", "holding-register:3:REAL").
+ Build()
+ require.NoError(t, err, "the bad address must not take the request down with it")
+ require.NotNil(t, request)
+
+ results := request.Execute(testutils.TestContext(t))
+
+ // The two good tags are adjacent, so they are read as one block of four registers; the bad one
+ // contributes nothing to it, because there is no address to ask about.
+ select {
+ case handlers := <-codec.handlers:
+ requestAdu, ok := handlers.message.(readWriteModel.ModbusTcpADU)
+ require.True(t, ok, "expected a ModbusTcpADU, got %T", handlers.message)
+ pdu := holdingRegisterRequestOf(t, requestAdu)
+ assert.Equal(t, uint16(0), pdu.GetStartingAddress())
+ assert.Equal(t, uint16(4), pdu.GetQuantity())
+ responseAdu := readWriteModel.NewModbusTcpADU(
+ requestAdu.GetTransactionIdentifier(), requestAdu.GetUnitIdentifier(),
+ readWriteModel.NewModbusPDUReadHoldingRegistersResponse(append(append([]byte{}, pi...), pi...)))
+ require.True(t, handlers.acceptsMessage(responseAdu), "the reader didn't accept the response to its own request")
+ require.NoError(t, handlers.handleMessage(responseAdu))
+ case <-time.After(time.Second):
+ t.Fatal("the good tags were never read")
+ }
+
+ var result apiModel.PlcReadRequestResult
+ select {
+ case result = <-results:
+ case <-time.After(2 * time.Second):
+ t.Fatal("no result was delivered")
+ }
+
+ // The rejection is reported as well as being visible per tag, the way a failed block is.
+ require.Error(t, result.GetErr())
+ assert.Contains(t, result.GetErr().Error(), "bad", "the error has to name the tag that was rejected")
+
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ assert.Equal(t, apiModel.PlcResponseCode_INVALID_ADDRESS, response.GetResponseCode("bad"))
+ value := response.GetValue("bad")
+ require.NotNil(t, value)
+ assert.Equal(t, apiValues.NULL, value.GetPlcValueType())
+
+ for _, tagName := range []string{"good1", "good2"} {
+ assert.Equal(t, apiModel.PlcResponseCode_OK, response.GetResponseCode(tagName), tagName)
+ assert.InDelta(t, 3.1415927, response.GetValue(tagName).GetFloat32(), 0.0000001, tagName)
+ }
+}
+
+// The rejected tag keeps the address it was given, so that a caller holding only the response can
+// still say which address was refused.
+func TestReadRequestBuilder_theRejectedTagKeepsItsAddress(t *testing.T) {
+ codec := newCaptureCodec(nil)
+ connection := testConnection(DefaultConfiguration(), codec)
+
+ request, err := connection.ReadRequestBuilder().
+ AddTagAddress("bad", "not-an-address").
+ Build()
+ require.NoError(t, err)
+
+ assert.Equal(t, []string{"bad"}, request.GetTagNames())
+ assert.Equal(t, "not-an-address", request.GetTag("bad").GetAddressString())
+ assert.Equal(t, apiValues.NULL, request.GetTag("bad").GetValueType())
+}
diff --git a/plc4go/internal/modbus/Reader.go b/plc4go/internal/modbus/Reader.go
index 7941ac2280..fd9edd1511 100644
--- a/plc4go/internal/modbus/Reader.go
+++ b/plc4go/internal/modbus/Reader.go
@@ -36,6 +36,7 @@ import (
spiModel "github.com/apache/plc4x/plc4go/spi/model"
"github.com/apache/plc4x/plc4go/spi/options"
"github.com/apache/plc4x/plc4go/spi/utils"
+ "github.com/apache/plc4x/plc4go/spi/values"
)
type Reader struct {
@@ -61,32 +62,27 @@ func NewReader(configuration Configuration, messageCodec spi.MessageCodec, _opti
}
}
-// readRequestPdu is the request that reads what the tag addresses (plc4j
-// ModbusTcpConnection.getReadRequestPdu).
-func readRequestPdu(tag modbusTag) (readWriteModel.ModbusPDU, error) {
- switch tag.TagType {
+// blockRequestPdu is the request that reads one run of addresses in one area (plc4j
+// ModbusTcpConnection.getReadRequestPdu, which builds the same PDUs from a single tag).
+//
+// quantity counts what the area addresses: coils in the two bit areas, registers in the other
+// three. That is the same count a tag's own read carries, which is why a block covering exactly
+// one tag produces byte for byte the request that tag was read with before anything was merged.
+func blockRequestPdu(tagType TagType, address uint16, quantity uint16) (readWriteModel.ModbusPDU, error) {
+ switch tagType {
case Coil:
- return readWriteModel.NewModbusPDUReadCoilsRequest(tag.Address, tag.Quantity), nil
+ return readWriteModel.NewModbusPDUReadCoilsRequest(address, quantity), nil
case DiscreteInput:
- return readWriteModel.NewModbusPDUReadDiscreteInputsRequest(tag.Address, tag.Quantity), nil
- case InputRegister, HoldingRegister:
- numWords, err := tag.lengthWords()
- if err != nil {
- return nil, err
- }
- if tag.TagType == InputRegister {
- return readWriteModel.NewModbusPDUReadInputRegistersRequest(tag.Address, numWords), nil
- }
- return readWriteModel.NewModbusPDUReadHoldingRegistersRequest(tag.Address, numWords), nil
+ return readWriteModel.NewModbusPDUReadDiscreteInputsRequest(address, quantity), nil
+ case InputRegister:
+ return readWriteModel.NewModbusPDUReadInputRegistersRequest(address, quantity), nil
+ case HoldingRegister:
+ return readWriteModel.NewModbusPDUReadHoldingRegistersRequest(address, quantity), nil
case ExtendedRegister:
// The extended register area is read with FC 0x14, which addresses it as a set of files
// rather than flat, so the address turns into one item per file it touches (plc4j
// ModbusTcpConnection.getReadRequestPdu).
- numWords, err := tag.lengthWords()
- if err != nil {
- return nil, err
- }
- groups := splitExtendedRegister(tag.Address, numWords)
+ groups := splitExtendedRegister(address, quantity)
items := make([]readWriteModel.ModbusPDUReadFileRecordRequestItem, len(groups))
for i, group := range groups {
items[i] = readWriteModel.NewModbusPDUReadFileRecordRequestItem(
@@ -94,12 +90,24 @@ func readRequestPdu(tag modbusTag) (readWriteModel.ModbusPDU, error) {
}
return readWriteModel.NewModbusPDUReadFileRecordRequest(items), nil
default:
- return nil, errors.Errorf("unsupported tag type %x", tag.TagType)
+ return nil, errors.Errorf("unsupported tag type %x", tagType)
}
}
+// readRequestPdu is the request that reads what one tag addresses. Ping is what is left that reads
+// a lone tag; a read request goes through the optimizer and asks for blocks instead.
+func readRequestPdu(tag modbusTag) (readWriteModel.ModbusPDU, error) {
+ // The same count the optimizer measures a block in, so that this and blockRequestPdu can't
+ // drift apart. tagSpan doesn't police the 16 bit quantity field, so that is checked here the
+ // way lengthWords used to check it.
+ span := tagSpan(tag)
+ if span > maxWireAddress {
+ return nil, errors.Errorf("the requested %d registers don't fit into a request, at most %d do", span, maxWireAddress)
+ }
+ return blockRequestPdu(tag.TagType, tag.Address, uint16(span))
+}
+
func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest) <-chan apiModel.PlcReadRequestResult {
- // TODO: handle ctx
m.log.Trace().Msg("Reading")
result := make(chan apiModel.PlcReadRequestResult, 1)
m.wg.Go(func() {
@@ -108,151 +116,321 @@ func (m *Reader) Read(ctx context.Context, readRequest apiModel.PlcReadRequest)
utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, errors.Errorf("panic-ed %v. Stack: %s", err, debug.Stack())))
}
}()
- // deliver hands one result to the caller. It is only set up once the request is under way;
- // until then the few rejections below deliver directly.
- var deliver func(response apiModel.PlcReadResponse, err error)
- if len(readRequest.GetTagNames()) != 1 {
- utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, nil, errors.New("modbus only supports single-item requests")))
- m.log.Debug().Int("nTags", len(readRequest.GetTagNames())).Msg("modbus only supports single-item requests. Got nTags tags")
- return
+ response, err := m.read(ctx, readRequest)
+ utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, response, err))
+ })
+ return result
+}
+
+// read answers the whole request: the tags are merged into the blocks that have to be read, the
+// blocks are read one after another, and every block's response is split back apart over the tags
+// it covers.
+//
+// The blocks go out one at a time rather than all at once. Only the TCP flavor carries a
+// transaction identifier, so the serial flavors correlate a response by station address and
+// function code alone (see rtuAduFactory.acceptsResponse) and two overlapping reads of the same
+// area would be indistinguishable. plc4j chains its requests for the same reason.
+//
+// A response is returned even when some of it failed, with the failures reported both as the
+// response code of the tags they took down and as the returned error, the way the single-item
+// interceptor reports a partial result. A read of twenty tags where one register is unreadable is
+// still nineteen values the caller asked for.
+func (m *Reader) read(ctx context.Context, readRequest apiModel.PlcReadRequest) (apiModel.PlcReadResponse, error) {
+ tags, invalid := modbusTagsOf(readRequest)
+ if len(tags)+len(invalid) == 0 {
+ return nil, errors.New("a read request has to ask for at least one tag")
+ }
+
+ responseCodes := make(map[string]apiModel.PlcResponseCode, len(tags)+len(invalid))
+ plcValues := make(map[string]apiValues.PlcValue, len(tags)+len(invalid))
+ var failures []error
+ for _, entry := range invalid {
+ // A tag this driver can't make sense of fails on its own; there is nothing to ask the
+ // device about it, and the rest of the request is still answerable.
+ m.log.Debug().Err(entry.err).Str("tagName", entry.name).Msg("Invalid tag")
+ responseCodes[entry.name] = apiModel.PlcResponseCode_INVALID_ADDRESS
+ // A null rather than nothing, for the same reason a tag whose block failed carries one:
+ // the response hands out whatever it was given, so a missing value is a nil to
+ // dereference.
+ plcValues[entry.name] = values.NewPlcNULL()
+ failures = append(failures, entry.err)
+ }
+
+ ctxForModel := options.GetLoggerContextForModel(ctx, m.log, options.WithPassLoggerToModel(m.passLogToModel))
+ // Blocks go out ONE AT A TIME, and that is the point rather than an oversight.
+ //
+ // The path this replaces handed every per-tag sub-read to the dispatcher at once, spaced only
+ // by a hardcoded 4ms sleep and with no in-flight cap, so a read of N tags put N transactions
+ // on the wire together. A direct-TCP device shrugs that off; a TCP<->RTU gateway, which has to
+ // serialise each one onto a slow serial line, resets the connection instead - and the reset
+ // takes every in-flight request with it, so the read comes back with nothing at all.
+ //
+ // plc4j reaches the same place from the other direction: its Modbus connections override
+ // getMaxConcurrentRequests() to 1, so one PDU is on the wire at a time per connection.
+ //
+ // The cost is latency for tags the optimizer cannot merge: worst case N round trips where the
+ // old path overlapped them. Merging is what buys that back - a profile of 84 tags spanning a
+ // few register clusters becomes a handful of blocks, not 84 of anything.
+ for _, block := range optimizeReads(tags, m.configuration.maxCoilsPerRequest, m.configuration.maxRegistersPerRequest) {
+ // A cancelled read must stop here rather than walk the remaining blocks. Blocks are
+ // issued one at a time, so without this a request cancelled after the first block still
+ // puts every later block on the wire -- which is worse than the path this replaced,
+ // where the sub-reads were already in flight by the time cancellation could be noticed.
+ if err := ctx.Err(); err != nil {
+ failures = append(failures, err)
+ break
}
- // If we are requesting only one tag, use a
- tagName := readRequest.GetTagNames()[0]
- tag := readRequest.GetTag(tagName)
- modbusTagVar, err := castToModbusTagFromPlcTag(tag)
- if err != nil {
- utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(
- readRequest,
- nil,
- errors.Wrap(err, "invalid tag item type"),
- ))
- m.log.Debug().Type("tagType", tag).Msg("Invalid tag item type")
- return
+
+ outcome := m.readBlock(ctx, block)
+ if outcome.err != nil {
+ m.log.Debug().Err(outcome.err).
+ Stringer("tagType", block.tagType).
+ Uint16("address", block.address).
+ Uint16("quantity", block.quantity).
+ Msg("Reading a block failed, so every tag in it fails")
+ failures = append(failures, outcome.err)
+ }
+ for _, tagResult := range splitBlockResponse(ctxForModel, block, outcome.responseCode, outcome.data, m.configuration.defaultPayloadByteOrder) {
+ responseCodes[tagResult.name] = tagResult.responseCode
+ // A tag that failed still gets an entry, so that a caller reading the value of a
+ // failed tag finds a null rather than a nil to dereference (as s7's reader does).
+ plcValues[tagResult.name] = valueOrNull(tagResult)
+ if tagResult.err != nil {
+ // The response carries only the code, so this is the one place the reason is
+ // known.
+ m.log.Debug().Err(tagResult.err).Str("tagName", tagResult.name).Msg("Couldn't answer a tag out of the block's response")
+ failures = append(failures, tagResult.err)
+ }
}
+ }
- // A request that nobody answers must not wait forever; the codec derives the lifetime of the
- // expectation it registers from the deadline of the context it is handed.
- // The cancel function must not be deferred here: SendRequest only registers the
- // expectation and returns, so this goroutine finishes long before the response arrives, and
- // cancelling here would take the expectation down with it. It is called once a result has
- // been delivered instead, which every path below does; a request that is never answered is
- // released by its own deadline.
- requestCtx, cancelRequest := withRequestTimeout(ctx, m.configuration.requestTimeout)
- deliver = func(response apiModel.PlcReadResponse, err error) {
- cancelRequest()
- utils.DeliverResult(m.log, result, spiModel.NewDefaultPlcReadRequestResult(readRequest, response, err))
+ m.log.Trace().Msg("Returning the response")
+ return spiModel.NewDefaultPlcReadResponse(readRequest, responseCodes, plcValues), errors.Join(failures...)
+}
+
+// invalidTag is a tag of the request that isn't a modbus tag at all, kept in the order the request
+// named it so that the failures are reported in a stable one.
+type invalidTag struct {
+ name string
+ err error
+}
+
+// modbusTagsOf reads the request's tags, separating the ones this driver can work with from the
+// ones it can't.
+func modbusTagsOf(readRequest apiModel.PlcReadRequest) ([]namedTag, []invalidTag) {
+ tagNames := readRequest.GetTagNames()
+ tags := make([]namedTag, 0, len(tagNames))
+ var invalid []invalidTag
+ for _, tagName := range tagNames {
+ tag := readRequest.GetTag(tagName)
+ if unparsed, ok := tag.(unparsedTag); ok {
+ // An address the builder couldn't parse, kept in the request so that it can be
+ // reported rather than taking the request down (see ReadRequestBuilder.go). Its own
+ // error says what was wrong with the address, which a failed cast wouldn't.
+ invalid = append(invalid, invalidTag{name: tagName, err: errors.Wrapf(unparsed.err, "invalid address for '%s'", tagName)})
+ continue
}
- pdu, err := readRequestPdu(modbusTagVar)
+ modbusTagVar, err := castToModbusTagFromPlcTag(tag)
if err != nil {
- deliver(nil, err)
- m.log.Debug().Err(err).Stringer("tagType", modbusTagVar.TagType).Msg("Couldn't build a read request")
- return
+ invalid = append(invalid, invalidTag{name: tagName, err: errors.Wrapf(err, "invalid tag item type for '%s'", tagName)})
+ continue
}
+ tags = append(tags, namedTag{name: tagName, tag: modbusTagVar})
+ }
+ return tags, invalid
+}
+
+// valueOrNull is the value a tag ended up with; a tag that failed carries a null rather than
+// nothing, so that a caller who reads the value before looking at the code gets a PlcValue either
+// way.
+func valueOrNull(result tagReadResult) apiValues.PlcValue {
+ if result.value == nil {
+ return values.NewPlcNULL()
+ }
+ return result.value
+}
+
+// blockOutcome is how one block read ended: the payload the device answered with, or the code
+// every tag of the block gets instead. err is why the code is not OK, kept for logging - the
+// response itself carries only the code.
+type blockOutcome struct {
+ responseCode apiModel.PlcResponseCode
+ data []byte
+ err error
+}
- // Calculate a new transaction identifier
- transactionIdentifier := atomic.AddInt32(&m.transactionIdentifier, 1)
- if transactionIdentifier > math.MaxUint8 {
- transactionIdentifier = 1
- atomic.StoreInt32(&m.transactionIdentifier, 1)
+// readBlock sends one block's request and waits for its answer.
+func (m *Reader) readBlock(ctx context.Context, block readBlock) blockOutcome {
+ pdu, err := blockRequestPdu(block.tagType, block.address, block.quantity)
+ if err != nil {
+ m.log.Debug().Err(err).Stringer("tagType", block.tagType).Msg("Couldn't build a read request")
+ return blockOutcome{responseCode: apiModel.PlcResponseCode_UNSUPPORTED, err: err}
+ }
+
+ // Calculate a new transaction identifier
+ transactionIdentifier := atomic.AddInt32(&m.transactionIdentifier, 1)
+ if transactionIdentifier > math.MaxUint8 {
+ transactionIdentifier = 1
+ atomic.StoreInt32(&m.transactionIdentifier, 1)
+ }
+ m.log.Debug().Int32("transactionIdentifier", transactionIdentifier).Msg("Calculated transaction identifier")
+
+ // Assemble the finished ADU. Which one that is depends on the flavor the connection
+ // speaks - the TCP one carries the transaction identifier, the RTU one a CRC.
+ m.log.Trace().Msg("Assemble ADU")
+ adus := m.configuration.adus()
+ unitIdentifier := m.configuration.unitIdentifier
+ if block.unitId != nil {
+ // Every tag of the block named this unit; a block read reaches exactly one device.
+ unitIdentifier = *block.unitId
+ }
+ requestAdu := adus.buildRequest(uint16(transactionIdentifier), unitIdentifier, pdu)
+
+ // A request that nobody answers must not wait forever; the codec derives the lifetime of the
+ // expectation it registers from the deadline of the context it is handed. Cancelling once the
+ // block is done releases the timer and drops an expectation that is somehow still registered,
+ // which is safe here because this waits for the answer rather than returning ahead of it.
+ requestCtx, cancelRequest := withRequestTimeout(ctx, m.configuration.requestTimeout)
+ defer cancelRequest()
+
+ // Buffered and written to without blocking, so that a handler firing for a block that is
+ // already accounted for returns straight away - a codec that reports a send failure and then
+ // times the expectation out anyway would otherwise leave that handler blocked forever, and
+ // blocked handlers pile up in the codec's WaitGroup and wedge Disconnect.
+ outcomes := make(chan blockOutcome, 1)
+ complete := func(outcome blockOutcome) {
+ select {
+ case outcomes <- outcome:
+ default:
+ m.log.Debug().Msg("dropping a second outcome for a block that is already answered")
}
- m.log.Debug().Int32("transactionIdentifier", transactionIdentifier).Msg("Calculated transaction identifier")
-
- // Assemble the finished ADU. Which one that is depends on the flavor the connection
- // speaks - the TCP one carries the transaction identifier, the RTU one a CRC.
- m.log.Trace().Msg("Assemble ADU")
- adus := m.configuration.adus()
- requestAdu := adus.buildRequest(uint16(transactionIdentifier), modbusTagVar.resolveUnitId(m.configuration.unitIdentifier), pdu)
-
- // Send the ADU over the wire
- m.log.Trace().Msg("Send ADU")
- if err = m.messageCodec.SendRequest(requestCtx, "read", requestAdu, func(message spi.Message) bool {
- return adus.acceptsResponse(requestAdu, message)
- }, func(message spi.Message) error {
- // Convert the response into an ADU
- m.log.Trace().Msg("convert response to ADU")
- responsePdu, err := adus.extractPdu(message)
- if err != nil {
- deliver(nil, err)
- return nil
- }
- // Convert the modbus response into a PLC4X response
- m.log.Trace().Msg("convert response to PLC4X response")
- readResponse, err := m.toPlc4xReadResponse(responsePdu, readRequest)
-
- if err != nil {
- deliver(nil, errors.Wrap(err, "Error decoding response"))
- // TODO: should we return the error here?
- return nil
- }
- deliver(readResponse, nil)
- return nil
- }, func(err error) error {
- deliver(nil, errors.Wrap(err, "got timeout while waiting for response"))
+ }
+
+ m.log.Trace().Msg("Send ADU")
+ if err := m.messageCodec.SendRequest(requestCtx, "read", requestAdu, func(message spi.Message) bool {
+ return adus.acceptsResponse(requestAdu, message)
+ }, func(message spi.Message) error {
+ // Convert the response into an ADU
+ m.log.Trace().Msg("convert response to ADU")
+ responsePdu, err := adus.extractPdu(message)
+ if err != nil {
+ complete(blockOutcome{responseCode: apiModel.PlcResponseCode_INTERNAL_ERROR, err: err})
return nil
- }); err != nil {
- deliver(nil, errors.Wrap(err, "error sending message"))
}
- })
- return result
-}
+ data, responseCode, err := extractResponseData(responsePdu)
+ complete(blockOutcome{responseCode: responseCode, data: data, err: err})
+ return nil
+ }, func(err error) error {
+ complete(blockOutcome{
+ responseCode: apiModel.PlcResponseCode_REQUEST_TIMEOUT,
+ err: errors.Wrap(err, "got timeout while waiting for response"),
+ })
+ return nil
+ }); err != nil {
+ complete(blockOutcome{
+ responseCode: apiModel.PlcResponseCode_INTERNAL_ERROR,
+ err: errors.Wrap(err, "error sending message"),
+ })
+ }
-// ToPlc4xReadResponse turns the ADU a device answered with into a PLC4X response. It takes the
-// ADU through the little bit of it this needs, so that it serves every flavor - the generated
-// ModbusADU parent type carries no PDU accessor of its own.
-func (m *Reader) ToPlc4xReadResponse(responseAdu aduWithPdu, readRequest apiModel.PlcReadRequest) (apiModel.PlcReadResponse, error) {
- return m.toPlc4xReadResponse(responseAdu.GetPdu(), readRequest)
+ select {
+ case outcome := <-outcomes:
+ return outcome
+ case <-requestCtx.Done():
+ // The codec answers every expectation it registers, so this is the caller giving up rather
+ // than the device being slow. An outcome that landed in the same instant is preferred over
+ // it - select picks at random between two ready cases, and a delivered answer is the
+ // truthful one.
+ select {
+ case outcome := <-outcomes:
+ return outcome
+ default:
+ }
+ return blockOutcome{responseCode: apiModel.PlcResponseCode_REQUEST_TIMEOUT, err: requestCtx.Err()}
+ }
}
-func (m *Reader) toPlc4xReadResponse(responsePdu readWriteModel.ModbusPDU, readRequest apiModel.PlcReadRequest) (apiModel.PlcReadResponse, error) {
- ctx := context.TODO()
- var data []uint8
+// extractResponseData pulls the payload out of a read response: the packed bits for the two bit
+// areas, the registers for the other three. The response code is what every tag the request
+// covered gets; it is OK exactly when there is a payload to answer them out of.
+func extractResponseData(responsePdu readWriteModel.ModbusPDU) ([]byte, apiModel.PlcResponseCode, error) {
switch pdu := responsePdu.(type) {
case readWriteModel.ModbusPDUReadDiscreteInputsResponse:
- data = pdu.GetValue()
- // Pure Boolean ...
+ return pdu.GetValue(), apiModel.PlcResponseCode_OK, nil
case readWriteModel.ModbusPDUReadCoilsResponse:
- data = pdu.GetValue()
- // Pure Boolean ...
+ return pdu.GetValue(), apiModel.PlcResponseCode_OK, nil
case readWriteModel.ModbusPDUReadInputRegistersResponse:
- data = pdu.GetValue()
- // DataIo ...
+ return pdu.GetValue(), apiModel.PlcResponseCode_OK, nil
case readWriteModel.ModbusPDUReadHoldingRegistersResponse:
- data = pdu.GetValue()
+ return pdu.GetValue(), apiModel.PlcResponseCode_OK, nil
case readWriteModel.ModbusPDUReadFileRecordResponse:
// A request that crossed a file boundary was sent as one item per file, and the response
// carries one item per request item, so the value is the registers of all of them in
// order. plc4j's extractResponseData only looks at the first item, which silently drops
// everything behind the boundary.
+ var data []byte
for _, item := range pdu.GetItems() {
data = append(data, item.GetData()...)
}
+ return data, apiModel.PlcResponseCode_OK, nil
case readWriteModel.ModbusPDUError:
- return nil, errors.Errorf("got an error from remote. Errorcode %x", pdu.GetExceptionCode())
+ responseCode, mapped := responseCodeOf(pdu.GetExceptionCode())
+ switch {
+ case !mapped:
+ // An exception code the specification doesn't define. The device refused; which way it
+ // spelled that is all that is unknown.
+ responseCode = apiModel.PlcResponseCode_REMOTE_ERROR
+ case responseCode == apiModel.PlcResponseCode_OK:
+ // ACKNOWLEDGE is the one exception plc4j maps to OK: the device took the request and
+ // will take its time over it. For a write that is a fair answer, for a read it is not
+ // - an exception response carries no registers, so OK would promise the caller a value
+ // that never arrived.
+ responseCode = apiModel.PlcResponseCode_REMOTE_BUSY
+ }
+ return nil, responseCode, errors.Errorf("got an error from remote. Errorcode %x", pdu.GetExceptionCode())
default:
- return nil, errors.Errorf("unsupported response type %T", pdu)
+ return nil, apiModel.PlcResponseCode_INTERNAL_ERROR, errors.Errorf("unsupported response type %T", pdu)
}
+}
+
+// ToPlc4xReadResponse turns the ADU a device answered with into a PLC4X response. It takes the
+// ADU through the little bit of it this needs, so that it serves every flavor - the generated
+// ModbusADU parent type carries no PDU accessor of its own.
+//
+// One ADU answers one block, so this only serves a request the optimizer merges into a single one.
+func (m *Reader) ToPlc4xReadResponse(responseAdu aduWithPdu, readRequest apiModel.PlcReadRequest) (apiModel.PlcReadResponse, error) {
+ return m.toPlc4xReadResponse(responseAdu.GetPdu(), readRequest)
+}
- // Get the tag from the request
- m.log.Trace().Msg("get a tag from request")
- tagName := readRequest.GetTagNames()[0]
- tag, err := castToModbusTagFromPlcTag(readRequest.GetTag(tagName))
+func (m *Reader) toPlc4xReadResponse(responsePdu readWriteModel.ModbusPDU, readRequest apiModel.PlcReadRequest) (apiModel.PlcReadResponse, error) {
+ tags, invalid := modbusTagsOf(readRequest)
+ if len(invalid) > 0 {
+ return nil, invalid[0].err
+ }
+ blocks := optimizeReads(tags, m.configuration.maxCoilsPerRequest, m.configuration.maxRegistersPerRequest)
+ if len(blocks) != 1 {
+ return nil, errors.Errorf("this request is read as %d blocks, so a single response can't answer it", len(blocks))
+ }
+ data, responseCode, err := extractResponseData(responsePdu)
if err != nil {
- return nil, errors.Wrap(err, "error casting to modbus-tag")
+ return nil, err
}
// Decode the data according to the information from the request
m.log.Trace().Msg("decode data")
- ctxForModel := options.GetLoggerContextForModel(ctx, m.log, options.WithPassLoggerToModel(m.passLogToModel))
- value, err := ParseRegisters(ctxForModel, data, tag.Datatype, tag.Quantity, tag.resolveByteOrder(m.configuration.defaultPayloadByteOrder), tag.StringLength)
- if err != nil {
- return nil, errors.Wrap(err, "Error parsing data item")
- }
+ ctxForModel := options.GetLoggerContextForModel(context.TODO(), m.log, options.WithPassLoggerToModel(m.passLogToModel))
responseCodes := map[string]apiModel.PlcResponseCode{}
plcValues := map[string]apiValues.PlcValue{}
- plcValues[tagName] = value
- responseCodes[tagName] = apiModel.PlcResponseCode_OK
+ var failures []error
+ for _, tagResult := range splitBlockResponse(ctxForModel, blocks[0], responseCode, data, m.configuration.defaultPayloadByteOrder) {
+ responseCodes[tagResult.name] = tagResult.responseCode
+ plcValues[tagResult.name] = valueOrNull(tagResult)
+ if tagResult.err != nil {
+ failures = append(failures, tagResult.err)
+ }
+ }
// Return the response
m.log.Trace().Msg("Returning the response")
- return spiModel.NewDefaultPlcReadResponse(readRequest, responseCodes, plcValues), nil
+ return spiModel.NewDefaultPlcReadResponse(readRequest, responseCodes, plcValues), errors.Join(failures...)
}
diff --git a/plc4go/internal/modbus/Reader_test.go b/plc4go/internal/modbus/Reader_test.go
index c9fe7fc3d6..f5f7686203 100644
--- a/plc4go/internal/modbus/Reader_test.go
+++ b/plc4go/internal/modbus/Reader_test.go
@@ -20,6 +20,7 @@
package modbus
import (
+ "encoding/hex"
"testing"
"time"
@@ -27,10 +28,12 @@ import (
"github.com/stretchr/testify/require"
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/modbus/readwrite/model"
"github.com/apache/plc4x/plc4go/spi/errors"
spiModel "github.com/apache/plc4x/plc4go/spi/model"
"github.com/apache/plc4x/plc4go/spi/testutils"
+ "github.com/apache/plc4x/plc4go/spi/utils"
)
// A caller whose context expired abandons the result channel without reading.
@@ -201,3 +204,302 @@ func TestReader_decodesInTheResolvedByteOrder(t *testing.T) {
})
}
}
+
+// addressedTag is one entry of a read request, kept in a slice so that the order the request names
+// its tags in - which is the order the optimizer sees them in - is part of the test.
+type addressedTag struct {
+ name string
+ address string
+}
+
+// runRead performs a read of the given addresses and answers every block that goes out with the
+// next of the given PDUs. It returns the result the caller ends up with and the requests that went
+// onto the wire, so that a test can check both what was asked and what came back.
+//
+// One response per block, in order: the reader sends the blocks one at a time, so the n-th
+// response answers the n-th request.
+func runRead(t *testing.T, configuration Configuration, requested []addressedTag, responses ...readWriteModel.ModbusPDU) (apiModel.PlcReadRequestResult, []readWriteModel.ModbusTcpADU) {
+ t.Helper()
+ codec := newCaptureCodec(nil)
+ reader := NewReader(configuration, codec)
+ tags := make(map[string]apiModel.PlcTag, len(requested))
+ tagNames := make([]string, 0, len(requested))
+ for _, entry := range requested {
+ tags[entry.name] = parseTag(t, entry.address)
+ tagNames = append(tagNames, entry.name)
+ }
+ results := reader.Read(testutils.TestContext(t), spiModel.NewDefaultPlcReadRequest(tags, tagNames, reader, nil))
+
+ sent := make([]readWriteModel.ModbusTcpADU, 0, len(responses))
+ for _, response := range responses {
+ select {
+ case handlers := <-codec.handlers:
+ requestAdu, ok := handlers.message.(readWriteModel.ModbusTcpADU)
+ require.True(t, ok, "expected a ModbusTcpADU, got %T", handlers.message)
+ sent = append(sent, requestAdu)
+ responseAdu := readWriteModel.NewModbusTcpADU(requestAdu.GetTransactionIdentifier(), requestAdu.GetUnitIdentifier(), response)
+ require.True(t, handlers.acceptsMessage(responseAdu), "the reader didn't accept the response to its own request")
+ require.NoError(t, handlers.handleMessage(responseAdu))
+ case <-time.After(time.Second):
+ t.Fatalf("only %d of the %d expected block requests were sent", len(sent), len(responses))
+ }
+ }
+
+ select {
+ case result := <-results:
+ return result, sent
+ case <-time.After(2 * time.Second):
+ t.Fatal("no result was delivered")
+ return nil, nil
+ }
+}
+
+func holdingRegisterRequestOf(t *testing.T, adu readWriteModel.ModbusTcpADU) readWriteModel.ModbusPDUReadHoldingRegistersRequest {
+ t.Helper()
+ pdu, ok := adu.GetPdu().(readWriteModel.ModbusPDUReadHoldingRegistersRequest)
+ require.True(t, ok, "expected a read-holding-registers request, got %T", adu.GetPdu())
+ return pdu
+}
+
+// pi is 3.1415927 as a REAL occupies two registers, the value the driver testsuite reads.
+var pi = []byte{0x40, 0x49, 0x0f, 0xdb}
+
+// Tags that one request can cover are read with one request, and the single response answers all
+// of them. This is what the optimizer is for: without it every tag costs a round trip, which is
+// ruinous behind a TCP<->RTU gateway.
+func TestReader_mergesTagsIntoOneRequest(t *testing.T) {
+ result, sent := runRead(t, DefaultConfiguration(),
+ []addressedTag{{"first", "holding-register:1:REAL"}, {"second", "holding-register:3:REAL"}},
+ readWriteModel.NewModbusPDUReadHoldingRegistersResponse(append(append([]byte{}, pi...), pi...)))
+
+ require.NoError(t, result.GetErr())
+ require.Len(t, sent, 1, "both tags fit into one request")
+ pdu := holdingRegisterRequestOf(t, sent[0])
+ assert.Equal(t, uint16(0), pdu.GetStartingAddress())
+ assert.Equal(t, uint16(4), pdu.GetQuantity())
+
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ for _, tagName := range []string{"first", "second"} {
+ assert.Equal(t, apiModel.PlcResponseCode_OK, response.GetResponseCode(tagName), tagName)
+ assert.InDelta(t, 3.1415927, response.GetValue(tagName).GetFloat32(), 0.0000001, tagName)
+ }
+}
+
+// Tags nothing can cover with one request are read with one request each, in the order the
+// optimizer emits the blocks.
+func TestReader_asksOncePerBlock(t *testing.T) {
+ // Two registers apart is inside the 125 register window; 500 registers apart is not.
+ result, sent := runRead(t, DefaultConfiguration(),
+ []addressedTag{{"near", "holding-register:1:REAL"}, {"far", "holding-register:501:REAL"}},
+ readWriteModel.NewModbusPDUReadHoldingRegistersResponse(pi),
+ readWriteModel.NewModbusPDUReadHoldingRegistersResponse(pi))
+
+ require.NoError(t, result.GetErr())
+ require.Len(t, sent, 2)
+ assert.Equal(t, uint16(0), holdingRegisterRequestOf(t, sent[0]).GetStartingAddress())
+ assert.Equal(t, uint16(2), holdingRegisterRequestOf(t, sent[0]).GetQuantity())
+ assert.Equal(t, uint16(500), holdingRegisterRequestOf(t, sent[1]).GetStartingAddress())
+ assert.Equal(t, uint16(2), holdingRegisterRequestOf(t, sent[1]).GetQuantity())
+
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ assert.Equal(t, apiModel.PlcResponseCode_OK, response.GetResponseCode("near"))
+ assert.Equal(t, apiModel.PlcResponseCode_OK, response.GetResponseCode("far"))
+}
+
+// Merging reads means one rejected address takes its whole block down - that is the price of the
+// optimization. What it must not do is take the rest of the request down with it.
+func TestReader_aFailedBlockFailsOnlyItsOwnTags(t *testing.T) {
+ result, sent := runRead(t, DefaultConfiguration(),
+ []addressedTag{
+ {"merged1", "holding-register:1:REAL"},
+ {"merged2", "holding-register:3:REAL"},
+ {"separate", "holding-register:501:REAL"},
+ },
+ readWriteModel.NewModbusPDUError(readWriteModel.ModbusErrorCode_ILLEGAL_DATA_ADDRESS),
+ readWriteModel.NewModbusPDUReadHoldingRegistersResponse(pi))
+
+ require.Len(t, sent, 2)
+ // The failure is reported as well as being visible per tag, the way the single-item
+ // interceptor reports a partial result.
+ require.Error(t, result.GetErr())
+
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ assert.Equal(t, apiModel.PlcResponseCode_INVALID_ADDRESS, response.GetResponseCode("merged1"))
+ assert.Equal(t, apiModel.PlcResponseCode_INVALID_ADDRESS, response.GetResponseCode("merged2"))
+ assert.Equal(t, apiModel.PlcResponseCode_OK, response.GetResponseCode("separate"))
+ assert.InDelta(t, 3.1415927, response.GetValue("separate").GetFloat32(), 0.0000001)
+}
+
+// A tag the device never answers takes its block down with a timeout, and the rest of the request
+// is still delivered.
+func TestReader_aBlockThatIsNeverAnsweredTimesOut(t *testing.T) {
+ configuration := DefaultConfiguration()
+ configuration.requestTimeout = 50 * time.Millisecond
+ codec := newCaptureCodec(nil)
+ reader := NewReader(configuration, codec)
+ tag := parseTag(t, "holding-register:1:REAL")
+ request := spiModel.NewDefaultPlcReadRequest(
+ map[string]apiModel.PlcTag{"tag": tag}, []string{"tag"}, reader, nil)
+
+ results := reader.Read(testutils.TestContext(t), request)
+ select {
+ case <-codec.handlers:
+ case <-time.After(time.Second):
+ t.Fatal("SendRequest was never invoked")
+ }
+
+ select {
+ case result := <-results:
+ require.Error(t, result.GetErr())
+ require.NotNil(t, result.GetResponse())
+ assert.Equal(t, apiModel.PlcResponseCode_REQUEST_TIMEOUT, result.GetResponse().GetResponseCode("tag"))
+ case <-time.After(2 * time.Second):
+ t.Fatal("the unanswered block never released the request")
+ }
+}
+
+// Merging is bounded by max-registers-per-request, so a device that can't answer a full width
+// request is asked for less rather than for something it rejects.
+func TestReader_honoursTheConfiguredCeilings(t *testing.T) {
+ configuration := DefaultConfiguration()
+ configuration.maxRegistersPerRequest = 4
+
+ result, sent := runRead(t, configuration,
+ []addressedTag{
+ {"first", "holding-register:1:REAL"},
+ {"second", "holding-register:3:REAL"},
+ {"third", "holding-register:5:REAL"},
+ },
+ readWriteModel.NewModbusPDUReadHoldingRegistersResponse(append(append([]byte{}, pi...), pi...)),
+ readWriteModel.NewModbusPDUReadHoldingRegistersResponse(pi))
+
+ require.NoError(t, result.GetErr())
+ require.Len(t, sent, 2, "four registers is all one request may carry here")
+ assert.Equal(t, uint16(4), holdingRegisterRequestOf(t, sent[0]).GetQuantity())
+ assert.Equal(t, uint16(2), holdingRegisterRequestOf(t, sent[1]).GetQuantity())
+}
+
+// A one-tag read asks for exactly what it asked for before anything was merged: the block the
+// optimizer builds around a lone tag is that tag's own read.
+func TestReader_aSingleTagRequestIsUnchanged(t *testing.T) {
+ // GOLDEN BYTES, captured from the implementation as it stood BEFORE the optimizer
+ // was wired in (commit 1033e8b897, via a detached worktree). They are written out
+ // literally on purpose.
+ //
+ // The obvious way to write this test - build the expected PDU with readRequestPdu
+ // and compare - is worthless, because wiring the optimizer in rewrote that function
+ // to share the block-building code. Both sides of the comparison then move together,
+ // so the test passes however wrong the shared code becomes. That version of this test
+ // was checked by mutation: making tagSpan return one register too many for every tag
+ // left all seven cases green.
+ for _, tc := range []struct {
+ address string
+ pdu string
+ }{
+ {"coil:1:BOOL", "0100000001"},
+ {"coil:1[0..4]:BOOL", "0100000005"},
+ {"discrete-input:1:BOOL", "0200000001"},
+ {"holding-register:1:REAL", "0300000002"},
+ {"holding-register:1:STRING(20)", "030000000a"},
+ {"input-register:7:UDINT", "0400060002"},
+ {"extended-register:9998:DINT", "1407060001270e0002"},
+ } {
+ t.Run(tc.address, func(t *testing.T) {
+ adu := capturedReadRequest(t, DefaultConfiguration(), parseTag(t, tc.address))
+
+ wb := utils.NewWriteBufferByteBased()
+ require.NoError(t, adu.GetPdu().SerializeWithWriteBuffer(t.Context(), wb))
+
+ assert.Equal(t, tc.pdu, hex.EncodeToString(wb.GetBytes()),
+ "a single-tag read must put exactly the bytes on the wire it did before the optimizer")
+ })
+ }
+}
+
+// Coils are packed least significant bit first: the first coil of the response sits in bit 0 of
+// the first byte, and a run of them crosses into the next byte at bit 8 rather than restarting
+// there.
+//
+// This is a behaviour CHANGE. The unoptimized path decoded coils through ParseRegisters, which
+// reads bits most significant first, so it returned the block reversed - and for a single coil it
+// skipped 15 bits into a one byte response and failed outright. Both are fixed by reading the
+// response the way the specification packs it.
+func TestReader_readsCoilsLeastSignificantBitFirst(t *testing.T) {
+ t.Run("a single coil", func(t *testing.T) {
+ result, sent := runRead(t, DefaultConfiguration(),
+ []addressedTag{{"tag", "coil:1:BOOL"}},
+ readWriteModel.NewModbusPDUReadCoilsResponse([]byte{0x01}))
+
+ require.NoError(t, result.GetErr())
+ require.Len(t, sent, 1)
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ assert.Equal(t, apiModel.PlcResponseCode_OK, response.GetResponseCode("tag"))
+ assert.True(t, response.GetValue("tag").GetBool())
+ })
+
+ t.Run("a run of coils", func(t *testing.T) {
+ // 0x05 is coils 1, 2 and 3 reading on, off, on.
+ result, _ := runRead(t, DefaultConfiguration(),
+ []addressedTag{{"tag", "coil:1[0..2]:BOOL"}},
+ readWriteModel.NewModbusPDUReadCoilsResponse([]byte{0x05}))
+
+ require.NoError(t, result.GetErr())
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ require.Equal(t, apiModel.PlcResponseCode_OK, response.GetResponseCode("tag"))
+ read := response.GetValue("tag").GetList()
+ require.Len(t, read, 3)
+ assert.Equal(t, []bool{true, false, true},
+ []bool{read[0].GetBool(), read[1].GetBool(), read[2].GetBool()})
+ })
+
+ t.Run("coils of separate tags out of one block", func(t *testing.T) {
+ // Two coils eight apart, merged into one nine-coil read: the first sits in bit 0 of the
+ // first byte, the second in bit 0 of the second.
+ result, sent := runRead(t, DefaultConfiguration(),
+ []addressedTag{{"low", "coil:1:BOOL"}, {"high", "coil:9:BOOL"}},
+ readWriteModel.NewModbusPDUReadCoilsResponse([]byte{0x01, 0x00}))
+
+ require.NoError(t, result.GetErr())
+ require.Len(t, sent, 1)
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ assert.True(t, response.GetValue("low").GetBool())
+ assert.False(t, response.GetValue("high").GetBool())
+ })
+}
+
+// An exception code the specification doesn't define still says the device refused; only the way
+// it spelled that is unknown. The tag has to come back as REMOTE_ERROR rather than as a code that
+// claims to know more than the response does.
+func TestReader_anUnmappedExceptionCodeIsARemoteError(t *testing.T) {
+ result, _ := runRead(t, DefaultConfiguration(),
+ []addressedTag{{"tag", "holding-register:1:REAL"}},
+ // 0x09 is one of the gaps the specification leaves between the codes it defines.
+ readWriteModel.NewModbusPDUError(readWriteModel.ModbusErrorCode(9)))
+
+ require.Error(t, result.GetErr())
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ assert.Equal(t, apiModel.PlcResponseCode_REMOTE_ERROR, response.GetResponseCode("tag"))
+}
+
+// A tag that failed still carries a value, a null one, so that a caller who reads the value before
+// looking at the code gets a PlcValue rather than something to dereference (as s7's reader does).
+func TestReader_aFailedTagCarriesANullValue(t *testing.T) {
+ result, _ := runRead(t, DefaultConfiguration(),
+ []addressedTag{{"tag", "holding-register:1:REAL"}},
+ readWriteModel.NewModbusPDUError(readWriteModel.ModbusErrorCode_ILLEGAL_DATA_ADDRESS))
+
+ require.Error(t, result.GetErr())
+ response := result.GetResponse()
+ require.NotNil(t, response)
+ require.Equal(t, apiModel.PlcResponseCode_INVALID_ADDRESS, response.GetResponseCode("tag"))
+ value := response.GetValue("tag")
+ require.NotNil(t, value)
+ assert.Equal(t, apiValues.NULL, value.GetPlcValueType())
+}
diff --git a/plc4go/tests/configparity/parameter_names_test.go b/plc4go/tests/configparity/parameter_names_test.go
index a6c622bc87..678567843d 100644
--- a/plc4go/tests/configparity/parameter_names_test.go
+++ b/plc4go/tests/configparity/parameter_names_test.go
@@ -75,6 +75,8 @@ func TestTheCanonicalNamesAreRecognisedHere(t *testing.T) {
"default-unit-identifier": {"3"},
"request-timeout-ms": {"5000"},
"default-payload-byte-order": {"LITTLE_ENDIAN"},
+ "max-coils-per-request": {"2000"},
+ "max-registers-per-request": {"125"},
})
assert.NotContains(t, logged, "not known", "every canonical modbus name must be read here")
diff --git a/protocols/modbus/src/test/resources/protocols/modbus/tcp/DriverTestsuite.xml b/protocols/modbus/src/test/resources/protocols/modbus/tcp/DriverTestsuite.xml
index cc6976df48..2a175e94ec 100644
--- a/protocols/modbus/src/test/resources/protocols/modbus/tcp/DriverTestsuite.xml
+++ b/protocols/modbus/src/test/resources/protocols/modbus/tcp/DriverTestsuite.xml
@@ -235,7 +235,7 @@
</tags>
</TestReadRequest>
</api-request>
- <outgoing-plc-message name="Send First Item Modbus Input-Register Read Request">
+ <outgoing-plc-message name="Send Read Request for optimized chunk of data">
<parser-arguments>
<driverType>MODBUS_TCP</driverType>
<response>false</response>
@@ -252,14 +252,14 @@
<functionFlag dataType="uint" bitLength="7">3</functionFlag>
<ModbusPDUReadHoldingRegistersRequest>
<startingAddress dataType="uint" bitLength="16">0</startingAddress>
- <quantity dataType="uint" bitLength="16">2</quantity>
+ <quantity dataType="uint" bitLength="16">4</quantity>
</ModbusPDUReadHoldingRegistersRequest>
</ModbusPDU>
</pdu>
</ModbusTcpADU>
</ModbusADU>
</outgoing-plc-message>
- <incoming-plc-message name="Receive First Item Modbus Input-Register Read Response">
+ <incoming-plc-message name="Send Read Response for optimized chunk of data">
<parser-arguments>
<driverType>MODBUS_TCP</driverType>
<response>true</response>
@@ -268,63 +268,15 @@
<ModbusTcpADU>
<transactionIdentifier dataType="uint" bitLength="16">1</transactionIdentifier>
<protocolIdentifier dataType="uint" bitLength="16">0</protocolIdentifier>
- <length dataType="uint" bitLength="16">7</length>
- <unitIdentifier dataType="uint" bitLength="8">1</unitIdentifier>
- <pdu>
- <ModbusPDU>
- <errorFlag dataType="bit" bitLength="1">false</errorFlag>
- <functionFlag dataType="uint" bitLength="7">3</functionFlag>
- <ModbusPDUReadHoldingRegistersResponse>
- <byteCount dataType="uint" bitLength="8">4</byteCount>
- <value dataType="byte" bitLength="32">0x40490fdb</value>
- </ModbusPDUReadHoldingRegistersResponse>
- </ModbusPDU>
- </pdu>
- </ModbusTcpADU>
- </ModbusADU>
- </incoming-plc-message>
- <outgoing-plc-message name="Send Second Item Modbus Input-Register Read Request">
- <parser-arguments>
- <driverType>MODBUS_TCP</driverType>
- <response>false</response>
- </parser-arguments>
- <ModbusADU>
- <ModbusTcpADU>
- <transactionIdentifier dataType="uint" bitLength="16">2</transactionIdentifier>
- <protocolIdentifier dataType="uint" bitLength="16">0</protocolIdentifier>
- <length dataType="uint" bitLength="16">6</length>
- <unitIdentifier dataType="uint" bitLength="8">1</unitIdentifier>
- <pdu>
- <ModbusPDU>
- <errorFlag dataType="bit" bitLength="1">false</errorFlag>
- <functionFlag dataType="uint" bitLength="7">3</functionFlag>
- <ModbusPDUReadHoldingRegistersRequest>
- <startingAddress dataType="uint" bitLength="16">2</startingAddress>
- <quantity dataType="uint" bitLength="16">2</quantity>
- </ModbusPDUReadHoldingRegistersRequest>
- </ModbusPDU>
- </pdu>
- </ModbusTcpADU>
- </ModbusADU>
- </outgoing-plc-message>
- <incoming-plc-message name="Receive Second Item Modbus Input-Register Read Response">
- <parser-arguments>
- <driverType>MODBUS_TCP</driverType>
- <response>true</response>
- </parser-arguments>
- <ModbusADU>
- <ModbusTcpADU>
- <transactionIdentifier dataType="uint" bitLength="16">2</transactionIdentifier>
- <protocolIdentifier dataType="uint" bitLength="16">0</protocolIdentifier>
- <length dataType="uint" bitLength="16">7</length>
+ <length dataType="uint" bitLength="16">11</length>
<unitIdentifier dataType="uint" bitLength="8">1</unitIdentifier>
<pdu>
<ModbusPDU>
<errorFlag dataType="bit" bitLength="1">false</errorFlag>
<functionFlag dataType="uint" bitLength="7">3</functionFlag>
<ModbusPDUReadHoldingRegistersResponse>
- <byteCount dataType="uint" bitLength="8">4</byteCount>
- <value dataType="byte" bitLength="32">0x40490fdb</value>
+ <byteCount dataType="uint" bitLength="8">8</byteCount>
+ <value dataType="byte" bitLength="64">0x40490fdb40490fdb</value>
</ModbusPDUReadHoldingRegistersResponse>
</ModbusPDU>
</pdu>