You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by sr...@apache.org on 2026/08/19 14:47:23 UTC
(plc4x) 05/07: fix(plc4go/knxnetip): repair the codec dispatch and the connection lifecycle
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 2e482832b81d4d8a57c4fbfa9d84a821873acdb7
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Wed Aug 19 16:29:03 2026 +0200
fix(plc4go/knxnetip): repair the codec dispatch and the connection lifecycle
CustomMessageHandling asserted the incoming message to a tunneling type with a
single-value assertion, so every other KNXnet/IP frame - search, connect,
connection-state and disconnect responses - panicked the receive worker. The
default codec swallows that panic and restarts the worker, so those frames were
silently dropped and the handshake and keepalive simply timed out. The dispatch
is a type switch now, and a tunneling ACK is deliberately left unhandled so the
expectations waiting for it can still see it.
The keepalive timer was declared and stopped but never created, so no
connection-state request was ever sent and gateways dropped the tunnel. It runs
on a fixed interval now, independent of incoming traffic, the way the heartbeat
requires; resetting it on every frame would starve it on a busy bus.
Also: IsConnected fell through to false on the success path, the metadata
derived both the multicast and the MAC address from the serial number, the
connection type was pinned to LINK_LAYER instead of following the configured
one, and the reserved-byte skip on the group-value read path consumed a byte
the datapoint parser already accounts for.
---
plc4go/internal/knxnetip/Connection.go | 150 +++++++++--
.../knxnetip/ConnectionDriverSpecificOperations.go | 97 ++++++-
plc4go/internal/knxnetip/ConnectionHelper.go | 83 +++++-
.../knxnetip/ConnectionInternalOperations.go | 153 ++++++++++-
plc4go/internal/knxnetip/Connection_test.go | 273 ++++++++++++++++++++
plc4go/internal/knxnetip/MessageCodec.go | 37 +--
plc4go/internal/knxnetip/MessageCodec_test.go | 279 +++++++++++++++++++++
7 files changed, 1015 insertions(+), 57 deletions(-)
diff --git a/plc4go/internal/knxnetip/Connection.go b/plc4go/internal/knxnetip/Connection.go
index edae39b0d0..b48984f885 100644
--- a/plc4go/internal/knxnetip/Connection.go
+++ b/plc4go/internal/knxnetip/Connection.go
@@ -71,8 +71,8 @@ func (m *ConnectionMetadata) GetConnectionAttributes() map[string]string {
"ProjectNumber": strconv.Itoa(int(m.ProjectNumber)),
"InstallationNumber": strconv.Itoa(int(m.InstallationNumber)),
"DeviceSerialNumber": ByteArrayToString(m.DeviceSerialNumber, " "),
- "DeviceMulticastAddress": ByteArrayToString(m.DeviceSerialNumber, "."),
- "DeviceMacAddress": ByteArrayToString(m.DeviceSerialNumber, ":"),
+ "DeviceMulticastAddress": ByteArrayToString(m.DeviceMulticastAddress, "."),
+ "DeviceMacAddress": ByteArrayToString(m.DeviceMacAddress, ":"),
"SupportedServices": strings.Join(m.SupportedServices, ", "),
}
}
@@ -104,14 +104,36 @@ type KnxMemoryReadFragment struct {
startingAddress uint16
}
+// connectionStateInterval is how often a ConnectionStateRequest is sent to the
+// gateway in order to keep the tunneling connection alive. Gateways typically
+// drop a tunnel after 120s of silence, the KNX spec recommends 60s.
+// (Java: KnxNetIpConnection#HEARTBEAT_INTERVAL_MS)
+const connectionStateInterval = 60 * time.Second
+
+// defaultRequestTimeout mirrors the "request-timeout" default of the java driver
+// (KnxNetIpConfiguration#requestTimeout = 10_000ms).
+const defaultRequestTimeout = 10 * time.Second
+
type Connection struct {
- messageCodec spi.MessageCodec
- options map[string][]string
- tagHandler spi.PlcTagHandler
- valueHandler spi.PlcValueHandler
+ messageCodec spi.MessageCodec
+ options map[string][]string
+ tagHandler spi.PlcTagHandler
+ valueHandler spi.PlcValueHandler
+ // connectionStateLock guards connectionStateTimer and quitConnectionStateTimer,
+ // which are touched both by the codec worker (on every incoming message) and by
+ // the connect/close paths.
+ connectionStateLock sync.Mutex
connectionStateTimer *time.Ticker
quitConnectionStateTimer chan struct{}
- subscribers []*Subscriber
+ // keepaliveInterval is the interval of the keepalive-ticker. A zero value means
+ // connectionStateInterval; it is only ever set to something else by the tests, which
+ // can't wait a minute for a heartbeat.
+ keepaliveInterval time.Duration
+ // subscribersMutex guards subscribers, which is appended to / removed from by the
+ // (un)subscription paths while the codec worker iterates it on every incoming
+ // group-value write.
+ subscribersMutex sync.RWMutex
+ subscribers []*Subscriber
valueCache map[uint16][]byte
valueCacheMutex sync.RWMutex
@@ -160,6 +182,12 @@ type KnxReadResult struct {
err error
}
+// KnxWriteResult is the outcome of a single group-address write. A nil err means the
+// gateway acknowledged and confirmed the frame.
+type KnxWriteResult struct {
+ err error
+}
+
type KnxDeviceConnectResult struct {
connection *KnxDeviceConnection
err error
@@ -359,6 +387,10 @@ func (m *Connection) Connect(ctx context.Context) error {
m.log.Warn().Msg("Tunneling handler shat down")
})
+ // Start sending periodic ConnectionStateRequests, otherwise the
+ // gateway will drop the tunneling connection after a while.
+ m.startConnectionStateTimer()
+
// Fire the "connected" event
case driverModel.Status_NO_MORE_CONNECTIONS:
return m.doSomethingAndClose(func() error { return errors.New("no more connections") })
@@ -380,15 +412,86 @@ func (m *Connection) doSomethingAndClose(something func() error) error {
return errors.Join(err, m.messageCodec.Disconnect())
}
+// startConnectionStateTimer starts the keepalive-ticker which periodically sends
+// a ConnectionStateRequest to the gateway. Calling it more than once is a no-op.
+func (m *Connection) startConnectionStateTimer() {
+ m.connectionStateLock.Lock()
+ if m.connectionStateTimer != nil {
+ m.connectionStateLock.Unlock()
+ return
+ }
+ ticker := time.NewTicker(m.getKeepaliveInterval())
+ quit := make(chan struct{})
+ m.connectionStateTimer = ticker
+ m.quitConnectionStateTimer = quit
+ m.connectionStateLock.Unlock()
+
+ m.wg.Go(func() {
+ defer func() {
+ if err := recover(); err != nil {
+ m.log.Error().
+ Str("stack", string(debug.Stack())).
+ Interface("err", err).
+ Msg("panic-ed")
+ }
+ }()
+ for {
+ select {
+ case <-quit:
+ m.log.Debug().Msg("Stopping connection-state timer")
+ return
+ case <-ticker.C:
+ m.sendKeepalive()
+ }
+ }
+ })
+}
+
+// getKeepaliveInterval returns the interval the keepalive-ticker runs at.
+func (m *Connection) getKeepaliveInterval() time.Duration {
+ if m.keepaliveInterval > 0 {
+ return m.keepaliveInterval
+ }
+ return connectionStateInterval
+}
+
+// sendKeepalive sends a single ConnectionStateRequest and logs the outcome.
+func (m *Connection) sendKeepalive() {
+ ctx, cancelFunc := utils.WithNamedTimeout(context.Background(), "connection state request timeout", m.getRequestTimeout())
+ defer cancelFunc()
+ connectionStateResponse, err := m.sendConnectionStateRequest(ctx)
+ if err != nil {
+ m.log.Warn().Err(err).Msg("Error sending connection state request")
+ return
+ }
+ if connectionStateResponse.GetStatus() != driverModel.Status_NO_ERROR {
+ m.log.Warn().
+ Stringer("status", connectionStateResponse.GetStatus()).
+ Msg("Got a non-OK status for the connection state request")
+ }
+}
+
+// stopConnectionStateTimer stops the keepalive-ticker. Calling it more than once
+// (or without a prior start) is a no-op.
+func (m *Connection) stopConnectionStateTimer() {
+ m.connectionStateLock.Lock()
+ defer m.connectionStateLock.Unlock()
+ if m.connectionStateTimer == nil {
+ return
+ }
+ m.connectionStateTimer.Stop()
+ m.connectionStateTimer = nil
+ close(m.quitConnectionStateTimer)
+ m.quitConnectionStateTimer = nil
+}
+
func (m *Connection) Close() error {
ctx := context.TODO()
- ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection close timeout", 5*time.Second)
+ ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection close timeout", m.getRequestTimeout())
defer cancelFunc()
// Stop the connection-state checker.
- if m.connectionStateTimer != nil {
- m.connectionStateTimer.Stop()
- }
+ m.stopConnectionStateTimer()
// Disconnect from all knx devices we are still connected to.
for targetAddress := range m.DeviceConnections {
@@ -411,19 +514,21 @@ func (m *Connection) Close() error {
}
func (m *Connection) IsConnected() bool {
+ if m.messageCodec == nil {
+ return false
+ }
+
ctx := context.TODO()
- ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection status check timeout", 5*time.Second)
+ ctx, cancelFunc := utils.WithNamedTimeout(ctx, "connection status check timeout", m.getRequestTimeout())
defer cancelFunc()
- if m.messageCodec != nil {
- if err := m.Ping(ctx); err != nil {
- if errors.Is(err, context.DeadlineExceeded) {
- m.handleTimeout()
- }
- return false
+ if err := m.Ping(ctx); err != nil {
+ if errors.Is(err, context.DeadlineExceeded) {
+ m.handleTimeout()
}
+ return false
}
- return false
+ return true
}
func (m *Connection) Ping(ctx context.Context) error {
@@ -462,7 +567,7 @@ func (m *Connection) ReadRequestBuilder() apiModel.PlcReadRequestBuilder {
func (m *Connection) WriteRequestBuilder() apiModel.PlcWriteRequestBuilder {
return spiModel.NewDefaultPlcWriteRequestBuilder(
- m.tagHandler, m.valueHandler, NewWriter(m.messageCodec, options.WithCustomLogger(m.log)))
+ m.tagHandler, m.valueHandler, NewWriter(m, options.WithCustomLogger(m.log)))
}
func (m *Connection) SubscriptionRequestBuilder() apiModel.PlcSubscriptionRequestBuilder {
@@ -481,8 +586,9 @@ func (m *Connection) BrowseRequestBuilder() apiModel.PlcBrowseRequestBuilder {
}
func (m *Connection) UnsubscriptionRequestBuilder() apiModel.PlcUnsubscriptionRequestBuilder {
- return nil /*spiModel.NewDefaultPlcUnsubscriptionRequestBuilder(
- m.tagHandler, m.valueHandler, NewSubscriber(m.messageCodec))*/
+ // KNX has no wire-level subscribe, the handles carry the subscriber they belong to,
+ // so the default builder is all we need here.
+ return spiModel.NewDefaultPlcUnsubscriptionRequestBuilder()
}
func (m *Connection) GetTransportInstance() transports.TransportInstance {
diff --git a/plc4go/internal/knxnetip/ConnectionDriverSpecificOperations.go b/plc4go/internal/knxnetip/ConnectionDriverSpecificOperations.go
index 5ac4b5e5dc..63ff2b8fe3 100644
--- a/plc4go/internal/knxnetip/ConnectionDriverSpecificOperations.go
+++ b/plc4go/internal/knxnetip/ConnectionDriverSpecificOperations.go
@@ -82,12 +82,12 @@ func (m *Connection) ReadGroupAddress(ctx context.Context, groupAddress []byte,
payload = append(payload, byte(groupAddressReadResponse.GetDataFirstByte()))
payload = append(payload, groupAddressReadResponse.GetData()...)
- // Parse the response data.
+ // Parse the response data. The payload is handed to the datapoint parser as-is:
+ // the generated parser reads the reserved bits/byte of the datapoint-type itself,
+ // so skipping the first byte here would consume the value of a small
+ // datapoint-type and shift a bigger one by a byte.
+ // (Java: KnxNetIpConnection hands the payload to KnxDatapoint.staticParse unchanged)
rb := utils.NewReadBufferByteBased(payload)
- // If the size of the tag is greater than 6, we have to skip the first byte
- if datapointType.DatapointMainType().SizeInBits() > 6 {
- _, _ = rb.ReadUint8("datapointType", 8)
- }
// Set a default datatype if none is provided
if *datapointType == driverModel.KnxDatapointType_DPT_UNKNOWN {
defaultDatapointType := driverModel.KnxDatapointType_USINT
@@ -107,6 +107,93 @@ func (m *Connection) ReadGroupAddress(ctx context.Context, groupAddress []byte,
return result
}
+// WriteGroupAddress serializes the given value according to the given datapoint-type and
+// sends it as a GroupValueWrite to the given group address.
+//
+// The returned channel is always completed exactly once: with a nil error if the gateway
+// acknowledged and confirmed the frame, and with an error on any failure, including the
+// request-timeout which is applied here. (Java: KnxNetIpConnection#onWrite)
+func (m *Connection) WriteGroupAddress(ctx context.Context, groupAddress []byte, datapointType *driverModel.KnxDatapointType, value values.PlcValue) <-chan KnxWriteResult {
+ result := make(chan KnxWriteResult, 1)
+
+ sendResponse := func(err error) {
+ select {
+ case result <- KnxWriteResult{err: err}:
+ default:
+ m.log.Trace().Err(err).Msg("dropping write result")
+ }
+ }
+
+ m.wg.Go(func() {
+ defer func() {
+ if err := recover(); err != nil {
+ m.log.Error().
+ Str("stack", string(debug.Stack())).
+ Interface("err", err).
+ Msg("panic-ed")
+ // Always complete the result channel, otherwise callers
+ // blocked on it would hang forever after a recovered panic.
+ sendResponse(errors.Errorf("panic-ed %v", err))
+ }
+ }()
+
+ dataFirstByte, data, err := SerializeGroupValue(value, datapointType)
+ if err != nil {
+ sendResponse(err)
+ return
+ }
+
+ // Don't wait forever for a gateway which never answers.
+ // (Java: ackFuture.orTimeout(getConfiguration().getRequestTimeout(), MILLISECONDS))
+ writeContext, cancel := context.WithTimeout(ctx, m.getRequestTimeout())
+ defer cancel()
+ if err := m.sendGroupAddressWriteRequest(writeContext, groupAddress, dataFirstByte, data); err != nil {
+ sendResponse(errors.Wrap(err, "error writing group address"))
+ return
+ }
+
+ sendResponse(nil)
+ })
+
+ return result
+}
+
+// SerializeGroupValue turns a plc-value into the payload of a GroupValueWrite APDU: the
+// 6 bit "data first byte" which is part of the APDU itself, and the data bytes which follow
+// it.
+//
+// The generated datapoint serialization already produces exactly that layout: datapoint-types
+// of up to 6 bits are packed into a single byte (which then is the small-payload), everything
+// bigger is preceded by an empty reserved byte which becomes the (zero) first byte. Splitting
+// the serialized bytes into first-byte and rest is therefore the exact inverse of the
+// read-path, which glues them back together and hands the result to the datapoint parser
+// unchanged.
+// (Java: KnxNetIpConnection#serializePayload)
+func SerializeGroupValue(value values.PlcValue, datapointType *driverModel.KnxDatapointType) (int8, []byte, error) {
+ if value == nil {
+ return 0, nil, errors.New("no value given to write")
+ }
+ if datapointType == nil || *datapointType == driverModel.KnxDatapointType_DPT_UNKNOWN {
+ return 0, nil, errors.New("no datapoint-type given, unable to serialize the value")
+ }
+ serialized, err := driverModel.KnxDatapointSerialize(value, *datapointType)
+ if err != nil {
+ return 0, nil, errors.Wrap(err, "error serializing value")
+ }
+ if len(serialized) == 0 {
+ return 0, nil, errors.Errorf("serializing a value of type %s didn't produce any data", datapointType)
+ }
+ // The apdu only has room for 6 bits here, so anything bigger would silently be truncated.
+ // (Java: KnxNetIpConnection#checkSixBitByte)
+ if serialized[0] > 0x3F {
+ return 0, nil, errors.Errorf("the first byte of a value of type %s doesn't fit into the 6 bit small-payload", datapointType)
+ }
+ if len(serialized) == 1 {
+ return int8(serialized[0]), nil, nil
+ }
+ return int8(serialized[0]), serialized[1:], nil
+}
+
func (m *Connection) DeviceConnect(ctx context.Context, targetAddress driverModel.KnxAddress) <-chan KnxDeviceConnectResult {
result := make(chan KnxDeviceConnectResult, 1)
diff --git a/plc4go/internal/knxnetip/ConnectionHelper.go b/plc4go/internal/knxnetip/ConnectionHelper.go
index 674ed7ce2c..faf81eb023 100644
--- a/plc4go/internal/knxnetip/ConnectionHelper.go
+++ b/plc4go/internal/knxnetip/ConnectionHelper.go
@@ -27,6 +27,7 @@ import (
"runtime/debug"
"slices"
"strconv"
+ "strings"
"sync/atomic"
"time"
@@ -41,12 +42,16 @@ import (
// Internal helper functions
///////////////////////////////////////////////////////////////////////////////////////////////////////
+// interceptIncomingMessage is called by the codec for every frame which arrives.
+//
+// The keepalive ticker is deliberately NOT reset here: the KNXnet/IP heartbeat has to
+// run at a fixed interval, independent of the traffic on the bus. Resetting it on every
+// incoming frame means that on any installation with group telegrams more often than
+// every 60s the ConnectionStateRequest would never be sent at all and the gateway would
+// drop the tunnel after 120s.
+// (Java: KnxNetIpConnection uses scheduleAtFixedRate with a fixed heartbeat interval)
func (m *Connection) interceptIncomingMessage(spi.Message) {
m.resetTimeout()
- if m.connectionStateTimer != nil {
- // Reset the timer for sending the ConnectionStateRequest
- m.connectionStateTimer.Reset(60 * time.Second)
- }
}
func (m *Connection) castIpToKnxAddress(ip net.IP) driverModel.IPAddress {
@@ -129,10 +134,8 @@ func (m *Connection) handleValueCacheUpdate(ctx context.Context, destinationAddr
m.valueCacheMutex.Unlock()
changed = true
}
- if m.subscribers != nil {
- for _, subscriber := range m.subscribers {
- subscriber.handleValueChange(ctx, destinationAddress, payload, changed)
- }
+ for _, subscriber := range m.getSubscribers() {
+ subscriber.handleValueChange(ctx, destinationAddress, payload, changed)
}
}
@@ -174,20 +177,74 @@ func (m *Connection) getGroupAddressNumLevels() uint8 {
return 3
}
+// getTunnelConnectionType evaluates the "connection-type" connection option and
+// maps it to the KnxLayer used in the ConnectionRequest.
+// (Java: KnxNetIpConfiguration#connectionType)
+func (m *Connection) getTunnelConnectionType() driverModel.KnxLayer {
+ if val, ok := m.options["connection-type"]; ok && len(val) > 0 {
+ switch strings.ToUpper(strings.TrimSpace(val[0])) {
+ case "LINK_LAYER", "TUNNEL_LINK_LAYER":
+ return driverModel.KnxLayer_TUNNEL_LINK_LAYER
+ case "RAW", "TUNNEL_RAW":
+ return driverModel.KnxLayer_TUNNEL_RAW
+ case "BUSMONITOR", "TUNNEL_BUSMONITOR":
+ return driverModel.KnxLayer_TUNNEL_BUSMONITOR
+ default:
+ m.log.Warn().Str("connection-type", val[0]).Msg("Invalid value for connection-type, falling back to LINK_LAYER")
+ }
+ }
+ return driverModel.KnxLayer_TUNNEL_LINK_LAYER
+}
+
+// getRequestTimeout evaluates the "request-timeout" connection option (in
+// milliseconds) which limits how long we wait for a gateway reply.
+// (Java: KnxNetIpConfiguration#requestTimeout)
+func (m *Connection) getRequestTimeout() time.Duration {
+ if val, ok := m.options["request-timeout"]; ok && len(val) > 0 {
+ requestTimeout, err := strconv.ParseUint(val[0], 10, 32)
+ if err == nil && requestTimeout > 0 {
+ return time.Duration(requestTimeout) * time.Millisecond
+ }
+ m.log.Warn().Str("request-timeout", val[0]).Msg("Invalid value for request-timeout, falling back to the default")
+ }
+ return defaultRequestTimeout
+}
+
func (m *Connection) addSubscriber(subscriber *Subscriber) {
+ m.subscribersMutex.Lock()
+ defer m.subscribersMutex.Unlock()
if slices.Contains(m.subscribers, subscriber) {
- m.log.Debug().Interface("subscriber", subscriber).Msg("Subscriber %v already added")
+ m.log.Debug().Interface("subscriber", subscriber).Msg("Subscriber already added")
return
}
m.subscribers = append(m.subscribers, subscriber)
}
func (m *Connection) removeSubscriber(subscriber *Subscriber) {
- for i, sub := range m.subscribers {
- if sub == subscriber {
- m.subscribers = append(m.subscribers[:i], m.subscribers[i+1:]...)
- }
+ m.subscribersMutex.Lock()
+ defer m.subscribersMutex.Unlock()
+ m.subscribers = slices.DeleteFunc(m.subscribers, func(other *Subscriber) bool {
+ return other == subscriber
+ })
+}
+
+// getSubscribers returns a snapshot of the currently registered subscribers so the
+// callers can iterate them without holding the lock while calling into them.
+func (m *Connection) getSubscribers() []*Subscriber {
+ m.subscribersMutex.RLock()
+ defer m.subscribersMutex.RUnlock()
+ return slices.Clone(m.subscribers)
+}
+
+// knxAddressEqual compares two individual knx addresses by value. The generated model
+// types are interfaces backed by pointers, so "==" would compare identities instead.
+func knxAddressEqual(a, b driverModel.KnxAddress) bool {
+ if a == nil || b == nil {
+ return a == nil && b == nil
}
+ return a.GetMainGroup() == b.GetMainGroup() &&
+ a.GetMiddleGroup() == b.GetMiddleGroup() &&
+ a.GetSubGroup() == b.GetSubGroup()
}
func (m *Connection) sliceEqual(a, b []byte) bool {
diff --git a/plc4go/internal/knxnetip/ConnectionInternalOperations.go b/plc4go/internal/knxnetip/ConnectionInternalOperations.go
index d85d5893f8..4434ea5774 100644
--- a/plc4go/internal/knxnetip/ConnectionInternalOperations.go
+++ b/plc4go/internal/knxnetip/ConnectionInternalOperations.go
@@ -96,7 +96,7 @@ func (m *Connection) sendGatewayConnectionRequest(ctx context.Context) (driverMo
connectionRequest := driverModel.NewConnectionRequest(
driverModel.NewHPAIDiscoveryEndpoint(driverModel.HostProtocolCode_IPV4_UDP, localAddr, uint16(localAddress.Port)),
driverModel.NewHPAIDataEndpoint(driverModel.HostProtocolCode_IPV4_UDP, localAddr, uint16(localAddress.Port)),
- driverModel.NewConnectionRequestInformationTunnelConnection(driverModel.KnxLayer_TUNNEL_LINK_LAYER),
+ driverModel.NewConnectionRequestInformationTunnelConnection(m.getTunnelConnectionType()),
)
result := make(chan driverModel.ConnectionResponse, 1)
@@ -297,6 +297,157 @@ func (m *Connection) sendGroupAddressReadRequest(ctx context.Context, groupAddre
}
}
+// sendGroupAddressWriteRequest puts a GroupValueWrite for the given group address onto the
+// bus. A gateway answers such a tunneling-request twice: first with a TunnelingResponse
+// acknowledging it accepted the frame, and then with an LDataCon echoing the frame back as
+// soon as it made it onto the bus. Both are correlated here, so a write only reports success
+// if the frame was really sent.
+// (Java: KnxNetIpConnection#onWrite)
+func (m *Connection) sendGroupAddressWriteRequest(ctx context.Context, groupAddress []byte, dataFirstByte int8, data []byte) error {
+ sequenceCounter := m.getNewSequenceCounter()
+ groupAddressWriteRequest := driverModel.NewTunnelingRequest(
+ driverModel.NewTunnelingRequestDataBlock(m.CommunicationChannelId, sequenceCounter),
+ driverModel.NewLDataReq(
+ 0,
+ nil,
+ driverModel.NewLDataExtended(
+ true,
+ false,
+ driverModel.CEMIPriority_LOW,
+ false,
+ false,
+ true,
+ 6,
+ 0,
+ m.ClientKnxAddress, groupAddress,
+ driverModel.NewApduDataContainer(
+ false,
+ 0,
+ driverModel.NewApduDataGroupValueWrite(dataFirstByte, data),
+ ),
+ ),
+ ),
+ )
+
+ // The confirmation expectation gets a context of its own so it can be dropped again as
+ // soon as this write is done. A stale expectation - e.g. one left behind by a write
+ // which already failed on the ack - would otherwise stay registered until its ttl runs
+ // out and consume the confirmation of a later write, since HandleMessages fans a
+ // message out to every expectation accepting it.
+ confirmationCtx, cancelConfirmation := context.WithCancel(ctx)
+ defer cancelConfirmation()
+
+ // Register the expectation for the confirmation before sending anything, otherwise a
+ // fast gateway could confirm the frame before we started listening for it.
+ confirmationResult := make(chan error, 1)
+ m.messageCodec.Expect(
+ confirmationCtx,
+ "group_address_write_confirmation",
+ func(message spi.Message) bool {
+ // A canceled expectation is only removed the next time the expire-worker runs,
+ // so stop accepting anything the moment this write is over.
+ if confirmationCtx.Err() != nil {
+ return false
+ }
+ tunnelingRequest, ok := message.(driverModel.TunnelingRequest)
+ if !ok || tunnelingRequest.GetTunnelingRequestDataBlock().GetCommunicationChannelId() != m.CommunicationChannelId {
+ return false
+ }
+ lDataCon, ok := tunnelingRequest.GetCemi().(driverModel.LDataCon)
+ if !ok {
+ return false
+ }
+ dataFrameExt, ok := lDataCon.GetDataFrame().(driverModel.LDataExtended)
+ if !ok {
+ return false
+ }
+ // Only the confirmation of the very frame we sent is of interest here: the
+ // gateway echoes it back verbatim, so source- and destination-address as well
+ // as the payload have to match, otherwise concurrent writes (to the same group
+ // address, or from another client on the same tunnel) satisfy each other.
+ if !dataFrameExt.GetGroupAddress() ||
+ !m.sliceEqual(dataFrameExt.GetDestinationAddress(), groupAddress) ||
+ !knxAddressEqual(dataFrameExt.GetSourceAddress(), m.ClientKnxAddress) {
+ return false
+ }
+ dataContainer, ok := dataFrameExt.GetApdu().(driverModel.ApduDataContainer)
+ if !ok {
+ return false
+ }
+ groupValueWrite, ok := dataContainer.GetDataApdu().(driverModel.ApduDataGroupValueWrite)
+ if !ok {
+ return false
+ }
+ return groupValueWrite.GetDataFirstByte() == dataFirstByte &&
+ m.sliceEqual(groupValueWrite.GetData(), data)
+ },
+ func(message spi.Message) error {
+ lDataCon := message.(driverModel.TunnelingRequest).GetCemi().(driverModel.LDataCon)
+ if lDataCon.GetDataFrame().GetErrorFlag() {
+ confirmationResult <- errors.New("the gateway reported an error confirming the group value write")
+ return nil
+ }
+ confirmationResult <- nil
+ return nil
+ },
+ func(err error) error {
+ // If this is a timeout, do a check if the connection requires a reconnection
+ var timeoutError utils.TimeoutError
+ if errors.As(err, &timeoutError) {
+ m.handleTimeout()
+ }
+ confirmationResult <- errors.Wrap(err, "got error waiting for the write confirmation")
+ return nil
+ },
+ )
+
+ ackResult := make(chan error, 1)
+ if err := m.messageCodec.SendRequest(ctx, "send_group_address_write_request", groupAddressWriteRequest, func(message spi.Message) bool {
+ tunnelingResponse, ok := message.(driverModel.TunnelingResponse)
+ if !ok {
+ return false
+ }
+ dataBlock := tunnelingResponse.GetTunnelingResponseDataBlock()
+ return dataBlock.GetCommunicationChannelId() == m.CommunicationChannelId &&
+ dataBlock.GetSequenceCounter() == sequenceCounter
+ }, func(message spi.Message) error {
+ tunnelingResponse := message.(driverModel.TunnelingResponse)
+ if status := tunnelingResponse.GetTunnelingResponseDataBlock().GetStatus(); status != driverModel.Status_NO_ERROR {
+ ackResult <- errors.Errorf("got a return status of: %s", status)
+ return nil
+ }
+ ackResult <- nil
+ return nil
+ }, func(err error) error {
+ // If this is a timeout, do a check if the connection requires a reconnection
+ var timeoutError utils.TimeoutError
+ if errors.As(err, &timeoutError) {
+ m.handleTimeout()
+ }
+ ackResult <- errors.Wrap(err, "got error processing request")
+ return nil
+ }); err != nil {
+ return errors.Wrap(err, "got error sending request")
+ }
+
+ // First wait for the gateway to acknowledge the tunneling-request itself ...
+ select {
+ case err := <-ackResult:
+ if err != nil {
+ return err
+ }
+ case <-ctx.Done():
+ return errors.Wrap(ctx.Err(), "context done while waiting for the tunneling ack")
+ }
+ // ... and then for the confirmation that the frame was put onto the bus.
+ select {
+ case err := <-confirmationResult:
+ return err
+ case <-ctx.Done():
+ return errors.Wrap(ctx.Err(), "context done while waiting for the write confirmation")
+ }
+}
+
func (m *Connection) sendDeviceConnectionRequest(ctx context.Context, targetAddress driverModel.KnxAddress) (driverModel.ApduControlConnect, error) {
// Send a connection request to the individual KNX device
deviceConnectionRequest := driverModel.NewTunnelingRequest(
diff --git a/plc4go/internal/knxnetip/Connection_test.go b/plc4go/internal/knxnetip/Connection_test.go
new file mode 100644
index 0000000000..c79b5ec11f
--- /dev/null
+++ b/plc4go/internal/knxnetip/Connection_test.go
@@ -0,0 +1,273 @@
+/*
+ * 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 knxnetip
+
+import (
+ "context"
+ "net"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ driverModel "github.com/apache/plc4x/plc4go/protocols/knxnetip/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi"
+ "github.com/apache/plc4x/plc4go/spi/errors"
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+ "github.com/apache/plc4x/plc4go/spi/transports"
+ "github.com/apache/plc4x/plc4go/spi/transports/udp"
+)
+
+// stubCodec is a minimal spi.MessageCodec test double. Only the methods which
+// the connection-lifecycle actually needs are implemented; everything else is
+// inherited from the (nil) embedded interface and would panic if it ever got
+// called, which is exactly what we want from a test double.
+type stubCodec struct {
+ spi.MessageCodec
+
+ transportInstance transports.TransportInstance
+ sentRequests []spi.Message
+ // respond is invoked for every SendRequest. Returning an error makes the
+ // codec report a send-failure, otherwise the (optional) reply is handed to
+ // the handleMessage / handleError callbacks.
+ respond func(message spi.Message, handleMessage spi.HandleMessage, handleError spi.HandleError) error
+}
+
+func (s *stubCodec) GetTransportInstance() transports.TransportInstance {
+ return s.transportInstance
+}
+
+func (s *stubCodec) SendRequest(_ context.Context, _ string, message spi.Message, _ spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError) error {
+ s.sentRequests = append(s.sentRequests, message)
+ if s.respond == nil {
+ return nil
+ }
+ return s.respond(message, handleMessage, handleError)
+}
+
+// newStubbedConnection builds a Connection which talks to a stubCodec. The udp
+// transport-instance is never connected, it only supplies the local address the
+// request-builders need.
+func newStubbedConnection(t *testing.T, connectionOptions map[string][]string, respond func(message spi.Message, handleMessage spi.HandleMessage, handleError spi.HandleError) error) (*Connection, *stubCodec) {
+ t.Helper()
+ transportInstance := udp.NewTransportInstance(
+ &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 3671}, nil, false, nil)
+ codec := &stubCodec{transportInstance: transportInstance, respond: respond}
+ connection := &Connection{
+ messageCodec: codec,
+ options: connectionOptions,
+ metadata: &ConnectionMetadata{},
+ log: testutils.ProduceTestingLogger(t),
+ }
+ return connection, codec
+}
+
+func respondWithConnectionState(status driverModel.Status) func(spi.Message, spi.HandleMessage, spi.HandleError) error {
+ return func(_ spi.Message, handleMessage spi.HandleMessage, _ spi.HandleError) error {
+ return handleMessage(driverModel.NewConnectionStateResponse(1, status))
+ }
+}
+
+func Test_Connection_IsConnected(t *testing.T) {
+ t.Run("no codec at all", func(t *testing.T) {
+ connection := &Connection{log: testutils.ProduceTestingLogger(t)}
+ assert.False(t, connection.IsConnected())
+ })
+ t.Run("gateway answers", func(t *testing.T) {
+ connection, codec := newStubbedConnection(t, nil, respondWithConnectionState(driverModel.Status_NO_ERROR))
+ assert.True(t, connection.IsConnected(), "a successful ping means we are connected")
+ require.Len(t, codec.sentRequests, 1)
+ assert.IsType(t, driverModel.NewConnectionStateRequest(0, newTestHPAIControlEndpoint()), codec.sentRequests[0])
+ })
+ t.Run("gateway reports an error", func(t *testing.T) {
+ connection, _ := newStubbedConnection(t, nil,
+ func(_ spi.Message, _ spi.HandleMessage, handleError spi.HandleError) error {
+ return handleError(errors.New("nope"))
+ })
+ assert.False(t, connection.IsConnected())
+ })
+ t.Run("sending fails", func(t *testing.T) {
+ connection, _ := newStubbedConnection(t, nil,
+ func(spi.Message, spi.HandleMessage, spi.HandleError) error {
+ return errors.New("transport is gone")
+ })
+ assert.False(t, connection.IsConnected())
+ })
+ t.Run("invalidated connection", func(t *testing.T) {
+ connection, codec := newStubbedConnection(t, nil, respondWithConnectionState(driverModel.Status_NO_ERROR))
+ connection.invalidated.Store(true)
+ assert.False(t, connection.IsConnected())
+ assert.Empty(t, codec.sentRequests, "an invalidated connection must not even try to ping")
+ })
+}
+
+func Test_Connection_connectionStateTimer(t *testing.T) {
+ t.Run("start creates the ticker and stop tears it down", func(t *testing.T) {
+ connection, _ := newStubbedConnection(t, nil, respondWithConnectionState(driverModel.Status_NO_ERROR))
+
+ connection.startConnectionStateTimer()
+ require.NotNil(t, connection.connectionStateTimer, "a keepalive ticker must be created on connect")
+ require.NotNil(t, connection.quitConnectionStateTimer)
+ quit := connection.quitConnectionStateTimer
+
+ // Starting twice must not replace the running ticker.
+ connection.startConnectionStateTimer()
+ assert.True(t, quit == connection.quitConnectionStateTimer, "starting twice must not swap the quit channel")
+
+ connection.stopConnectionStateTimer()
+ assert.Nil(t, connection.connectionStateTimer)
+ assert.Nil(t, connection.quitConnectionStateTimer)
+ select {
+ case <-quit:
+ default:
+ t.Error("the keepalive goroutine was never told to quit")
+ }
+ // Stopping twice must be safe.
+ assert.NotPanics(t, connection.stopConnectionStateTimer)
+ connection.wg.Wait()
+ })
+ // The keepalive used to be reset by interceptIncomingMessage, so on any installation
+ // with group telegrams more often than every 60s the ConnectionStateRequest was never
+ // sent and the gateway dropped the tunnel. The heartbeat has to run at a fixed rate.
+ t.Run("incoming traffic must not postpone the keepalive", func(t *testing.T) {
+ var keepalives atomic.Int32
+ connection, _ := newStubbedConnection(t, nil,
+ func(message spi.Message, handleMessage spi.HandleMessage, _ spi.HandleError) error {
+ if _, ok := message.(driverModel.ConnectionStateRequest); ok {
+ keepalives.Add(1)
+ }
+ return handleMessage(driverModel.NewConnectionStateResponse(1, driverModel.Status_NO_ERROR))
+ })
+ // Without a ticker the interceptor must be a no-op instead of a nil-dereference.
+ assert.NotPanics(t, func() { connection.interceptIncomingMessage(nil) })
+
+ connection.keepaliveInterval = 20 * time.Millisecond
+ connection.startConnectionStateTimer()
+ t.Cleanup(func() {
+ connection.stopConnectionStateTimer()
+ connection.wg.Wait()
+ })
+
+ // Feed the interceptor far more often than the keepalive interval.
+ const wantKeepalives = 3
+ deadline := time.Now().Add(10 * time.Second)
+ for time.Now().Before(deadline) && keepalives.Load() < wantKeepalives {
+ assert.NotPanics(t, func() { connection.interceptIncomingMessage(nil) })
+ time.Sleep(time.Millisecond)
+ }
+ assert.GreaterOrEqual(t, keepalives.Load(), int32(wantKeepalives),
+ "the heartbeat has to run at a fixed interval, independent of the incoming traffic")
+ })
+ t.Run("keepalive sends a connection state request", func(t *testing.T) {
+ var gotStatusRequest atomic.Bool
+ connection, codec := newStubbedConnection(t, nil,
+ func(message spi.Message, handleMessage spi.HandleMessage, _ spi.HandleError) error {
+ if _, ok := message.(driverModel.ConnectionStateRequest); ok {
+ gotStatusRequest.Store(true)
+ }
+ return handleMessage(driverModel.NewConnectionStateResponse(1, driverModel.Status_NO_ERROR))
+ })
+ connection.CommunicationChannelId = 42
+
+ connection.sendKeepalive()
+
+ assert.True(t, gotStatusRequest.Load())
+ require.Len(t, codec.sentRequests, 1)
+ connectionStateRequest, ok := codec.sentRequests[0].(driverModel.ConnectionStateRequest)
+ require.True(t, ok)
+ assert.Equal(t, uint8(42), connectionStateRequest.GetCommunicationChannelId())
+ })
+ t.Run("keepalive survives a failing gateway", func(t *testing.T) {
+ connection, _ := newStubbedConnection(t, nil,
+ func(spi.Message, spi.HandleMessage, spi.HandleError) error {
+ return errors.New("gateway is gone")
+ })
+ assert.NotPanics(t, connection.sendKeepalive)
+ })
+}
+
+func Test_Connection_getTunnelConnectionType(t *testing.T) {
+ tests := []struct {
+ name string
+ options map[string][]string
+ want driverModel.KnxLayer
+ }{
+ {"no option", nil, driverModel.KnxLayer_TUNNEL_LINK_LAYER},
+ {"empty option", map[string][]string{"connection-type": {}}, driverModel.KnxLayer_TUNNEL_LINK_LAYER},
+ {"link layer", map[string][]string{"connection-type": {"LINK_LAYER"}}, driverModel.KnxLayer_TUNNEL_LINK_LAYER},
+ {"lowercase", map[string][]string{"connection-type": {"link_layer"}}, driverModel.KnxLayer_TUNNEL_LINK_LAYER},
+ {"raw", map[string][]string{"connection-type": {"RAW"}}, driverModel.KnxLayer_TUNNEL_RAW},
+ {"busmonitor", map[string][]string{"connection-type": {"BUSMONITOR"}}, driverModel.KnxLayer_TUNNEL_BUSMONITOR},
+ {"prefixed", map[string][]string{"connection-type": {"TUNNEL_BUSMONITOR"}}, driverModel.KnxLayer_TUNNEL_BUSMONITOR},
+ {"garbage falls back", map[string][]string{"connection-type": {"nonsense"}}, driverModel.KnxLayer_TUNNEL_LINK_LAYER},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ connection, _ := newStubbedConnection(t, tt.options, nil)
+ assert.Equal(t, tt.want, connection.getTunnelConnectionType())
+ })
+ }
+}
+
+func Test_Connection_getRequestTimeout(t *testing.T) {
+ tests := []struct {
+ name string
+ options map[string][]string
+ want time.Duration
+ }{
+ {"no option", nil, defaultRequestTimeout},
+ {"empty option", map[string][]string{"request-timeout": {}}, defaultRequestTimeout},
+ {"explicit value", map[string][]string{"request-timeout": {"1500"}}, 1500 * time.Millisecond},
+ {"zero falls back", map[string][]string{"request-timeout": {"0"}}, defaultRequestTimeout},
+ {"garbage falls back", map[string][]string{"request-timeout": {"soon"}}, defaultRequestTimeout},
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ connection, _ := newStubbedConnection(t, tt.options, nil)
+ assert.Equal(t, tt.want, connection.getRequestTimeout())
+ })
+ }
+}
+
+// Test_ConnectionMetadata_GetConnectionAttributes guards against the copy-paste
+// bug which rendered the multicast- and mac-address from the serial number.
+func Test_ConnectionMetadata_GetConnectionAttributes(t *testing.T) {
+ metadata := &ConnectionMetadata{
+ KnxMedium: driverModel.KnxMedium_MEDIUM_TP1,
+ GatewayName: "gateway",
+ GatewayKnxAddress: "1.1.0",
+ ClientKnxAddress: "1.1.255",
+ ProjectNumber: 1,
+ InstallationNumber: 2,
+ DeviceSerialNumber: []byte{1, 2, 3, 4, 5, 6},
+ DeviceMulticastAddress: []byte{224, 0, 23, 12},
+ DeviceMacAddress: []byte{10, 20, 30, 40, 50, 60},
+ SupportedServices: []string{"core", "tunneling"},
+ }
+
+ attributes := metadata.GetConnectionAttributes()
+
+ assert.Equal(t, "1 2 3 4 5 6", attributes["DeviceSerialNumber"])
+ assert.Equal(t, "224.0.23.12", attributes["DeviceMulticastAddress"])
+ assert.Equal(t, "10:20:30:40:50:60", attributes["DeviceMacAddress"])
+ assert.Equal(t, "core, tunneling", attributes["SupportedServices"])
+}
diff --git a/plc4go/internal/knxnetip/MessageCodec.go b/plc4go/internal/knxnetip/MessageCodec.go
index 366a331630..3a2eb32c42 100644
--- a/plc4go/internal/knxnetip/MessageCodec.go
+++ b/plc4go/internal/knxnetip/MessageCodec.go
@@ -131,32 +131,37 @@ func (m *MessageCodec) Receive(ctx context.Context) (spi.Message, error) {
func CustomMessageHandling(localLog zerolog.Logger) _default.CustomMessageHandler {
return func(ctx context.Context, codec _default.DefaultCodecRequirements, message spi.Message) bool {
- // If this message is a simple KNXNet/IP UDP Ack, ignore it for now
- tunnelingResponse := message.(model.TunnelingResponse)
- if tunnelingResponse != nil {
- return true
- }
-
- // If this is an incoming tunneling request, automatically send a tunneling ACK back to the gateway
- tunnelingRequest := message.(model.TunnelingRequest)
- if tunnelingRequest != nil {
+ // Only the two tunneling frame types get special treatment here. Every
+ // other KNXnet/IP frame (SearchResponse, ConnectionResponse,
+ // ConnectionStateResponse, DisconnectRequest/Response, ...) has to fall
+ // through unhandled, otherwise the handshake- and keepalive-expectations
+ // registered on the codec would never see their responses.
+ switch typedMessage := message.(type) {
+ case model.TunnelingResponse:
+ // A TunnelingResponse is the gateway's ACK for a tunneling-request we sent.
+ // It must NOT be reported as "handled" here: DefaultCodec.ReceiveWork skips
+ // the expectations entirely for a message the custom handler claims, and
+ // everything which correlates its ACK (e.g. the group-address write) waits
+ // for exactly this frame. An ACK nobody waits for ends up on the default
+ // incoming-message channel, where the connection logs it.
+ localLog.Trace().Msg("Passing the tunneling ACK on to the expectations")
+ case model.TunnelingRequest:
+ // If this is an incoming tunneling request, automatically send a tunneling ACK back to the gateway
response := model.NewTunnelingResponse(
model.NewTunnelingResponseDataBlock(
- tunnelingRequest.GetTunnelingRequestDataBlock().GetCommunicationChannelId(),
- tunnelingRequest.GetTunnelingRequestDataBlock().GetSequenceCounter(),
+ typedMessage.GetTunnelingRequestDataBlock().GetCommunicationChannelId(),
+ typedMessage.GetTunnelingRequestDataBlock().GetSequenceCounter(),
model.Status_NO_ERROR),
)
- err := codec.Send(ctx, "tunneling_request", response) // TODO: where is a good place to get this timeout from?
- if err != nil {
+ if err := codec.Send(ctx, "tunneling_request", response); err != nil { // TODO: where is a good place to get this timeout from?
localLog.Warn().Err(err).Msg("got an error sending ACK from transport")
}
}
- localCodec := codec.(*MessageCodec)
// Handle the packet itself
// Give a message interceptor a chance to intercept
- if (*localCodec).messageInterceptor != nil {
- (*localCodec).messageInterceptor(message)
+ if localCodec, ok := codec.(*MessageCodec); ok && localCodec.messageInterceptor != nil {
+ localCodec.messageInterceptor(message)
}
return false
}
diff --git a/plc4go/internal/knxnetip/MessageCodec_test.go b/plc4go/internal/knxnetip/MessageCodec_test.go
new file mode 100644
index 0000000000..ec2ea69898
--- /dev/null
+++ b/plc4go/internal/knxnetip/MessageCodec_test.go
@@ -0,0 +1,279 @@
+/*
+ * 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 knxnetip
+
+import (
+ "context"
+ "net/url"
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/plc4x/plc4go/protocols/knxnetip/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi"
+ "github.com/apache/plc4x/plc4go/spi/testutils"
+ "github.com/apache/plc4x/plc4go/spi/transports/test"
+)
+
+func newTestHPAIControlEndpoint() model.HPAIControlEndpoint {
+ return model.NewHPAIControlEndpoint(
+ model.HostProtocolCode_IPV4_UDP, model.NewIPAddress([]byte{192, 168, 42, 11}), 3671)
+}
+
+func newTestSearchResponse() model.SearchResponse {
+ return model.NewSearchResponse(
+ newTestHPAIControlEndpoint(),
+ model.NewDIBDeviceInfo(
+ 54,
+ model.KnxMedium_MEDIUM_TP1,
+ model.NewDeviceStatus(false),
+ model.NewKnxAddress(1, 1, 0),
+ model.NewProjectInstallationIdentifier(1, 2),
+ []byte{1, 2, 3, 4, 5, 6},
+ model.NewIPAddress([]byte{224, 0, 23, 12}),
+ model.NewMACAddress([]byte{0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff}),
+ []byte("test-gateway"),
+ ),
+ model.NewDIBSuppSvcFamilies(2, nil),
+ )
+}
+
+func newTestTunnelingRequest(communicationChannelId, sequenceCounter uint8) model.TunnelingRequest {
+ return model.NewTunnelingRequest(
+ model.NewTunnelingRequestDataBlock(communicationChannelId, sequenceCounter),
+ model.NewLDataInd(0, nil,
+ model.NewLDataExtended(
+ true, true, model.CEMIPriority_LOW, false, false, true, 6, 0,
+ model.NewKnxAddress(1, 1, 1),
+ []byte{0x08, 0x01},
+ model.NewApduDataContainer(false, 0,
+ model.NewApduDataGroupValueWrite(1, nil)),
+ ),
+ ),
+ )
+}
+
+// newTestMessageCodec spins up a MessageCodec on top of a (connected, but not
+// running) test-transport. The returned transport instance can be used to check
+// what the codec wrote and the returned slice-pointer collects every message the
+// message-interceptor got to see.
+func newTestMessageCodec(t *testing.T) (*MessageCodec, *test.TransportInstance, *[]spi.Message) {
+ t.Helper()
+ _options := testutils.EnrichOptionsWithOptionsForTesting(t)
+
+ transport := test.NewTransport(_options...)
+ transportInstance, err := transport.CreateTransportInstance(url.URL{Scheme: "test"}, nil, _options...)
+ require.NoError(t, err)
+ testTransportInstance, ok := transportInstance.(*test.TransportInstance)
+ require.True(t, ok)
+ require.NoError(t, testTransportInstance.Connect(t.Context()))
+ t.Cleanup(func() {
+ assert.NoError(t, testTransportInstance.Close())
+ })
+
+ var intercepted []spi.Message
+ codec := NewMessageCodec(testTransportInstance, func(message spi.Message) {
+ intercepted = append(intercepted, message)
+ }, _options...)
+ return codec, testTransportInstance, &intercepted
+}
+
+// Test_CustomMessageHandling_passesNonTunnelingFramesOn is the regression test for
+// the single-value type-assertions which used to panic (and thus silently drop)
+// every frame which was neither a TunnelingRequest nor a TunnelingResponse. Those
+// are exactly the frames the connect-handshake and the keepalive wait for.
+func Test_CustomMessageHandling_passesNonTunnelingFramesOn(t *testing.T) {
+ tests := []struct {
+ name string
+ message spi.Message
+ }{
+ {
+ name: "SearchResponse",
+ message: newTestSearchResponse(),
+ },
+ {
+ name: "ConnectionResponse",
+ message: model.NewConnectionResponse(1, model.Status_NO_ERROR, nil, nil),
+ },
+ {
+ name: "ConnectionStateResponse",
+ message: model.NewConnectionStateResponse(1, model.Status_NO_ERROR),
+ },
+ {
+ name: "DisconnectResponse",
+ message: model.NewDisconnectResponse(1, model.Status_NO_ERROR),
+ },
+ {
+ name: "DisconnectRequest",
+ message: model.NewDisconnectRequest(1, newTestHPAIControlEndpoint()),
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ codec, transportInstance, intercepted := newTestMessageCodec(t)
+
+ handler := CustomMessageHandling(testutils.ProduceTestingLogger(t))
+
+ assert.NotPanics(t, func() {
+ assert.False(t, handler(t.Context(), codec, tt.message),
+ "non-tunneling frames must not be marked as handled, otherwise no expectation ever sees them")
+ })
+
+ assert.Equal(t, []spi.Message{tt.message}, *intercepted,
+ "the message interceptor must see the frame")
+ assert.Zero(t, transportInstance.GetNumDrainableBytes(),
+ "no response should have been sent for a non-tunneling frame")
+ })
+ }
+}
+
+// Test_CustomMessageHandling_tunnelingResponseIsPassedOn is the regression test for the
+// tunneling ACK which used to be swallowed here: DefaultCodec.ReceiveWork skips the
+// expectations completely for a message the custom handler reports as handled, so a
+// swallowed ACK made every correlated tunneling-request (e.g. a group-address write) run
+// into its request-timeout.
+func Test_CustomMessageHandling_tunnelingResponseIsPassedOn(t *testing.T) {
+ codec, transportInstance, intercepted := newTestMessageCodec(t)
+
+ handler := CustomMessageHandling(testutils.ProduceTestingLogger(t))
+
+ tunnelingResponse := model.NewTunnelingResponse(
+ model.NewTunnelingResponseDataBlock(1, 2, model.Status_NO_ERROR))
+
+ assert.False(t, handler(t.Context(), codec, tunnelingResponse),
+ "an ACK must reach the expectations, otherwise no tunneling request can ever be correlated")
+ assert.Equal(t, []spi.Message{spi.Message(tunnelingResponse)}, *intercepted)
+ assert.Zero(t, transportInstance.GetNumDrainableBytes(), "an ACK is not answered")
+}
+
+// Test_MessageCodec_tunnelingAckSatisfiesAnExpectation drives the very same case through
+// the running codec, which is what the write-path relies on.
+func Test_MessageCodec_tunnelingAckSatisfiesAnExpectation(t *testing.T) {
+ codec, transportInstance, _ := newTestMessageCodec(t)
+ require.NoError(t, codec.Connect(t.Context()))
+ t.Cleanup(func() { _ = codec.Disconnect() })
+
+ tunnelingResponse := model.NewTunnelingResponse(
+ model.NewTunnelingResponseDataBlock(0x2a, 0x0b, model.Status_NO_ERROR))
+
+ handled := make(chan model.TunnelingResponse, 1)
+ codec.Expect(t.Context(), "tunneling_ack",
+ func(message spi.Message) bool {
+ _, ok := message.(model.TunnelingResponse)
+ return ok
+ },
+ func(message spi.Message) error {
+ handled <- message.(model.TunnelingResponse)
+ return nil
+ },
+ func(err error) error { return nil },
+ )
+
+ theBytes, err := tunnelingResponse.Serialize()
+ require.NoError(t, err)
+ transportInstance.FillReadBuffer(theBytes)
+ // The test transport only moves the bytes from its channel into the read buffer on an
+ // actual read, while the codec only reads once it sees them as available.
+ _, err = transportInstance.PeekReadableBytes(t.Context(), uint32(len(theBytes)))
+ require.NoError(t, err)
+
+ select {
+ case got := <-handled:
+ assert.Equal(t, uint8(0x2a), got.GetTunnelingResponseDataBlock().GetCommunicationChannelId())
+ assert.Equal(t, uint8(0x0b), got.GetTunnelingResponseDataBlock().GetSequenceCounter())
+ case <-time.After(10 * time.Second):
+ t.Fatal("the expectation never saw the tunneling ack")
+ }
+}
+
+func Test_CustomMessageHandling_tunnelingRequestIsAcknowledged(t *testing.T) {
+ codec, transportInstance, intercepted := newTestMessageCodec(t)
+
+ handler := CustomMessageHandling(testutils.ProduceTestingLogger(t))
+
+ tunnelingRequest := newTestTunnelingRequest(0x2a, 0x0b)
+
+ assert.False(t, handler(t.Context(), codec, tunnelingRequest),
+ "a tunneling request still has to be passed on for further handling")
+ assert.Equal(t, []spi.Message{spi.Message(tunnelingRequest)}, *intercepted)
+
+ // The gateway must have gotten an ACK for the request.
+ numDrainable := transportInstance.GetNumDrainableBytes()
+ require.NotZero(t, numDrainable)
+ rawAck := transportInstance.DrainWriteBuffer(numDrainable)
+ parsedAck, err := model.KnxNetIpMessageParse[model.KnxNetIpMessage](context.Background(), rawAck)
+ require.NoError(t, err)
+ ack, ok := parsedAck.(model.TunnelingResponse)
+ require.True(t, ok, "expected a TunnelingResponse, got %T", parsedAck)
+ assert.Equal(t, uint8(0x2a), ack.GetTunnelingResponseDataBlock().GetCommunicationChannelId())
+ assert.Equal(t, uint8(0x0b), ack.GetTunnelingResponseDataBlock().GetSequenceCounter())
+ assert.Equal(t, model.Status_NO_ERROR, ack.GetTunnelingResponseDataBlock().GetStatus())
+}
+
+// foreignCodec is a _default.DefaultCodecRequirements implementation which is NOT our
+// *MessageCodec, which is what the comma-ok type assertion in CustomMessageHandling has to
+// cope with.
+type foreignCodec struct {
+ sent []spi.Message
+}
+
+func (f *foreignCodec) GetCodec() spi.MessageCodec { return nil }
+
+func (f *foreignCodec) Send(_ context.Context, _ string, message spi.Message) error {
+ f.sent = append(f.sent, message)
+ return nil
+}
+
+func (f *foreignCodec) Receive(_ context.Context) (spi.Message, error) { return nil, nil }
+
+// Test_CustomMessageHandling_foreignCodec makes sure we don't blow up if the
+// handler is ever attached to something which isn't our own codec.
+func Test_CustomMessageHandling_foreignCodec(t *testing.T) {
+ handler := CustomMessageHandling(testutils.ProduceTestingLogger(t))
+
+ t.Run("a frame without an interceptor to call", func(t *testing.T) {
+ codec := &foreignCodec{}
+ assert.NotPanics(t, func() {
+ assert.False(t, handler(t.Context(), codec, newTestSearchResponse()))
+ })
+ assert.Empty(t, codec.sent)
+ })
+ t.Run("a tunneling request is still acknowledged", func(t *testing.T) {
+ codec := &foreignCodec{}
+ tunnelingRequest := newTestTunnelingRequest(0x2a, 0x0b)
+ assert.NotPanics(t, func() {
+ assert.False(t, handler(t.Context(), codec, tunnelingRequest))
+ })
+ require.Len(t, codec.sent, 1)
+ ack, ok := codec.sent[0].(model.TunnelingResponse)
+ require.True(t, ok, "expected a TunnelingResponse, got %T", codec.sent[0])
+ assert.Equal(t, uint8(0x2a), ack.GetTunnelingResponseDataBlock().GetCommunicationChannelId())
+ assert.Equal(t, uint8(0x0b), ack.GetTunnelingResponseDataBlock().GetSequenceCounter())
+ })
+ t.Run("our own codec without an interceptor", func(t *testing.T) {
+ codec, _, _ := newTestMessageCodec(t)
+ codec.messageInterceptor = nil
+ assert.NotPanics(t, func() {
+ assert.False(t, handler(t.Context(), codec, newTestSearchResponse()))
+ })
+ })
+}