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/20 08:16:56 UTC
(plc4x) 03/03: test(plc4x): cover the paths the new drivers' fixes left unasserted
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 33435692ef0d17b4315c7185223d133cab70cf41
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Thu Aug 20 10:14:22 2026 +0200
test(plc4x): cover the paths the new drivers' fixes left unasserted
A review of the firmata and ab-eth ports raised several defects that had in fact
already been fixed while the review was being written, but whose fixes nothing
pinned. Filling those gaps rather than trusting the fix:
- firmata asks the board for its firmware report instead of only waiting for a
volunteered one, which is what makes the TCP transport work at all. Tests now
assert the exact bytes, that a board answering only the request completes the
handshake, and that a failed request fails the connect instead of waiting out
the timeout.
- firmata gives a pin claim back when part of its batch fails to send, so a retry
re-sends the whole batch rather than finding the pin claimed and doing nothing.
Tests now fail a send part way through, not just entirely.
- ab-eth drains the codec's default incoming channel, so a late or duplicate
packet no longer fills a 100 slot buffer and turns every later stray packet
into a warning for the life of the connection.
EncodingBCD's javadoc gains the other half of its alignment contract: the buffers
run byteOrder.process between the encoding and the bit stream, so reversing a
right-aligned array moves the padding nibble into the middle of the field and BCD
does not survive LITTLE_ENDIAN. Nothing pairs the two today, and this is not a
regression - the previous algorithm failed the same pipeline - but the class now
says so rather than implying a contract it cannot keep.
---
plc4go/internal/abeth/Connection.go | 44 ++++++---
plc4go/internal/abeth/Connection_test.go | 108 +++++++++++++++++++++
plc4go/internal/abeth/endToEnd_test.go | 28 ------
plc4go/internal/firmata/Connection_test.go | 81 ++++++++++++++++
plc4go/internal/firmata/Subscriber_test.go | 28 ++++++
plc4go/internal/firmata/Writer_test.go | 28 ++++++
plc4go/internal/firmata/codec_stub_test.go | 41 ++++++--
.../buffers/bytebased/encoding/EncodingBCD.java | 5 +
8 files changed, 313 insertions(+), 50 deletions(-)
diff --git a/plc4go/internal/abeth/Connection.go b/plc4go/internal/abeth/Connection.go
index 12966c3315..76ccd8f3b3 100644
--- a/plc4go/internal/abeth/Connection.go
+++ b/plc4go/internal/abeth/Connection.go
@@ -68,10 +68,11 @@ type Connection struct {
connectionId string
tracer tracer.Tracer
- // lifecycleMutex guards the stray-message drain's cancel function so Close can be called
- // concurrently with, and without, a preceding Connect.
+ // lifecycleMutex guards the stray-message drain's handles so Close can be called concurrently
+ // with, and without, a preceding Connect.
lifecycleMutex sync.Mutex
drainCancel context.CancelFunc
+ drainDone chan struct{}
wg sync.WaitGroup // use to track spawned go routines
@@ -177,6 +178,9 @@ func (c *Connection) Connect(ctx context.Context) error {
}
if err := c.handshake(ctx); err != nil {
+ // The caller gets an error instead of a connection and will never call Close on it, so the
+ // drain has to be reeled back in here or its goroutine outlives the failed attempt.
+ c.stopStrayMessageDrain()
if disconnectErr := c.messageCodec.Disconnect(); disconnectErr != nil {
c.log.Debug().Err(disconnectErr).Msg("error disconnecting after a failed handshake")
}
@@ -217,13 +221,7 @@ func (c *Connection) handshake(ctx context.Context) error {
// AbEthConnection just closes the socket - so there is nothing to send here.
func (c *Connection) Close() error {
c.log.Trace().Msg("Closing")
- c.lifecycleMutex.Lock()
- drainCancel := c.drainCancel
- c.drainCancel = nil
- c.lifecycleMutex.Unlock()
- if drainCancel != nil {
- drainCancel()
- }
+ c.stopStrayMessageDrain()
if c.subscriber != nil {
c.subscriber.Close()
}
@@ -240,18 +238,21 @@ func (c *Connection) Close() error {
// so on a flaky link a handful of late responses turn into permanent log noise. Draining costs one
// goroutine and turns that into a single debug line per stray packet.
func (c *Connection) startStrayMessageDrain() {
+ // A second Connect on the same connection must not leave the first drain running.
+ c.stopStrayMessageDrain()
+
drainCtx, drainCancel := context.WithCancel(context.Background())
+ drainDone := make(chan struct{})
c.lifecycleMutex.Lock()
- previous := c.drainCancel
c.drainCancel = drainCancel
+ c.drainDone = drainDone
c.lifecycleMutex.Unlock()
- // A second Connect on the same connection must not leave the first drain running.
- if previous != nil {
- previous()
- }
incomingMessageChannel := c.messageCodec.GetDefaultIncomingMessageChannel()
c.wg.Go(func() {
+ // Closing this last, after the recover below, is what lets stopStrayMessageDrain wait for
+ // the goroutine to really be gone.
+ defer close(drainDone)
defer func() {
if err := recover(); err != nil {
c.log.Error().
@@ -274,6 +275,21 @@ func (c *Connection) startStrayMessageDrain() {
})
}
+// stopStrayMessageDrain ends the drain and waits for its goroutine to be gone, so that a caller
+// which stops the drain can rely on nothing reading the channel any more. Calling it without a
+// running drain, or twice, is harmless.
+func (c *Connection) stopStrayMessageDrain() {
+ c.lifecycleMutex.Lock()
+ drainCancel, drainDone := c.drainCancel, c.drainDone
+ c.drainCancel, c.drainDone = nil, nil
+ c.lifecycleMutex.Unlock()
+ if drainCancel == nil {
+ return
+ }
+ drainCancel()
+ <-drainDone
+}
+
////////////////////////////////////////////////////////////////////////////////////////////////////
// Metadata and builders
////////////////////////////////////////////////////////////////////////////////////////////////////
diff --git a/plc4go/internal/abeth/Connection_test.go b/plc4go/internal/abeth/Connection_test.go
index 29a730611c..0d60cf434c 100644
--- a/plc4go/internal/abeth/Connection_test.go
+++ b/plc4go/internal/abeth/Connection_test.go
@@ -29,6 +29,7 @@ import (
apiModel "github.com/apache/plc4x/plc4go/pkg/api/model"
apiValues "github.com/apache/plc4x/plc4go/pkg/api/values"
readWriteModel "github.com/apache/plc4x/plc4go/protocols/abeth/readwrite/model"
+ "github.com/apache/plc4x/plc4go/spi"
"github.com/apache/plc4x/plc4go/spi/testutils"
"github.com/apache/plc4x/plc4go/spi/transactions"
spiValues "github.com/apache/plc4x/plc4go/spi/values"
@@ -393,3 +394,110 @@ func TestConnection_CloseWithoutConnect(t *testing.T) {
// failed connect attempt.
assert.NoError(t, connection.Close())
}
+
+// strayMessage is a packet no request is waiting for: what the codec hands to its default incoming
+// message channel and what the drain is there to throw away.
+func strayMessage() spi.Message {
+ return readWriteModel.NewCIPEncapsulationReadResponse(
+ testSessionHandle, 0, emptySenderContext, 0,
+ readWriteModel.NewDF1CommandResponseMessageProtectedTypedLogicalRead(
+ df1SourceAddress, DefaultConfiguration().station, 0, 1, []uint8{0}))
+}
+
+// requireNothingDrains insists that nothing is reading the codec's default incoming message channel
+// any more: a message handed to it stays there. Both the close and the failed-connect path stop the
+// drain synchronously, so a single check is enough - there is no goroutine left that could still
+// take the message afterwards.
+func requireNothingDrains(t *testing.T, codec *stubCodec) {
+ t.Helper()
+ incoming := codec.GetDefaultIncomingMessageChannel()
+ require.Empty(t, incoming, "the drain should have emptied the channel while it was running")
+ incoming <- strayMessage()
+ assert.Len(t, incoming, 1, "something is still draining the channel")
+}
+
+// TestConnection_StrayMessageDrainStopsOnClose pins the drain's lifetime to the connection's: it has
+// to be gone once Close returned, rather than sitting on the codec's channel forever.
+func TestConnection_StrayMessageDrainStopsOnClose(t *testing.T) {
+ connection, codec := newTestConnection(t, DefaultConfiguration())
+
+ connectResult := make(chan error, 1)
+ go func() {
+ connectResult <- connection.Connect(testutils.TestContext(t))
+ }()
+ connectHandshake(t, codec)
+ require.NoError(t, <-connectResult)
+
+ // While the connection is up the drain takes everything that lands on the channel.
+ codec.GetDefaultIncomingMessageChannel() <- strayMessage()
+ require.Eventually(t, func() bool {
+ return len(codec.GetDefaultIncomingMessageChannel()) == 0
+ }, 5*time.Second, time.Millisecond, "the stray message was never drained")
+
+ require.NoError(t, connection.Close())
+ requireNothingDrains(t, codec)
+}
+
+// TestConnection_StrayMessageDrainStopsWhenTheHandshakeFails covers the path with no Close in it: the
+// driver hands the caller an error instead of a connection, so nobody will ever close this one and
+// the drain has to have stopped itself.
+func TestConnection_StrayMessageDrainStopsWhenTheHandshakeFails(t *testing.T) {
+ connection, codec := newTestConnection(t, DefaultConfiguration())
+
+ connectResult := make(chan error, 1)
+ go func() {
+ connectResult <- connection.Connect(testutils.TestContext(t))
+ }()
+ request := codec.nextRequest(t)
+ require.NoError(t, request.handleError(assert.AnError))
+ require.Error(t, <-connectResult)
+
+ requireNothingDrains(t, codec)
+}
+
+// TestConnection_StrayMessageDrainIsNotDoubledByASecondConnect makes sure a reconnect replaces the
+// drain instead of stacking a second one on top of it, and that closing still stops all of them.
+func TestConnection_StrayMessageDrainIsNotDoubledByASecondConnect(t *testing.T) {
+ connection, codec := newTestConnection(t, DefaultConfiguration())
+
+ for i := 0; i < 3; i++ {
+ connectResult := make(chan error, 1)
+ go func() {
+ connectResult <- connection.Connect(testutils.TestContext(t))
+ }()
+ connectHandshake(t, codec)
+ require.NoError(t, <-connectResult)
+ }
+
+ require.NoError(t, connection.Close())
+ requireNothingDrains(t, codec)
+}
+
+// TestConnection_StrayMessageDrainKeepsUpWithMoreThanTheChannelHolds is the finding itself: without a
+// drain the channel filled up after its last slot was taken and the codec logged a warning for every
+// further stray packet for the rest of the connection's life. With one, far more packets than the
+// channel can hold go through it without it ever filling up.
+func TestConnection_StrayMessageDrainKeepsUpWithMoreThanTheChannelHolds(t *testing.T) {
+ connection, codec := newTestConnection(t, DefaultConfiguration())
+
+ connectResult := make(chan error, 1)
+ go func() {
+ connectResult <- connection.Connect(testutils.TestContext(t))
+ }()
+ connectHandshake(t, codec)
+ require.NoError(t, <-connectResult)
+
+ incoming := codec.GetDefaultIncomingMessageChannel()
+ for i := 0; i < 5*cap(incoming); i++ {
+ select {
+ case incoming <- strayMessage():
+ case <-time.After(5 * time.Second):
+ require.FailNow(t, "the channel filled up, so the stray messages aren't being drained")
+ }
+ }
+ require.Eventually(t, func() bool {
+ return len(incoming) == 0
+ }, 5*time.Second, time.Millisecond, "the stray messages were never drained")
+
+ require.NoError(t, connection.Close())
+}
diff --git a/plc4go/internal/abeth/endToEnd_test.go b/plc4go/internal/abeth/endToEnd_test.go
index c895f8d8bb..0fcf3a974e 100644
--- a/plc4go/internal/abeth/endToEnd_test.go
+++ b/plc4go/internal/abeth/endToEnd_test.go
@@ -126,31 +126,3 @@ func TestEndToEnd_ReadThroughTheRunningCodec(t *testing.T) {
require.FailNow(t, "the read never completed")
}
}
-
-// A packet no request is waiting for used to sit in the codec's default incoming message channel for
-// the life of the connection: nothing read that channel, so once its 100 slots were full every
-// further stray packet cost a warning. The drain keeps it empty.
-func TestEndToEnd_StrayPacketsAreDrained(t *testing.T) {
- connection, codec := newTestConnection(t, DefaultConfiguration())
-
- connectErrors := make(chan error, 1)
- go func() {
- connectErrors <- connection.Connect(testutils.TestContext(t))
- }()
- connectHandshake(t, codec)
- require.NoError(t, <-connectErrors)
-
- // A late read response - its request timed out long ago - for every slot the channel has.
- incoming := codec.GetDefaultIncomingMessageChannel()
- for i := 0; i < cap(incoming); i++ {
- incoming <- readWriteModel.NewCIPEncapsulationReadResponse(
- testSessionHandle, 0, emptySenderContext, 0,
- readWriteModel.NewDF1CommandResponseMessageProtectedTypedLogicalRead(
- df1SourceAddress, DefaultConfiguration().station, 0, uint16(i), []uint8{0}))
- }
- assert.Eventually(t, func() bool {
- return len(incoming) == 0
- }, 20*time.Second, time.Millisecond, "the stray packets were never drained")
-
- require.NoError(t, connection.Close())
-}
diff --git a/plc4go/internal/firmata/Connection_test.go b/plc4go/internal/firmata/Connection_test.go
index 9300f0be20..8efc1c7ef4 100644
--- a/plc4go/internal/firmata/Connection_test.go
+++ b/plc4go/internal/firmata/Connection_test.go
@@ -155,6 +155,87 @@ func TestConnection_ConnectFailsOnAHandshakeError(t *testing.T) {
assert.False(t, connection.IsConnected())
}
+// The firmware report is asked for rather than waited on. StandardFirmata's systemResetCallback
+// only clears the pin state - printFirmwareVersion runs from setup() - so a board volunteers the
+// report only when it really restarts. Over serial that is masked, as opening the port toggles DTR
+// and resets the AVR, but nothing resets a WiFi or Ethernet board when a TCP connection is opened,
+// which is the transport this driver newly speaks. plc4j's onConnect only ever waits.
+func TestConnection_HandshakeAsksForTheFirmwareReport(t *testing.T) {
+ codec := newStubCodec()
+ configuration := DefaultConfiguration()
+ configuration.requestTimeout = 250 * time.Millisecond
+ connection := NewConnection(configuration, codec, map[string][]string{}, NewTagHandler())
+
+ // Nothing is fed back, so the handshake runs into its timeout - what matters is what went out.
+ assert.Error(t, connection.Connect(testutils.TestContext(t)))
+
+ assert.Equal(t, []byte{
+ 0xFF, // system reset
+ 0xF0, 0x79, 0xF7, // report firmware: asked for, not waited for
+ }, codec.sentBytesOf(t))
+}
+
+// A board which only answers the request - the TCP case, where the reset never restarts it - still
+// completes the handshake.
+func TestConnection_HandshakeCompletesOnTheAnsweredRequest(t *testing.T) {
+ codec := newStubCodec()
+ connection := NewConnection(DefaultConfiguration(), codec, map[string][]string{}, NewTagHandler())
+
+ connectErrors := make(chan error, 1)
+ go func() {
+ connectErrors <- connection.Connect(testutils.TestContext(t))
+ }()
+
+ var request stubRequest
+ select {
+ case request = <-codec.requests:
+ case <-time.After(20 * time.Second):
+ require.FailNow(t, "no system reset was sent")
+ }
+ // Wait for the request to be out before answering it: it is the request the board reacts to.
+ require.Eventually(t, func() bool { return len(codec.getSent()) == 2 }, 20*time.Second, 10*time.Millisecond)
+
+ report := parseMessage(t, reportFirmwareFrame(0x02, 0x05, "StandardFirmataWiFi.ino"))
+ require.True(t, request.acceptsMessage(report))
+ require.NoError(t, request.handleMessage(report))
+
+ select {
+ case err := <-connectErrors:
+ require.NoError(t, err)
+ case <-time.After(20 * time.Second):
+ require.FailNow(t, "the connect never returned")
+ }
+ require.NoError(t, connection.Close())
+}
+
+// A transport which carries the reset but not the report-firmware request fails the connect right
+// away rather than waiting out the timeout for a report which was never asked for. Both halves of
+// that need pinning: the error has to name the send which failed (a handshake which swallowed it and
+// then timed out would report an error too), and the connect has to be back long before the request
+// timeout - deliberately set far above the bound below - could have expired.
+func TestConnection_ConnectFailsWhenTheFirmwareRequestCantBeSent(t *testing.T) {
+ codec := newStubCodec()
+ // The system reset gets out, the report-firmware request behind it doesn't.
+ codec.failSendsAfter(1)
+ configuration := DefaultConfiguration()
+ configuration.requestTimeout = time.Minute
+ connection := NewConnection(configuration, codec, map[string][]string{}, NewTagHandler())
+
+ started := time.Now()
+ err := connection.Connect(testutils.TestContext(t))
+ elapsed := time.Since(started)
+
+ require.Error(t, err)
+ assert.ErrorContains(t, err, "error requesting the firmware report",
+ "the connect has to fail on the send which failed")
+ assert.ErrorContains(t, err, "the transport is gone", "and it has to carry the transport's error")
+ assert.Less(t, elapsed, 10*time.Second,
+ "the connect has to fail on the failed send instead of waiting out the %s request timeout",
+ configuration.requestTimeout)
+ assert.False(t, connection.IsConnected())
+ assert.Len(t, codec.getSent(), 1)
+}
+
// A transport which is gone can't even carry the system reset.
func TestConnection_ConnectFailsWhenTheResetCantBeSent(t *testing.T) {
codec := newStubCodec()
diff --git a/plc4go/internal/firmata/Subscriber_test.go b/plc4go/internal/firmata/Subscriber_test.go
index fca357cc5b..4396a075a7 100644
--- a/plc4go/internal/firmata/Subscriber_test.go
+++ b/plc4go/internal/firmata/Subscriber_test.go
@@ -420,3 +420,31 @@ func TestSubscriber_SubscribeRefusesEventTags(t *testing.T) {
// Only the tag which was accepted was configured on the board.
assert.Equal(t, []byte{0xF4, 0x0A, 0x00, 0xD1, 0x01}, sentBytes(t, transportInstance))
}
+
+// Same for a subscription which dies part-way: the set-pin-mode gets out but the report-digital
+// which actually switches reporting on for the port does not, so the board says nothing about the
+// pin. A claim left standing would make the retry hand out a handle that never delivers.
+func TestSubscriber_APartialSendGivesThePinClaimBack(t *testing.T) {
+ codec := newStubCodec()
+ connection := NewConnection(DefaultConfiguration(), codec, map[string][]string{}, NewTagHandler())
+ codec.failSendsAfter(1)
+
+ handle, responseCode := subscribe(t, connection, "button", "digital:9")
+ require.Equal(t, apiModel.PlcResponseCode_INTERNAL_ERROR, responseCode)
+ assert.Nil(t, handle)
+ func() {
+ connection.pinMutex.Lock()
+ defer connection.pinMutex.Unlock()
+ assert.NotContains(t, connection.digitalPins, uint8(9), "a claim whose batch died part-way has to be given back")
+ }()
+
+ codec.allowSends()
+ handle, responseCode = subscribe(t, connection, "button", "digital:9")
+ require.Equal(t, apiModel.PlcResponseCode_OK, responseCode)
+ require.NotNil(t, handle)
+ assert.Equal(t, []byte{
+ 0xF4, 0x09, 0x00, // the set pin mode which did get out the first time
+ 0xF4, 0x09, 0x00, // and both messages again from the retry
+ 0xD1, 0x01,
+ }, codec.sentBytesOf(t))
+}
diff --git a/plc4go/internal/firmata/Writer_test.go b/plc4go/internal/firmata/Writer_test.go
index b601dfed1a..7f0e85084d 100644
--- a/plc4go/internal/firmata/Writer_test.go
+++ b/plc4go/internal/firmata/Writer_test.go
@@ -249,3 +249,31 @@ func TestWriter_AFailedSendGivesThePinClaimBack(t *testing.T) {
0xF5, 0x09, 0x01, // set digital pin value: pin 9, high
}, codec.sentBytesOf(t))
}
+
+// The interesting failure is the partial one: the set-pin-mode makes it onto the wire and the value
+// behind it doesn't. The pin is not usably configured either way - nothing was written to it - so
+// the claim still has to go back, and the retry re-sends both messages.
+func TestWriter_APartialSendGivesThePinClaimBack(t *testing.T) {
+ codec := newStubCodec()
+ connection := NewConnection(DefaultConfiguration(), codec, map[string][]string{}, NewTagHandler())
+ codec.failSendsAfter(1)
+
+ result := executeWrite(t, connection, "led", "digital:9", true)
+ require.NoError(t, result.GetErr())
+ require.Equal(t, apiModel.PlcResponseCode_INTERNAL_ERROR, result.GetResponse().GetResponseCode("led"))
+ func() {
+ connection.pinMutex.Lock()
+ defer connection.pinMutex.Unlock()
+ assert.NotContains(t, connection.digitalPins, uint8(9), "a claim whose batch died part-way has to be given back")
+ }()
+
+ codec.allowSends()
+ result = executeWrite(t, connection, "led", "digital:9", true)
+ require.NoError(t, result.GetErr())
+ assert.Equal(t, apiModel.PlcResponseCode_OK, result.GetResponse().GetResponseCode("led"))
+ assert.Equal(t, []byte{
+ 0xF4, 0x09, 0x01, // the set pin mode which did get out the first time
+ 0xF4, 0x09, 0x01, // and both messages again from the retry
+ 0xF5, 0x09, 0x01,
+ }, codec.sentBytesOf(t))
+}
diff --git a/plc4go/internal/firmata/codec_stub_test.go b/plc4go/internal/firmata/codec_stub_test.go
index 7948c74fa3..ca7589be88 100644
--- a/plc4go/internal/firmata/codec_stub_test.go
+++ b/plc4go/internal/firmata/codec_stub_test.go
@@ -36,11 +36,14 @@ import (
// transport in MessageCodec_test.go; here it would only get in the way, as driving a board means
// driving both ends of the same transport at once.
type stubCodec struct {
- mutex sync.Mutex
- sent []spi.Message
- sendErr error
- requests chan stubRequest
- incoming chan spi.Message
+ mutex sync.Mutex
+ sent []spi.Message
+ sendErr error
+ // remainingSends is how many further sends a failSendsAfter still lets through before sendErr
+ // starts to bite. It is what lets a test fail a batch of messages part-way.
+ remainingSends int
+ requests chan stubRequest
+ incoming chan spi.Message
}
type stubRequest struct {
@@ -66,8 +69,8 @@ func (c *stubCodec) IsRunning() bool { return true }
func (c *stubCodec) Send(_ context.Context, _ string, message spi.Message) error {
c.mutex.Lock()
defer c.mutex.Unlock()
- if c.sendErr != nil {
- return c.sendErr
+ if err := c.sendError(); err != nil {
+ return err
}
c.sent = append(c.sent, message)
return nil
@@ -78,7 +81,7 @@ func (c *stubCodec) Expect(context.Context, string, spi.AcceptsMessage, spi.Hand
func (c *stubCodec) SendRequest(_ context.Context, _ string, message spi.Message, acceptsMessage spi.AcceptsMessage, handleMessage spi.HandleMessage, handleError spi.HandleError) error {
c.mutex.Lock()
- sendErr := c.sendErr
+ sendErr := c.sendError()
if sendErr == nil {
c.sent = append(c.sent, message)
}
@@ -106,10 +109,31 @@ func (c *stubCodec) getSent() []spi.Message {
return append([]spi.Message(nil), c.sent...)
}
+// sendError reports whether this send is to fail, counting down the sends a failSendsAfter still
+// lets through. Callers hold the mutex.
+func (c *stubCodec) sendError() error {
+ if c.sendErr == nil {
+ return nil
+ }
+ if c.remainingSends > 0 {
+ c.remainingSends--
+ return nil
+ }
+ return c.sendErr
+}
+
// failSends makes every following send fail.
func (c *stubCodec) failSends() {
+ c.failSendsAfter(0)
+}
+
+// failSendsAfter lets the next n sends through and fails every send after them. It is how a test
+// drives a batch of messages which dies part-way - the case a claim recorded before its messages
+// went out has to survive.
+func (c *stubCodec) failSendsAfter(n int) {
c.mutex.Lock()
defer c.mutex.Unlock()
+ c.remainingSends = n
c.sendErr = errors.New("the transport is gone")
}
@@ -118,6 +142,7 @@ func (c *stubCodec) allowSends() {
c.mutex.Lock()
defer c.mutex.Unlock()
c.sendErr = nil
+ c.remainingSends = 0
}
// sentBytesOf serializes everything the connection has sent, which is what would have been on the
diff --git a/plc4j/spi/buffers/byte/src/main/java/org/apache/plc4x/java/spi/buffers/bytebased/encoding/EncodingBCD.java b/plc4j/spi/buffers/byte/src/main/java/org/apache/plc4x/java/spi/buffers/bytebased/encoding/EncodingBCD.java
index 3b13f5541a..ff6d03a5f8 100644
--- a/plc4j/spi/buffers/byte/src/main/java/org/apache/plc4x/java/spi/buffers/bytebased/encoding/EncodingBCD.java
+++ b/plc4j/spi/buffers/byte/src/main/java/org/apache/plc4x/java/spi/buffers/bytebased/encoding/EncodingBCD.java
@@ -44,6 +44,11 @@ import java.math.BigInteger;
* {@code LITTLE_ENDIAN} a 12 bit write of {@code 123} reads back as {@code 103}. Reversing whole bytes
* is simply not a meaningful operation on a nibble stream that does not start on a byte boundary; if a
* protocol ever needs byte-swapped BCD it has to swap complete BCD bytes, not the encoded field.
+ * <p>
+ * Rejecting that combination outright would be preferable to corrupting the value silently, but the
+ * {@code decode*}/{@code encode*} contract passes no byte order down here, so the guard would have to
+ * live in the buffers next to their {@code byteOrder.process} call - a behaviour change (a hard failure
+ * where callers see none today) and hence deliberately not made as part of the alignment fix.
*/
public class EncodingBCD extends BaseEncodingDefault {