dy-admin-temp/internal/pcmserver/bus/cabinet_pkg/initialize.go

package cabinet_pkg

import (
	"dy-admin/internal/pcmserver/bus/cabinet_pkg/tcp_server"
	"dy-admin/pkg/log"
	"encoding/hex"
	"fmt"
	"go.uber.org/zap"
	"net"
	"strings"
	"time"
)

var tcpServer *tcp_server.TCPServer

type tcpServe struct {
}

func (t *tcpServe) OnConnect(_ *tcp_server.TCPServer, conn net.Conn) {
	// 连接回调,与内存柜子信息关联
	if cabinets == nil {
		log.Error("cabinets is nil")
		return
	}

	remoteIP := strings.Split(conn.RemoteAddr().String(), ":")[0]

	cab, ok := cabinets.getCabinetByIP(remoteIP)
	if !ok {
		log.Error("remote ip not found in set", zap.String("remoteIP", remoteIP))
		tcpServer.WriteData(conn, []byte(fmt.Sprintf("remote ip %s not found in set", remoteIP)))
		time.Sleep(time.Second)
		//主动关闭客户端连接
		_ = conn.Close()
		createConnectLog(remoteIP, fmt.Sprintf("此ip未在平台登记,拒绝连接"))
		return
	}
	if cab.checkConnected() {
		log.Warn("remote ip has connected", zap.String("remoteIP", remoteIP))
		// 主动关闭连接
		conn.Close()
		createConnectLog(remoteIP, fmt.Sprintf("此ip已经被使用,拒绝连接"))
		return
	}

	// 张总要求
	time.Sleep(time.Second * 5)

	err := cab.syncCabinets(cab.memInfo.CabinetID, conn)
	if err != nil {
		log.Error("syncCabinets err", zap.Error(err))
		//柜子状态与平台同步失败,切断连接。重新连接
		cab.setUnconnected()
		conn.Close()
		createConnectLog(remoteIP, fmt.Sprintf("拒绝连接,同步柜子状态失败：%s", err))
		return
	}

	// 发送心跳
	go func() {
		if err := cab.heartBeat(cab.memInfo.CabinetID); err != nil {
			log.Error("get heartbeat err", zap.Error(err))
			cab.setUnconnected()
			conn.Close()
			createConnectLog(remoteIP, fmt.Sprintf("拒绝连接,发送心跳失败,：%s", err))
		}
	}()

	// 连接并且同步状态成功
	log.Info("cabinet connected success", zap.String("remoteIP", remoteIP))
	createConnectLog(remoteIP, fmt.Sprintf("连接成功"))
}

func (t *tcpServe) OnReceive(_ *tcp_server.TCPServer, conn net.Conn, data []byte, readSize int) {
	if cabinets == nil {
		log.Error("cabinets is nil")
		return
	}

	remoteIP := strings.Split(conn.RemoteAddr().String(), ":")[0]
	log.Info("receive message info", zap.String("remoteIP", remoteIP), zap.Int("dataLen", len(data)), zap.Int("readSize", readSize), zap.String("dataHex", hex.EncodeToString(data[:readSize])))

	cab, ok := cabinets.getCabinetByIP(remoteIP)
	if !ok {
		log.Warn("remote ip not found in set", zap.String("remoteIP", remoteIP))
		return
	}

	err := cab.packet.NewPacket(data[:readSize])
	if err != nil {
		log.Error(err.Error())
		return
	}
}

func (t *tcpServe) OnWrite(_ *tcp_server.TCPServer, conn net.Conn, writeSize int) {
	//remoteIP := strings.Split(conn.RemoteAddr().String(), ":")[0]
	//logrus.WithFields(logrus.Fields{"remoteIP": remoteIP, "writeSize": writeSize}).Info("send message info")
	return
}

func (t *tcpServe) OnReceiveError(_ *tcp_server.TCPServer, conn net.Conn, err error) {
	remoteIP := strings.Split(conn.RemoteAddr().String(), ":")[0]
	log.Error("receive error", zap.String("remoteIP", remoteIP), zap.Error(err))
	// 修改在线状态
	cab, ok := cabinets.getCabinetByIP(remoteIP)
	if !ok {
		log.Warn("remote ip not found in set", zap.String("remoteIP", remoteIP))
		return
	}

	cab.setUnconnected()
	log.Warn("remote ip offline", zap.String("remoteIP", remoteIP))
}

func (t *tcpServe) OnWriteError(_ *tcp_server.TCPServer, conn net.Conn, err error) {
	remoteIP := strings.Split(conn.RemoteAddr().String(), ":")[0]
	log.Error("send error", zap.String("remoteIP", remoteIP))

	// 修改在线状态
	cab, ok := cabinets.getCabinetByIP(remoteIP)
	if !ok {
		log.Warn("remote ip not found in set", zap.String("remoteIP", remoteIP))
		return
	}

	cab.setUnconnected()
	log.Warn("remote ip offline", zap.String("remoteIP", remoteIP))
}

func InitTcpServer(port string) {
	// 初始化tcpServer

	conf := tcp_server.NewTCPServerConfig()
	server, err := tcp_server.NewTCPServer(conf, "0.0.0.0:"+port, &tcpServe{})
	if err != nil {
		log.Error(err.Error())
		return
	}

	tcpServer = server
	log.Info("tcp server listening", zap.String("port", port))
}

func DestroyTcpServer() {
	tcp_server.DestroyTCPServer(tcpServer)
	tcpServer = nil
}