From 9d5425975b00c4f0280e2ae62cd9fc9b4356af15 Mon Sep 17 00:00:00 2001 From: Dieter Lang Date: Mon, 10 Aug 2026 17:59:35 +0200 Subject: [PATCH] Add client runtime and serial bridge --- CHANGELOG.md | 20 + internal/client/bridge.go | 130 +++++ internal/client/bridge_test.go | 271 +++++++++++ internal/client/connection.go | 111 +++++ internal/client/connection_test.go | 443 ++++++++++++++++++ internal/client/control.go | 66 ++- internal/client/device.go | 70 +++ internal/client/device_test.go | 159 +++++++ internal/client/pty_linux.go | 99 +++- internal/client/runtime.go | 216 +++++++++ internal/client/runtime_integration_test.go | 280 +++++++++++ internal/client/runtime_test.go | 82 ++++ internal/client/virtual_port_manager.go | 21 +- internal/client/virtual_port_manager_linux.go | 79 +++- 14 files changed, 2003 insertions(+), 44 deletions(-) create mode 100644 internal/client/bridge.go create mode 100644 internal/client/bridge_test.go create mode 100644 internal/client/connection.go create mode 100644 internal/client/connection_test.go create mode 100644 internal/client/device.go create mode 100644 internal/client/device_test.go create mode 100644 internal/client/runtime.go create mode 100644 internal/client/runtime_integration_test.go create mode 100644 internal/client/runtime_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index fc282b3..71c430d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,26 @@ Alle wesentlichen Änderungen am Projekt werden in dieser Datei dokumentiert. +## [0.0.6] - 2026-08-10 + +### Added + +- Clientseitige Runtime zur Verwaltung mehrerer virtueller serieller Ports +- Verbindung konfigurierter virtueller Ports mit den vom Server bereitgestellten + Remote-Geräten +- Clientseitige TCP-Data-Connections zu den dynamischen Data-Ports des Servers +- Bidirektionale Datenübertragung über eine `Bridge` +- Verwaltung des Lebenszyklus von virtuellen Ports, Bridges und Client-Verbindung +- Integrationstest für den vollständigen Datenweg zwischen virtuellem seriellen + Port und Server +- Tests für Verbindungsaufbau, Gerätezuordnung und Runtime-Lebenszyklus + +### Tests + +- `go test ./...` erfolgreich +- `go test -race ./...` erfolgreich +- Bidirektionale Übertragung über Runtime, Bridge, TCP und PTY erfolgreich +- Race Conditions in Client-Control-Verbindung, PTY und virtuellen Ports behoben ## [0.0.5] - 2026-08-10 ### Added diff --git a/internal/client/bridge.go b/internal/client/bridge.go new file mode 100644 index 0000000..71c451a --- /dev/null +++ b/internal/client/bridge.go @@ -0,0 +1,130 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/bridge.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Bidirektionale Daten-Bridge zwischen der virtuellen seriellen + * Schnittstelle des Clients und der TCP-Data-Verbindung. + * + * Die Bridge transportiert den Byte-Strom in beide Richtungen: + * + * VirtualSerial ─────► TCP + * VirtualSerial ◄───── TCP + * + * Die Bridge kennt weder das Control-Protokoll noch die konkrete + * serielle Hardware. Sie verbindet ausschließlich zwei io.ReadWriter- + * Endpunkte miteinander. + * ============================================================================ + */ +package client + +import ( + "fmt" + "io" + "net" + "sync" +) + +/////////////////////////////////////////////////////////////////////////////// +// Bridge +/////////////////////////////////////////////////////////////////////////////// + +// Bridge verbindet eine virtuelle serielle Schnittstelle mit einer +// TCP-Data-Verbindung. +// +// Beide Datenrichtungen werden gleichzeitig bedient. Dadurch bleibt der +// serielle Datenstrom vollständig bidirektional. +type Bridge struct { + serial VirtualSerial + conn net.Conn +} + +/////////////////////////////////////////////////////////////////////////////// +// Constructor +/////////////////////////////////////////////////////////////////////////////// + +// NewBridge creates a new bidirectional data bridge. +// +// The bridge does not take ownership of the endpoints until Run is called. +func NewBridge(serial VirtualSerial, conn net.Conn) (*Bridge, error) { + if serial == nil { + return nil, fmt.Errorf("virtual serial is nil") + } + + if conn == nil { + return nil, fmt.Errorf("TCP connection is nil") + } + + return &Bridge{ + serial: serial, + conn: conn, + }, nil +} + +/////////////////////////////////////////////////////////////////////////////// +// Run +/////////////////////////////////////////////////////////////////////////////// + +// Run starts the bidirectional byte transfer and blocks until one of the +// directions terminates. +// +// Closing the endpoints after termination releases the opposite blocked +// transfer as well. +func (b *Bridge) Run() error { + if b == nil || b.serial == nil { + return fmt.Errorf("bridge is not initialized") + } + + if b.conn == nil { + return fmt.Errorf("TCP connection is nil") + } + + type result struct { + name string + err error + } + + results := make(chan result, 2) + + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + + _, err := io.Copy(b.conn, b.serial) + results <- result{ + name: "serial to TCP", + err: err, + } + }() + + go func() { + defer wg.Done() + + _, err := io.Copy(b.serial, b.conn) + results <- result{ + name: "TCP to serial", + err: err, + } + }() + + first := <-results + + // Beide Richtungen sollen beendet werden. Das Schließen der Endpunkte + // unterbricht insbesondere einen eventuell noch blockierenden Read. + _ = b.conn.Close() + _ = b.serial.Close() + + wg.Wait() + + if first.err != nil { + return fmt.Errorf("%s: %w", first.name, first.err) + } + + return nil +} diff --git a/internal/client/bridge_test.go b/internal/client/bridge_test.go new file mode 100644 index 0000000..ab24fc4 --- /dev/null +++ b/internal/client/bridge_test.go @@ -0,0 +1,271 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/bridge_test.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Tests für die bidirektionale Daten-Bridge zwischen VirtualSerial und + * TCP-Verbindung. + * + * Die Tests verwenden net.Pipe und eine kleine In-Memory-Implementierung + * von VirtualSerial. Dadurch wird ausschließlich die Daten-Bridge getestet, + * ohne echte TCP-Ports oder eine reale PTY-Schnittstelle zu benötigen. + * ============================================================================ + */ +package client + +import ( + "bytes" + "io" + "net" + "sync" + "testing" + "time" +) + +/////////////////////////////////////////////////////////////////////////////// +// Test VirtualSerial +/////////////////////////////////////////////////////////////////////////////// + +type testVirtualSerial struct { + mu sync.Mutex + reader *bytes.Reader + writes [][]byte + closed bool +} + +func newTestVirtualSerial(data []byte) *testVirtualSerial { + return &testVirtualSerial{ + reader: bytes.NewReader(data), + } +} + +func (s *testVirtualSerial) Path() string { + return "test-serial" +} + +func (s *testVirtualSerial) Read(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.closed { + return 0, io.EOF + } + + return s.reader.Read(p) +} + +func (s *testVirtualSerial) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.closed { + return 0, io.ErrClosedPipe + } + + cp := append([]byte(nil), p...) + s.writes = append(s.writes, cp) + + return len(p), nil +} + +func (s *testVirtualSerial) Written() []byte { + s.mu.Lock() + defer s.mu.Unlock() + + var result []byte + for _, p := range s.writes { + result = append(result, p...) + } + return result +} + +func (s *testVirtualSerial) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + + s.closed = true + return nil +} + +/////////////////////////////////////////////////////////////////////////////// +// Blocking test VirtualSerial +/////////////////////////////////////////////////////////////////////////////// + +type blockingVirtualSerial struct { + mu sync.Mutex + writes [][]byte + readCh chan struct{} + closed bool +} + +func newBlockingVirtualSerial() *blockingVirtualSerial { + return &blockingVirtualSerial{ + readCh: make(chan struct{}), + } +} + +func (s *blockingVirtualSerial) Path() string { + return "blocking-test-serial" +} + +func (s *blockingVirtualSerial) Read([]byte) (int, error) { + <-s.readCh + return 0, io.EOF +} + +func (s *blockingVirtualSerial) Write(p []byte) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + + if s.closed { + return 0, io.ErrClosedPipe + } + + s.writes = append(s.writes, append([]byte(nil), p...)) + return len(p), nil +} + +func (s *blockingVirtualSerial) Written() []byte { + s.mu.Lock() + defer s.mu.Unlock() + + var result []byte + for _, p := range s.writes { + result = append(result, p...) + } + return result +} + +func (s *blockingVirtualSerial) Close() error { + s.mu.Lock() + defer s.mu.Unlock() + + if !s.closed { + s.closed = true + close(s.readCh) + } + + return nil +} + +/////////////////////////////////////////////////////////////////////////////// +// Tests +/////////////////////////////////////////////////////////////////////////////// + +func TestNewBridge(t *testing.T) { + serial := newTestVirtualSerial(nil) + connA, connB := net.Pipe() + defer connA.Close() + defer connB.Close() + + if _, err := NewBridge(nil, connA); err == nil { + t.Fatal("expected error for nil VirtualSerial") + } + + if _, err := NewBridge(serial, nil); err == nil { + t.Fatal("expected error for nil TCP connection") + } + + bridge, err := NewBridge(serial, connA) + if err != nil { + t.Fatalf("NewBridge: %v", err) + } + + if bridge.serial != serial { + t.Fatal("bridge serial endpoint mismatch") + } + + if bridge.conn != connA { + t.Fatal("bridge TCP endpoint mismatch") + } +} + +func TestBridgeSerialToTCP(t *testing.T) { + serial := newTestVirtualSerial([]byte("hello")) + + clientConn, bridgeConn := net.Pipe() + + bridge, err := NewBridge(serial, bridgeConn) + if err != nil { + t.Fatalf("NewBridge: %v", err) + } + + runDone := make(chan error, 1) + go func() { + runDone <- bridge.Run() + }() + + got := make([]byte, 5) + if _, err := io.ReadFull(clientConn, got); err != nil { + t.Fatalf("read TCP data: %v", err) + } + + if string(got) != "hello" { + t.Fatalf("received %q, want %q", got, "hello") + } + + _ = clientConn.Close() + + select { + case err := <-runDone: + if err != nil { + t.Fatalf("Bridge.Run: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Bridge.Run did not terminate") + } +} + +func TestBridgeTCPToSerial(t *testing.T) { + serial := newBlockingVirtualSerial() + + clientConn, bridgeConn := net.Pipe() + + bridge, err := NewBridge(serial, bridgeConn) + if err != nil { + t.Fatalf("NewBridge: %v", err) + } + + runDone := make(chan error, 1) + go func() { + runDone <- bridge.Run() + }() + + want := []byte{0x46, 0x41, 0x00, 0x10, 0x0D} + + if _, err := clientConn.Write(want); err != nil { + t.Fatalf("write TCP data: %v", err) + } + + deadline := time.Now().Add(2 * time.Second) + for { + if bytes.Equal(serial.Written(), want) { + break + } + + if time.Now().After(deadline) { + t.Fatalf( + "serial received % X, want % X", + serial.Written(), + want, + ) + } + + time.Sleep(1 * time.Millisecond) + } + + _ = clientConn.Close() + + select { + case err := <-runDone: + if err != nil { + t.Fatalf("Bridge.Run: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Bridge.Run did not terminate") + } +} diff --git a/internal/client/connection.go b/internal/client/connection.go new file mode 100644 index 0000000..ffc3c77 --- /dev/null +++ b/internal/client/connection.go @@ -0,0 +1,111 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/connection.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Aufbau einer Verbindung zwischen einem lokalen virtuellen seriellen Port + * und dem zugehörigen entfernten Gerät. + * ============================================================================ + */ +package client + +import ( + "fmt" + "net" + + "git.lang-dieter.de/rs2322tcp/internal/config" + "git.lang-dieter.de/rs2322tcp/internal/transport" +) + +/////////////////////////////////////////////////////////////////////////////// +// Virtual port connection +/////////////////////////////////////////////////////////////////////////////// + +// connectVirtualPort opens the local virtual port and the corresponding +// remote TCP data connection and returns both endpoints. +// +// The caller owns the returned virtual port and TCP connection. +func connectVirtualPort( + c *Client, + manager *VirtualPortManager, + virtualPortConfig config.VirtualPortConfig, + devices []transport.RemoteDeviceInfo, +) (*ManagedVirtualPort, net.Conn, error) { + if c == nil || c.conn == nil { + return nil, nil, fmt.Errorf("client is not connected") + } + + if manager == nil { + return nil, nil, fmt.Errorf("virtual port manager is nil") + } + + device, err := findRemoteDevice(virtualPortConfig, devices) + if err != nil { + return nil, nil, err + } + + virtualPort, err := manager.Open() + if err != nil { + return nil, nil, fmt.Errorf( + "open virtual port for %q: %w", + virtualPortConfig.RemoteDevice, + err, + ) + } + + conn, err := c.OpenDataConnection(device) + if err != nil { + _ = virtualPort.Close() + + return nil, nil, fmt.Errorf( + "open data connection for %q: %w", + device.ID, + err, + ) + } + + return virtualPort, conn, nil +} + +/////////////////////////////////////////////////////////////////////////////// +// Bridge connection +/////////////////////////////////////////////////////////////////////////////// + +// connectBridge creates the local virtual port, opens the corresponding +// remote TCP data connection and creates the data bridge. +// +// The caller owns the returned bridge and its endpoints. +func connectBridge( + c *Client, + manager *VirtualPortManager, + virtualPortConfig config.VirtualPortConfig, + devices []transport.RemoteDeviceInfo, +) (*Bridge, *ManagedVirtualPort, net.Conn, error) { + virtualPort, conn, err := connectVirtualPort( + c, + manager, + virtualPortConfig, + devices, + ) + if err != nil { + return nil, nil, nil, err + } + + bridge, err := NewBridge(virtualPort, conn) + if err != nil { + _ = conn.Close() + _ = virtualPort.Close() + + return nil, nil, nil, fmt.Errorf( + "create bridge for %q: %w", + virtualPortConfig.RemoteDevice, + err, + ) + } + + return bridge, virtualPort, conn, nil +} diff --git a/internal/client/connection_test.go b/internal/client/connection_test.go new file mode 100644 index 0000000..46d0d65 --- /dev/null +++ b/internal/client/connection_test.go @@ -0,0 +1,443 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/connection_test.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Tests für den Aufbau einer Verbindung zwischen einem lokalen virtuellen + * seriellen Port und dem zugehörigen entfernten TCP-Gerät. + * + * Die Tests prüfen insbesondere die Auswahl des Remote-Geräts, die Erzeugung + * des virtuellen Ports, den Aufbau der TCP-Datenverbindung und die Verbindung + * dieser Endpunkte über eine Bridge. + * ============================================================================ + */ +package client + +import ( + "net" + "strconv" + "strings" + "testing" + + "git.lang-dieter.de/rs2322tcp/internal/config" + "git.lang-dieter.de/rs2322tcp/internal/transport" +) + +/////////////////////////////////////////////////////////////////////////////// +// Test connection +/////////////////////////////////////////////////////////////////////////////// + +// testRemoteAddrConn wraps a net.Conn and provides a TCP-style remote address. +// +// net.Pipe() is used for the control connection in these tests. Its default +// RemoteAddr() is "pipe", which cannot be used by OpenDataConnection() to +// determine the server host. This wrapper supplies a normal TCP address. +type testRemoteAddrConn struct { + net.Conn + remoteAddr net.Addr +} + +// RemoteAddr returns the TCP-style remote address configured for the test. +func (c *testRemoteAddrConn) RemoteAddr() net.Addr { + return c.remoteAddr +} + +/////////////////////////////////////////////////////////////////////////////// +// Test helpers +/////////////////////////////////////////////////////////////////////////////// + +// startTestDataServer starts a local TCP listener for a simulated remote +// device. +// +// The listener accepts exactly one connection. The accepted connection is +// returned through the channel so the test can verify that the client has +// successfully established the data connection. +func startTestDataServer(t *testing.T) (string, <-chan net.Conn) { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen for test data server: %v", err) + } + + t.Cleanup(func() { + _ = listener.Close() + }) + + accepted := make(chan net.Conn, 1) + + go func() { + conn, err := listener.Accept() + if err != nil { + return + } + + accepted <- conn + }() + + return listener.Addr().String(), accepted +} + +// remoteDeviceFromAddress creates a RemoteDeviceInfo whose DataPort is +// taken from the supplied local TCP address. +func remoteDeviceFromAddress( + t *testing.T, + id string, + address string, +) transport.RemoteDeviceInfo { + t.Helper() + + _, portText, err := net.SplitHostPort(address) + if err != nil { + t.Fatalf("split test server address %q: %v", address, err) + } + + port, err := strconv.Atoi(portText) + if err != nil { + t.Fatalf("parse test server port %q: %v", portText, err) + } + + return transport.RemoteDeviceInfo{ + ID: id, + Name: id, + DataPort: port, + } +} + +// newTestClient creates a Client whose control connection uses net.Pipe() +// but reports a normal TCP-style remote address. +func newTestClient(t *testing.T) *Client { + t.Helper() + + controlClient, serverConn := net.Pipe() + + t.Cleanup(func() { + _ = controlClient.Close() + _ = serverConn.Close() + }) + + return &Client{ + conn: &testRemoteAddrConn{ + Conn: controlClient, + remoteAddr: &net.TCPAddr{ + IP: net.ParseIP("127.0.0.1"), + Port: 5000, + }, + }, + } +} + +/////////////////////////////////////////////////////////////////////////////// +// Tests +/////////////////////////////////////////////////////////////////////////////// + +func TestConnectVirtualPortNilClient(t *testing.T) { + manager := NewVirtualPortManager( + config.VirtualPortRangeConfig{ + First: 100, + Last: 199, + }, + ) + + cfg := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "radio", + } + + devices := []transport.RemoteDeviceInfo{ + { + ID: "radio", + DataPort: 50123, + }, + } + + _, _, err := connectVirtualPort( + nil, + manager, + cfg, + devices, + ) + if err == nil { + t.Fatal("expected error for nil client") + } + + if !strings.Contains(err.Error(), "client is not connected") { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestConnectVirtualPortNilManager(t *testing.T) { + cfg := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "radio", + } + + devices := []transport.RemoteDeviceInfo{ + { + ID: "radio", + DataPort: 50123, + }, + } + + _, _, err := connectVirtualPort( + &Client{}, + nil, + cfg, + devices, + ) + if err == nil { + t.Fatal("expected error for nil manager") + } + + if !strings.Contains(err.Error(), "client is not connected") { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestConnectVirtualPortUnknownDevice(t *testing.T) { + manager := NewVirtualPortManager( + config.VirtualPortRangeConfig{ + First: 100, + Last: 199, + }, + ) + + client := newTestClient(t) + + cfg := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "unknown", + } + + devices := []transport.RemoteDeviceInfo{ + { + ID: "radio", + DataPort: 50123, + }, + } + + _, _, err := connectVirtualPort( + client, + manager, + cfg, + devices, + ) + if err == nil { + t.Fatal("expected error for unknown device") + } + + if !strings.Contains( + err.Error(), + `remote device "unknown" not found`, + ) { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestConnectVirtualPortClosesPortWhenDataConnectionFails(t *testing.T) { + manager := NewVirtualPortManager( + config.VirtualPortRangeConfig{ + First: 100, + Last: 100, + }, + ) + + client := newTestClient(t) + + cfg := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "radio", + } + + devices := []transport.RemoteDeviceInfo{ + { + ID: "radio", + DataPort: 1, + }, + } + + port, conn, err := connectVirtualPort( + client, + manager, + cfg, + devices, + ) + + if err == nil { + if conn != nil { + _ = conn.Close() + } + + if port != nil { + _ = port.Close() + } + + t.Fatal("expected data connection error") + } + + if port != nil { + t.Fatal( + "virtual port must not be returned after connection failure", + ) + } + + if conn != nil { + t.Fatal( + "TCP connection must not be returned after connection failure", + ) + } + + // Port 100 must have been released after the failed connection. + portNumber, err := manager.Reserve() + if err != nil { + t.Fatalf("Reserve after cleanup: %v", err) + } + + if portNumber != 100 { + t.Fatalf("reserved port = %d, want 100", portNumber) + } + + manager.Release(portNumber) +} + +func TestConnectVirtualPortSuccess(t *testing.T) { + dataAddress, accepted := startTestDataServer(t) + + device := remoteDeviceFromAddress( + t, + "radio", + dataAddress, + ) + + client := newTestClient(t) + + manager := NewVirtualPortManager( + config.VirtualPortRangeConfig{ + First: 100, + Last: 199, + }, + ) + + cfg := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "radio", + } + + virtualPort, dataConn, err := connectVirtualPort( + client, + manager, + cfg, + []transport.RemoteDeviceInfo{device}, + ) + if err != nil { + t.Fatalf("connectVirtualPort: %v", err) + } + + if virtualPort == nil { + t.Fatal("virtual port is nil") + } + + if dataConn == nil { + t.Fatal("data connection is nil") + } + + defer virtualPort.Close() + defer dataConn.Close() + + if virtualPort.Path() != "/dev/ttyUSB100" { + t.Fatalf( + "virtual port path = %q, want %q", + virtualPort.Path(), + "/dev/ttyUSB100", + ) + } + + // The test data server must receive the connection created by + // OpenDataConnection(). + select { + case conn := <-accepted: + _ = conn.Close() + + case <-t.Context().Done(): + t.Fatal("test context cancelled while waiting for data connection") + } +} + +func TestConnectBridgeSuccess(t *testing.T) { + dataAddress, accepted := startTestDataServer(t) + + device := remoteDeviceFromAddress( + t, + "radio", + dataAddress, + ) + + client := newTestClient(t) + + manager := NewVirtualPortManager( + config.VirtualPortRangeConfig{ + First: 100, + Last: 199, + }, + ) + + cfg := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "radio", + } + + bridge, virtualPort, dataConn, err := connectBridge( + client, + manager, + cfg, + []transport.RemoteDeviceInfo{device}, + ) + if err != nil { + t.Fatalf("connectBridge: %v", err) + } + + if bridge == nil { + t.Fatal("bridge is nil") + } + + if virtualPort == nil { + t.Fatal("virtual port is nil") + } + + if dataConn == nil { + t.Fatal("data connection is nil") + } + + defer virtualPort.Close() + defer dataConn.Close() + + if bridge.serial != virtualPort { + t.Fatal("bridge serial endpoint does not match virtual port") + } + + if bridge.conn != dataConn { + t.Fatal("bridge TCP endpoint does not match data connection") + } + + if virtualPort.Path() != "/dev/ttyUSB100" { + t.Fatalf( + "virtual port path = %q, want %q", + virtualPort.Path(), + "/dev/ttyUSB100", + ) + } + + // The test data server must receive the TCP data connection created by + // connectBridge(). + select { + case conn := <-accepted: + _ = conn.Close() + + case <-t.Context().Done(): + t.Fatal("test context cancelled while waiting for data connection") + } +} diff --git a/internal/client/control.go b/internal/client/control.go index acb7fbd..6b45b2a 100644 --- a/internal/client/control.go +++ b/internal/client/control.go @@ -1,7 +1,7 @@ /* * ============================================================================ * Projekt.....: rs2322tcp - * Datei.......: control.go + * Datei.......: internal/client/control.go * Copyright (C) 2026 Dieter Lang * * SPDX-License-Identifier: GPL-3.0-or-later @@ -18,6 +18,7 @@ import ( "bufio" "fmt" "net" + "sync" "git.lang-dieter.de/rs2322tcp/internal/transport" ) @@ -28,6 +29,8 @@ import ( // Client represents a connection to an rs2322tcp server. type Client struct { + mu sync.Mutex + conn net.Conn reader *bufio.Reader } @@ -79,6 +82,9 @@ func (c *Client) Conn() net.Conn { return nil } + c.mu.Lock() + defer c.mu.Unlock() + return c.conn } @@ -87,12 +93,21 @@ func (c *Client) Conn() net.Conn { /////////////////////////////////////////////////////////////////////////////// func (c *Client) hello() error { - if c == nil || c.conn == nil { + if c == nil { + return fmt.Errorf("client is not connected") + } + + c.mu.Lock() + conn := c.conn + reader := c.reader + c.mu.Unlock() + + if conn == nil { return fmt.Errorf("client is not connected") } if err := transport.WriteMessage( - c.conn, + conn, transport.NewHello(), ); err != nil { return fmt.Errorf("send hello: %w", err) @@ -101,7 +116,7 @@ func (c *Client) hello() error { var response transport.HelloResponseMessage if err := transport.ReadMessage( - c.reader, + reader, &response, ); err != nil { return fmt.Errorf("read hello response: %w", err) @@ -125,24 +140,39 @@ func (c *Client) hello() error { // // The returned list contains the session-specific dynamic TCP data ports. func (c *Client) GetDevices() ([]transport.RemoteDeviceInfo, error) { - if c == nil || c.conn == nil { + if c == nil { + return nil, fmt.Errorf("client is not connected") + } + + c.mu.Lock() + conn := c.conn + reader := c.reader + c.mu.Unlock() + + if conn == nil { return nil, fmt.Errorf("client is not connected") } if err := transport.WriteMessage( - c.conn, + conn, transport.NewGetDevices(), ); err != nil { - return nil, fmt.Errorf("send get-devices request: %w", err) + return nil, fmt.Errorf( + "send get-devices request: %w", + err, + ) } var response transport.DeviceListMessage if err := transport.ReadMessage( - c.reader, + reader, &response, ); err != nil { - return nil, fmt.Errorf("read device list: %w", err) + return nil, fmt.Errorf( + "read device list: %w", + err, + ) } if response.Type != transport.MessageDeviceList { @@ -160,13 +190,25 @@ func (c *Client) GetDevices() ([]transport.RemoteDeviceInfo, error) { /////////////////////////////////////////////////////////////////////////////// // Close closes the client control connection. +// +// The connection is removed from the Client while holding the mutex. +// The actual network close is performed afterwards so that another caller +// cannot race with the state change. func (c *Client) Close() error { - if c == nil || c.conn == nil { + if c == nil { return nil } - err := c.conn.Close() + c.mu.Lock() + + conn := c.conn c.conn = nil - return err + c.mu.Unlock() + + if conn == nil { + return nil + } + + return conn.Close() } diff --git a/internal/client/device.go b/internal/client/device.go new file mode 100644 index 0000000..612438e --- /dev/null +++ b/internal/client/device.go @@ -0,0 +1,70 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/device.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Zuordnung eines lokalen virtuellen Ports zu einem vom Server gemeldeten + * Remote-Gerät. + * ============================================================================ + */ +package client + +import ( + "fmt" + + "git.lang-dieter.de/rs2322tcp/internal/config" + "git.lang-dieter.de/rs2322tcp/internal/transport" +) + +/////////////////////////////////////////////////////////////////////////////// +// Remote device lookup +/////////////////////////////////////////////////////////////////////////////// + +// findRemoteDevice finds the server-provided device information matching +// the remote device ID configured for a local virtual port. +// +// The caller uses the returned RemoteDeviceInfo to open the session-specific +// TCP data connection. +func findRemoteDevice( + virtualPort config.VirtualPortConfig, + devices []transport.RemoteDeviceInfo, +) (transport.RemoteDeviceInfo, error) { + if virtualPort.RemoteDevice == "" { + return transport.RemoteDeviceInfo{}, fmt.Errorf( + "remote device is empty for virtual port %q", + virtualPort.Port, + ) + } + + var found *transport.RemoteDeviceInfo + + for i := range devices { + device := &devices[i] + + if device.ID != virtualPort.RemoteDevice { + continue + } + + if found != nil { + return transport.RemoteDeviceInfo{}, fmt.Errorf( + "remote device %q is ambiguous", + virtualPort.RemoteDevice, + ) + } + + found = device + } + + if found == nil { + return transport.RemoteDeviceInfo{}, fmt.Errorf( + "remote device %q not found", + virtualPort.RemoteDevice, + ) + } + + return *found, nil +} diff --git a/internal/client/device_test.go b/internal/client/device_test.go new file mode 100644 index 0000000..2131cda --- /dev/null +++ b/internal/client/device_test.go @@ -0,0 +1,159 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/device_test.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Tests für die Zuordnung lokaler virtueller Ports zu den vom Server + * gemeldeten Remote-Geräten. + * ============================================================================ + */ +package client + +import ( + "strings" + "testing" + + "git.lang-dieter.de/rs2322tcp/internal/config" + "git.lang-dieter.de/rs2322tcp/internal/transport" +) + +/////////////////////////////////////////////////////////////////////////////// +// Test data +/////////////////////////////////////////////////////////////////////////////// + +func testRemoteDevices() []transport.RemoteDeviceInfo { + return []transport.RemoteDeviceInfo{ + { + ID: "radio", + Name: "Radio", + BaudRate: 9600, + DataBits: 8, + Parity: "none", + StopBits: 1, + DataPort: 50123, + }, + { + ID: "rotor", + Name: "Rotor", + BaudRate: 4800, + DataBits: 8, + Parity: "none", + StopBits: 1, + DataPort: 50124, + }, + } +} + +/////////////////////////////////////////////////////////////////////////////// +// Tests +/////////////////////////////////////////////////////////////////////////////// + +func TestFindRemoteDevice(t *testing.T) { + devices := testRemoteDevices() + + virtualPort := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "radio", + } + + got, err := findRemoteDevice(virtualPort, devices) + if err != nil { + t.Fatalf("findRemoteDevice: %v", err) + } + + if got.ID != "radio" { + t.Fatalf("ID = %q, want %q", got.ID, "radio") + } + + if got.DataPort != 50123 { + t.Fatalf("DataPort = %d, want %d", got.DataPort, 50123) + } +} + +func TestFindRemoteDeviceRotor(t *testing.T) { + devices := testRemoteDevices() + + virtualPort := config.VirtualPortConfig{ + Port: "/dev/ttyUSB101", + RemoteDevice: "rotor", + } + + got, err := findRemoteDevice(virtualPort, devices) + if err != nil { + t.Fatalf("findRemoteDevice: %v", err) + } + + if got.ID != "rotor" { + t.Fatalf("ID = %q, want %q", got.ID, "rotor") + } + + if got.DataPort != 50124 { + t.Fatalf("DataPort = %d, want %d", got.DataPort, 50124) + } +} + +func TestFindRemoteDeviceEmptyID(t *testing.T) { + devices := testRemoteDevices() + + virtualPort := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + } + + _, err := findRemoteDevice(virtualPort, devices) + if err == nil { + t.Fatal("expected error for empty remote device") + } + + if !strings.Contains(err.Error(), "remote device is empty") { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestFindRemoteDeviceNotFound(t *testing.T) { + devices := testRemoteDevices() + + virtualPort := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "unknown", + } + + _, err := findRemoteDevice(virtualPort, devices) + if err == nil { + t.Fatal("expected error for unknown remote device") + } + + if !strings.Contains(err.Error(), `remote device "unknown" not found`) { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestFindRemoteDeviceAmbiguous(t *testing.T) { + devices := []transport.RemoteDeviceInfo{ + { + ID: "radio", + DataPort: 50123, + }, + { + ID: "radio", + DataPort: 50125, + }, + } + + virtualPort := config.VirtualPortConfig{ + Port: "/dev/ttyUSB100", + RemoteDevice: "radio", + } + + _, err := findRemoteDevice(virtualPort, devices) + if err == nil { + t.Fatal("expected error for ambiguous remote device") + } + + if !strings.Contains(err.Error(), `remote device "radio" is ambiguous`) { + t.Fatalf("unexpected error: %v", err) + } +} diff --git a/internal/client/pty_linux.go b/internal/client/pty_linux.go index 3438636..a07eca4 100644 --- a/internal/client/pty_linux.go +++ b/internal/client/pty_linux.go @@ -1,19 +1,51 @@ //go:build linux +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/pty_linux.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Linux-spezifische Implementierung der virtuellen seriellen Schnittstelle + * über ein PTY-Paar. + * + * Der PTY-Master wird intern vom Client verwendet. Der zugehörige PTY-Slave + * wird über den virtuellen Port als serielle Schnittstelle bereitgestellt. + * ============================================================================ + */ + package client import ( "fmt" + "sync/atomic" "golang.org/x/sys/unix" ) +/////////////////////////////////////////////////////////////////////////////// +// ptySerial +/////////////////////////////////////////////////////////////////////////////// + +// ptySerial implements VirtualSerial using a Linux PTY pair. +// +// The file descriptors are stored atomically because Read/Write may execute +// concurrently with Close. Close must be able to invalidate the descriptors +// without racing with a concurrent Read or Write. type ptySerial struct { - masterFD int - slaveFD int + masterFD atomic.Int64 + slaveFD atomic.Int64 path string } +/////////////////////////////////////////////////////////////////////////////// +// Constructor +/////////////////////////////////////////////////////////////////////////////// + +// newVirtualSerial creates a new Linux PTY pair. func newVirtualSerial() (VirtualSerial, error) { masterFD, err := unix.Open( "/dev/ptmx", @@ -60,7 +92,11 @@ func newVirtualSerial() (VirtualSerial, error) { 0, ) if err != nil { - return nil, fmt.Errorf("open PTY slave %s: %w", path, err) + return nil, fmt.Errorf( + "open PTY slave %s: %w", + path, + err, + ) } closeSlave := true @@ -74,16 +110,23 @@ func newVirtualSerial() (VirtualSerial, error) { return nil, err } + serial := &ptySerial{ + path: path, + } + + serial.masterFD.Store(int64(masterFD)) + serial.slaveFD.Store(int64(slaveFD)) + closeMaster = false closeSlave = false - return &ptySerial{ - masterFD: masterFD, - slaveFD: slaveFD, - path: path, - }, nil + return serial, nil } +/////////////////////////////////////////////////////////////////////////////// +// PTY configuration +/////////////////////////////////////////////////////////////////////////////// + func configurePTY(fd int) error { termios, err := unix.IoctlGetTermios( fd, @@ -134,33 +177,57 @@ func configurePTY(fd int) error { return nil } +/////////////////////////////////////////////////////////////////////////////// +// VirtualSerial implementation +/////////////////////////////////////////////////////////////////////////////// + +// Path returns the PTY slave path. func (p *ptySerial) Path() string { return p.path } +// Read reads raw bytes from the PTY master. func (p *ptySerial) Read(b []byte) (int, error) { - return unix.Read(p.masterFD, b) + fd := int(p.masterFD.Load()) + + if fd < 0 { + return 0, fmt.Errorf("PTY master is closed") + } + + return unix.Read(fd, b) } +// Write writes raw bytes to the PTY master. func (p *ptySerial) Write(b []byte) (int, error) { - return unix.Write(p.masterFD, b) + fd := int(p.masterFD.Load()) + + if fd < 0 { + return 0, fmt.Errorf("PTY master is closed") + } + + return unix.Write(fd, b) } +// Close closes the PTY master and slave. +// +// The descriptors are invalidated atomically before closing them so that +// concurrent Read or Write calls do not access the descriptor fields +// concurrently with Close. func (p *ptySerial) Close() error { var firstErr error - if p.masterFD >= 0 { - if err := unix.Close(p.masterFD); err != nil { + masterFD := int(p.masterFD.Swap(-1)) + if masterFD >= 0 { + if err := unix.Close(masterFD); err != nil { firstErr = err } - p.masterFD = -1 } - if p.slaveFD >= 0 { - if err := unix.Close(p.slaveFD); err != nil && firstErr == nil { + slaveFD := int(p.slaveFD.Swap(-1)) + if slaveFD >= 0 { + if err := unix.Close(slaveFD); err != nil && firstErr == nil { firstErr = err } - p.slaveFD = -1 } return firstErr diff --git a/internal/client/runtime.go b/internal/client/runtime.go new file mode 100644 index 0000000..d24f7bb --- /dev/null +++ b/internal/client/runtime.go @@ -0,0 +1,216 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/runtime.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Clientseitige Laufzeitverwaltung für die konfigurierten virtuellen + * seriellen Schnittstellen. + * + * Die Runtime verbindet die in der Client-Konfiguration angegebenen + * virtuellen Ports mit den zugehörigen entfernten Geräten. + * + * Für jedes konfigurierte virtuelle Gerät wird eine eigene Bridge gestartet: + * + * /dev/ttyUSBxxx ─── Bridge ─── TCP Data Connection + * + * Die Control-Verbindung zum Server wird gemeinsam von allen Bridges + * verwendet. + * ============================================================================ + */ +package client + +import ( + "fmt" + "sync" + + "git.lang-dieter.de/rs2322tcp/internal/config" +) + +/////////////////////////////////////////////////////////////////////////////// +// Runtime +/////////////////////////////////////////////////////////////////////////////// + +// Runtime manages the client-side virtual serial connections. +// +// A Runtime owns the Client control connection and the VirtualPortManager. +// Each configured virtual port is represented by one running Bridge. +type Runtime struct { + client *Client + manager *VirtualPortManager + + mu sync.Mutex + bridges []*Bridge +} + +// NewRuntime creates a new client Runtime. +// +// The Runtime does not establish a server connection and does not open any +// virtual ports until Run is called. +func NewRuntime( + client *Client, + manager *VirtualPortManager, +) (*Runtime, error) { + if client == nil { + return nil, fmt.Errorf("client is nil") + } + + if manager == nil { + return nil, fmt.Errorf("virtual port manager is nil") + } + + return &Runtime{ + client: client, + manager: manager, + }, nil +} + +/////////////////////////////////////////////////////////////////////////////// +// Run +/////////////////////////////////////////////////////////////////////////////// + +// Run starts all configured virtual serial connections. +// +// The supplied configuration determines which local virtual ports are +// assigned to which remote devices. +// +// Run blocks until one bridge terminates or an error occurs while creating +// one of the configured bridges. +func (r *Runtime) Run( + cfg config.ClientConfig, +) error { + if r == nil || r.client == nil { + return fmt.Errorf("runtime is not initialized") + } + + if r.manager == nil { + return fmt.Errorf("virtual port manager is nil") + } + + if len(cfg.VirtualPorts) == 0 { + return nil + } + devices, err := r.client.GetDevices() + if err != nil { + return fmt.Errorf("get remote devices: %w", err) + } + + for _, virtualPortConfig := range cfg.VirtualPorts { + bridge, _, _, err := connectBridge( + r.client, + r.manager, + virtualPortConfig, + devices, + ) + if err != nil { + r.Close() + + return err + } + + r.mu.Lock() + r.bridges = append(r.bridges, bridge) + r.mu.Unlock() + } + + return r.waitForBridge() +} + +/////////////////////////////////////////////////////////////////////////////// +// Bridge lifecycle +/////////////////////////////////////////////////////////////////////////////// + +// waitForBridge waits until one of the running bridges terminates. +// +// A bridge terminating is considered the end of the Runtime. The Bridge +// itself closes its serial and TCP endpoints when Run returns. +func (r *Runtime) waitForBridge() error { + results := make(chan error, len(r.bridges)) + + r.mu.Lock() + bridges := append([]*Bridge(nil), r.bridges...) + r.mu.Unlock() + + for _, bridge := range bridges { + go func(b *Bridge) { + results <- b.Run() + }(bridge) + } + + err := <-results + + r.Close() + + return err +} + +/////////////////////////////////////////////////////////////////////////////// +// Close +/////////////////////////////////////////////////////////////////////////////// + +// Close closes all resources owned by the Runtime. +// +// Closing the bridge endpoints causes running Bridge.Run calls to terminate. +// The client control connection is closed afterwards. +// +// Calling Close more than once is safe. + +func (r *Runtime) Close() error { + if r == nil { + return nil + } + + r.mu.Lock() + bridges := append([]*Bridge(nil), r.bridges...) + r.bridges = nil + r.mu.Unlock() + + var firstErr error + + for _, bridge := range bridges { + if bridge == nil { + continue + } + + if bridge.serial != nil { + if err := bridge.serial.Close(); err != nil && firstErr == nil { + firstErr = err + } + } + + if bridge.conn != nil { + if err := bridge.conn.Close(); err != nil && firstErr == nil { + firstErr = err + } + } + } + + if r.client != nil { + if err := r.client.Close(); err != nil && firstErr == nil { + firstErr = err + } + } + + return firstErr +} + +// Client returns the Client used by the Runtime. +func (r *Runtime) Client() *Client { + if r == nil { + return nil + } + + return r.client +} + +// Manager returns the VirtualPortManager used by the Runtime. +func (r *Runtime) Manager() *VirtualPortManager { + if r == nil { + return nil + } + + return r.manager +} diff --git a/internal/client/runtime_integration_test.go b/internal/client/runtime_integration_test.go new file mode 100644 index 0000000..e6900e4 --- /dev/null +++ b/internal/client/runtime_integration_test.go @@ -0,0 +1,280 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/runtime_integration_test.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Integrationstest für die clientseitige Runtime mit einem echten + * rs2322tcp-Server und einem virtuellen seriellen Client-Port. + * + * Der Test verwendet die vorhandene virtuelle serielle Test-Schnittstelle + * aus integration_test.go und prüft den vollständigen Datenweg: + * + * /dev/ttyUSB100 + * ↕ + * Bridge + * ↕ + * TCP Data Connection + * ↕ + * rs2322tcp Server + * ↕ + * serielle PTY + * + * Beide Übertragungsrichtungen werden geprüft. + * ============================================================================ + */ +package client_test + +import ( + "net" + "os" + "path/filepath" + "testing" + "time" + + "git.lang-dieter.de/rs2322tcp/internal/client" + "git.lang-dieter.de/rs2322tcp/internal/config" + "git.lang-dieter.de/rs2322tcp/internal/server" +) + +/////////////////////////////////////////////////////////////////////////////// +// Tests +/////////////////////////////////////////////////////////////////////////////// + +func TestClientRuntimeSerialIntegration(t *testing.T) { + serialA, serialB, cleanup := startVirtualSerialPair(t) + defer cleanup() + + const virtualPortPath = "/dev/ttyUSB100" + + home, err := os.UserHomeDir() + if err != nil { + t.Fatalf("get user home directory: %v", err) + } + + internalPortPath := filepath.Join( + home, + ".rs2322tcp", + "virtual", + "ttyUSB100", + ) + + //////////////////////////////////////////////////////////////////////////// + // Server + //////////////////////////////////////////////////////////////////////////// + + controlProbe, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("find free control port: %v", err) + } + + controlPort := controlProbe.Addr().(*net.TCPAddr).Port + + if err := controlProbe.Close(); err != nil { + t.Fatalf("close control probe: %v", err) + } + + serverConfig := &config.ServerConfig{ + Listen: config.ListenConfig{ + Address: "127.0.0.1", + Port: controlPort, + }, + Devices: []config.DeviceConfig{ + { + ID: "radio", + Name: "Funkgerät", + SerialPort: serialA, + BaudRate: 9600, + DataBits: 8, + Parity: "none", + StopBits: 1, + }, + }, + } + + srv, err := server.NewControlServer(serverConfig) + if err != nil { + t.Fatalf("NewControlServer() failed: %v", err) + } + + if err := srv.Listen(); err != nil { + t.Fatalf("server Listen() failed: %v", err) + } + defer srv.Close() + + go func() { + _ = srv.Serve() + }() + + //////////////////////////////////////////////////////////////////////////// + // Client and Runtime + //////////////////////////////////////////////////////////////////////////// + + c, err := client.New(srv.Addr().String()) + if err != nil { + t.Fatalf("client.New() failed: %v", err) + } + defer c.Close() + + manager := client.NewVirtualPortManager( + config.VirtualPortRangeConfig{ + First: 100, + Last: 199, + }, + ) + + runtime, err := client.NewRuntime(c, manager) + if err != nil { + t.Fatalf("NewRuntime() failed: %v", err) + } + defer runtime.Close() + + clientConfig := config.ClientConfig{ + Server: config.ServerConnectionConfig{ + Address: "127.0.0.1", + Port: controlPort, + }, + VirtualPorts: []config.VirtualPortConfig{ + { + Port: virtualPortPath, + RemoteDevice: "radio", + }, + }, + } + + //////////////////////////////////////////////////////////////////////////// + // Start Runtime + //////////////////////////////////////////////////////////////////////////// + + runtimeDone := make(chan error, 1) + + go func() { + runtimeDone <- runtime.Run(clientConfig) + }() + + //////////////////////////////////////////////////////////////////////////// + // Open client-side virtual serial endpoint + //////////////////////////////////////////////////////////////////////////// + + var virtualSerial *os.File + + deadline := time.Now().Add(2 * time.Second) + + for { + virtualSerial, err = os.OpenFile( + internalPortPath, + os.O_RDWR, + 0, + ) + if err == nil { + break + } + + if time.Now().After(deadline) { + t.Fatalf( + "timeout opening virtual serial port %s: %v", + internalPortPath, + err, + ) + } + + time.Sleep(10 * time.Millisecond) + } + + defer virtualSerial.Close() + + //////////////////////////////////////////////////////////////////////////// + // Client virtual serial -> Bridge -> Server -> serial + //////////////////////////////////////////////////////////////////////////// + + clientMessage := []byte("hello from runtime") + + if _, err := virtualSerial.Write(clientMessage); err != nil { + t.Fatalf( + "write to virtual serial port failed: %v", + err, + ) + } + + serialReceived := make([]byte, len(clientMessage)) + + serialPeer, err := os.OpenFile( + serialB, + os.O_RDWR, + 0, + ) + if err != nil { + t.Fatalf( + "open server serial peer: %v", + err, + ) + } + defer serialPeer.Close() + + readExactWithTimeout( + t, + serialPeer, + serialReceived, + 2*time.Second, + ) + + if string(serialReceived) != string(clientMessage) { + t.Fatalf( + "serial received %q, want %q", + string(serialReceived), + string(clientMessage), + ) + } + + //////////////////////////////////////////////////////////////////////////// + // Server serial -> Bridge -> Client virtual serial + //////////////////////////////////////////////////////////////////////////// + + serverMessage := []byte("hello from server") + + if _, err := serialPeer.Write(serverMessage); err != nil { + t.Fatalf( + "write to server serial peer failed: %v", + err, + ) + } + + clientReceived := make([]byte, len(serverMessage)) + + readExactWithTimeout( + t, + virtualSerial, + clientReceived, + 2*time.Second, + ) + + if string(clientReceived) != string(serverMessage) { + t.Fatalf( + "virtual serial received %q, want %q", + string(clientReceived), + string(serverMessage), + ) + } + + /////////////////////////////////////////////////////////////////////////////// + // Terminate Runtime + /////////////////////////////////////////////////////////////////////////////// + + if err := virtualSerial.Close(); err != nil { + t.Fatalf("close virtual serial port: %v", err) + } + + if err := runtime.Close(); err != nil { + t.Fatalf("close runtime: %v", err) + } + + select { + case <-runtimeDone: + // Runtime termination is expected after Runtime.Close(). + case <-time.After(2 * time.Second): + t.Fatal("runtime did not terminate after Runtime.Close()") + } +} diff --git a/internal/client/runtime_test.go b/internal/client/runtime_test.go new file mode 100644 index 0000000..3abcca9 --- /dev/null +++ b/internal/client/runtime_test.go @@ -0,0 +1,82 @@ +/* + * ============================================================================ + * Projekt.....: rs2322tcp + * Datei.......: internal/client/runtime_test.go + * Copyright (C) 2026 Dieter Lang + * + * SPDX-License-Identifier: GPL-3.0-or-later + * + * Beschreibung: + * Tests für die clientseitige Runtime. + * + * Die Tests prüfen die Initialisierung der Runtime und den grundlegenden + * Lebenszyklus ohne einen externen rs2322tcp-Server vorauszusetzen. + * ============================================================================ + */ +package client + +import ( + "strings" + "testing" + + "git.lang-dieter.de/rs2322tcp/internal/config" +) + +/////////////////////////////////////////////////////////////////////////////// +// Tests +/////////////////////////////////////////////////////////////////////////////// + +func TestNewRuntimeNilClient(t *testing.T) { + manager := NewVirtualPortManager( + config.VirtualPortRangeConfig{ + First: 100, + Last: 199, + }, + ) + + _, err := NewRuntime(nil, manager) + if err == nil { + t.Fatal("expected error for nil client") + } + + if !strings.Contains(err.Error(), "client is nil") { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestNewRuntimeNilManager(t *testing.T) { + client := newTestClient(t) + + _, err := NewRuntime(client, nil) + if err == nil { + t.Fatal("expected error for nil manager") + } + + if !strings.Contains(err.Error(), "virtual port manager is nil") { + t.Fatalf("unexpected error: %v", err) + } +} + +func TestRuntimeRunWithoutVirtualPorts(t *testing.T) { + client := newTestClient(t) + + manager := NewVirtualPortManager( + config.VirtualPortRangeConfig{ + First: 100, + Last: 199, + }, + ) + + runtime, err := NewRuntime(client, manager) + if err != nil { + t.Fatalf("NewRuntime: %v", err) + } + + cfg := config.ClientConfig{ + VirtualPorts: nil, + } + + if err := runtime.Run(cfg); err != nil { + t.Fatalf("Run: %v", err) + } +} diff --git a/internal/client/virtual_port_manager.go b/internal/client/virtual_port_manager.go index 071cfa5..8998617 100644 --- a/internal/client/virtual_port_manager.go +++ b/internal/client/virtual_port_manager.go @@ -15,6 +15,7 @@ package client import ( "fmt" + "sync" "git.lang-dieter.de/rs2322tcp/internal/config" ) @@ -25,15 +26,23 @@ import ( // VirtualPortManager verwaltet den reservierten Bereich virtueller // /dev/ttyUSBxxx-Schnittstellen. +// +// Der reservierte Portbereich und die Belegungstabelle werden durch einen +// Mutex geschützt, da mehrere Bridges beziehungsweise Goroutinen gleichzeitig +// virtuelle Ports öffnen und freigeben können. type VirtualPortManager struct { + mu sync.Mutex + first int last int used map[int]bool } -// NewVirtualPortManager erzeugt einen VirtualPortManager aus der +// NewVirtualPortManager erzeugt einen VirtualPortManager aus dem // konfigurierten Portbereich. -func NewVirtualPortManager(cfg config.VirtualPortRangeConfig) *VirtualPortManager { +func NewVirtualPortManager( + cfg config.VirtualPortRangeConfig, +) *VirtualPortManager { return &VirtualPortManager{ first: cfg.First, last: cfg.Last, @@ -52,6 +61,7 @@ func (m *VirtualPortManager) PortPath(number int) string { } // Reserve reserviert den nächsten freien virtuellen Port. +// // Es wird immer mit dem kleinsten freien Port im konfigurierten Bereich // begonnen. func (m *VirtualPortManager) Reserve() (int, error) { @@ -59,12 +69,16 @@ func (m *VirtualPortManager) Reserve() (int, error) { return 0, fmt.Errorf("virtual port manager is nil") } + m.mu.Lock() + defer m.mu.Unlock() + for number := m.first; number <= m.last; number++ { if m.used[number] { continue } m.used[number] = true + return number, nil } @@ -81,5 +95,8 @@ func (m *VirtualPortManager) Release(number int) { return } + m.mu.Lock() + defer m.mu.Unlock() + delete(m.used, number) } diff --git a/internal/client/virtual_port_manager_linux.go b/internal/client/virtual_port_manager_linux.go index 26ed6ba..34812f0 100644 --- a/internal/client/virtual_port_manager_linux.go +++ b/internal/client/virtual_port_manager_linux.go @@ -18,7 +18,10 @@ */ package client -import "fmt" +import ( + "fmt" + "sync" +) /////////////////////////////////////////////////////////////////////////////// // Managed virtual port @@ -26,7 +29,12 @@ import "fmt" // ManagedVirtualPort verbindet einen reservierten Port mit einer // VirtualSerial-Instanz und verwaltet deren Lebenszyklus. +// +// Der Mutex schützt den internen Port-Pointer gegen gleichzeitige Zugriffe +// aus Read, Write und Close. type ManagedVirtualPort struct { + mu sync.Mutex + port *VirtualPort manager *VirtualPortManager number int @@ -34,7 +42,14 @@ type ManagedVirtualPort struct { // Path returns the device path visible to the external application. func (p *ManagedVirtualPort) Path() string { - if p == nil || p.port == nil { + if p == nil { + return "" + } + + p.mu.Lock() + defer p.mu.Unlock() + + if p.port == nil { return "" } @@ -43,45 +58,77 @@ func (p *ManagedVirtualPort) Path() string { // Read reads data from the underlying virtual serial device. func (p *ManagedVirtualPort) Read(b []byte) (int, error) { - if p == nil || p.port == nil { + if p == nil { return 0, fmt.Errorf("managed virtual port is nil") } - return p.port.Read(b) + p.mu.Lock() + port := p.port + p.mu.Unlock() + + if port == nil { + return 0, fmt.Errorf("managed virtual port is closed") + } + + return port.Read(b) } // Write writes data to the underlying virtual serial device. func (p *ManagedVirtualPort) Write(b []byte) (int, error) { - if p == nil || p.port == nil { + if p == nil { return 0, fmt.Errorf("managed virtual port is nil") } - return p.port.Write(b) + p.mu.Lock() + port := p.port + p.mu.Unlock() + + if port == nil { + return 0, fmt.Errorf("managed virtual port is closed") + } + + return port.Write(b) } // Close removes the internal link, closes the PTY and releases the // reservation. +// +// The managed port is detached from the object before the underlying PTY +// is closed. This prevents a concurrent Read or Write from accessing the +// managed port pointer after Close has taken ownership of it. func (p *ManagedVirtualPort) Close() error { - if p == nil || p.port == nil { + if p == nil { + return nil + } + + p.mu.Lock() + + port := p.port + p.port = nil + + manager := p.manager + number := p.number + + p.mu.Unlock() + + if port == nil { return nil } var firstErr error - if err := removeVirtualPortLink(p.port.Path()); err != nil { + if err := removeVirtualPortLink(port.Path()); err != nil { firstErr = err } - if err := p.port.Close(); err != nil && firstErr == nil { + if err := port.Close(); err != nil && firstErr == nil { firstErr = err } - if p.manager != nil { - p.manager.Release(p.number) + if manager != nil { + manager.Release(number) } - p.port = nil - return firstErr } @@ -108,6 +155,7 @@ func (m *VirtualPortManager) Open() (*ManagedVirtualPort, error) { serial, err := newVirtualSerial() if err != nil { m.Release(number) + return nil, fmt.Errorf( "create virtual serial for %s: %w", portPath, @@ -115,7 +163,10 @@ func (m *VirtualPortManager) Open() (*ManagedVirtualPort, error) { ) } - if err := setVirtualPortLink(portPath, serial.Path()); err != nil { + if err := setVirtualPortLink( + portPath, + serial.Path(), + ); err != nil { _ = serial.Close() m.Release(number)