--- /dev/null
+package m31
+
+import (
+ "bytes"
+ "context"
+ "encoding/binary"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "time"
+)
+
+const (
+ connectTimeout = 2 * time.Second
+ writeTimeout = time.Second
+ readTimeout = 2 * time.Second
+)
+
+type tcpClient struct {
+ conn net.Conn
+ unit byte
+ transaction uint16
+}
+
+// NewTCPClient supplies Modbus TCP functions 01, 02, 03, 05, 15, and 16 without
+// opening a connection. Connect has a two-second timeout. Writes have a
+// one-second timeout, and reads have a two-second timeout. Earlier caller
+// deadlines apply to each operation. The caller owns Close.
+func NewTCPClient() Client {
+ return &tcpClient{}
+}
+
+func (c *tcpClient) Connect(ctx context.Context, address string, unit byte) error {
+ if c.conn != nil {
+ return fmt.Errorf("modbus connection is already open")
+ }
+
+ if unit == 0 || unit > 247 {
+ return fmt.Errorf("modbus unit must be from one through 247")
+ }
+
+ ctx, cancel := context.WithTimeout(ctx, connectTimeout)
+ defer cancel()
+
+ conn, err := (&net.Dialer{}).DialContext(ctx, "tcp", address)
+ if err != nil {
+ return fmt.Errorf("connect Modbus transport: %w", err)
+ }
+
+ c.conn, c.unit = conn, unit
+
+ return nil
+}
+
+func (c *tcpClient) Close() error {
+ if c.conn == nil {
+ return nil
+ }
+
+ err := c.conn.Close()
+ c.conn = nil
+ if err != nil {
+ return fmt.Errorf("close Modbus transport: %w", err)
+ }
+
+ return nil
+}
+
+func (c *tcpClient) ReadInputs(ctx context.Context, count uint16) ([]bool, error) {
+ return c.readBits(ctx, 2, count)
+}
+
+func (c *tcpClient) ReadCoils(ctx context.Context, count uint16) ([]bool, error) {
+ return c.readBits(ctx, 1, count)
+}
+
+func (c *tcpClient) readBits(ctx context.Context, function byte, count uint16) ([]bool, error) {
+ if count == 0 || count > 2000 {
+ return nil, fmt.Errorf("bit read count must be from one through 2000")
+ }
+
+ data, err := c.exchange(ctx, function, words(0, count))
+ if err != nil {
+ return nil, fmt.Errorf("read Modbus bits: %w", err)
+ }
+
+ size := (int(count) + 7) / 8
+ if len(data) != size+1 || int(data[0]) != size {
+ return nil, fmt.Errorf("modbus bit response has an invalid length")
+ }
+
+ values := make([]bool, count)
+
+ for index := range values {
+ values[index] = data[1+index/8]&(1<<uint(index%8)) != 0
+ }
+
+ return values, nil
+}
+
+func (c *tcpClient) ReadRegisters(ctx context.Context, address uint16, count uint16) ([]uint16, error) {
+ if count == 0 || count > 125 || uint32(address)+uint32(count) > 65536 {
+ return nil, fmt.Errorf("invalid register read range")
+ }
+
+ data, err := c.exchange(ctx, 3, words(address, count))
+ if err != nil {
+ return nil, fmt.Errorf("read Modbus registers: %w", err)
+ }
+
+ if len(data) != int(count)*2+1 || int(data[0]) != int(count)*2 {
+ return nil, fmt.Errorf("modbus register response has an invalid length")
+ }
+
+ values := make([]uint16, count)
+
+ for index := range values {
+ values[index] = binary.BigEndian.Uint16(data[1+2*index:])
+ }
+
+ return values, nil
+}
+
+func (c *tcpClient) WriteCoil(ctx context.Context, address uint16, value bool) error {
+ var encoded uint16
+
+ if value {
+ encoded = 0xff00
+ }
+
+ return c.write(ctx, 5, words(address, encoded), words(address, encoded))
+}
+
+func (c *tcpClient) WriteCoils(ctx context.Context, address uint16, values []bool) error {
+ if len(values) == 0 || len(values) > 1968 || uint32(address)+uint32(len(values)) > 65536 {
+ return fmt.Errorf("invalid coil write range")
+ }
+
+ data := append(words(address, uint16(len(values))), byte((len(values)+7)/8))
+ data = append(data, make([]byte, (len(values)+7)/8)...)
+
+ for index, value := range values {
+ if value {
+ data[5+index/8] |= 1 << uint(index%8)
+ }
+ }
+
+ return c.write(ctx, 15, data, words(address, uint16(len(values))))
+}
+
+func (c *tcpClient) WriteRegister(ctx context.Context, address uint16, value uint16) error {
+ data := append(words(address, 1), 2)
+ data = append(data, words(value)...)
+
+ return c.write(ctx, 16, data, words(address, 1))
+}
+
+func (c *tcpClient) write(ctx context.Context, function byte, request []byte, expected []byte) error {
+ data, err := c.exchange(ctx, function, request)
+ if err != nil {
+ return fmt.Errorf("write Modbus value: %w", err)
+ }
+
+ if !bytes.Equal(data, expected) {
+ return fmt.Errorf("modbus write acknowledgement differs from request")
+ }
+
+ return nil
+}
+
+func (c *tcpClient) exchange(ctx context.Context, function byte, data []byte) (response []byte, err error) {
+ if ctx.Err() != nil {
+ return nil, ctx.Err()
+ }
+
+ if c.conn == nil {
+ return nil, fmt.Errorf("modbus transport is closed")
+ }
+
+ c.transaction++
+ request := append(words(c.transaction, 0, uint16(len(data)+2)), c.unit, function)
+ request = append(request, data...)
+ err = c.conn.SetWriteDeadline(operationDeadline(ctx, writeTimeout))
+ if err != nil {
+ return nil, fmt.Errorf("set Modbus write deadline: %w", err)
+ }
+
+ err = c.conn.SetReadDeadline(operationDeadline(ctx, readTimeout))
+ if err != nil {
+ return nil, fmt.Errorf("set Modbus read deadline: %w", err)
+ }
+
+ interrupted := make(chan struct{})
+
+ var interruptErr error
+
+ stop := context.AfterFunc(ctx, func() {
+ interruptErr = c.conn.SetDeadline(time.Now())
+
+ close(interrupted)
+ })
+ defer func() {
+ if !stop() {
+ <-interrupted
+
+ err = errors.Join(err, interruptErr, ctx.Err())
+ }
+ }()
+
+ count, err := c.conn.Write(request)
+ if err != nil {
+ return nil, fmt.Errorf("write Modbus request: %w", err)
+ }
+
+ if count != len(request) {
+ return nil, fmt.Errorf("write Modbus request: %w", io.ErrShortWrite)
+ }
+
+ var header [7]byte
+
+ _, err = io.ReadFull(c.conn, header[:])
+ if err != nil {
+ return nil, fmt.Errorf("read Modbus header: %w", err)
+ }
+
+ length := binary.BigEndian.Uint16(header[4:6])
+ if binary.BigEndian.Uint16(header[:2]) != c.transaction || binary.BigEndian.Uint16(header[2:4]) != 0 || header[6] != c.unit || length < 2 || length > 254 {
+ return nil, fmt.Errorf("invalid Modbus response header")
+ }
+
+ body := make([]byte, length-1)
+ _, err = io.ReadFull(c.conn, body)
+ if err != nil {
+ return nil, fmt.Errorf("read Modbus body: %w", err)
+ }
+
+ if body[0] == function|0x80 && len(body) == 2 {
+ return nil, fmt.Errorf("modbus exception %d", body[1])
+ }
+
+ if body[0] != function {
+ return nil, fmt.Errorf("invalid Modbus response function")
+ }
+
+ return body[1:], nil
+}
+
+func words(values ...uint16) []byte {
+ data := make([]byte, 2*len(values))
+
+ for index, value := range values {
+ binary.BigEndian.PutUint16(data[2*index:], value)
+ }
+
+ return data
+}
+
+func operationDeadline(ctx context.Context, timeout time.Duration) time.Time {
+ deadline := time.Now().Add(timeout)
+ caller, exists := ctx.Deadline()
+ if exists && caller.Before(deadline) {
+ return caller
+ }
+
+ return deadline
+}
--- /dev/null
+package m31
+
+import (
+ "context"
+ "encoding/binary"
+ "errors"
+ "io"
+ "net"
+ "testing"
+ "testing/synctest"
+ "time"
+
+ "github.com/stretchr/testify/require"
+)
+
+func TestTCPClientEncodesAndChecksModbusOperations(t *testing.T) {
+ testCases := []struct {
+ name string
+ call func(*tcpClient) error
+ request []byte
+ reply []byte
+ }{
+ {
+ name: "reads low bit first across byte boundaries",
+ call: func(c *tcpClient) error {
+ value, err := c.ReadInputs(t.Context(), 9)
+ require.Equal(t, []bool{true, false, true, false, false, false, false, false, true}, value)
+
+ return err
+ },
+ request: []byte{2, 0, 0, 0, 9},
+ reply: []byte{2, 2, 5, 1},
+ },
+ {
+ name: "reads coil state",
+ call: func(c *tcpClient) error {
+ value, err := c.ReadCoils(t.Context(), 1)
+ require.Equal(t, []bool{true}, value)
+
+ return err
+ },
+ request: []byte{1, 0, 0, 0, 1},
+ reply: []byte{1, 1, 1},
+ },
+ {
+ name: "reads full sixteen bit registers",
+ call: func(c *tcpClient) error {
+ value, err := c.ReadRegisters(t.Context(), 8, 2)
+ require.Equal(t, []uint16{10000, 65535}, value)
+
+ return err
+ },
+ request: []byte{3, 0, 8, 0, 2},
+ reply: []byte{3, 4, 0x27, 0x10, 0xff, 0xff},
+ },
+ {
+ name: "writes true with function five",
+ call: func(c *tcpClient) error {
+ return c.WriteCoil(t.Context(), 7, true)
+ },
+ request: []byte{5, 0, 7, 0xff, 0},
+ reply: []byte{5, 0, 7, 0xff, 0},
+ },
+ {
+ name: "writes false with function five",
+ call: func(c *tcpClient) error {
+ return c.WriteCoil(t.Context(), 7, false)
+ },
+ request: []byte{5, 0, 7, 0, 0},
+ reply: []byte{5, 0, 7, 0, 0},
+ },
+ {
+ name: "writes a packed contiguous coil range",
+ call: func(c *tcpClient) error {
+ return c.WriteCoils(t.Context(), 6, []bool{true, false, true, false, false, false, false, false, true})
+ },
+ request: []byte{15, 0, 6, 0, 9, 2, 5, 1},
+ reply: []byte{15, 0, 6, 0, 9},
+ },
+ {
+ name: "writes a holding register with function sixteen",
+ call: func(c *tcpClient) error {
+ return c.WriteRegister(t.Context(), 0x30d4, 200)
+ },
+ request: []byte{16, 0x30, 0xd4, 0, 1, 2, 0, 200},
+ reply: []byte{16, 0x30, 0xd4, 0, 1},
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ request, err := wireCall(t, tc.call, func(header []byte) []byte {
+ return append(append(header[:4:4], words(uint16(len(tc.reply)+1))...), append([]byte{7}, tc.reply...)...)
+ })
+ require.NoError(t, err)
+ require.Equal(t, append([]byte{0, 1, 0, 0, 0, byte(len(tc.request) + 1), 7}, tc.request...), request)
+ })
+ }
+}
+
+func TestTCPClientRejectsInvalidAndTruncatedResponses(t *testing.T) {
+ testCases := []struct {
+ name string
+ reply []byte
+ }{
+ {name: "rejects another transaction", reply: []byte{0, 2, 0, 0, 0, 4, 7, 1, 1, 0}},
+ {name: "rejects another protocol", reply: []byte{0, 1, 0, 1, 0, 4, 7, 1, 1, 0}},
+ {name: "rejects another unit", reply: []byte{0, 1, 0, 0, 0, 4, 8, 1, 1, 0}},
+ {name: "rejects empty PDU", reply: []byte{0, 1, 0, 0, 0, 1, 7}},
+ {name: "rejects excessive PDU", reply: []byte{0, 1, 0, 0, 0, 255, 7}},
+ {name: "rejects truncated header", reply: []byte{0, 1}},
+ {name: "rejects truncated body", reply: []byte{0, 1, 0, 0, 0, 4, 7, 1}},
+ {name: "rejects wrong function", reply: []byte{0, 1, 0, 0, 0, 4, 7, 2, 1, 0}},
+ {name: "rejects exception", reply: []byte{0, 1, 0, 0, 0, 3, 7, 0x81, 2}},
+ {name: "rejects wrong byte count", reply: []byte{0, 1, 0, 0, 0, 4, 7, 1, 2, 0}},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ _, err := wireCall(t, func(c *tcpClient) error {
+ _, err := c.ReadCoils(t.Context(), 1)
+ return err
+ }, func([]byte) []byte {
+ return tc.reply
+ })
+ require.Error(t, err)
+ })
+ }
+}
+
+func TestTCPClientRejectsInvalidRangesAndAcknowledgements(t *testing.T) {
+ c := &tcpClient{}
+ _, err := c.ReadInputs(t.Context(), 0)
+ require.Error(t, err)
+
+ _, err = c.ReadCoils(t.Context(), 2001)
+ require.Error(t, err)
+
+ _, err = c.ReadRegisters(t.Context(), 0, 0)
+ require.Error(t, err)
+
+ _, err = c.ReadRegisters(t.Context(), 0, 126)
+ require.Error(t, err)
+
+ _, err = c.ReadRegisters(t.Context(), 65535, 2)
+ require.Error(t, err)
+ require.Error(t, c.WriteCoils(t.Context(), 0, nil))
+ require.Error(t, c.WriteCoils(t.Context(), 65535, []bool{true, true}))
+ require.Error(t, c.WriteCoils(t.Context(), 0, make([]bool, 1969)))
+ require.Error(t, c.WriteCoil(t.Context(), 0, true))
+ require.NoError(t, c.Close())
+ require.Error(t, c.Connect(t.Context(), "", 0))
+ require.Error(t, c.Connect(t.Context(), "", 248))
+ require.Error(t, c.Connect(t.Context(), "", 1))
+
+ _, err = wireCall(t, func(c *tcpClient) error {
+ return c.WriteRegister(t.Context(), 0, 20)
+ }, func(header []byte) []byte {
+ return append(header[:7:7], 16, 0, 0, 0, 2)
+ })
+ require.Error(t, err)
+
+ _, err = wireCall(t, func(c *tcpClient) error {
+ _, readErr := c.ReadRegisters(t.Context(), 0, 1)
+ return readErr
+ }, func(header []byte) []byte {
+ return append(header[:7:7], 3, 3, 0, 0, 0)
+ })
+ require.Error(t, err)
+}
+
+func TestTCPClientCancellationAndTimeouts(t *testing.T) {
+ testCases := []struct {
+ name string
+ cancel bool
+ write bool
+ }{
+ {name: "cancels a blocked write", cancel: true, write: true},
+ {name: "cancels a blocked read", cancel: true},
+ {name: "limits a blocked write", write: true},
+ {name: "limits a blocked read"},
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ synctest.Test(t, func(t *testing.T) {
+ conn, peer := net.Pipe()
+ c := &tcpClient{conn: conn, unit: 1}
+ t.Cleanup(func() {
+ require.NoError(t, c.Close())
+ require.NoError(t, peer.Close())
+ })
+
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+
+ result := make(chan error, 1)
+
+ if !tc.write {
+ go func() {
+ var request [12]byte
+
+ _, err := io.ReadFull(peer, request[:])
+ result <- err
+ }()
+ }
+
+ done := make(chan error, 1)
+ go func() {
+ _, err := c.ReadCoils(ctx, 1)
+ done <- err
+ }()
+
+ synctest.Wait()
+
+ if !tc.write {
+ require.NoError(t, <-result)
+ }
+
+ if tc.cancel {
+ cancel()
+ } else {
+ time.Sleep(2 * time.Second)
+ }
+
+ err := <-done
+ require.Error(t, err)
+
+ if tc.cancel {
+ require.ErrorIs(t, err, context.Canceled)
+ }
+ })
+ })
+ }
+}
+
+func wireCall(t *testing.T, call func(*tcpClient) error, response func([]byte) []byte) ([]byte, error) {
+ t.Helper()
+
+ conn, peer := net.Pipe()
+ c := &tcpClient{conn: conn, unit: 7}
+ t.Cleanup(func() {
+ require.NoError(t, c.Close())
+ require.NoError(t, peer.Close())
+ })
+
+ requests := make(chan []byte, 1)
+ done := make(chan error, 1)
+
+ go func() {
+ var header [7]byte
+
+ _, err := io.ReadFull(peer, header[:])
+ if err != nil {
+ done <- err
+ return
+ }
+
+ body := make([]byte, int(binary.BigEndian.Uint16(header[4:6]))-1)
+ _, err = io.ReadFull(peer, body)
+ requests <- append(header[:7:7], body...)
+ if err == nil {
+ _, err = peer.Write(response(header[:]))
+ }
+
+ if errors.Is(err, io.ErrClosedPipe) {
+ err = nil
+ }
+
+ err = errors.Join(err, peer.Close())
+ done <- err
+ }()
+
+ err := call(c)
+ require.NoError(t, c.Close())
+ require.NoError(t, <-done)
+
+ return <-requests, err
+}
projects / onehouse / commitdiff