jl-iot/protocol/modbus/client.go

package modbus

import (
	"context"
	"fmt"
	"log/slog"
	"net"
	"os"
	"time"

	"github.com/simonvetter/modbus"
)

type ConnectState int

const ()

type ClientConfig struct {
	Url     string
	Timeout uint32 // 超时时间,单位ms
}

type client struct {
	url       string
	cli       *modbus.ModbusClient
	started   bool // 是否启动
	connected bool // 是否连接
	cancel    context.CancelFunc
}

func newClient(conf *ClientConfig) (MasterClient, error) {
	cli, err := modbus.NewClient(&modbus.ClientConfiguration{
		URL:     conf.Url,
		Timeout: time.Duration(conf.Timeout) * time.Millisecond,
	})
	if err != nil {
		return nil, err
	}
	c := &client{
		url:       conf.Url,
		cli:       cli,
		connected: false,
	}
	ctx, cancel := context.WithCancel(context.Background())
	go c.connManage(ctx)
	c.cancel = cancel
	return c, err
}

func (c *client) SetUnitId(id uint8) error {
	return c.cli.SetUnitId(id)
}

func (c *client) SetEndianness(endianness Endianness) error {
	switch endianness {
	case BigEndian:
		return c.cli.SetEncoding(modbus.BIG_ENDIAN, modbus.HIGH_WORD_FIRST)
	case LittleEndian:
		return c.cli.SetEncoding(modbus.LITTLE_ENDIAN, modbus.HIGH_WORD_FIRST)
	default:
		return fmt.Errorf("unknown endianness value %v", endianness)
	}
}

func (c *client) Start() error {
	c.started = true
	return nil
}

func (c *client) Stop() error {
	c.started = false
	return nil
}

// 客户端连接管理
func (c *client) connManage(ctx context.Context) {
	for {
		select {
		case <-ctx.Done():
			slog.Info("modbus客户端连接管理线程退出", "url", c.url)
			return
		default:
		}
		if c.started && !c.connected { // 已经启动, 尝试重连
			err := c.cli.Open()
			if err != nil {
				slog.Error("modbus客户端连接失败", "url", c.url, "err", err)
			} else {
				c.connected = true
				slog.Info("modbus客户端连接成功", "url", c.url)
			}
		} else if !c.started && c.connected { // 未启动，尝试关闭
			err := c.cli.Close()
			if err != nil {
				slog.Error("modbus客户端关闭错误", "url", c.url, "err", err)
			}
			c.connected = false
			slog.Info("modbus客户端关闭", "url", c.url)
		}
		time.Sleep(time.Second)
	}
}
func (c *client) IsConnected() bool {
	return c.connected
}

// Close implements MasterClient.
func (c *client) Close() error {
	c.cancel()
	if c.connected {
		c.connected = false
	}
	return c.cli.Close()
}

// 读请求执行
func readExecute[T any](c *client, req func() (T, error)) (T, error) {
	if !c.started {
		return *new(T), fmt.Errorf("modbus客户端未启动")
	}
	if !c.connected {
		return *new(T), fmt.Errorf("modbus客户端未连接或连接断开")
	}
	res, err := req()
	if err != nil {
		if newErr, ok := err.(*net.OpError); ok {
			if se, ok := newErr.Err.(*os.SyscallError); ok {
				c.connected = false
				return res, se
			}
		}
	}
	return res, err
}

// 写请求执行
func writeExecute(c *client, req func() error) error {
	if !c.started {
		return fmt.Errorf("modbus客户端未启动")
	}
	if !c.connected {
		return fmt.Errorf("modbus客户端未连接或连接断开")
	}
	return req()
}

// ReadCoil implements MasterClient.
func (c *client) ReadCoil(addr uint16, quantity uint16) ([]bool, error) {
	return readExecute[[]bool](c, func() ([]bool, error) {
		return c.cli.ReadCoils(addr, quantity)
	})
}

func (c *client) ReadCoilBit(addr uint16) (bool, error) {
	return readExecute[bool](c, func() (bool, error) {
		return c.cli.ReadCoil(addr)
	})
}

// ReadDiscreteInput implements MasterClient.
func (c *client) ReadDiscreteInput(addr uint16, quantity uint16) ([]bool, error) {
	return readExecute[[]bool](c, func() ([]bool, error) {
		return c.cli.ReadDiscreteInputs(addr, quantity)
	})
}

// ReadHoldingRegister implements MasterClient.
func (c *client) ReadHoldingRegister(addr uint16, quantity uint16) ([]uint16, error) {
	return readExecute[[]uint16](c, func() ([]uint16, error) {
		return c.cli.ReadRegisters(addr, quantity, modbus.HOLDING_REGISTER)
	})
}

func (c *client) ReadHoldingRegisterBytes(addr uint16, quantity uint16) ([]byte, error) {
	return readExecute[[]byte](c, func() ([]byte, error) {
		return c.cli.ReadBytes(addr, quantity, modbus.HOLDING_REGISTER)
	})
}

// ReadHoldingRegisterUint16 implements MasterClient.
func (c *client) ReadHoldingRegisterUint16(addr uint16) (uint16, error) {
	return readExecute[uint16](c, func() (uint16, error) {
		return c.cli.ReadRegister(addr, modbus.HOLDING_REGISTER)
	})
}

// ReadInputRegister implements MasterClient.
func (c *client) ReadInputRegister(addr uint16, quantity uint16) ([]uint16, error) {
	return readExecute[[]uint16](c, func() ([]uint16, error) {
		return c.cli.ReadRegisters(addr, quantity, modbus.INPUT_REGISTER)
	})
}

func (c *client) ReadInputRegisterBytes(addr uint16, quantity uint16) ([]byte, error) {
	return readExecute[[]byte](c, func() ([]byte, error) {
		return c.cli.ReadBytes(addr, quantity, modbus.INPUT_REGISTER)
	})
}

// WriteCoil implements MasterClient.
func (c *client) WriteCoil(addr uint16, value bool) error {
	return writeExecute(c, func() error {
		return c.cli.WriteCoil(addr, value)
	})
}

// WriteCoils implements MasterClient.
func (c *client) WriteCoils(addr uint16, values []bool) error {
	return writeExecute(c, func() error {
		return c.cli.WriteCoils(addr, values)
	})
}

// WriteRegister implements MasterClient.
func (c *client) WriteRegister(addr uint16, value uint16) error {
	return writeExecute(c, func() error {
		return c.cli.WriteRegister(addr, value)
	})
}

// WriteRegisters implements MasterClient.
func (c *client) WriteRegisters(addr uint16, values []uint16) error {
	return writeExecute(c, func() error {
		return c.cli.WriteRegisters(addr, values)
	})
}

func (c *client) WriteRegisterBytes(addr uint16, values []byte) error {
	return writeExecute(c, func() error {
		return c.cli.WriteBytes(addr, values)
	})
}