feat: introduce serial link recovery

ascii_transport_test.go

@@ -0,0 +1,89 @@
+//go:build darwin || linux || freebsd || openbsd || netbsd
+// +build darwin linux freebsd openbsd netbsd
+
+// Copyright 2014 Quoc-Viet Nguyen. All rights reserved.
+// This software may be modified and distributed under the terms
+// of the BSD license. See the LICENSE file for details.
+
+package modbus
+
+import (
+	"bytes"
+	"context"
+	"testing"
+	"time"
+)
+
+func TestASCIISerialTransporter_Send_PTY(t *testing.T) {
+	master, slavePath, err := openPTY()
+	if err != nil {
+		t.Skipf("Skipping PTY test: %v", err)
+	}
+	defer master.Close()
+
+	// Request: 01 03 00 00 00 01 (Read Holding Registers)
+	// ASCII: :010300000001FB\r\n
+	reqASCII := []byte(":010300000001FB\r\n")
+
+	// Response: 01 03 02 00 00
+	// ASCII: :0103020000FA\r\n
+	respASCII := []byte(":0103020000FA\r\n")
+
+	transporter := &asciiSerialTransporter{}
+	transporter.Address = slavePath
+	transporter.BaudRate = 19200
+	transporter.Timeout = 1 * time.Second
+	transporter.IdleTimeout = serialIdleTimeout
+
+	// Start a goroutine to read request and write response to master
+	go func() {
+		buf := make([]byte, 1024)
+		n, err := master.Read(buf)
+		if err != nil {
+			return
+		}
+		if !bytes.Equal(buf[:n], reqASCII) {
+			// t.Errorf would be racy here, just log or ignore
+			return
+		}
+		// Write response
+		_, err = master.Write(respASCII)
+		if err != nil {
+			t.Errorf("Failed to write response: %v", err)
+		}
+	}()
+
+	ctx := context.Background()
+	aduResponse, err := transporter.Send(ctx, reqASCII)
+	if err != nil {
+		t.Fatalf("Send failed: %v", err)
+	}
+
+	if !bytes.Equal(aduResponse, respASCII) {
+		t.Errorf("Expected response %s, got %s", respASCII, aduResponse)
+	}
+}
+
+func TestASCIISerialTransporter_Timeout_PTY(t *testing.T) {
+	master, slavePath, err := openPTY()
+	if err != nil {
+		t.Skipf("Skipping PTY test: %v", err)
+	}
+	defer master.Close()
+
+	reqASCII := []byte(":010300000001FB\r\n")
+
+	transporter := &asciiSerialTransporter{}
+	transporter.Address = slavePath
+	transporter.BaudRate = 19200
+	transporter.Timeout = 100 * time.Millisecond
+	transporter.IdleTimeout = serialIdleTimeout
+
+	// Don't write anything to master
+
+	ctx := context.Background()
+	_, err = transporter.Send(ctx, reqASCII)
+	if err == nil {
+		t.Fatal("Expected timeout error, got nil")
+	}
+}

asciiclient.go

@@ -9,6 +9,8 @@ import (
 	"context"
 	"encoding/hex"
 	"fmt"
+	"io"
+	"syscall"
 	"time"
 )
 
@@ -181,17 +183,76 @@ func (mb *asciiSerialTransporter) Send(ctx context.Context, aduRequest []byte) (
 	mb.lastActivity = time.Now()
 	mb.startCloseTimer()
 
-	// Send the request
-	mb.logf("modbus: send % x\n", aduRequest)
-	if _, err = mb.port.Write(aduRequest); err != nil {
+	connDeadline := time.Now().Add(mb.Timeout)
+	linkRecoveryDeadline := time.Now().Add(mb.LinkRecoveryTimeout)
+
+	for {
+		// Send the request
+		mb.logf("modbus: send % x\n", aduRequest)
+		if _, err = mb.port.Write(aduRequest); err != nil {
+			if err == io.EOF || err == io.ErrUnexpectedEOF || err == syscall.ECONNRESET {
+				if time.Now().After(linkRecoveryDeadline) {
+					err = fmt.Errorf("modbus: link recovery timeout reached: %w", err)
+					return
+				}
+				// reconnect on connection reset
+				mb.logf("modbus: connection reset, reconnecting")
+				if cerr := mb.close(); cerr != nil {
+					mb.logf("modbus: error closing connection: %v", cerr)
+					return
+				}
+				if cerr := mb.connect(ctx); cerr != nil {
+					mb.logf("modbus: error reconnecting: %v", cerr)
+					return
+				}
+				// retry the communication
+				continue
+			}
+
+			return
+		}
+		// Get the response
+		aduResponse, err = readASCII(mb.port, connDeadline)
+		mb.logf("modbus: recv % x\n", aduResponse)
+		if err != nil {
+			if err == io.EOF || err == io.ErrUnexpectedEOF || err == syscall.ECONNRESET {
+				if time.Now().After(linkRecoveryDeadline) {
+					err = fmt.Errorf("modbus: link recovery timeout reached: %w", err)
+					return
+				}
+				// reconnect on connection reset
+				mb.logf("modbus: connection reset, reconnecting")
+				if cerr := mb.close(); cerr != nil {
+					mb.logf("modbus: error closing connection: %v", cerr)
+					return
+				}
+				if cerr := mb.connect(ctx); cerr != nil {
+					mb.logf("modbus: error reconnecting: %v", cerr)
+					return
+				}
+				// retry the communication
+				continue
+			}
+			// Unknown error
+			mb.logf("modbus: read error: %v", err)
+			return
+		}
+
 		return
 	}
-	// Get the response
+}
+
+func readASCII(r io.Reader, deadline time.Time) ([]byte, error) {
 	var n, length int
 	var data [asciiMaxSize]byte
+	var err error
+
 	for {
-		if n, err = mb.port.Read(data[length:]); err != nil {
-			return
+		if time.Now().After(deadline) {
+			return nil, context.DeadlineExceeded
+		}
+		if n, err = r.Read(data[length:]); err != nil {
+			return nil, err
 		}
 		length += n
 		if length >= asciiMaxSize || n == 0 {
@@ -204,9 +265,8 @@ func (mb *asciiSerialTransporter) Send(ctx context.Context, aduRequest []byte) (
 			}
 		}
 	}
-	aduResponse = data[:length]
-	mb.logf("modbus: recv % x\n", aduResponse)
-	return
+
+	return data[:length], nil
 }
 
 // writeHex encodes byte to string in hexadecimal, e.g. 0xA5 => "A5"

rtu_transport_test.go

@@ -0,0 +1,117 @@
+//go:build darwin || linux || freebsd || openbsd || netbsd
+// +build darwin linux freebsd openbsd netbsd
+
+// Copyright 2014 Quoc-Viet Nguyen. All rights reserved.
+// This software may be modified and distributed under the terms
+// of the BSD license. See the LICENSE file for details.
+
+package modbus
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"os"
+	"syscall"
+	"testing"
+	"time"
+	"unsafe"
+)
+
+// openPTY opens a PTY pair and returns the master file and the slave path.
+func openPTY() (master *os.File, slavePath string, err error) {
+	master, err = os.OpenFile("/dev/ptmx", os.O_RDWR, 0)
+	if err != nil {
+		return nil, "", err
+	}
+
+	// unlockpt
+	var unlock int32
+	// TIOCSPTLCK
+	_, _, errno := syscall.Syscall(syscall.SYS_IOCTL, master.Fd(), syscall.TIOCSPTLCK, uintptr(unsafe.Pointer(&unlock)))
+	if errno != 0 {
+		master.Close()
+		return nil, "", errno
+	}
+
+	// ptsname
+	var ptyno int32
+	// TIOCGPTN
+	_, _, errno = syscall.Syscall(syscall.SYS_IOCTL, master.Fd(), syscall.TIOCGPTN, uintptr(unsafe.Pointer(&ptyno)))
+	if errno != 0 {
+		master.Close()
+		return nil, "", errno
+	}
+
+	slavePath = fmt.Sprintf("/dev/pts/%d", ptyno)
+	return master, slavePath, nil
+}
+
+func TestRTUSerialTransporter_Send_PTY(t *testing.T) {
+	master, slavePath, err := openPTY()
+	if err != nil {
+		t.Skipf("Skipping PTY test: %v", err)
+	}
+	defer master.Close()
+
+	// Request: 01 03 00 00 00 01 84 0A (Read Holding Registers)
+	// Response: 01 03 02 00 00 B8 44
+	req := []byte{0x01, 0x03, 0x00, 0x00, 0x00, 0x01, 0x84, 0x0A}
+	resp := []byte{0x01, 0x03, 0x02, 0x00, 0x00, 0xB8, 0x44}
+
+	transporter := &rtuSerialTransporter{}
+	transporter.Address = slavePath
+	transporter.BaudRate = 19200
+	transporter.Timeout = 1 * time.Second
+
+	// Start a goroutine to read request and write response to master
+	go func() {
+		buf := make([]byte, 1024)
+		n, err := master.Read(buf)
+		if err != nil {
+			return
+		}
+		if !bytes.Equal(buf[:n], req) {
+			// t.Errorf would be racy here, just log or ignore
+			return
+		}
+		// Write response
+		_, err = master.Write(resp)
+		if err != nil {
+			t.Errorf("Failed to write response: %v", err)
+		}
+	}()
+
+	ctx := context.Background()
+	aduResponse, err := transporter.Send(ctx, req)
+	if err != nil {
+		t.Fatalf("Send failed: %v", err)
+	}
+
+	if !bytes.Equal(aduResponse, resp) {
+		t.Errorf("Expected response %x, got %x", resp, aduResponse)
+	}
+}
+
+func TestRTUSerialTransporter_Timeout_PTY(t *testing.T) {
+	master, slavePath, err := openPTY()
+	if err != nil {
+		t.Skipf("Skipping PTY test: %v", err)
+	}
+	defer master.Close()
+
+	req := []byte{0x01, 0x03, 0x00, 0x00, 0x00, 0x01, 0x84, 0x0A}
+
+	transporter := &rtuSerialTransporter{}
+	transporter.Address = slavePath
+	transporter.BaudRate = 19200
+	transporter.Timeout = 100 * time.Millisecond
+
+	// Don't write anything to master
+
+	ctx := context.Background()
+	_, err = transporter.Send(ctx, req)
+	if err == nil {
+		t.Fatal("Expected timeout error, got nil")
+	}
+}