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 b5259d7916f6bec449613ea18225f876d2ed0f62 Author: Sebastian Rühl <[email protected]> AuthorDate: Mon Jul 6 16:24:59 2026 +0200 feat(plc4go): shared serial ports and inter-frame write pacing New serial options: "reuse-port" shares one physical port between multiple connections (multi-slave Modbus RTU) — broadcast reads into per-subscriber bounded drop-oldest buffers, serialized writes, a refcounted per-transport registry with hard errors on config mismatch — and "interframe-delay" enforces minimum write silence measured from the later of last-write-end and last-received-byte, on shared and dedicated ports alike. Subscriptions implement the serialport.Port interface, so the deadline/buffering/codec layers are unchanged. --- RELEASE_NOTES | 4 + plc4go/spi/transports/serial/Transport.go | 5 +- plc4go/spi/transports/serial/TransportInstance.go | 63 ++-- .../serial/TransportInstance_pty_test.go | 71 +++++ .../transports/serial/TransportInstance_test.go | 17 ++ plc4go/spi/transports/serial/Transport_test.go | 3 +- plc4go/spi/transports/serial/deadline_reader.go | 11 +- .../spi/transports/serial/deadline_reader_test.go | 10 +- plc4go/spi/transports/serial/options.go | 21 ++ plc4go/spi/transports/serial/options_test.go | 28 ++ plc4go/spi/transports/serial/pacer.go | 77 +++++ plc4go/spi/transports/serial/pacer_test.go | 80 +++++ plc4go/spi/transports/serial/ring_buffer.go | 173 +++++++++++ plc4go/spi/transports/serial/ring_buffer_test.go | 122 ++++++++ plc4go/spi/transports/serial/shared_port.go | 337 +++++++++++++++++++++ .../spi/transports/serial/shared_port_pty_test.go | 175 +++++++++++ 16 files changed, 1168 insertions(+), 29 deletions(-) diff --git a/RELEASE_NOTES b/RELEASE_NOTES index dffd8b8e0e..9d76d24faf 100644 --- a/RELEASE_NOTES +++ b/RELEASE_NOTES @@ -19,6 +19,10 @@ New Features - 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. +- The Go serial transport supports sharing one physical serial port + between multiple connections ("reuse-port", e.g. multi-slave Modbus + RTU) with broadcast reads and serialized writes, plus inter-frame + write pacing ("interframe-delay") on both shared and dedicated ports. Incompatible changes -------------------- diff --git a/plc4go/spi/transports/serial/Transport.go b/plc4go/spi/transports/serial/Transport.go index 700e4ee04b..7cc828b1c3 100644 --- a/plc4go/spi/transports/serial/Transport.go +++ b/plc4go/spi/transports/serial/Transport.go @@ -31,13 +31,16 @@ import ( ) type Transport struct { + registry *sharedPortRegistry + log zerolog.Logger } func NewTransport(_options ...options.WithOption) *Transport { customLogger := options.ExtractCustomLoggerOrDefaultToGlobal(_options...) return &Transport{ - log: customLogger, + registry: newSharedPortRegistry(customLogger), + log: customLogger, } } diff --git a/plc4go/spi/transports/serial/TransportInstance.go b/plc4go/spi/transports/serial/TransportInstance.go index 562fc43b0a..f99fc91284 100644 --- a/plc4go/spi/transports/serial/TransportInstance.go +++ b/plc4go/spi/transports/serial/TransportInstance.go @@ -53,6 +53,7 @@ type TransportInstance struct { transport *Transport serialPort serialport.Port armer *deadlineReader + pacer *pacer reader *bufio.Reader log zerolog.Logger @@ -89,13 +90,35 @@ func (m *TransportInstance) Connect(ctx context.Context) error { return errors.New("Already connected") } - serialPort, err := serialport.Open(m.SerialPortName, m.cfg.port) - if err != nil { - return errors.Wrap(err, "error connecting to serial port") + var serialPort serialport.Port + var err error + if m.cfg.reusePort { + if m.transport == nil { + return errors.New("reuse-port requires a transport-managed instance") + } + serialPort, err = m.transport.registry.acquire(m.SerialPortName, sharedPortConfig{ + port: m.cfg.port, + dtr: m.cfg.dtr, + rts: m.cfg.rts, + interframeDelay: m.cfg.interframeDelay, + }) + if err != nil { + return errors.Wrap(err, "error acquiring shared serial port") + } + m.serialPort = serialPort + // dtr/rts are applied by the registry at open; pacing lives inside + // the shared port's write path. + m.pacer = nil + } else { + 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.applyModemLines(serialPort) + m.pacer = newPacer(m.cfg.interframeDelay) } - m.serialPort = serialPort - m.applyModemLines(serialPort) - m.armer = newDeadlineReader(serialPort, m.cfg.readTimeout) + m.armer = newDeadlineReader(m.serialPort, m.cfg.readTimeout, m.pacer) m.reader = bufio.NewReader(m.armer) m.connected.Store(true) return nil @@ -134,13 +157,13 @@ func (m *TransportInstance) Close() error { return nil } err := m.serialPort.Close() - if err != nil { - return errors.Wrap(err, "error closing serial port") - } m.serialPort = nil m.armer = nil - + m.pacer = nil m.connected.Store(false) + if err != nil { + return errors.Wrap(err, "error closing serial port") + } return nil } @@ -155,18 +178,20 @@ func (m *TransportInstance) Write(ctx context.Context, data []byte) error { if m.serialPort == nil { return errors.New("error writing to transport. No writer available") } - if deadline, ok := ctx.Deadline(); ok { - if err := m.serialPort.SetWriteDeadline(deadline); err != nil { - return errors.Wrap(err, "error setting write deadline") - } + var deadline time.Time + if ctxDeadline, ok := ctx.Deadline(); ok { + deadline = ctxDeadline } 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") + deadline = time.Now().Add(m.cfg.writeTimeout) + } + if err := m.pacer.waitTurn(deadline); err != nil { + return errors.Wrap(err, "error pacing write") + } + if err := m.serialPort.SetWriteDeadline(deadline); err != nil { + return errors.Wrap(err, "error setting write deadline") } num, err := m.serialPort.Write(data) + m.pacer.noteActivity() if err != nil { return errors.Wrap(err, "error writing") } diff --git a/plc4go/spi/transports/serial/TransportInstance_pty_test.go b/plc4go/spi/transports/serial/TransportInstance_pty_test.go index d5a94bae8b..a601d13f14 100644 --- a/plc4go/spi/transports/serial/TransportInstance_pty_test.go +++ b/plc4go/spi/transports/serial/TransportInstance_pty_test.go @@ -32,6 +32,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "golang.org/x/sys/unix" + + "github.com/apache/plc4x/plc4go/spi/transports" ) // openPTY allocates a pseudo-terminal pair; the slave path stands in for a @@ -173,3 +175,72 @@ func TestTransportInstance_ExplicitCtxDeadlineBeatsFallback(t *testing.T) { require.Error(t, err) assert.Less(t, elapsed, 5*time.Second, "the 100ms ctx deadline must win over the 60s fallback") } + +func TestTransportInstance_ReusePortSharesOnePTY(t *testing.T) { + master, slavePath := openPTY(t) + transport := NewTransport() + options := map[string][]string{"reuse-port": {"true"}} + + first, err := transport.CreateTransportInstance(url.URL{Scheme: "serial", Path: slavePath}, options) + require.NoError(t, err) + second, err := transport.CreateTransportInstance(url.URL{Scheme: "serial", Path: slavePath}, options) + require.NoError(t, err) + + require.NoError(t, first.Connect(context.Background())) + t.Cleanup(func() { _ = first.Close() }) + require.NoError(t, second.Connect(context.Background())) + t.Cleanup(func() { _ = second.Close() }) + + _, err = master.Write([]byte{0xCA, 0xFE}) + require.NoError(t, err) + + for _, instance := range []transports.TransportInstance{first, second} { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + data, err := instance.Read(ctx, 2) + cancel() + require.NoError(t, err) + assert.Equal(t, []byte{0xCA, 0xFE}, data, "both connections see the broadcast") + } +} + +func TestTransportInstance_ReusePortConfigMismatch(t *testing.T) { + _, slavePath := openPTY(t) + transport := NewTransport() + + first, err := transport.CreateTransportInstance(url.URL{Scheme: "serial", Path: slavePath}, + map[string][]string{"reuse-port": {"true"}, "baud-rate": {"9600"}}) + require.NoError(t, err) + require.NoError(t, first.Connect(context.Background())) + t.Cleanup(func() { _ = first.Close() }) + + second, err := transport.CreateTransportInstance(url.URL{Scheme: "serial", Path: slavePath}, + map[string][]string{"reuse-port": {"true"}, "baud-rate": {"19200"}}) + require.NoError(t, err) + err = second.Connect(context.Background()) + require.Error(t, err) + assert.Contains(t, err.Error(), slavePath) +} + +func TestTransportInstance_DedicatedInterframeDelayOnPTY(t *testing.T) { + master, slavePath := openPTY(t) + transport := NewTransport() + instance, err := transport.CreateTransportInstance(url.URL{Scheme: "serial", Path: slavePath}, + map[string][]string{"interframe-delay": {"60"}}) + require.NoError(t, err) + require.NoError(t, instance.Connect(context.Background())) + t.Cleanup(func() { _ = instance.Close() }) + + require.NoError(t, instance.Write(context.Background(), []byte{0x01})) + start := time.Now() + require.NoError(t, instance.Write(context.Background(), []byte{0x02})) + assert.GreaterOrEqual(t, time.Since(start), 50*time.Millisecond) + + buf := make([]byte, 2) + total := 0 + for total < 2 { + n, err := master.Read(buf[total:]) + require.NoError(t, err) + total += n + } + assert.Equal(t, []byte{0x01, 0x02}, buf) +} diff --git a/plc4go/spi/transports/serial/TransportInstance_test.go b/plc4go/spi/transports/serial/TransportInstance_test.go index 33e93cd013..808da7105e 100644 --- a/plc4go/spi/transports/serial/TransportInstance_test.go +++ b/plc4go/spi/transports/serial/TransportInstance_test.go @@ -371,3 +371,20 @@ func TestWrite_ZeroWriteTimeoutClearsDeadline(t *testing.T) { require.Len(t, port.writeDeadlines, 1) assert.True(t, port.writeDeadlines[0].IsZero()) } + +func TestWrite_DedicatedInterframePacing(t *testing.T) { + port := &recordingWritePort{} + m := NewTransportInstanceWithConfig("test", func() serialConfig { + c := defaultSerialConfig() + c.interframeDelay = 50 * time.Millisecond + return c + }(), NewTransport()) + m.serialPort = port + m.pacer = newPacer(m.cfg.interframeDelay) + m.connected.Store(true) + + require.NoError(t, m.Write(context.Background(), []byte{0x01})) + start := time.Now() + require.NoError(t, m.Write(context.Background(), []byte{0x02})) + assert.GreaterOrEqual(t, time.Since(start), 40*time.Millisecond, "second write must wait out the gap") +} diff --git a/plc4go/spi/transports/serial/Transport_test.go b/plc4go/spi/transports/serial/Transport_test.go index f9af03e011..e996e28579 100644 --- a/plc4go/spi/transports/serial/Transport_test.go +++ b/plc4go/spi/transports/serial/Transport_test.go @@ -38,7 +38,8 @@ func TestNewTransport(t *testing.T) { { name: "create it", want: &Transport{ - log: log.Logger, + registry: newSharedPortRegistry(log.Logger), + log: log.Logger, }, }, } diff --git a/plc4go/spi/transports/serial/deadline_reader.go b/plc4go/spi/transports/serial/deadline_reader.go index 78e6cd5080..97b3c6f3e4 100644 --- a/plc4go/spi/transports/serial/deadline_reader.go +++ b/plc4go/spi/transports/serial/deadline_reader.go @@ -37,11 +37,12 @@ import ( type deadlineReader struct { port serialport.Port fallback time.Duration + activity *pacer // notified on successful reads; may be nil explicit atomic.Value // time.Time; zero value = none } -func newDeadlineReader(port serialport.Port, fallback time.Duration) *deadlineReader { - r := &deadlineReader{port: port, fallback: fallback} +func newDeadlineReader(port serialport.Port, fallback time.Duration, activity *pacer) *deadlineReader { + r := &deadlineReader{port: port, fallback: fallback, activity: activity} r.explicit.Store(time.Time{}) return r } @@ -76,5 +77,9 @@ func (r *deadlineReader) Read(p []byte) (int, error) { return 0, err } } - return r.port.Read(p) + n, err := r.port.Read(p) + if n > 0 { + r.activity.noteActivity() + } + return n, err } diff --git a/plc4go/spi/transports/serial/deadline_reader_test.go b/plc4go/spi/transports/serial/deadline_reader_test.go index cb5aa3e247..92872b2ee5 100644 --- a/plc4go/spi/transports/serial/deadline_reader_test.go +++ b/plc4go/spi/transports/serial/deadline_reader_test.go @@ -40,7 +40,7 @@ func (r *recordingPort) SetReadDeadline(t time.Time) error { func TestDeadlineReader_FallbackArmsRelativeDeadline(t *testing.T) { port := &recordingPort{} - reader := newDeadlineReader(port, 200*time.Millisecond) + reader := newDeadlineReader(port, 200*time.Millisecond, nil) before := time.Now() _, err := reader.Read(make([]byte, 1)) @@ -54,7 +54,7 @@ func TestDeadlineReader_FallbackArmsRelativeDeadline(t *testing.T) { func TestDeadlineReader_ZeroFallbackClearsDeadline(t *testing.T) { port := &recordingPort{} - reader := newDeadlineReader(port, 0) + reader := newDeadlineReader(port, 0, nil) _, err := reader.Read(make([]byte, 1)) require.NoError(t, err) @@ -65,7 +65,7 @@ func TestDeadlineReader_ZeroFallbackClearsDeadline(t *testing.T) { func TestDeadlineReader_ExplicitDeadlineWins(t *testing.T) { port := &recordingPort{} - reader := newDeadlineReader(port, time.Hour) + reader := newDeadlineReader(port, time.Hour, nil) explicit := time.Now().Add(30 * time.Millisecond) reader.setExplicitDeadline(explicit) @@ -78,7 +78,7 @@ func TestDeadlineReader_ExplicitDeadlineWins(t *testing.T) { func TestDeadlineReader_ExpiredExplicitHonoredOnceThenFallback(t *testing.T) { port := &recordingPort{} - reader := newDeadlineReader(port, time.Hour) + reader := newDeadlineReader(port, time.Hour, nil) expired := time.Now().Add(-time.Second) reader.setExplicitDeadline(expired) @@ -98,7 +98,7 @@ func TestDeadlineReader_ExpiredExplicitHonoredOnceThenFallback(t *testing.T) { func TestDeadlineReader_ClearingExplicitRestoresFallback(t *testing.T) { port := &recordingPort{} - reader := newDeadlineReader(port, time.Minute) + reader := newDeadlineReader(port, time.Minute, nil) reader.setExplicitDeadline(time.Now().Add(time.Hour)) _, _ = reader.Read(make([]byte, 1)) diff --git a/plc4go/spi/transports/serial/options.go b/plc4go/spi/transports/serial/options.go index 82bd9f34d4..dc44ceaf54 100644 --- a/plc4go/spi/transports/serial/options.go +++ b/plc4go/spi/transports/serial/options.go @@ -42,6 +42,13 @@ type serialConfig struct { // 0 means no write timeout. writeTimeout time.Duration connectTimeout uint32 // milliseconds + // reusePort shares one physical port between transport instances with + // identical configuration (broadcast reads, serialized paced writes). + reusePort bool + // interframeDelay is the minimum silence enforced before each write, + // measured from the later of last-write-end and last-received-byte. + // 0 disables pacing. Works on shared and dedicated ports alike. + interframeDelay time.Duration } func defaultSerialConfig() serialConfig { @@ -147,6 +154,20 @@ func parseSerialOptions(options map[string][]string) (serialConfig, error) { } cfg.connectTimeout = uint32(millis) } + if raw, ok := firstValue(options, "reuse-port"); ok { + value, err := strconv.ParseBool(raw) + if err != nil { + return cfg, optionError("reuse-port", raw, "must be true or false") + } + cfg.reusePort = value + } + if raw, ok := firstValue(options, "interframe-delay"); ok { + millis, err := strconv.ParseUint(raw, 10, 32) + if err != nil { + return cfg, optionError("interframe-delay", raw, "must be a non-negative integer (milliseconds)") + } + cfg.interframeDelay = time.Duration(millis) * time.Millisecond + } return cfg, nil } diff --git a/plc4go/spi/transports/serial/options_test.go b/plc4go/spi/transports/serial/options_test.go index 5c22ad6253..b0a7afb606 100644 --- a/plc4go/spi/transports/serial/options_test.go +++ b/plc4go/spi/transports/serial/options_test.go @@ -144,6 +144,34 @@ func TestParseSerialOptions(t *testing.T) { options: map[string][]string{"break-enabled": {"true"}, "no-such-thing": {"1"}}, want: defaultSerialConfig(), }, + { + name: "reuse-port accepted", + options: map[string][]string{"reuse-port": {"true"}}, + want: func() serialConfig { + c := defaultSerialConfig() + c.reusePort = true + return c + }(), + }, + { + name: "interframe-delay accepted", + options: map[string][]string{"interframe-delay": {"50"}}, + want: func() serialConfig { + c := defaultSerialConfig() + c.interframeDelay = 50 * time.Millisecond + return c + }(), + }, + { + name: "invalid reuse-port", + options: map[string][]string{"reuse-port": {"maybe"}}, + wantErr: `"reuse-port"`, + }, + { + name: "invalid interframe-delay", + options: map[string][]string{"interframe-delay": {"-1"}}, + wantErr: `"interframe-delay"`, + }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/plc4go/spi/transports/serial/pacer.go b/plc4go/spi/transports/serial/pacer.go new file mode 100644 index 0000000000..cf28671527 --- /dev/null +++ b/plc4go/spi/transports/serial/pacer.go @@ -0,0 +1,77 @@ +/* + * 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 ( + "os" + "sync" + "time" +) + +// pacer enforces a minimum silence gap before writes on a serial line, +// measured from the later of the last write end and the last observed +// read activity (RTU t3.5-style semantics). A nil pacer or a zero delay +// disables pacing entirely. A pacer only tracks timestamps — it does not +// serialize concurrent writers; callers must hold their own write lock +// around waitTurn+write for the gap guarantee to hold. +type pacer struct { + delay time.Duration + mu sync.Mutex + last time.Time +} + +func newPacer(delay time.Duration) *pacer { + return &pacer{delay: delay} +} + +// noteActivity records line activity (a completed write or received data). +func (p *pacer) noteActivity() { + if p == nil || p.delay <= 0 { + return + } + p.mu.Lock() + if now := time.Now(); now.After(p.last) { + p.last = now + } + p.mu.Unlock() +} + +// waitTurn sleeps until the configured gap since the last activity has +// elapsed. A non-zero deadline bounds the wait: if it would expire before +// the gap does, waitTurn returns os.ErrDeadlineExceeded without sleeping +// the gap out. +func (p *pacer) waitTurn(deadline time.Time) error { + if p == nil || p.delay <= 0 { + return nil + } + for { + p.mu.Lock() + ready := p.last.Add(p.delay) + p.mu.Unlock() + wait := time.Until(ready) + if wait <= 0 { + return nil + } + if !deadline.IsZero() && ready.After(deadline) { + return os.ErrDeadlineExceeded + } + time.Sleep(wait) + } +} diff --git a/plc4go/spi/transports/serial/pacer_test.go b/plc4go/spi/transports/serial/pacer_test.go new file mode 100644 index 0000000000..acae8f5312 --- /dev/null +++ b/plc4go/spi/transports/serial/pacer_test.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 ( + "os" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestPacer_NilAndZeroDelayAreNoops(t *testing.T) { + var nilPacer *pacer + nilPacer.noteActivity() + require.NoError(t, nilPacer.waitTurn(time.Time{})) + + zero := newPacer(0) + zero.noteActivity() + start := time.Now() + require.NoError(t, zero.waitTurn(time.Time{})) + assert.Less(t, time.Since(start), 20*time.Millisecond) +} + +func TestPacer_EnforcesGapAfterActivity(t *testing.T) { + p := newPacer(60 * time.Millisecond) + p.noteActivity() + start := time.Now() + require.NoError(t, p.waitTurn(time.Time{})) + assert.GreaterOrEqual(t, time.Since(start), 50*time.Millisecond, "must wait out the gap") +} + +func TestPacer_NoWaitWhenGapAlreadyElapsed(t *testing.T) { + p := newPacer(30 * time.Millisecond) + p.noteActivity() + time.Sleep(40 * time.Millisecond) + start := time.Now() + require.NoError(t, p.waitTurn(time.Time{})) + assert.Less(t, time.Since(start), 20*time.Millisecond) +} + +func TestPacer_DeadlineShorterThanGapFailsFast(t *testing.T) { + p := newPacer(500 * time.Millisecond) + p.noteActivity() + start := time.Now() + err := p.waitTurn(time.Now().Add(20 * time.Millisecond)) + require.ErrorIs(t, err, os.ErrDeadlineExceeded) + assert.Less(t, time.Since(start), 200*time.Millisecond, "must not sleep out the full gap") +} + +func TestPacer_ActivityDuringWaitExtendsGap(t *testing.T) { + p := newPacer(80 * time.Millisecond) + p.noteActivity() + go func() { + time.Sleep(40 * time.Millisecond) + p.noteActivity() // traffic arrives while a writer is waiting its turn + }() + start := time.Now() + require.NoError(t, p.waitTurn(time.Time{})) + assert.GreaterOrEqual(t, time.Since(start), 110*time.Millisecond, + "the gap must restart from the mid-wait activity (40ms + 80ms)") +} diff --git a/plc4go/spi/transports/serial/ring_buffer.go b/plc4go/spi/transports/serial/ring_buffer.go new file mode 100644 index 0000000000..40400bc555 --- /dev/null +++ b/plc4go/spi/transports/serial/ring_buffer.go @@ -0,0 +1,173 @@ +/* + * 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 ( + "os" + "sync" + "time" +) + +// byteRing is a bounded FIFO byte buffer connecting the shared-port reader +// (append side) to one subscription (read side). On overflow the OLDEST +// bytes are dropped — a slow subscriber loses history but can never block +// the shared reader; protocol codecs resynchronize on the resulting gap. +// +// Contract: exactly one goroutine blocks in read at a time (wake-ups are +// coalesced through a one-slot channel). +type byteRing struct { + mu sync.Mutex + buf []byte + start int + length int + dropped uint64 + closed bool + err error + notify chan struct{} +} + +func newByteRing(capacity int) *byteRing { + return &byteRing{ + buf: make([]byte, capacity), + notify: make(chan struct{}, 1), + } +} + +func (r *byteRing) signal() { + select { + case r.notify <- struct{}{}: + default: + } +} + +// append copies p into the ring, dropping the oldest bytes when capacity +// is exceeded, and returns how many bytes were dropped. Appends after +// close are discarded. +func (r *byteRing) append(p []byte) int { + r.mu.Lock() + if r.closed || len(p) == 0 { + r.mu.Unlock() + return 0 + } + dropped := 0 + if len(p) >= len(r.buf) { + dropped = r.length + len(p) - len(r.buf) + p = p[len(p)-len(r.buf):] + r.start, r.length = 0, 0 + } else if over := r.length + len(p) - len(r.buf); over > 0 { + dropped = over + r.start = (r.start + over) % len(r.buf) + r.length -= over + } + for i := range p { + r.buf[(r.start+r.length+i)%len(r.buf)] = p[i] + } + r.length += len(p) + r.dropped += uint64(dropped) + r.mu.Unlock() + r.signal() + return dropped +} + +// fail records a fatal error. Buffered data remains readable; once it is +// drained, reads return err. +func (r *byteRing) fail(err error) { + r.mu.Lock() + if r.err == nil { + r.err = err + } + r.mu.Unlock() + r.signal() +} + +// close makes all subsequent reads fail with os.ErrClosed (buffered data +// is NOT drained first — matching a closed port's behavior). +func (r *byteRing) close() { + r.mu.Lock() + r.closed = true + r.mu.Unlock() + r.signal() +} + +func (r *byteRing) isClosed() bool { + r.mu.Lock() + defer r.mu.Unlock() + return r.closed +} + +func (r *byteRing) droppedTotal() uint64 { + r.mu.Lock() + defer r.mu.Unlock() + return r.dropped +} + +// read blocks until data is available, the deadline expires +// (os.ErrDeadlineExceeded; immediately if already past), the ring is +// closed (os.ErrClosed), or a recorded failure surfaces after the buffer +// drains. A zero deadline blocks indefinitely. +func (r *byteRing) read(p []byte, deadline time.Time) (int, error) { + if len(p) == 0 { + return 0, nil + } + var timeout <-chan time.Time + if !deadline.IsZero() { + wait := time.Until(deadline) + if wait <= 0 { + r.mu.Lock() + closed := r.closed + r.mu.Unlock() + if closed { + return 0, os.ErrClosed + } + return 0, os.ErrDeadlineExceeded + } + timer := time.NewTimer(wait) + defer timer.Stop() + timeout = timer.C + } + for { + r.mu.Lock() + if r.closed { + r.mu.Unlock() + return 0, os.ErrClosed + } + if r.length > 0 { + n := min(len(p), r.length) + for i := 0; i < n; i++ { + p[i] = r.buf[(r.start+i)%len(r.buf)] + } + r.start = (r.start + n) % len(r.buf) + r.length -= n + r.mu.Unlock() + return n, nil + } + if r.err != nil { + err := r.err + r.mu.Unlock() + return 0, err + } + r.mu.Unlock() + select { + case <-r.notify: + case <-timeout: + return 0, os.ErrDeadlineExceeded + } + } +} diff --git a/plc4go/spi/transports/serial/ring_buffer_test.go b/plc4go/spi/transports/serial/ring_buffer_test.go new file mode 100644 index 0000000000..556b3d1478 --- /dev/null +++ b/plc4go/spi/transports/serial/ring_buffer_test.go @@ -0,0 +1,122 @@ +/* + * 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 ( + "os" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/apache/plc4x/plc4go/spi/errors" +) + +func TestByteRing_FIFOAndWraparound(t *testing.T) { + r := newByteRing(8) + assert.Zero(t, r.append([]byte("abcde"))) + buf := make([]byte, 3) + n, err := r.read(buf, time.Time{}) + require.NoError(t, err) + assert.Equal(t, "abc", string(buf[:n])) + // wraps around the 8-byte capacity + assert.Zero(t, r.append([]byte("fghi"))) + got := make([]byte, 0, 6) + for len(got) < 6 { + n, err = r.read(buf, time.Time{}) + require.NoError(t, err) + got = append(got, buf[:n]...) + } + assert.Equal(t, "defghi", string(got)) +} + +func TestByteRing_BlockingReadWokenByAppend(t *testing.T) { + r := newByteRing(8) + go func() { + time.Sleep(20 * time.Millisecond) + r.append([]byte{0x42}) + }() + buf := make([]byte, 1) + n, err := r.read(buf, time.Now().Add(5*time.Second)) + require.NoError(t, err) + require.Equal(t, 1, n) + assert.Equal(t, byte(0x42), buf[0]) +} + +func TestByteRing_DeadlineExpiry(t *testing.T) { + r := newByteRing(8) + start := time.Now() + _, err := r.read(make([]byte, 1), time.Now().Add(40*time.Millisecond)) + require.ErrorIs(t, err, os.ErrDeadlineExceeded) + assert.GreaterOrEqual(t, time.Since(start), 40*time.Millisecond) + + // already-expired deadline errors immediately, even with data buffered + r.append([]byte{0x01}) + _, err = r.read(make([]byte, 1), time.Now().Add(-time.Second)) + require.ErrorIs(t, err, os.ErrDeadlineExceeded) +} + +func TestByteRing_DropOldestOnOverflow(t *testing.T) { + r := newByteRing(4) + assert.Zero(t, r.append([]byte("ab"))) + assert.Equal(t, 2, r.append([]byte("cdef")), "two oldest bytes dropped") + assert.EqualValues(t, 2, r.droppedTotal()) + buf := make([]byte, 4) + n, err := r.read(buf, time.Time{}) + require.NoError(t, err) + assert.Equal(t, "cdef", string(buf[:n])) + + // a chunk larger than capacity keeps only its tail + dropped := r.append([]byte("0123456789")) + assert.Equal(t, 6, dropped) + n, err = r.read(buf, time.Time{}) + require.NoError(t, err) + assert.Equal(t, "6789", string(buf[:n])) +} + +func TestByteRing_CloseBeatsBufferedData(t *testing.T) { + r := newByteRing(8) + r.append([]byte("data")) + r.close() + assert.True(t, r.isClosed()) + _, err := r.read(make([]byte, 4), time.Time{}) + require.ErrorIs(t, err, os.ErrClosed) +} + +func TestByteRing_FailDrainsDataThenSurfacesError(t *testing.T) { + r := newByteRing(8) + r.append([]byte("ok")) + portErr := errors.New("port exploded") + r.fail(portErr) + buf := make([]byte, 8) + n, err := r.read(buf, time.Time{}) + require.NoError(t, err, "buffered data drains before the error") + assert.Equal(t, "ok", string(buf[:n])) + _, err = r.read(buf, time.Time{}) + require.ErrorContains(t, err, "port exploded") +} + +func TestByteRing_ClosedBeatsExpiredDeadline(t *testing.T) { + r := newByteRing(8) + r.close() + _, err := r.read(make([]byte, 1), time.Now().Add(-time.Second)) + require.ErrorIs(t, err, os.ErrClosed, "closed must take precedence over an expired deadline") +} diff --git a/plc4go/spi/transports/serial/shared_port.go b/plc4go/spi/transports/serial/shared_port.go new file mode 100644 index 0000000000..4bc01762dd --- /dev/null +++ b/plc4go/spi/transports/serial/shared_port.go @@ -0,0 +1,337 @@ +/* + * 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 ( + "os" + "sync" + "sync/atomic" + "time" + + "github.com/rs/zerolog" + + "github.com/apache/plc4x/plc4go/spi/errors" + "github.com/apache/plc4x/plc4go/spi/transports" + "github.com/apache/plc4x/plc4go/spi/transports/serial/serialport" +) + +const ( + sharedSubscriberBufferCapacity = 64 << 10 + sharedReaderSlice = 100 * time.Millisecond + sharedReaderChunk = 4096 +) + +// sharedPortConfig is the full tuple that must be identical for two +// connections to share one physical port. It is comparable by design. +type sharedPortConfig struct { + port serialport.Config + dtr bool + rts bool + interframeDelay time.Duration +} + +// sharedPortRegistry tracks the shared ports of one Transport. It is not +// global: sharing works within one driver manager; separate managers +// opening the same device remain unsupported (as without sharing). +type sharedPortRegistry struct { + mu sync.Mutex + ports map[string]*sharedPort + bufferCapacity int // per-subscriber ring size; overridable in tests + log zerolog.Logger +} + +func newSharedPortRegistry(log zerolog.Logger) *sharedPortRegistry { + return &sharedPortRegistry{ + ports: map[string]*sharedPort{}, + bufferCapacity: sharedSubscriberBufferCapacity, + log: log, + } +} + +// acquire joins the existing shared port (config must match exactly) or +// opens the device and starts its reader. +func (r *sharedPortRegistry) acquire(portName string, cfg sharedPortConfig) (serialport.Port, error) { + r.mu.Lock() + defer r.mu.Unlock() + if existing, ok := r.ports[portName]; ok { + if existing.cfg != cfg { + return nil, errors.Errorf( + "serial port %s is already shared with a different configuration (existing: %+v, requested: %+v)", + portName, existing.cfg, cfg) + } + if sub, ok := existing.subscribe(r.bufferCapacity); ok { + return sub, nil + } + // The entry closed concurrently; drop it and open fresh below. + delete(r.ports, portName) + } + port, err := serialport.Open(portName, cfg.port) + if err != nil { + return nil, errors.Wrap(err, "error opening shared serial port") + } + applySharedModemLines(port, cfg, r.log) + sp := &sharedPort{ + registry: r, + portName: portName, + cfg: cfg, + port: port, + pacer: newPacer(cfg.interframeDelay), + log: r.log, + } + r.ports[portName] = sp + sub, ok := sp.subscribe(r.bufferCapacity) + if !ok { + // Cannot happen: the port was created closed==false and nothing + // else references it yet. + delete(r.ports, portName) + _ = port.Close() + return nil, errors.New("serialport: freshly opened shared port already closed") + } + go sp.readLoop() + return sub, nil +} + +// remove deletes the registry entry if it still maps to sp. +func (r *sharedPortRegistry) remove(sp *sharedPort) { + r.mu.Lock() + defer r.mu.Unlock() + if r.ports[sp.portName] == sp { + delete(r.ports, sp.portName) + } +} + +// applySharedModemLines asserts DTR/RTS once at open, mirroring the +// dedicated path's warn-don't-fail semantics. +func applySharedModemLines(port serialport.Port, cfg sharedPortConfig, log zerolog.Logger) { + controlPort, ok := port.(serialport.ControlPort) + if !ok { + return + } + if cfg.dtr { + if err := controlPort.SetDTR(true); err != nil { + log.Warn().Err(err).Msg("could not assert DTR on shared port") + } + } + if cfg.rts { + if err := controlPort.SetRTS(true); err != nil { + log.Warn().Err(err).Msg("could not assert RTS on shared port") + } + } +} + +// sharedPort owns one physical port: a broadcast reader, the subscriber +// list, and the serialized paced write path. +// +// Locking: stateMu guards subs/closed; writeMu serializes writers. writeMu may +// acquire stateMu momentarily (via isClosed) but never the reverse — the nesting +// is strictly one-directional, so no cycle is possible. Never hold stateMu while +// taking registry.mu; on last-unsubscribe the physical port is closed BEFORE the +// registry entry is removed, so a concurrent acquire either joins a live port or +// re-opens a fully released device. +type sharedPort struct { + registry *sharedPortRegistry + portName string + cfg sharedPortConfig + port serialport.Port + pacer *pacer + log zerolog.Logger + + stateMu sync.Mutex + subs []*byteRing + closed bool + + writeMu sync.Mutex +} + +// subscribe adds a subscriber; ok is false when the port already closed. +func (sp *sharedPort) subscribe(capacity int) (*subscription, bool) { + sp.stateMu.Lock() + defer sp.stateMu.Unlock() + if sp.closed { + return nil, false + } + ring := newByteRing(capacity) + sp.subs = append(sp.subs, ring) + sub := &subscription{sp: sp, ring: ring} + sub.readDeadline.Store(time.Time{}) + sub.writeDeadline.Store(time.Time{}) + return sub, true +} + +func (sp *sharedPort) isClosed() bool { + sp.stateMu.Lock() + defer sp.stateMu.Unlock() + return sp.closed +} + +// unsubscribe detaches ring; the last subscriber closes the port. +func (sp *sharedPort) unsubscribe(ring *byteRing) error { + ring.close() + sp.stateMu.Lock() + for i, s := range sp.subs { + if s == ring { + sp.subs = append(sp.subs[:i], sp.subs[i+1:]...) + break + } + } + last := len(sp.subs) == 0 && !sp.closed + if last { + sp.closed = true + } + sp.stateMu.Unlock() + if !last { + return nil + } + err := sp.port.Close() // unblocks the reader; close before unmapping + sp.registry.remove(sp) + return err +} + +// fail propagates a fatal port error to all subscribers and evicts the +// port so a reconnect re-opens the device. +func (sp *sharedPort) fail(err error) { + sp.log.Warn().Err(err).Str("port", sp.portName).Msg("shared serial port failed") + sp.stateMu.Lock() + alreadyClosed := sp.closed + sp.closed = true + subs := sp.subs + sp.subs = nil + sp.stateMu.Unlock() + for _, ring := range subs { + ring.fail(err) + } + if !alreadyClosed { + _ = sp.port.Close() + } + sp.registry.remove(sp) +} + +// readLoop broadcasts everything the port delivers into every +// subscriber's ring. Deadline slices keep shutdown prompt. +func (sp *sharedPort) readLoop() { + buf := make([]byte, sharedReaderChunk) + warnedDrops := map[*byteRing]uint64{} + for { + if sp.isClosed() { + return + } + _ = sp.port.SetReadDeadline(time.Now().Add(sharedReaderSlice)) + n, err := sp.port.Read(buf) + if n > 0 { + sp.pacer.noteActivity() + sp.stateMu.Lock() + subs := append([]*byteRing(nil), sp.subs...) + sp.stateMu.Unlock() + for _, ring := range subs { + if dropped := ring.append(buf[:n]); dropped > 0 { + total := ring.droppedTotal() + if last := warnedDrops[ring]; last == 0 || total >= 2*last { + warnedDrops[ring] = total + sp.log.Warn(). + Int("dropped", dropped). + Uint64("totalDropped", total). + Str("port", sp.portName). + Msg("shared serial subscriber buffer overflow, oldest bytes dropped") + } + } + } + } + if err != nil && !transports.ErrorIs(err, os.ErrDeadlineExceeded) { + if sp.isClosed() { + return // expected: the last unsubscribe closed the port + } + sp.fail(err) + return + } + } +} + +// write serializes and paces writes from all subscribers. +func (sp *sharedPort) write(ring *byteRing, p []byte, deadline time.Time) (int, error) { + sp.writeMu.Lock() + defer sp.writeMu.Unlock() + // The subscription may close while a Write is waiting on writeMu; recheck under the lock + // so a closed connection's bytes never reach the shared line. + if ring.isClosed() { + return 0, os.ErrClosed + } + if sp.isClosed() { + return 0, os.ErrClosed + } + if err := sp.pacer.waitTurn(deadline); err != nil { + return 0, err + } + if err := sp.port.SetWriteDeadline(deadline); err != nil { + return 0, err + } + n, err := sp.port.Write(p) + sp.pacer.noteActivity() + return n, err +} + +// subscription is one connection's view of a shared port. It implements +// serialport.Port (and deliberately NOT serialport.ControlPort — modem +// line changes would cross-talk between connections). +type subscription struct { + sp *sharedPort + ring *byteRing + readDeadline atomic.Value // time.Time + writeDeadline atomic.Value // time.Time + closeOnce sync.Once + closeErr error +} + +var _ serialport.Port = (*subscription)(nil) + +func (s *subscription) Read(p []byte) (int, error) { + deadline, _ := s.readDeadline.Load().(time.Time) + return s.ring.read(p, deadline) +} + +func (s *subscription) Write(p []byte) (int, error) { + if s.ring.isClosed() { + return 0, os.ErrClosed + } + deadline, _ := s.writeDeadline.Load().(time.Time) + return s.sp.write(s.ring, p, deadline) +} + +func (s *subscription) SetReadDeadline(t time.Time) error { + if s.ring.isClosed() { + return os.ErrClosed + } + s.readDeadline.Store(t) + return nil +} + +func (s *subscription) SetWriteDeadline(t time.Time) error { + if s.ring.isClosed() { + return os.ErrClosed + } + s.writeDeadline.Store(t) + return nil +} + +func (s *subscription) Close() error { + s.closeOnce.Do(func() { + s.closeErr = s.sp.unsubscribe(s.ring) + }) + return s.closeErr +} diff --git a/plc4go/spi/transports/serial/shared_port_pty_test.go b/plc4go/spi/transports/serial/shared_port_pty_test.go new file mode 100644 index 0000000000..3ab510282d --- /dev/null +++ b/plc4go/spi/transports/serial/shared_port_pty_test.go @@ -0,0 +1,175 @@ +//go:build linux + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package serial + +import ( + "os" + "sync" + "testing" + "time" + + "github.com/rs/zerolog" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/apache/plc4x/plc4go/spi/transports/serial/serialport" +) + +func acquireTwo(t *testing.T, registry *sharedPortRegistry, slavePath string, cfg sharedPortConfig) (serialport.Port, serialport.Port) { + t.Helper() + first, err := registry.acquire(slavePath, cfg) + require.NoError(t, err) + second, err := registry.acquire(slavePath, cfg) + require.NoError(t, err) + return first, second +} + +func TestSharedPort_BroadcastToAllSubscribers(t *testing.T) { + master, slavePath := openPTY(t) + registry := newSharedPortRegistry(zerolog.Nop()) + cfg := sharedPortConfig{port: serialport.Config{BaudRate: 9600}} + first, second := acquireTwo(t, registry, slavePath, cfg) + defer first.Close() + defer second.Close() + + _, err := master.Write([]byte("hello")) + require.NoError(t, err) + + for _, sub := range []serialport.Port{first, second} { + require.NoError(t, sub.SetReadDeadline(time.Now().Add(5*time.Second))) + buf := make([]byte, 16) + n, err := sub.Read(buf) + require.NoError(t, err) + assert.Equal(t, "hello", string(buf[:n]), "every subscriber sees the full traffic") + } +} + +func TestSharedPort_WritesSerializedAndPaced(t *testing.T) { + master, slavePath := openPTY(t) + registry := newSharedPortRegistry(zerolog.Nop()) + cfg := sharedPortConfig{port: serialport.Config{BaudRate: 9600}, interframeDelay: 60 * time.Millisecond} + first, second := acquireTwo(t, registry, slavePath, cfg) + defer first.Close() + defer second.Close() + + start := time.Now() + var wg sync.WaitGroup + for _, sub := range []serialport.Port{first, second} { + wg.Add(1) + go func(p serialport.Port) { + defer wg.Done() + _, err := p.Write([]byte("XXXX")) + assert.NoError(t, err) + }(sub) + } + wg.Wait() + elapsed := time.Since(start) + + buf := make([]byte, 8) + total := 0 + for total < 8 { + n, err := master.Read(buf[total:]) + require.NoError(t, err) + total += n + } + assert.Equal(t, "XXXXXXXX", string(buf), "both payloads intact") + assert.GreaterOrEqual(t, elapsed, 60*time.Millisecond, "second write must wait out the inter-frame gap") +} + +func TestSharedPort_ConfigMismatchFails(t *testing.T) { + _, slavePath := openPTY(t) + registry := newSharedPortRegistry(zerolog.Nop()) + first, err := registry.acquire(slavePath, sharedPortConfig{port: serialport.Config{BaudRate: 9600}}) + require.NoError(t, err) + defer first.Close() + + _, err = registry.acquire(slavePath, sharedPortConfig{port: serialport.Config{BaudRate: 19200}}) + require.Error(t, err) + assert.Contains(t, err.Error(), slavePath) +} + +func TestSharedPort_RefcountLifecycle(t *testing.T) { + master, slavePath := openPTY(t) + registry := newSharedPortRegistry(zerolog.Nop()) + cfg := sharedPortConfig{port: serialport.Config{BaudRate: 9600}} + first, second := acquireTwo(t, registry, slavePath, cfg) + + // First close: the second subscriber keeps working. + require.NoError(t, first.Close()) + _, err := master.Write([]byte{0x01}) + require.NoError(t, err) + require.NoError(t, second.SetReadDeadline(time.Now().Add(5*time.Second))) + buf := make([]byte, 1) + _, err = second.Read(buf) + require.NoError(t, err) + + // Reads on the closed subscription fail with os.ErrClosed. + _, err = first.Read(buf) + require.ErrorIs(t, err, os.ErrClosed) + + // Last close releases the port; a fresh acquire re-opens it. + require.NoError(t, second.Close()) + third, err := registry.acquire(slavePath, cfg) + require.NoError(t, err) + require.NoError(t, third.Close()) +} + +func TestSharedPort_OverflowDropsOldest(t *testing.T) { + master, slavePath := openPTY(t) + registry := newSharedPortRegistry(zerolog.Nop()) + registry.bufferCapacity = 8 // tiny buffer to force overflow + cfg := sharedPortConfig{port: serialport.Config{BaudRate: 9600}} + sub, err := registry.acquire(slavePath, cfg) + require.NoError(t, err) + defer sub.Close() + + _, err = master.Write([]byte("0123456789ABCDEF")) + require.NoError(t, err) + + // Wait for delivery, then confirm reads still work and end with the + // freshest bytes (drop-oldest, never drop-newest). + require.NoError(t, sub.SetReadDeadline(time.Now().Add(5*time.Second))) + collected := make([]byte, 0, 16) + buf := make([]byte, 16) + for { + require.NoError(t, sub.SetReadDeadline(time.Now().Add(200*time.Millisecond))) + n, err := sub.Read(buf) + if err != nil { + break // deadline: stream drained + } + collected = append(collected, buf[:n]...) + } + require.NotEmpty(t, collected) + assert.Equal(t, byte('F'), collected[len(collected)-1], "newest byte must survive") + assert.LessOrEqual(t, len(collected), 16) + assert.Positive(t, sub.(*subscription).ring.droppedTotal(), "drop counter must have risen") +} + +func TestSharedPort_SubscriptionIsNotControlPort(t *testing.T) { + _, slavePath := openPTY(t) + registry := newSharedPortRegistry(zerolog.Nop()) + sub, err := registry.acquire(slavePath, sharedPortConfig{port: serialport.Config{BaudRate: 9600}}) + require.NoError(t, err) + defer sub.Close() + _, isControl := sub.(serialport.ControlPort) + assert.False(t, isControl, "modem control must not leak through shared subscriptions") +}
