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
The following commit(s) were added to refs/heads/develop by this push:
new 711fb31206 feat(plc4go): decode BACnet array/bit-string properties,
multi-target WhoIs, and codec receive fixes
711fb31206 is described below
commit 711fb312069e2f6c3dd60cdbb706d9949fe5c370
Author: Sebastian Rühl <[email protected]>
AuthorDate: Wed Jun 24 13:05:38 2026 +0200
feat(plc4go): decode BACnet array/bit-string properties, multi-target
WhoIs, and codec receive fixes
Further hardening of the pure-Go BACnet/IP driver so it can drive real
device discovery, capability detection, and COV.
- ValueDecoder: decode array-valued ConstructedData (OBJECT_LIST, ...) to a
PlcList — and an array-index-0 read to the element count — instead of
stringifying it; decode tagged bit-strings (PROTOCOL_SERVICES_SUPPORTED,
STATUS_FLAGS, ...) to a packed byte array. Object enumeration and
services-supported detection now work. (+ValueDecoder_test)
- Discoverer: add the `remote-addresses` option — a broadcast WhoIs plus a
directed unicast WhoIs to each listed host from a single socket — for
environments where a broadcast IAm is not routed back to the sender. New
DiscovererControl_{linux,other} tunes the socket
(SO_REUSEADDR/REUSEPORT/BROADCAST/BINDTODEVICE). (+Discoverer_test)
- udp TransportInstance: set a fresh short read deadline before peeking the
buffer; the request-context deadline is sticky, so a deadline-less peek
kept
failing with i/o timeout and stranded late/unsolicited responses.
- Connection: block with a 100ms timeout in the incoming-message poll loop
instead of a non-blocking default that pegged a CPU core and starved the
codec receive worker, so responses were never read.
- Add ReadRoundtrip_test covering a read against a fake UDP device.
---
plc4go/internal/bacnetip/Connection.go | 7 +-
plc4go/internal/bacnetip/Discoverer.go | 177 ++++++++++++-----
.../internal/bacnetip/DiscovererControl_linux.go | 57 ++++++
.../internal/bacnetip/DiscovererControl_other.go | 32 +++
plc4go/internal/bacnetip/Discoverer_test.go | 50 +++++
plc4go/internal/bacnetip/ReadRoundtrip_test.go | 220 +++++++++++++++++++++
plc4go/internal/bacnetip/ValueDecoder.go | 86 ++++++++
plc4go/internal/bacnetip/ValueDecoder_test.go | 74 +++++++
plc4go/spi/transports/udp/TransportInstance.go | 7 +
9 files changed, 657 insertions(+), 53 deletions(-)
diff --git a/plc4go/internal/bacnetip/Connection.go
b/plc4go/internal/bacnetip/Connection.go
index 2277ad0d14..d995a7d329 100644
--- a/plc4go/internal/bacnetip/Connection.go
+++ b/plc4go/internal/bacnetip/Connection.go
@@ -25,6 +25,7 @@ import (
"runtime/debug"
"slices"
"sync"
+ "time"
"github.com/rs/zerolog"
@@ -128,10 +129,14 @@ func (c *Connection) Connect(ctx context.Context) error {
func (c *Connection) passToDefaultIncomingMessageChannel() {
incomingMessageChannel :=
c.messageCodec.GetDefaultIncomingMessageChannel()
+ // Block (with a short timeout so the Connect loop can re-check
IsConnected
+ // for shutdown) rather than busy-spinning with a default case. The
previous
+ // non-blocking select pegged a CPU core and starved the codec's receive
+ // worker, so request responses were never read from the socket.
select {
case message := <-incomingMessageChannel:
c.routeIncomingMessage(message)
- default:
+ case <-time.After(100 * time.Millisecond):
c.log.Trace().Msg("no incoming message")
}
}
diff --git a/plc4go/internal/bacnetip/Discoverer.go
b/plc4go/internal/bacnetip/Discoverer.go
index f7926a1eb7..d7616c56c5 100644
--- a/plc4go/internal/bacnetip/Discoverer.go
+++ b/plc4go/internal/bacnetip/Discoverer.go
@@ -27,6 +27,7 @@ import (
"strconv"
"strings"
"sync"
+ "syscall"
"time"
"github.com/rs/zerolog"
@@ -98,7 +99,13 @@ func (d *Discoverer) Discover(ctx context.Context, callback
func(event apiModel.
if err != nil {
return errors.Wrap(err, "error broadcasting and discovering")
}
- d.handleIncomingBVLCs(ctx, callback, incomingBVLCChannel)
+ // Dispatch received IAm/IHave to the callback in the background so this
+ // function can enforce the discovery window and tear down cleanly.
Running
+ // it synchronously here would block until ctx cancellation AND its
select
+ // had no ctx.Done branch, so Discover never returned.
+ d.wg.Go(func() {
+ d.handleIncomingBVLCs(ctx, callback, incomingBVLCChannel)
+ })
// Wait for the discovery window to elapse OR for the caller to cancel.
select {
case <-time.After(timeout):
@@ -111,7 +118,7 @@ func (d *Discoverer) Discover(ctx context.Context, callback
func(event apiModel.
}
func (d *Discoverer) broadcastAndDiscover(ctx context.Context,
communicationChannels []communicationChannel, specificOptions
*protocolSpecificOptions) (chan receivedBvlcMessage, error) {
- incomingBVLCChannel := make(chan receivedBvlcMessage)
+ incomingBVLCChannel := make(chan receivedBvlcMessage, 32)
for _, communicationChannelInstance := range communicationChannels {
if err := ctx.Err(); err != nil {
return incomingBVLCChannel, err
@@ -124,32 +131,55 @@ func (d *Discoverer) broadcastAndDiscover(ctx
context.Context, communicationChan
lowLimit =
driverModel.CreateBACnetContextTagUnsignedInteger(0, whoIsOptions.limits.low)
highLimit =
driverModel.CreateBACnetContextTagUnsignedInteger(1, whoIsOptions.limits.high)
}
- requestWhoIs :=
driverModel.NewBACnetUnconfirmedServiceRequestWhoIs(lowLimit, highLimit)
- apdu :=
driverModel.NewAPDUUnconfirmedRequest(requestWhoIs)
- control := driverModel.NewNPDUControl(false, false,
false, false, driverModel.NPDUNetworkPriority_NORMAL_MESSAGE)
- npdu := driverModel.NewNPDU(1, control, nil, nil, nil,
nil, nil, nil, nil, nil, apdu)
- bvlc := driverModel.NewBVLCOriginalUnicastNPDU(npdu)
+ // serializeWhoIs frames a WhoIs BVLC: an
Original-Broadcast-NPDU for a
+ // broadcast (so peers reply with a broadcast IAm), an
+ // Original-Unicast-NPDU for a directed WhoIs.
+ serializeWhoIs := func(directed bool) ([]byte, error) {
+ requestWhoIs :=
driverModel.NewBACnetUnconfirmedServiceRequestWhoIs(lowLimit, highLimit)
+ apdu :=
driverModel.NewAPDUUnconfirmedRequest(requestWhoIs)
+ control := driverModel.NewNPDUControl(false,
false, false, false, driverModel.NPDUNetworkPriority_NORMAL_MESSAGE)
+ npdu := driverModel.NewNPDU(1, control, nil,
nil, nil, nil, nil, nil, nil, nil, apdu)
+ var bvlc driverModel.BVLC
+ if directed {
+ bvlc =
driverModel.NewBVLCOriginalUnicastNPDU(npdu)
+ } else {
+ bvlc =
driverModel.NewBVLCOriginalBroadcastNPDU(npdu)
+ }
+ return bvlc.Serialize()
+ }
- // Send the search request.
- theBytes, err := bvlc.Serialize()
- if err != nil {
+ // Always broadcast on the interface (standard
discovery on a local
+ // subnet).
+ if theBytes, err := serializeWhoIs(false); err != nil {
return nil, err
+ } else if _, err :=
communicationChannelInstance.broadcastConnection.WriteTo(theBytes,
communicationChannelInstance.broadcastTarget); err != nil {
+ d.log.Debug().Err(err).Msg("Error sending
broadcast WhoIs")
}
- // Directed (unicast) WhoIs when a remote address is
supplied,
- // otherwise broadcast on the interface.
- sendConn :=
communicationChannelInstance.broadcastConnection
- target :=
communicationChannelInstance.broadcastConnection.LocalAddr()
+
+ // Additionally send a directed unicast WhoIs to each
explicit target.
+ // See remoteAddress / remoteAddresses for why this is
needed where a
+ // broadcast IAm is not routed back to the sender.
+ var directedTargets []string
if specificOptions.remoteAddress != "" {
- if udpAddr, rerr :=
resolveBacnetUDPAddr(specificOptions.remoteAddress,
specificOptions.bacNetPort); rerr == nil {
- sendConn =
communicationChannelInstance.unicastConnection
- target = udpAddr
- } else {
-
d.log.Warn().Err(rerr).Str("remoteAddress",
specificOptions.remoteAddress).Msg("invalid remote-address; falling back to
broadcast")
- }
+ directedTargets = append(directedTargets,
specificOptions.remoteAddress)
}
- if _, err := sendConn.WriteTo(theBytes, target); err !=
nil {
- d.log.Debug().Err(err).Msg("Error sending
WhoIs")
+ directedTargets = append(directedTargets,
specificOptions.remoteAddresses...)
+ if len(directedTargets) > 0 {
+ theBytes, err := serializeWhoIs(true)
+ if err != nil {
+ return nil, err
+ }
+ for _, ra := range directedTargets {
+ udpAddr, rerr :=
resolveBacnetUDPAddr(ra, specificOptions.bacNetPort)
+ if rerr != nil {
+
d.log.Warn().Err(rerr).Str("remoteAddress", ra).Msg("invalid directed WhoIs
target; skipping")
+ continue
+ }
+ if _, err :=
communicationChannelInstance.unicastConnection.WriteTo(theBytes, udpAddr); err
!= nil {
+
d.log.Debug().Err(err).Stringer("target", udpAddr).Msg("Error sending directed
WhoIs")
+ }
+ }
}
}
if whoHasOptions := specificOptions.whoHasOptions;
whoHasOptions != nil {
@@ -196,7 +226,7 @@ func (d *Discoverer) broadcastAndDiscover(ctx
context.Context, communicationChan
if err != nil {
return nil, err
}
- if _, err :=
communicationChannelInstance.broadcastConnection.WriteTo(theBytes,
communicationChannelInstance.broadcastConnection.LocalAddr()); err != nil {
+ if _, err :=
communicationChannelInstance.broadcastConnection.WriteTo(theBytes,
communicationChannelInstance.broadcastTarget); err != nil {
d.log.Debug().Err(err).Msg("Error sending
broadcast")
}
}
@@ -216,7 +246,7 @@ func (d *Discoverer) broadcastAndDiscover(ctx
context.Context, communicationChan
blockingReadChan <- false
return
}
- d.log.Debug().Stringer("addr",
addr).Msg("Received broadcast bvlc")
+ d.log.Debug().Stringer("addr",
addr).Msg("Received unicast bvlc")
ctxForModel :=
options.GetLoggerContextForModel(ctx, d.log,
options.WithPassLoggerToModel(d.passLogToModel))
incomingBvlc, err :=
driverModel.BVLCParse[driverModel.BVLC](ctxForModel, buf[:n])
if err != nil {
@@ -285,11 +315,10 @@ func (d *Discoverer) broadcastAndDiscover(ctx
context.Context, communicationChan
func (d *Discoverer) handleIncomingBVLCs(ctx context.Context, callback
func(event apiModel.PlcDiscoveryItem), incomingBVLCChannel chan
receivedBvlcMessage) {
for {
- if err := ctx.Err(); err != nil {
- // TODO: maybe we log something, but maybe it is fine
- return
- }
select {
+ case <-ctx.Done():
+ d.log.Debug().Err(ctx.Err()).Msg("Ending incoming BVLC
handling")
+ return
case receivedBvlc := <-incomingBVLCChannel:
var npdu driverModel.NPDU
if bvlc, ok := receivedBvlc.bvlc.(interface{ GetNpdu()
driverModel.NPDU }); ok {
@@ -344,9 +373,6 @@ func (d *Discoverer) handleIncomingBVLCs(ctx
context.Context, callback func(even
// Pass the event back to the callback
callback(discoveryEvent)
}
- case <-ctx.Done():
- d.log.Debug().Err(ctx.Err()).Msg("Ending unicast
receive")
- return
}
}
}
@@ -400,30 +426,39 @@ func (d *Discoverer) buildupCommunicationChannels(ctx
context.Context, interface
// requires a userland demultiplexer, not socket
options. If you
// hit "address already in use" here, stop whatever
else is bound
// to the BACnet/IP UDP port.
- var lc net.ListenConfig
- unicastConnection, err := lc.ListenPacket(ctx, "udp4",
fmt.Sprintf("%v:%d", ipAddr, bacNetPort))
+ // Bind to the wildcard address (0.0.0.0) rather than
the specific
+ // interface IP. A socket bound to a specific IP fails
to receive the
+ // unicast IAm replies on some interfaces (notably the
virtual
+ // interfaces used in tests) even though the packets
arrive at that IP
+ // — which is why the conventional BACnet/IP stack (and
gobacnet) bind
+ // the wildcard. SO_BROADCAST lets the same socket send
the WhoIs to
+ // the subnet broadcast address.
+ ifName := networkInterface.Name
+ lc := net.ListenConfig{Control: func(_ string, _
string, c syscall.RawConn) error {
+ return controlDiscoverySocket(c, ifName)
+ }}
+ conn, err := lc.ListenPacket(ctx, "udp4",
fmt.Sprintf("0.0.0.0:%d", bacNetPort))
if err != nil {
- d.log.Debug().Err(err).Msg("Error building
unicast Port")
+ d.log.Debug().Err(err).Msg("Error building
discovery socket")
continue
}
_, cidr, _ := net.ParseCIDR(unicastAddress.String())
- broadcastAddr := make(net.IP, len(cidr.IP))
- for i := range broadcastAddr {
- broadcastAddr[i] = cidr.IP[i] | ^cidr.Mask[i]
- }
- broadcastConnection, err := lc.ListenPacket(ctx,
"udp4", fmt.Sprintf("%v:%d", broadcastAddr, bacNetPort))
- if err != nil {
- if err := unicastConnection.Close(); err != nil
{
- d.log.Debug().Err(err).Msg("Error
closing transport instance")
- }
- d.log.Debug().Err(err).Msg("Error building
broadcast Port")
- continue
+ broadcastIP := make(net.IP, len(cidr.IP))
+ for i := range broadcastIP {
+ broadcastIP[i] = cidr.IP[i] | ^cidr.Mask[i]
}
+ broadcastTarget := &net.UDPAddr{IP: broadcastIP, Port:
bacNetPort}
+ d.log.Debug().
+ Str("interface", networkInterface.Name).
+ Stringer("local", conn.LocalAddr()).
+ Stringer("broadcastTarget", broadcastTarget).
+ Msg("discovery channel bound")
communicationChannels = append(communicationChannels,
communicationChannel{
networkInterface: networkInterface,
- unicastConnection: unicastConnection,
- broadcastConnection: broadcastConnection,
+ unicastConnection: conn,
+ broadcastConnection: conn,
+ broadcastTarget: broadcastTarget,
log: d.log,
})
}
@@ -447,13 +482,20 @@ type communicationChannel struct {
networkInterface net.Interface
unicastConnection net.PacketConn
broadcastConnection net.PacketConn
- log zerolog.Logger
+ // broadcastTarget is the subnet broadcast address (e.g.
192.168.100.255:47808)
+ // to which WhoIs is sent. The socket itself is bound to the wildcard
address.
+ broadcastTarget net.Addr
+ log zerolog.Logger
}
func (c communicationChannel) Close() error {
defer utils.StopWarn(c.log)()
_ = c.unicastConnection.Close()
- _ = c.broadcastConnection.Close()
+ // unicastConnection and broadcastConnection are the same socket;
closing
+ // twice is harmless (the second returns an already-closed error).
+ if c.broadcastConnection != c.unicastConnection {
+ _ = c.broadcastConnection.Close()
+ }
return nil
}
@@ -486,11 +528,19 @@ func extractInterfaces(discoveryOptions
[]options.WithDiscoveryOption) ([]net.In
type protocolSpecificOptions struct {
bacNetPort int
// remoteAddress, when set, sends the WhoIs as a directed unicast to
this
- // host (instead of the interface broadcast), enabling targeted
discovery of
- // a specific device or subnet. Host only or host:port; port defaults to
+ // host (in addition to the interface broadcast), enabling targeted
discovery
+ // of a specific device or subnet. Host only or host:port; port
defaults to
// bacNetPort.
remoteAddress string
- whoIsOptions *struct {
+ // remoteAddresses, when non-empty, sends a directed unicast WhoIs to
each
+ // listed host (in addition to the broadcast and remoteAddress). This
lets a
+ // caller sweep a list of candidate device IPs. A directed unicast
forces the
+ // sender's OS to ARP each target first, which seeds the reverse path
so the
+ // IAm reply can be delivered even where a broadcast IAm would not be
routed
+ // back (through a router/BBMD, or on stacks that learn MACs only from
ARP).
+ // Each entry is host only or host:port; port defaults to bacNetPort.
+ remoteAddresses []string
+ whoIsOptions *struct {
limits *struct {
low uint
high uint
@@ -545,6 +595,13 @@ func remoteAddress(addr string) option {
}
}
+func remoteAddresses(addrs []string) option {
+ return func(specificOptions *protocolSpecificOptions) error {
+ specificOptions.remoteAddresses = addrs
+ return nil
+ }
+}
+
func whoIsLimits(whoIsLowLimit, whoIsHighLimit uint) option {
return func(specificOptions *protocolSpecificOptions) error {
specificOptions.whoIsOptions = &struct {
@@ -691,6 +748,22 @@ func extractProtocolSpecificOptions(discoveryOptions
[]options.WithDiscoveryOpti
collectedOptions = append(collectedOptions, remoteAddress(addr))
}
+ if _, ok := filteredOptionMap["remote-addresses"]; ok {
+ joined, err := OneString(filteredOptionMap, "remote-addresses")
+ if err != nil {
+ return nil, err
+ }
+ var addrs []string
+ for _, a := range strings.Split(joined, ",") {
+ if a = strings.TrimSpace(a); a != "" {
+ addrs = append(addrs, a)
+ }
+ }
+ if len(addrs) > 0 {
+ collectedOptions = append(collectedOptions,
remoteAddresses(addrs))
+ }
+ }
+
if whoIsLow, whoIsHigh, ok, err := func() (whoIsLowLimit uint,
whoIsHighLimit uint, ok bool, err error) {
if _, limitPresent := filteredOptionMap["who-is-low-limit"];
!limitPresent {
return
diff --git a/plc4go/internal/bacnetip/DiscovererControl_linux.go
b/plc4go/internal/bacnetip/DiscovererControl_linux.go
new file mode 100644
index 0000000000..f0c563d998
--- /dev/null
+++ b/plc4go/internal/bacnetip/DiscovererControl_linux.go
@@ -0,0 +1,57 @@
+//go:build linux
+
+/*
+ * 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 bacnetip
+
+import (
+ "syscall"
+
+ "golang.org/x/sys/unix"
+)
+
+// controlDiscoverySocket configures the discovery UDP socket. SO_BINDTODEVICE
+// ties the wildcard-bound socket to one NIC so it reliably receives unicast
+// IAm replies arriving on that interface (notably on virtual/test interfaces,
+// where a plain wildcard socket may not be handed the unicast datagrams).
+// SO_REUSEADDR/SO_REUSEPORT allow rebinding across discovery sweeps and
+// co-existing with other BACnet listeners; SO_BROADCAST permits sending the
+// WhoIs to the subnet broadcast address. This mirrors gobacnet's datalink.
+func controlDiscoverySocket(c syscall.RawConn, interfaceName string) error {
+ var sockErr error
+ ctrlErr := c.Control(func(fd uintptr) {
+ if sockErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET,
unix.SO_REUSEADDR, 1); sockErr != nil {
+ return
+ }
+ if sockErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET,
unix.SO_REUSEPORT, 1); sockErr != nil {
+ return
+ }
+ if sockErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET,
unix.SO_BROADCAST, 1); sockErr != nil {
+ return
+ }
+ if interfaceName != "" {
+ sockErr = unix.SetsockoptString(int(fd),
unix.SOL_SOCKET, unix.SO_BINDTODEVICE, interfaceName)
+ }
+ })
+ if ctrlErr != nil {
+ return ctrlErr
+ }
+ return sockErr
+}
diff --git a/plc4go/internal/bacnetip/DiscovererControl_other.go
b/plc4go/internal/bacnetip/DiscovererControl_other.go
new file mode 100644
index 0000000000..66d4e47d52
--- /dev/null
+++ b/plc4go/internal/bacnetip/DiscovererControl_other.go
@@ -0,0 +1,32 @@
+//go:build !linux
+
+/*
+ * 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 bacnetip
+
+import "syscall"
+
+// controlDiscoverySocket is a no-op on non-Linux platforms. SO_BINDTODEVICE
+// (the option that makes the wildcard-bound socket reliably receive unicast
IAm
+// replies on a specific NIC) is Linux-specific, so discovery socket tuning is
+// only applied there for now.
+func controlDiscoverySocket(_ syscall.RawConn, _ string) error {
+ return nil
+}
diff --git a/plc4go/internal/bacnetip/Discoverer_test.go
b/plc4go/internal/bacnetip/Discoverer_test.go
index 655f4ed97a..bf773c8048 100644
--- a/plc4go/internal/bacnetip/Discoverer_test.go
+++ b/plc4go/internal/bacnetip/Discoverer_test.go
@@ -20,12 +20,16 @@
package bacnetip
import (
+ "context"
+ "net"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ driverModel
"github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
"github.com/apache/plc4x/plc4go/spi/options"
)
@@ -52,6 +56,52 @@ func TestSetDiscoveryTimeout_NegativeFallsBackToDefault(t
*testing.T) {
assert.Equal(t, 5*time.Second, d.discoveryTimeout)
}
+func TestHandleIncomingBVLCs_DispatchesIAm(t *testing.T) {
+ // Real IAm captured from a device advertising instance 3001 (vendor
999):
+ // BVLC(Original-Broadcast-NPDU) / NPDU / APDU(unconfirmed, IAm).
+ iamBytes := []byte{
+ 0x81, 0x0b, 0x00, 0x15, // BVLC
+ 0x01, 0x00, // NPDU
+ 0x10, 0x00, // APDU unconfirmed, service IAm
+ 0xc4, 0x02, 0x00, 0x0b, 0xb9, // object-id: device 3001
+ 0x22, 0x05, 0xc4, // max-apdu 1476
+ 0x91, 0x00, // segmentation: both
+ 0x22, 0x03, 0xe7, // vendor 999
+ }
+ bvlc, err :=
driverModel.BVLCParse[driverModel.BVLC](context.Background(), iamBytes)
+ require.NoError(t, err)
+
+ d := NewDiscoverer()
+ ch := make(chan receivedBvlcMessage, 1)
+ ch <- receivedBvlcMessage{bvlc: bvlc, addr: &net.UDPAddr{IP:
net.IPv4(192, 168, 100, 2), Port: 47808}}
+
+ got := make(chan apiModel.PlcDiscoveryItem, 1)
+ ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+ defer cancel()
+ done := make(chan struct{})
+ go func() {
+ d.handleIncomingBVLCs(ctx, func(item apiModel.PlcDiscoveryItem)
{ got <- item }, ch)
+ close(done)
+ }()
+
+ select {
+ case item := <-got:
+ assert.Contains(t, item.GetName(), "3001")
+ transportURL := item.GetTransportUrl()
+ assert.Equal(t, "192.168.100.2", transportURL.Hostname())
+ case <-time.After(2 * time.Second):
+ t.Fatal("callback was not invoked for IAm")
+ }
+
+ // Cancelling the context must make the handler return (no hang).
+ cancel()
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("handleIncomingBVLCs did not return after context
cancellation")
+ }
+}
+
func TestResolveBacnetUDPAddr(t *testing.T) {
// Host only — default port applies.
addr, err := resolveBacnetUDPAddr("192.168.1.50", 47808)
diff --git a/plc4go/internal/bacnetip/ReadRoundtrip_test.go
b/plc4go/internal/bacnetip/ReadRoundtrip_test.go
new file mode 100644
index 0000000000..28af60378d
--- /dev/null
+++ b/plc4go/internal/bacnetip/ReadRoundtrip_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 bacnetip
+
+import (
+ "context"
+ "fmt"
+ "net"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/rs/zerolog"
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ plc4go "github.com/apache/plc4x/plc4go/pkg/api"
+ "github.com/apache/plc4x/plc4go/pkg/api/cache"
+ apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
+ apiTransports "github.com/apache/plc4x/plc4go/pkg/api/transports"
+ model
"github.com/apache/plc4x/plc4go/protocols/bacnetip/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi/options"
+)
+
+// fakeBacnetDevice is a minimal UDP BACnet/IP device used to exercise the real
+// driver's read round-trip. On every confirmed request it replies with a
+// ReadProperty ComplexAck carrying ANALOG_INPUT,1/PRESENT_VALUE = 23.5,
echoing
+// the request's invoke id so the Reader's expectation matches.
+type fakeBacnetDevice struct {
+ conn *net.UDPConn
+ wg sync.WaitGroup
+ log zerolog.Logger
+ gotReqs int
+ mu sync.Mutex
+}
+
+func startFakeBacnetDevice(t *testing.T, log zerolog.Logger) *fakeBacnetDevice
{
+ t.Helper()
+ addr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
+ conn, err := net.ListenUDP("udp4", addr)
+ require.NoError(t, err)
+ d := &fakeBacnetDevice{conn: conn, log: log}
+ d.wg.Add(1)
+ go d.serve()
+ return d
+}
+
+func (d *fakeBacnetDevice) port() int { return
d.conn.LocalAddr().(*net.UDPAddr).Port }
+
+func (d *fakeBacnetDevice) requestCount() int {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ return d.gotReqs
+}
+
+func (d *fakeBacnetDevice) serve() {
+ defer d.wg.Done()
+ buf := make([]byte, 4096)
+ for {
+ n, src, err := d.conn.ReadFromUDP(buf)
+ if err != nil {
+ return // socket closed
+ }
+ data := make([]byte, n)
+ copy(data, buf[:n])
+ d.log.Info().Int("bytes", n).Stringer("src", src).Msg("fake
device received packet")
+
+ invokeId, ok := d.extractInvokeId(data)
+ if !ok {
+ d.log.Warn().Msg("fake device: could not extract invoke
id; ignoring")
+ continue
+ }
+ d.mu.Lock()
+ d.gotReqs++
+ d.mu.Unlock()
+
+ resp := d.buildReadPropertyAck(invokeId)
+ theBytes, err := resp.Serialize()
+ if err != nil {
+ d.log.Error().Err(err).Msg("fake device: serialize
response")
+ continue
+ }
+ if _, err := d.conn.WriteToUDP(theBytes, src); err != nil {
+ d.log.Error().Err(err).Msg("fake device: write
response")
+ continue
+ }
+ d.log.Info().Uint8("invokeId", invokeId).Int("bytes",
len(theBytes)).Stringer("dst", src).Msg("fake device sent ComplexAck")
+ }
+}
+
+func (d *fakeBacnetDevice) extractInvokeId(data []byte) (uint8, bool) {
+ bvlc, err := model.BVLCParse[model.BVLC](context.Background(), data)
+ if err != nil {
+ d.log.Error().Err(err).Msg("fake device: parse BVLC")
+ return 0, false
+ }
+ npduRetriever, ok := bvlc.(interface{ GetNpdu() model.NPDU })
+ if !ok {
+ return 0, false
+ }
+ apdu := npduRetriever.GetNpdu().GetApdu()
+ cr, ok := apdu.(model.APDUConfirmedRequest)
+ if !ok {
+ d.log.Warn().Msgf("fake device: not a confirmed request: %T",
apdu)
+ return 0, false
+ }
+ return cr.GetInvokeId(), true
+}
+
+func (d *fakeBacnetDevice) buildReadPropertyAck(invokeId uint8) model.BVLC {
+ serviceAck := model.NewBACnetServiceAckReadProperty(
+ 0,
+ model.CreateBACnetContextTagObjectIdentifier(0,
uint16(model.BACnetObjectType_ANALOG_INPUT), 1),
+ model.CreateBACnetPropertyIdentifierTagged(1,
uint32(model.BACnetPropertyIdentifier_PRESENT_VALUE)),
+ nil,
+
constructedDataFromTag(model.CreateBACnetApplicationTagReal(23.5)),
+ )
+ apdu := model.NewAPDUComplexAck(false, false, invokeId, nil, nil,
serviceAck, nil, nil)
+ return wrapAPDU(apdu, false)
+}
+
+func (d *fakeBacnetDevice) stop() {
+ _ = d.conn.Close()
+ d.wg.Wait()
+}
+
+func traceLogger(t *testing.T) zerolog.Logger {
+ return
zerolog.New(zerolog.NewTestWriter(t)).Level(zerolog.TraceLevel).With().Timestamp().Logger()
+}
+
+// readPresentValue issues a single ReadRequest for
ANALOG_INPUT,1/PRESENT_VALUE
+// against the given connection and returns the response code + float value.
+func readPresentValue(t *testing.T, conn plc4go.PlcConnection)
(apiModel.PlcResponseCode, float32, bool) {
+ t.Helper()
+ rr, err := conn.ReadRequestBuilder().AddTagAddress("pv",
"ANALOG_INPUT,1/PRESENT_VALUE").Build()
+ require.NoError(t, err)
+ ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
+ defer cancel()
+ select {
+ case <-ctx.Done():
+ return 0, 0, false
+ case res := <-rr.Execute(ctx):
+ if res.GetErr() != nil {
+ t.Logf("read error: %v", res.GetErr())
+ return 0, 0, false
+ }
+ resp := res.GetResponse()
+ code := resp.GetResponseCode("pv")
+ v := resp.GetValue("pv")
+ if v != nil && v.IsFloat32() {
+ return code, v.GetFloat32(), true
+ }
+ return code, 0, true
+ }
+}
+
+func TestNativeBacnetRead_DirectConnection(t *testing.T) {
+ log := traceLogger(t)
+ device := startFakeBacnetDevice(t, log)
+ defer device.stop()
+
+ dm := plc4go.NewPlcDriverManager(options.WithCustomLogger(log))
+ dm.RegisterDriver(NewDriver(options.WithCustomLogger(log)))
+ apiTransports.RegisterUdpTransport(dm)
+
+ connStr :=
fmt.Sprintf("bacnet-ip:udp://127.0.0.1:%d?local-port=0&ApduTimeoutMs=3000",
device.port())
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ conn, err := dm.GetConnection(ctx, connStr)
+ require.NoError(t, err)
+ defer conn.Close()
+
+ code, val, ok := readPresentValue(t, conn)
+ t.Logf("device received %d request(s)", device.requestCount())
+ require.True(t, ok, "read did not complete (timed out waiting for
response)")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+ assert.InDelta(t, 23.5, val, 0.001)
+}
+
+func TestNativeBacnetRead_ViaCache(t *testing.T) {
+ log := traceLogger(t)
+ device := startFakeBacnetDevice(t, log)
+ defer device.stop()
+
+ dm := plc4go.NewPlcDriverManager(options.WithCustomLogger(log))
+ dm.RegisterDriver(NewDriver(options.WithCustomLogger(log)))
+ apiTransports.RegisterUdpTransport(dm)
+ connCache := cache.NewPlcConnectionCache(dm,
cache.WithCustomLogger(log))
+ defer connCache.Close()
+
+ connStr :=
fmt.Sprintf("bacnet-ip:udp://127.0.0.1:%d?local-port=0&ApduTimeoutMs=3000",
device.port())
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ conn, err := connCache.GetConnection(ctx, connStr)
+ require.NoError(t, err)
+ defer conn.Close()
+
+ code, val, ok := readPresentValue(t, conn)
+ t.Logf("device received %d request(s)", device.requestCount())
+ require.True(t, ok, "read via cache did not complete (timed out waiting
for response)")
+ assert.Equal(t, apiModel.PlcResponseCode_OK, code)
+ assert.InDelta(t, 23.5, val, 0.001)
+}
diff --git a/plc4go/internal/bacnetip/ValueDecoder.go
b/plc4go/internal/bacnetip/ValueDecoder.go
index 48f983ca0a..dcd0bc98bc 100644
--- a/plc4go/internal/bacnetip/ValueDecoder.go
+++ b/plc4go/internal/bacnetip/ValueDecoder.go
@@ -22,6 +22,7 @@ package bacnetip
import (
"fmt"
"reflect"
+ "strings"
"time"
apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
@@ -119,10 +120,23 @@ func constructedDataToPlcValue(data
model.BACnetConstructedData) apiValues.PlcVa
if u, ok := data.(model.BACnetConstructedDataUnspecified); ok {
return elementsToPlcValue(u.GetData())
}
+ // Array-valued properties (OBJECT_LIST, PRIORITY_ARRAY, STATE_TEXT,
...) have
+ // no single GetActualValue; decode them to a PlcList (or, for an
array-index-0
+ // read, the element count) before falling through to the scalar path.
+ if pv, ok := arrayConstructedDataToPlcValue(data); ok {
+ return pv
+ }
if v, ok := callGetActualValue(data); ok {
if tag, ok := v.(model.BACnetApplicationTag); ok {
return appTagToPlcValue(tag)
}
+ // Tagged bit strings (PROTOCOL_SERVICES_SUPPORTED,
STATUS_FLAGS,
+ // LIMIT_ENABLE, OBJECT_TYPES_SUPPORTED, ...) carry a
bit-string payload
+ // rather than a single scalar; surface the packed bytes
(MSB-first), the
+ // same representation used for a BitString application tag.
+ if pv, ok := taggedBitStringToPlcValue(v); ok {
+ return pv
+ }
if pv, ok := taggedEnumToPlcValue(v); ok {
return pv
}
@@ -133,6 +147,78 @@ func constructedDataToPlcValue(data
model.BACnetConstructedData) apiValues.PlcVa
return spiValues.NewPlcSTRING(fmt.Sprintf("%T:%v", data, data))
}
+// arrayConstructedDataToPlcValue handles array-valued constructed data such as
+// OBJECT_LIST. BACnet array properties expose a NumberOfDataElements accessor
+// (set only when the client read array index 0, where the device returns just
+// the element count) plus a per-property element slice. A whole-array read
+// yields a PlcList of the decoded elements; an array-index-0 read yields the
+// count as an unsigned PlcValue. Returns ok=false for non-array constructed
+// data (no NumberOfDataElements accessor), so the caller falls through to the
+// scalar GetActualValue path.
+func arrayConstructedDataToPlcValue(data model.BACnetConstructedData)
(apiValues.PlcValue, bool) {
+ rv := reflect.ValueOf(data)
+ if !rv.IsValid() {
+ return nil, false
+ }
+ countMethod := rv.MethodByName("GetNumberOfDataElements")
+ if !countMethod.IsValid() || countMethod.Type().NumIn() != 0 ||
countMethod.Type().NumOut() != 1 {
+ return nil, false // not an array-valued constructed data
+ }
+ // Array-index-0 read: only the element count is present.
+ if out := countMethod.Call(nil)[0]; out.IsValid() && !out.IsNil() {
+ if tag, ok := out.Interface().(model.BACnetApplicationTag); ok {
+ return appTagToPlcValue(tag), true
+ }
+ }
+ // Whole-array read: find the element slice accessor and decode each
entry.
+ for i := 0; i < rv.NumMethod(); i++ {
+ name := rv.Type().Method(i).Name
+ if !strings.HasPrefix(name, "Get") || name ==
"GetNumberOfDataElements" {
+ continue
+ }
+ method := rv.Method(i)
+ mt := method.Type()
+ if mt.NumIn() != 0 || mt.NumOut() != 1 || mt.Out(0).Kind() !=
reflect.Slice {
+ continue
+ }
+ out := method.Call(nil)[0]
+ elems := make([]apiValues.PlcValue, 0, out.Len())
+ for j := 0; j < out.Len(); j++ {
+ switch el := out.Index(j).Interface().(type) {
+ case model.BACnetApplicationTag:
+ elems = append(elems, appTagToPlcValue(el))
+ case model.BACnetConstructedData:
+ elems = append(elems,
constructedDataToPlcValue(el))
+ default:
+ elems = append(elems,
spiValues.NewPlcSTRING(fmt.Sprintf("%v", el)))
+ }
+ }
+ return spiValues.NewPlcList(elems), true
+ }
+ return nil, false
+}
+
+// taggedBitStringToPlcValue handles BACnet *Tagged bit-string wrappers
+// (BACnetServicesSupportedTagged, BACnetStatusFlagsTagged,
BACnetLimitEnableTagged,
+// BACnetObjectTypesSupportedTagged, ...) returned by GetActualValue on typed
+// ConstructedData subtypes. These expose a GetPayload() returning a
+// BACnetTagPayloadBitString whose GetData() is the unpacked bit slice. We pack
+// it MSB-first into bytes and surface it as a PlcRawByteArray — identical to
how
+// a BitString application tag is decoded — so callers can walk the bits.
+func taggedBitStringToPlcValue(v any) (apiValues.PlcValue, bool) {
+ bs, ok := v.(interface {
+ GetPayload() model.BACnetTagPayloadBitString
+ })
+ if !ok {
+ return nil, false
+ }
+ payload := bs.GetPayload()
+ if payload == nil {
+ return nil, false
+ }
+ return spiValues.NewPlcRawByteArray(bitsToBytes(payload.GetData())),
true
+}
+
// taggedEnumToPlcValue handles BACnet's *Tagged enum wrappers (BACnetBinaryPV,
// BACnetReliability, BACnetEventState, ...) returned by GetActualValue on
// typed ConstructedData subtypes. These don't satisfy BACnetApplicationTag
diff --git a/plc4go/internal/bacnetip/ValueDecoder_test.go
b/plc4go/internal/bacnetip/ValueDecoder_test.go
index d793034c06..e39f194621 100644
--- a/plc4go/internal/bacnetip/ValueDecoder_test.go
+++ b/plc4go/internal/bacnetip/ValueDecoder_test.go
@@ -136,3 +136,77 @@ func TestTaggedEnumToPlcValue_RejectsUnsupportedKind(t
*testing.T) {
type complexReturnStub struct{}
func (complexReturnStub) GetValue() []byte { return []byte{1, 2, 3} }
+
+// bitStringPayloadStub satisfies the `GetPayload() BACnetTagPayloadBitString`
+// shape taggedBitStringToPlcValue looks for, standing in for the generated
+// *Tagged bit-string wrappers (ServicesSupported, StatusFlags, ...).
+type bitStringPayloadStub struct {
+ p readWriteModel.BACnetTagPayloadBitString
+}
+
+func (s bitStringPayloadStub) GetPayload()
readWriteModel.BACnetTagPayloadBitString { return s.p }
+
+func TestTaggedBitStringToPlcValue_PacksBitsMsbFirst(t *testing.T) {
+ // PROTOCOL_SERVICES_SUPPORTED arrives as a tagged bit string. Bit 5
+ // (subscribe-cov) and bit 15 (write-property) set must pack MSB-first
to
+ // bytes [0x04, 0x01], so the agent's positional bit→service decode
detects
+ // the right capabilities. A miss here is what made writes report
"device is
+ // not writeable".
+ bits := make([]bool, 16)
+ bits[5] = true // subscribe-cov
+ bits[15] = true // write-property
+ payload :=
readWriteModel.CreateBACnetApplicationTagBitString(bits).GetPayload()
+
+ got, ok := taggedBitStringToPlcValue(bitStringPayloadStub{payload})
+ require.True(t, ok, "a tagged bit-string payload should be recognized")
+ require.NotNil(t, got)
+ assert.Equal(t, []byte{0x04, 0x01}, got.GetRaw())
+}
+
+func TestTaggedBitStringToPlcValue_RejectsNonBitString(t *testing.T) {
+ got, ok := taggedBitStringToPlcValue(noGetValueStub{})
+ assert.False(t, ok)
+ assert.Nil(t, got)
+}
+
+func TestConstructedDataToPlcValue_ObjectList_WholeArray(t *testing.T) {
+ // A read of OBJECT_LIST (no array index) returns the full element list.
+ // It must decode to a PlcList of "<type>,<instance>" strings (not a
+ // stringified blob), so device discovery can enumerate the objects.
+ objs := []readWriteModel.BACnetApplicationTagObjectIdentifier{
+
readWriteModel.CreateBACnetApplicationTagObjectIdentifier(uint16(readWriteModel.BACnetObjectType_DEVICE),
3001),
+
readWriteModel.CreateBACnetApplicationTagObjectIdentifier(uint16(readWriteModel.BACnetObjectType_ANALOG_OUTPUT),
1),
+
readWriteModel.CreateBACnetApplicationTagObjectIdentifier(uint16(readWriteModel.BACnetObjectType_ANALOG_INPUT),
1),
+ }
+ data := readWriteModel.NewBACnetConstructedDataObjectList(
+ readWriteModel.CreateBACnetOpeningTag(1),
+ readWriteModel.NewBACnetTagHeader(9, 0, 1, nil, nil, nil, nil),
+ readWriteModel.CreateBACnetClosingTag(1),
+ nil, objs)
+
+ got := constructedDataToPlcValue(data)
+ require.NotNil(t, got)
+ require.True(t, got.IsList(), "OBJECT_LIST should decode to a PlcList,
got %T", got)
+ list := got.GetList()
+ require.Len(t, list, 3)
+ assert.Equal(t, "DEVICE,3001", list[0].GetString())
+ assert.Equal(t, "ANALOG_OUTPUT,1", list[1].GetString())
+ assert.Equal(t, "ANALOG_INPUT,1", list[2].GetString())
+}
+
+func TestConstructedDataToPlcValue_ObjectList_Count(t *testing.T) {
+ // A read of OBJECT_LIST[0] returns only the element count as an
unsigned
+ // integer. It must decode to a numeric PlcValue (not a string), so
callers
+ // can read the array length for an indexed fallback.
+ count := readWriteModel.CreateBACnetApplicationTagUnsignedInteger(5)
+ data := readWriteModel.NewBACnetConstructedDataObjectList(
+ readWriteModel.CreateBACnetOpeningTag(1),
+ readWriteModel.NewBACnetTagHeader(9, 0, 1, nil, nil, nil, nil),
+ readWriteModel.CreateBACnetClosingTag(1),
+ count, nil)
+
+ got := constructedDataToPlcValue(data)
+ require.NotNil(t, got)
+ assert.True(t, got.IsUint32(), "OBJECT_LIST[0] count should be numeric,
got %T", got)
+ assert.Equal(t, uint32(5), got.GetUint32())
+}
diff --git a/plc4go/spi/transports/udp/TransportInstance.go
b/plc4go/spi/transports/udp/TransportInstance.go
index e29343a71d..f6a0f594da 100644
--- a/plc4go/spi/transports/udp/TransportInstance.go
+++ b/plc4go/spi/transports/udp/TransportInstance.go
@@ -160,6 +160,13 @@ func (m *TransportInstance) GetNumBytesAvailableInBuffer()
(uint32, error) {
if m.reader == nil {
return 0, nil
}
+ // Use a fresh, short read deadline for this poll.
Read/PeekReadableBytes set
+ // a sticky SetReadDeadline from the request context; once that
deadline has
+ // passed, a deadline-less Peek here would keep failing with i/o
timeout and
+ // the codec would never observe further inbound datagrams.
+ if m.udpConn != nil {
+ _ = m.udpConn.SetReadDeadline(time.Now().Add(10 *
time.Millisecond))
+ }
_, _ = m.reader.Peek(1)
return uint32(m.reader.Buffered()), nil
}
