backend/watchdog/tcpClient.go

package watchdog

import (
	"bufio"
	"fmt"
	"net"
	"strings"
	"super-frpc/postLog"
	"super-frpc/utils"
	"sync"
	"time"
)

var (
	tcpConn      net.Conn
	tcpConnMutex sync.Mutex
	isConnected  bool
	recvChan     chan string
	stopRecvChan chan struct{}
	recvWg       sync.WaitGroup
)

func Init() error {
	tcpConnMutex.Lock()
	defer tcpConnMutex.Unlock()

	if tcpConn != nil {
		return fmt.Errorf("TCP client already initialized")
	}

	recvChan = make(chan string, 100)
	stopRecvChan = make(chan struct{})
	isConnected = false

	return nil
}

func tcpConnect(ipaddr string, port int) error {
	tcpConnMutex.Lock()
	defer tcpConnMutex.Unlock()

	if tcpConn != nil {
		return fmt.Errorf("already connected")
	}

	address := net.JoinHostPort(ipaddr, fmt.Sprintf("%d", port))
	conn, err := net.DialTimeout("tcp", address, 3*time.Second)
	if err != nil {
		return fmt.Errorf("failed to connect to %s: %v", address, err)
	}

	tcpConn = conn
	isConnected = true

	recvWg.Add(1)
	go recvMsg()

	return nil
}

func sendMsg(message string, timeout int) (string, error) {
	tcpConnMutex.Lock()

	if tcpConn == nil {
		tcpConnMutex.Unlock()
		return "", fmt.Errorf("not connected")
	}

	_, err := tcpConn.Write([]byte(message + "\n"))
	if err != nil {
		tcpConnMutex.Unlock()
		return "", fmt.Errorf("failed to send message: %v", err)
	}

	tcpConnMutex.Unlock()

	select {
	case response := <-recvChan:
		for len(recvChan) > 0 {
			<-recvChan
		}
		return strings.TrimSpace(response), nil
	case <-time.After(time.Duration(timeout) * time.Second):
		return "", fmt.Errorf("timeout waiting for response")
	}
}

func recvMsg() {
	defer recvWg.Done()

	tcpConnMutex.Lock()
	if tcpConn == nil {
		tcpConnMutex.Unlock()
		return
	}
	conn := tcpConn
	tcpConnMutex.Unlock()

	reader := bufio.NewReader(conn)

	for {
		select {
		case <-stopRecvChan:
			return
		default:
			conn.SetReadDeadline(time.Now().Add(100 * time.Millisecond))

			data, err := reader.ReadBytes('\n')

			if len(data) > 0 {
				line := strings.TrimSpace(string(data))
				if len(line) > 0 {
					select {
					case recvChan <- line: 
					default:
						// drop the message
					}
					// Here add logic to handle the message
					if !isResponseMessage(line) {
						postLog.Debug(fmt.Sprintf("[Watchdog] TCP Socket received message: %s", line))
						cmdType := utils.GetCmdType(line)
						if cmdType == "Exception" {
							exceptionType := utils.GetCmdParams(line, "exceptionType")
							serviceName := utils.GetCmdParams(line, "serviceName")
							errorMsg := utils.GetCmdParams(line, "errorMsg")
							postLog.Error(fmt.Sprintf("[Watchdog] Exception[%s]: %s, %s", serviceName, exceptionType, errorMsg))
							exceptionHandle(exceptionType, serviceName, errorMsg)
						}
					}
				}
			}

			if err != nil {
				if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
					continue
				}

				tcpConnMutex.Lock()
				if tcpConn != nil {
					tcpConn.Close()
					tcpConn = nil
					isConnected = false
				}
				tcpConnMutex.Unlock()
				return
			}
		}
	}
}

func isResponseMessage(msg string) bool {
	return msg == "success" || msg == "failed"
}

func Destroy() error {
	tcpConnMutex.Lock()

	if stopRecvChan != nil {
		close(stopRecvChan)
		stopRecvChan = nil
	}

	tcpConnMutex.Unlock()

	recvWg.Wait()

	tcpConnMutex.Lock()
	defer tcpConnMutex.Unlock()

	if tcpConn != nil {
		err := tcpConn.Close()
		tcpConn = nil
		isConnected = false
		if err != nil {
			return fmt.Errorf("failed to close connection: %v", err)
		}
	}

	if recvChan != nil {
		close(recvChan)
		recvChan = nil
	}

	return nil
}

func IsConnected() bool {
	tcpConnMutex.Lock()
	defer tcpConnMutex.Unlock()
	return isConnected && tcpConn != nil
}