golib/v4/gnet/modbus/conn.go

package modbus

import (
	"context"
	"io"
	"net"
	"strings"
	"sync"
	"time"

	"golib/v4/gio"
	"golib/v4/gnet"
	"golib/v4/log"
)

// Conn PLC 主控连接
type Conn interface {
	// ConnStat 连接状态
	gnet.ConnStat
	// IsLocked 表示当前有其他线程正在与 PLC 交互
	IsLocked() bool
	// WriteResponse 向 PLC 发送数据并等待 PLC 响应
	WriteResponse(b []byte) ([]byte, error)
	// Closer 关闭与 PLC 主控的连接
	io.Closer
}

const (
	MaxReadBuffSize = 1024
)

type Dialer struct {
	conn   net.Conn
	buf    []byte
	logger log.Logger
	mu     sync.Mutex
	lock   bool
}

func (w *Dialer) IsConnected() bool {
	if w.conn == nil {
		return false
	}
	return w.conn.(gnet.ConnStat).IsConnected()
}

func (w *Dialer) IsClosed() bool {
	if w.conn == nil {
		return true
	}
	return w.conn.(gnet.ConnStat).IsClosed()
}

func (w *Dialer) IsLocked() bool {
	return w.lock
}

// WriteResponse 写入并读取下一次的数据
func (w *Dialer) WriteResponse(b []byte) ([]byte, error) {
	if w.conn == nil {
		return nil, gnet.ErrConnNotFound
	}
	w.mu.Lock()
	defer w.mu.Unlock()
	w.lock = true
	defer func() {
		w.lock = false
	}()
	w.logger.Debug("Write: %s", gnet.Bytes(b).HexTo())
	if i, err := w.conn.Write(b); err != nil {
		w.logger.Error("Write err: %d->%d %s", len(b), i, err)
		return nil, err
	}
	clear(w.buf)
	n, err := w.conn.Read(w.buf)
	if err != nil {
		w.logger.Error("Read err: %s", err)
		return nil, err
	}
	w.logger.Debug("Read: %s", gnet.Bytes(w.buf[:n]).HexTo())
	return w.buf[:n], nil
}

func (w *Dialer) Close() error {
	if w.conn == nil {
		return nil
	}
	return w.conn.Close()
}

func (w *Dialer) CloseWith(ctx context.Context) {
	<-ctx.Done()
	w.logger.Warn("DialContext: %s", ctx.Err())
	_ = w.Close()
}

func (w *Dialer) DialContext(ctx context.Context, address string, logger log.Logger) (Conn, error) {
	// 由于现场网络环境比较差, 因此加大超时时间以防止频繁掉线重连
	config := &gnet.Config{
		Timeout:     60 * time.Second,
		DialTimeout: 10 * time.Second,
		Reconnect:   true,
	}
	deadline := time.Now().Add(config.DialTimeout)
	if conn, err := gnet.DialTCPConfig(address, config); err == nil {
		w.conn = conn
	} else {
		if timeout := deadline.Sub(time.Now()); timeout > 0 {
			gio.RandSleep(0, timeout)
		}
		logger.Debug("DialContext: %s", err)
		return nil, err
	}
	go func() {
		w.CloseWith(ctx)
	}()
	w.buf = make([]byte, MaxReadBuffSize)
	w.logger = log.Part(logger, "conn", strings.ReplaceAll(address, ":", "_"))
	return w, nil
}

func DialContext(ctx context.Context, address string, logger log.Logger) (Conn, error) {
	var dialer Dialer
	return dialer.DialContext(ctx, address, logger)
}

func Dial(address string, logger log.Logger) (Conn, error) {
	return DialContext(context.Background(), address, logger)
}