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/07/06 18:53:51 UTC
(plc4x) 06/09: feat(plc4go): full serial option support in connection strings
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 b257386f0714952175c3dd039a6b0dfa140040fa
Author: Sebastian Rühl <sr...@apache.org>
AuthorDate: Mon Jul 6 14:49:47 2026 +0200
feat(plc4go): full serial option support in connection strings
The serial transport now parses data-bits, stop-bits, parity (incl.
mark/space), flow-control (rts-cts, xon-xoff), dtr, rts, read-timeout
and write-timeout. Timeouts act as fallback deadlines: an explicit
context deadline always wins; deadline-less reads/writes are bounded by
the configured timeout (default 1000 ms, 0 = blocking). Invalid option
values fail connection creation naming the option; enum values are
case-insensitive with "-"/"_" interchangeable.
Breaking: default baud-rate changed from 115200 to 9600; deadline-less
I/O no longer blocks indefinitely by default (see RELEASE_NOTES).
---
RELEASE_NOTES | 12 ++
plc4go/spi/transports/serial/Transport.go | 24 +--
plc4go/spi/transports/serial/TransportInstance.go | 55 +++++--
.../serial/TransportInstance_pty_test.go | 68 +++++++++
.../transports/serial/TransportInstance_test.go | 87 +++++++++++
plc4go/spi/transports/serial/deadline_reader.go | 80 ++++++++++
.../spi/transports/serial/deadline_reader_test.go | 110 ++++++++++++++
plc4go/spi/transports/serial/options.go | 167 +++++++++++++++++++++
plc4go/spi/transports/serial/options_test.go | 160 ++++++++++++++++++++
9 files changed, 734 insertions(+), 29 deletions(-)
diff --git a/RELEASE_NOTES b/RELEASE_NOTES
index 8742246cdd..8278a7be72 100644
--- a/RELEASE_NOTES
+++ b/RELEASE_NOTES
@@ -16,12 +16,24 @@ New Features
requires mandatory username/password authentication
(configured via the new "username" and "password" connection
parameters).
+- The Go serial transport now supports the full set of serial
+ options in the connection string: data-bits, stop-bits, parity,
+ flow-control, dtr, rts, read-timeout and write-timeout.
Incompatible changes
--------------------
- Dropped support for Java 11, new baseline Java version is
Java 21.
+- The Go serial transport's default baud-rate changed from 115200
+ to 9600 (aligning with common serial defaults and the Java
+ transport). Specify baud-rate explicitly if you relied on the
+ previous default.
+- Go serial reads/writes without an explicit context deadline are
+ now bounded by the new read-timeout/write-timeout options
+ (default 1000 ms; set to 0 for the previous blocking behavior),
+ and invalid serial option values now fail connection creation
+ instead of being silently ignored.
- The 'plc4x' proxy driver now defaults to the TLS transport
instead of plaintext TCP. Existing plaintext connections must
switch to an explicit transport prefix (e.g. "plc4x:tcp://...").
diff --git a/plc4go/spi/transports/serial/Transport.go b/plc4go/spi/transports/serial/Transport.go
index 6716b54434..700e4ee04b 100644
--- a/plc4go/spi/transports/serial/Transport.go
+++ b/plc4go/spi/transports/serial/Transport.go
@@ -22,7 +22,6 @@ package serial
import (
"net"
"net/url"
- "strconv"
"github.com/rs/zerolog"
@@ -57,27 +56,12 @@ func (m *Transport) CreateTransportInstance(transportUrl url.URL, options map[st
func (m *Transport) CreateTransportInstanceForLocalAddress(transportUrl url.URL, options map[string][]string, _ *net.UDPAddr, _options ...options.WithOption) (transports.TransportInstance, error) {
var serialPortName = transportUrl.Path
- var baudRate = uint(115200)
- if val, ok := options["baud-rate"]; ok {
- parsedBaudRate, err := strconv.ParseUint(val[0], 10, 32)
- if err != nil {
- return nil, errors.Wrap(err, "error setting connect-timeout")
- } else {
- baudRate = uint(parsedBaudRate)
- }
+ cfg, err := parseSerialOptions(options)
+ if err != nil {
+ return nil, errors.Wrap(err, "error parsing serial options")
}
- var connectTimeout uint32 = 1000
- if val, ok := options["connect-timeout"]; ok {
- parsedConnectTimeout, err := strconv.ParseUint(val[0], 10, 32)
- if err != nil {
- return nil, errors.Wrap(err, "error setting connect-timeout")
- } else {
- connectTimeout = uint32(parsedConnectTimeout)
- }
- }
-
- return NewTransportInstance(serialPortName, baudRate, connectTimeout, m, _options...), nil
+ return NewTransportInstanceWithConfig(serialPortName, cfg, m, _options...), nil
}
func (m *Transport) Close() error {
diff --git a/plc4go/spi/transports/serial/TransportInstance.go b/plc4go/spi/transports/serial/TransportInstance.go
index 772ae3b9f4..562fc43b0a 100644
--- a/plc4go/spi/transports/serial/TransportInstance.go
+++ b/plc4go/spi/transports/serial/TransportInstance.go
@@ -45,12 +45,14 @@ type TransportInstance struct {
SerialPortName string
BaudRate uint
ConnectTimeout uint32
+ cfg serialConfig
connected atomic.Bool
stateChangeMutex sync.Mutex
transport *Transport
serialPort serialport.Port
+ armer *deadlineReader
reader *bufio.Reader
log zerolog.Logger
@@ -59,11 +61,19 @@ type TransportInstance struct {
var _ transports.TransportInstance = (*TransportInstance)(nil)
func NewTransportInstance(serialPortName string, baudRate uint, connectTimeout uint32, transport *Transport, _options ...options.WithOption) *TransportInstance {
+ cfg := defaultSerialConfig()
+ cfg.port.BaudRate = baudRate
+ cfg.connectTimeout = connectTimeout
+ return NewTransportInstanceWithConfig(serialPortName, cfg, transport, _options...)
+}
+
+func NewTransportInstanceWithConfig(serialPortName string, cfg serialConfig, transport *Transport, _options ...options.WithOption) *TransportInstance {
customLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...)
transportInstance := &TransportInstance{
SerialPortName: serialPortName,
- BaudRate: baudRate,
- ConnectTimeout: connectTimeout,
+ BaudRate: cfg.port.BaudRate,
+ ConnectTimeout: cfg.connectTimeout,
+ cfg: cfg,
transport: transport,
log: customLogger,
@@ -79,17 +89,38 @@ func (m *TransportInstance) Connect(ctx context.Context) error {
return errors.New("Already connected")
}
- serialPort, err := serialport.Open(m.SerialPortName, serialport.Config{BaudRate: m.BaudRate})
+ serialPort, err := serialport.Open(m.SerialPortName, m.cfg.port)
if err != nil {
return errors.Wrap(err, "error connecting to serial port")
}
m.serialPort = serialPort
- m.reader = bufio.NewReader(m.serialPort)
+ m.applyModemLines(serialPort)
+ m.armer = newDeadlineReader(serialPort, m.cfg.readTimeout)
+ m.reader = bufio.NewReader(m.armer)
m.connected.Store(true)
-
return nil
}
+// applyModemLines asserts DTR/RTS when configured. Failures are warnings,
+// not connection errors: ptys and adapters without modem lines must stay
+// usable, and a missing line rarely prevents communication.
+func (m *TransportInstance) applyModemLines(port serialport.Port) {
+ controlPort, ok := port.(serialport.ControlPort)
+ if !ok {
+ return
+ }
+ if m.cfg.dtr {
+ if err := controlPort.SetDTR(true); err != nil {
+ m.log.Warn().Err(err).Msg("could not assert DTR")
+ }
+ }
+ if m.cfg.rts {
+ if err := controlPort.SetRTS(true); err != nil {
+ m.log.Warn().Err(err).Msg("could not assert RTS")
+ }
+ }
+}
+
func (m *TransportInstance) Reset() {
// No-Op
}
@@ -107,6 +138,7 @@ func (m *TransportInstance) Close() error {
return errors.Wrap(err, "error closing serial port")
}
m.serialPort = nil
+ m.armer = nil
m.connected.Store(false)
return nil
@@ -127,6 +159,10 @@ func (m *TransportInstance) Write(ctx context.Context, data []byte) error {
if err := m.serialPort.SetWriteDeadline(deadline); err != nil {
return errors.Wrap(err, "error setting write deadline")
}
+ } else if m.cfg.writeTimeout > 0 {
+ if err := m.serialPort.SetWriteDeadline(time.Now().Add(m.cfg.writeTimeout)); err != nil {
+ return errors.Wrap(err, "error setting write deadline")
+ }
} else if err := m.serialPort.SetWriteDeadline(time.Time{}); err != nil {
return errors.Wrap(err, "error clearing write deadline")
}
@@ -145,11 +181,12 @@ func (m *TransportInstance) GetReader() transports.ExtendedReader {
}
func (m *TransportInstance) SetReadDeadline(deadline time.Time) error {
- serialPort := m.serialPort
- if serialPort == nil {
- return errors.New("error setting read deadline. No serial port available")
+ armer := m.armer
+ if armer == nil {
+ return errors.New("error setting read deadline. Not connected")
}
- return serialPort.SetReadDeadline(deadline)
+ armer.setExplicitDeadline(deadline)
+ return nil
}
func (m *TransportInstance) String() string {
diff --git a/plc4go/spi/transports/serial/TransportInstance_pty_test.go b/plc4go/spi/transports/serial/TransportInstance_pty_test.go
index 04b3255e9a..d5a94bae8b 100644
--- a/plc4go/spi/transports/serial/TransportInstance_pty_test.go
+++ b/plc4go/spi/transports/serial/TransportInstance_pty_test.go
@@ -24,6 +24,7 @@ package serial
import (
"context"
"fmt"
+ "net/url"
"os"
"testing"
"time"
@@ -105,3 +106,70 @@ func TestTransportInstance_WriteDeadlineDoesNotStick(t *testing.T) {
// A deadline-less write must not inherit the lapsed deadline.
require.NoError(t, instance.Write(context.Background(), []byte{0x02}))
}
+
+func TestTransportInstance_OptionsAppliedOnPTY(t *testing.T) {
+ master, slavePath := openPTY(t)
+
+ transport := NewTransport()
+ instance, err := transport.CreateTransportInstance(
+ url.URL{Scheme: "serial", Path: slavePath},
+ map[string][]string{
+ "data-bits": {"7"},
+ "parity": {"EVEN"}, // deliberately non-canonical case
+ "stop-bits": {"2"},
+ "dtr": {"true"}, // must warn, not fail, on a pty
+ },
+ )
+ require.NoError(t, err)
+ require.NoError(t, instance.Connect(context.Background()))
+ t.Cleanup(func() { _ = instance.Close() })
+
+ // The pty pair shares one termios; reading it back on the MASTER side
+ // proves the options reached the kernel.
+ readBack, err := unix.IoctlGetTermios(int(master.Fd()), unix.TCGETS)
+ require.NoError(t, err)
+ assert.NotZero(t, readBack.Cflag&unix.CS7, "CS7")
+ assert.NotZero(t, readBack.Cflag&unix.CSTOPB, "CSTOPB")
+ // PARENB deliberately not asserted (known pty parity quirk).
+}
+
+func TestTransportInstance_FallbackReadDeadlineBoundsSilentRead(t *testing.T) {
+ _, slavePath := openPTY(t)
+ transport := NewTransport()
+ instance, err := transport.CreateTransportInstance(
+ url.URL{Scheme: "serial", Path: slavePath},
+ map[string][]string{"read-timeout": {"200"}},
+ )
+ require.NoError(t, err)
+ require.NoError(t, instance.Connect(context.Background()))
+ t.Cleanup(func() { _ = instance.Close() })
+
+ start := time.Now()
+ _, err = instance.Read(context.Background(), 1) // no ctx deadline
+ elapsed := time.Since(start)
+
+ require.Error(t, err, "silent line must time out via the fallback deadline")
+ assert.GreaterOrEqual(t, elapsed, 200*time.Millisecond)
+ assert.Less(t, elapsed, 5*time.Second)
+}
+
+func TestTransportInstance_ExplicitCtxDeadlineBeatsFallback(t *testing.T) {
+ _, slavePath := openPTY(t)
+ transport := NewTransport()
+ instance, err := transport.CreateTransportInstance(
+ url.URL{Scheme: "serial", Path: slavePath},
+ map[string][]string{"read-timeout": {"60000"}},
+ )
+ require.NoError(t, err)
+ require.NoError(t, instance.Connect(context.Background()))
+ t.Cleanup(func() { _ = instance.Close() })
+
+ ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+ defer cancel()
+ start := time.Now()
+ _, err = instance.Read(ctx, 1)
+ elapsed := time.Since(start)
+
+ require.Error(t, err)
+ assert.Less(t, elapsed, 5*time.Second, "the 100ms ctx deadline must win over the 60s fallback")
+}
diff --git a/plc4go/spi/transports/serial/TransportInstance_test.go b/plc4go/spi/transports/serial/TransportInstance_test.go
index 6690958468..33e93cd013 100644
--- a/plc4go/spi/transports/serial/TransportInstance_test.go
+++ b/plc4go/spi/transports/serial/TransportInstance_test.go
@@ -21,10 +21,13 @@ package serial
import (
"bufio"
+ "context"
+ "net/url"
"testing"
"time"
"github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
"github.com/apache/plc4x/plc4go/spi/transports/serial/serialport"
transportUtils "github.com/apache/plc4x/plc4go/spi/transports/utils"
@@ -277,6 +280,17 @@ func (stubPort) Close() error { return nil }
func (stubPort) SetReadDeadline(time.Time) error { return nil }
func (stubPort) SetWriteDeadline(time.Time) error { return nil }
+// recordingWritePort records SetWriteDeadline calls.
+type recordingWritePort struct {
+ stubPort
+ writeDeadlines []time.Time
+}
+
+func (r *recordingWritePort) SetWriteDeadline(t time.Time) error {
+ r.writeDeadlines = append(r.writeDeadlines, t)
+ return nil
+}
+
func TestTransportInstance_IsConnected_keysOnConnectedFlag(t *testing.T) {
m := &TransportInstance{}
m.serialPort = stubPort{}
@@ -284,3 +298,76 @@ func TestTransportInstance_IsConnected_keysOnConnectedFlag(t *testing.T) {
m.connected.Store(true)
assert.True(t, m.IsConnected())
}
+
+func TestParseAndCreate_OptionsReachInstance(t *testing.T) {
+ transport := NewTransport()
+ instance, err := transport.CreateTransportInstance(
+ url.URL{Scheme: "serial", Path: "/dev/ttyTest0"},
+ map[string][]string{
+ "baud-rate": {"19200"},
+ "data-bits": {"7"},
+ "parity": {"even"},
+ "stop-bits": {"2"},
+ "read-timeout": {"250"},
+ },
+ )
+ require.NoError(t, err)
+ serialInstance := instance.(*TransportInstance)
+ assert.Equal(t, "/dev/ttyTest0", serialInstance.SerialPortName)
+ assert.Equal(t, uint(19200), serialInstance.BaudRate, "legacy mirror field")
+ assert.Equal(t, uint(19200), serialInstance.cfg.port.BaudRate)
+ assert.Equal(t, uint(7), serialInstance.cfg.port.DataBits)
+ assert.Equal(t, serialport.ParityEven, serialInstance.cfg.port.Parity)
+ assert.Equal(t, serialport.StopBitsTwo, serialInstance.cfg.port.StopBits)
+ assert.Equal(t, 250*time.Millisecond, serialInstance.cfg.readTimeout)
+}
+
+func TestParseAndCreate_InvalidOptionFailsFast(t *testing.T) {
+ transport := NewTransport()
+ _, err := transport.CreateTransportInstance(
+ url.URL{Scheme: "serial", Path: "/dev/ttyTest0"},
+ map[string][]string{"parity": {"strong"}},
+ )
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), `"parity"`)
+}
+
+func TestWrite_FallbackWriteDeadline(t *testing.T) {
+ port := &recordingWritePort{}
+ m := NewTransportInstanceWithConfig("test", func() serialConfig {
+ c := defaultSerialConfig()
+ c.writeTimeout = 300 * time.Millisecond
+ return c
+ }(), NewTransport())
+ m.serialPort = port
+ m.connected.Store(true)
+
+ // No ctx deadline: fallback must be armed.
+ before := time.Now()
+ require.NoError(t, m.Write(context.Background(), []byte{0x01}))
+ require.Len(t, port.writeDeadlines, 1)
+ assert.False(t, port.writeDeadlines[0].Before(before.Add(300*time.Millisecond)))
+
+ // Explicit ctx deadline wins.
+ deadline := time.Now().Add(42 * time.Second)
+ ctx, cancel := context.WithDeadline(context.Background(), deadline)
+ defer cancel()
+ require.NoError(t, m.Write(ctx, []byte{0x02}))
+ require.Len(t, port.writeDeadlines, 2)
+ assert.Equal(t, deadline, port.writeDeadlines[1])
+}
+
+func TestWrite_ZeroWriteTimeoutClearsDeadline(t *testing.T) {
+ port := &recordingWritePort{}
+ m := NewTransportInstanceWithConfig("test", func() serialConfig {
+ c := defaultSerialConfig()
+ c.writeTimeout = 0
+ return c
+ }(), NewTransport())
+ m.serialPort = port
+ m.connected.Store(true)
+
+ require.NoError(t, m.Write(context.Background(), []byte{0x01}))
+ require.Len(t, port.writeDeadlines, 1)
+ assert.True(t, port.writeDeadlines[0].IsZero())
+}
diff --git a/plc4go/spi/transports/serial/deadline_reader.go b/plc4go/spi/transports/serial/deadline_reader.go
new file mode 100644
index 0000000000..78e6cd5080
--- /dev/null
+++ b/plc4go/spi/transports/serial/deadline_reader.go
@@ -0,0 +1,80 @@
+/*
+ * 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 serial
+
+import (
+ "sync/atomic"
+ "time"
+
+ "github.com/apache/plc4x/plc4go/spi/transports/serial/serialport"
+)
+
+// deadlineReader arms the port's read deadline before every Read. An
+// explicit deadline set via setExplicitDeadline (the context deadline of
+// the current transport operation) always wins; an explicit deadline
+// observed as already expired is honored for exactly one read — so the
+// bounded caller gets its timeout error — and then auto-clears. Without an
+// active explicit deadline the configured fallback applies (0 = blocking),
+// which keeps bare buffered-layer polls bounded instead of hanging forever
+// on a silent line.
+type deadlineReader struct {
+ port serialport.Port
+ fallback time.Duration
+ explicit atomic.Value // time.Time; zero value = none
+}
+
+func newDeadlineReader(port serialport.Port, fallback time.Duration) *deadlineReader {
+ r := &deadlineReader{port: port, fallback: fallback}
+ r.explicit.Store(time.Time{})
+ return r
+}
+
+// setExplicitDeadline sets (or, with a zero time, clears) the explicit
+// read deadline. Safe for concurrent use with Read.
+func (r *deadlineReader) setExplicitDeadline(t time.Time) {
+ r.explicit.Store(t)
+}
+
+func (r *deadlineReader) Read(p []byte) (int, error) {
+ deadline, _ := r.explicit.Load().(time.Time)
+ switch {
+ case !deadline.IsZero():
+ if err := r.port.SetReadDeadline(deadline); err != nil {
+ return 0, err
+ }
+ if !time.Now().Before(deadline) {
+ // Expired: honored for this read, cleared for the next. CAS so
+ // a concurrent setExplicitDeadline is never overwritten. The old
+ // value MUST be the exact Load()ed time.Time (never reconstructed
+ // or derived): CompareAndSwap uses interface ==, which includes
+ // time.Time's monotonic-clock reading.
+ r.explicit.CompareAndSwap(deadline, time.Time{})
+ }
+ case r.fallback > 0:
+ if err := r.port.SetReadDeadline(time.Now().Add(r.fallback)); err != nil {
+ return 0, err
+ }
+ default:
+ if err := r.port.SetReadDeadline(time.Time{}); err != nil {
+ return 0, err
+ }
+ }
+ return r.port.Read(p)
+}
diff --git a/plc4go/spi/transports/serial/deadline_reader_test.go b/plc4go/spi/transports/serial/deadline_reader_test.go
new file mode 100644
index 0000000000..cb5aa3e247
--- /dev/null
+++ b/plc4go/spi/transports/serial/deadline_reader_test.go
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package serial
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+)
+
+// recordingPort implements serialport.Port and records SetReadDeadline calls.
+type recordingPort struct {
+ stubPort // from TransportInstance_test.go: no-op serialport.Port
+ readDeadlines []time.Time
+}
+
+func (r *recordingPort) SetReadDeadline(t time.Time) error {
+ r.readDeadlines = append(r.readDeadlines, t)
+ return nil
+}
+
+func TestDeadlineReader_FallbackArmsRelativeDeadline(t *testing.T) {
+ port := &recordingPort{}
+ reader := newDeadlineReader(port, 200*time.Millisecond)
+
+ before := time.Now()
+ _, err := reader.Read(make([]byte, 1))
+ require.NoError(t, err)
+
+ require.Len(t, port.readDeadlines, 1)
+ armed := port.readDeadlines[0]
+ assert.False(t, armed.Before(before.Add(200*time.Millisecond)), "deadline must be >= now+fallback")
+ assert.True(t, armed.Before(before.Add(2*time.Second)), "deadline must be near now+fallback")
+}
+
+func TestDeadlineReader_ZeroFallbackClearsDeadline(t *testing.T) {
+ port := &recordingPort{}
+ reader := newDeadlineReader(port, 0)
+
+ _, err := reader.Read(make([]byte, 1))
+ require.NoError(t, err)
+
+ require.Len(t, port.readDeadlines, 1)
+ assert.True(t, port.readDeadlines[0].IsZero(), "no fallback means the deadline is cleared")
+}
+
+func TestDeadlineReader_ExplicitDeadlineWins(t *testing.T) {
+ port := &recordingPort{}
+ reader := newDeadlineReader(port, time.Hour)
+
+ explicit := time.Now().Add(30 * time.Millisecond)
+ reader.setExplicitDeadline(explicit)
+ _, err := reader.Read(make([]byte, 1))
+ require.NoError(t, err)
+
+ require.Len(t, port.readDeadlines, 1)
+ assert.Equal(t, explicit, port.readDeadlines[0])
+}
+
+func TestDeadlineReader_ExpiredExplicitHonoredOnceThenFallback(t *testing.T) {
+ port := &recordingPort{}
+ reader := newDeadlineReader(port, time.Hour)
+
+ expired := time.Now().Add(-time.Second)
+ reader.setExplicitDeadline(expired)
+
+ // First read still sees the expired explicit deadline (the ctx-bounded
+ // caller must get its timeout), ...
+ _, _ = reader.Read(make([]byte, 1))
+ require.Len(t, port.readDeadlines, 1)
+ assert.Equal(t, expired, port.readDeadlines[0])
+
+ // ... the next read falls back.
+ _, _ = reader.Read(make([]byte, 1))
+ require.Len(t, port.readDeadlines, 2)
+ assert.False(t, port.readDeadlines[1].Equal(expired), "expired explicit deadline must auto-clear")
+ assert.False(t, port.readDeadlines[1].IsZero(), "fallback must be armed")
+}
+
+func TestDeadlineReader_ClearingExplicitRestoresFallback(t *testing.T) {
+ port := &recordingPort{}
+ reader := newDeadlineReader(port, time.Minute)
+
+ reader.setExplicitDeadline(time.Now().Add(time.Hour))
+ _, _ = reader.Read(make([]byte, 1))
+ reader.setExplicitDeadline(time.Time{})
+ _, _ = reader.Read(make([]byte, 1))
+
+ require.Len(t, port.readDeadlines, 2)
+ assert.False(t, port.readDeadlines[1].Equal(port.readDeadlines[0]), "cleared explicit must not persist")
+}
diff --git a/plc4go/spi/transports/serial/options.go b/plc4go/spi/transports/serial/options.go
new file mode 100644
index 0000000000..82bd9f34d4
--- /dev/null
+++ b/plc4go/spi/transports/serial/options.go
@@ -0,0 +1,167 @@
+/*
+ * 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 serial
+
+import (
+ "fmt"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/apache/plc4x/plc4go/spi/transports/serial/serialport"
+)
+
+// serialConfig bundles everything the serial transport configures on a
+// connection: the port settings handed to serialport.Open, the modem lines
+// asserted after open, and the transport-level timeouts.
+type serialConfig struct {
+ port serialport.Config
+ dtr bool
+ rts bool
+ // readTimeout bounds reads that carry no explicit context deadline;
+ // 0 means blocking reads.
+ readTimeout time.Duration
+ // writeTimeout bounds writes that carry no explicit context deadline;
+ // 0 means no write timeout.
+ writeTimeout time.Duration
+ connectTimeout uint32 // milliseconds
+}
+
+func defaultSerialConfig() serialConfig {
+ return serialConfig{
+ port: serialport.Config{BaudRate: 9600}, // 8N1 via serialport defaults
+ readTimeout: time.Second,
+ writeTimeout: time.Second,
+ connectTimeout: 1000,
+ }
+}
+
+// parseSerialOptions translates the connection-string options map into a
+// serialConfig. Enum values are case-insensitive and accept "-" or "_" as
+// separator; invalid values are hard errors naming the option; unknown
+// options and empty value slices are ignored.
+func parseSerialOptions(options map[string][]string) (serialConfig, error) {
+ cfg := defaultSerialConfig()
+
+ if raw, ok := firstValue(options, "baud-rate"); ok {
+ value, err := strconv.ParseUint(raw, 10, 32)
+ if err != nil || value == 0 {
+ return cfg, optionError("baud-rate", raw, "must be a positive integer")
+ }
+ cfg.port.BaudRate = uint(value)
+ }
+ if raw, ok := firstValue(options, "data-bits"); ok {
+ value, err := strconv.ParseUint(raw, 10, 8)
+ if err != nil || value < 5 || value > 8 {
+ return cfg, optionError("data-bits", raw, "must be 5..8")
+ }
+ cfg.port.DataBits = uint(value)
+ }
+ if raw, ok := firstValue(options, "stop-bits"); ok {
+ switch raw {
+ case "1":
+ cfg.port.StopBits = serialport.StopBitsOne
+ case "2":
+ cfg.port.StopBits = serialport.StopBitsTwo
+ default:
+ return cfg, optionError("stop-bits", raw, "must be 1 or 2")
+ }
+ }
+ if raw, ok := firstValue(options, "parity"); ok {
+ switch normalizeEnum(raw) {
+ case "none":
+ cfg.port.Parity = serialport.ParityNone
+ case "odd":
+ cfg.port.Parity = serialport.ParityOdd
+ case "even":
+ cfg.port.Parity = serialport.ParityEven
+ case "mark":
+ cfg.port.Parity = serialport.ParityMark
+ case "space":
+ cfg.port.Parity = serialport.ParitySpace
+ default:
+ return cfg, optionError("parity", raw, "must be one of none, odd, even, mark, space")
+ }
+ }
+ if raw, ok := firstValue(options, "flow-control"); ok {
+ switch normalizeEnum(raw) {
+ case "none":
+ // defaults already off
+ case "rts-cts":
+ cfg.port.RTSCTSFlowControl = true
+ case "xon-xoff":
+ cfg.port.XONXOFFFlowControl = true
+ default:
+ return cfg, optionError("flow-control", raw, "must be one of none, rts-cts, xon-xoff")
+ }
+ }
+ if raw, ok := firstValue(options, "dtr"); ok {
+ value, err := strconv.ParseBool(raw)
+ if err != nil {
+ return cfg, optionError("dtr", raw, "must be true or false")
+ }
+ cfg.dtr = value
+ }
+ if raw, ok := firstValue(options, "rts"); ok {
+ value, err := strconv.ParseBool(raw)
+ if err != nil {
+ return cfg, optionError("rts", raw, "must be true or false")
+ }
+ cfg.rts = value
+ }
+ if raw, ok := firstValue(options, "read-timeout"); ok {
+ millis, err := strconv.ParseUint(raw, 10, 32)
+ if err != nil {
+ return cfg, optionError("read-timeout", raw, "must be a non-negative integer (milliseconds)")
+ }
+ cfg.readTimeout = time.Duration(millis) * time.Millisecond
+ }
+ if raw, ok := firstValue(options, "write-timeout"); ok {
+ millis, err := strconv.ParseUint(raw, 10, 32)
+ if err != nil {
+ return cfg, optionError("write-timeout", raw, "must be a non-negative integer (milliseconds)")
+ }
+ cfg.writeTimeout = time.Duration(millis) * time.Millisecond
+ }
+ if raw, ok := firstValue(options, "connect-timeout"); ok {
+ millis, err := strconv.ParseUint(raw, 10, 32)
+ if err != nil {
+ return cfg, optionError("connect-timeout", raw, "must be a non-negative integer (milliseconds)")
+ }
+ cfg.connectTimeout = uint32(millis)
+ }
+ return cfg, nil
+}
+
+func firstValue(options map[string][]string, key string) (string, bool) {
+ values, ok := options[key]
+ if !ok || len(values) == 0 {
+ return "", false
+ }
+ return values[0], true
+}
+
+func normalizeEnum(value string) string {
+ return strings.ReplaceAll(strings.ToLower(value), "_", "-")
+}
+
+func optionError(option, value, requirement string) error {
+ return fmt.Errorf("error parsing option %q: invalid value %q (%s)", option, value, requirement)
+}
diff --git a/plc4go/spi/transports/serial/options_test.go b/plc4go/spi/transports/serial/options_test.go
new file mode 100644
index 0000000000..5c22ad6253
--- /dev/null
+++ b/plc4go/spi/transports/serial/options_test.go
@@ -0,0 +1,160 @@
+/*
+ * 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 serial
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/plc4x/plc4go/spi/transports/serial/serialport"
+)
+
+func TestParseSerialOptions(t *testing.T) {
+ tests := []struct {
+ name string
+ options map[string][]string
+ want serialConfig
+ wantErr string // substring; empty = expect success
+ }{
+ {
+ name: "empty map yields all defaults",
+ options: map[string][]string{},
+ want: defaultSerialConfig(),
+ },
+ {
+ name: "full happy path",
+ options: map[string][]string{
+ "baud-rate": {"19200"},
+ "data-bits": {"7"},
+ "stop-bits": {"2"},
+ "parity": {"even"},
+ "flow-control": {"rts-cts"},
+ "dtr": {"true"},
+ "rts": {"true"},
+ "read-timeout": {"250"},
+ "write-timeout": {"0"},
+ "connect-timeout": {"5000"},
+ },
+ want: serialConfig{
+ port: serialport.Config{
+ BaudRate: 19200, DataBits: 7,
+ StopBits: serialport.StopBitsTwo, Parity: serialport.ParityEven,
+ RTSCTSFlowControl: true,
+ },
+ dtr: true, rts: true,
+ readTimeout: 250 * time.Millisecond, writeTimeout: 0,
+ connectTimeout: 5000,
+ },
+ },
+ {
+ name: "enum values are case-insensitive and accept underscores",
+ options: map[string][]string{"parity": {"EVEN"}, "flow-control": {"XON_XOFF"}},
+ want: func() serialConfig {
+ c := defaultSerialConfig()
+ c.port.Parity = serialport.ParityEven
+ c.port.XONXOFFFlowControl = true
+ return c
+ }(),
+ },
+ {
+ name: "mark and space parity",
+ options: map[string][]string{"parity": {"Mark"}},
+ want: func() serialConfig {
+ c := defaultSerialConfig()
+ c.port.Parity = serialport.ParityMark
+ return c
+ }(),
+ },
+ {
+ name: "read-timeout zero means blocking",
+ options: map[string][]string{"read-timeout": {"0"}},
+ want: func() serialConfig {
+ c := defaultSerialConfig()
+ c.readTimeout = 0
+ return c
+ }(),
+ },
+ {
+ name: "invalid baud rate",
+ options: map[string][]string{"baud-rate": {"fast"}},
+ wantErr: `"baud-rate"`,
+ },
+ {
+ name: "zero baud rate rejected",
+ options: map[string][]string{"baud-rate": {"0"}},
+ wantErr: `"baud-rate"`,
+ },
+ {
+ name: "data-bits out of range",
+ options: map[string][]string{"data-bits": {"9"}},
+ wantErr: `"data-bits"`,
+ },
+ {
+ name: "stop-bits out of range",
+ options: map[string][]string{"stop-bits": {"3"}},
+ wantErr: `"stop-bits"`,
+ },
+ {
+ name: "unknown parity value",
+ options: map[string][]string{"parity": {"strong"}},
+ wantErr: `"parity"`,
+ },
+ {
+ name: "combined flow control value rejected",
+ options: map[string][]string{"flow-control": {"rts-cts-xon-xoff"}},
+ wantErr: `"flow-control"`,
+ },
+ {
+ name: "invalid dtr boolean",
+ options: map[string][]string{"dtr": {"yes-please"}},
+ wantErr: `"dtr"`,
+ },
+ {
+ name: "invalid read-timeout",
+ options: map[string][]string{"read-timeout": {"-5"}},
+ wantErr: `"read-timeout"`,
+ },
+ {
+ name: "empty value slice ignored like absent option",
+ options: map[string][]string{"parity": {}},
+ want: defaultSerialConfig(),
+ },
+ {
+ name: "unknown options are ignored",
+ options: map[string][]string{"break-enabled": {"true"}, "no-such-thing": {"1"}},
+ want: defaultSerialConfig(),
+ },
+ }
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ got, err := parseSerialOptions(tt.options)
+ if tt.wantErr != "" {
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), tt.wantErr)
+ return
+ }
+ require.NoError(t, err)
+ assert.Equal(t, tt.want, got)
+ })
+ }
+}