crrc/internal/app/stc/index.go

package stc

import (
	"context"
	"fmt"
	"net"
	"pbx-api-gin/internal/app/stc/active"
	"pbx-api-gin/internal/app/stc/broadcast"
	msgdata "pbx-api-gin/internal/app/stc/data"
	"pbx-api-gin/internal/app/stc/socket"
	"pbx-api-gin/pkg/lfshook"
	"pbx-api-gin/pkg/utils"
	"sync"
	"syscall"

	"time"
)

func StartStcConnection(conn net.Conn, cab string) {

	var connMux sync.Mutex // 保护 conn 的读写
	var conn1 net.Conn
	var err error
	var logTag = 0

	for {
		// 尝试建立连接MC
		conn1, err = CreateConnection(cab)
		if err != nil || conn1 == nil {
			time.Sleep(2 * time.Second)
			lfshook.NewLogger().Logger.Infof("===========Reconnecting====To Cab:%s=======", cab)
			continue
		}

		//set connection log
		if logTag == 0 {
			utils.Logger.Printf("Train Information: CabNumber %s, Message: Connection to Cab%s STC is up !", active.ActivedCab, cab)
			logTag = 1
		}

		connMux.Lock()
		oldConn := conn
		if cab == "1" {
			socket.Conn = conn1
		} else {
			socket.Conn8 = conn1
		}
		connMux.Unlock()

		// 关闭旧连接（如果存在）
		if oldConn != nil {
			oldConn.Close()
			//lfshook.NewLogger().Logger.Infof("Closed previous connection")
		}

		// 使用 context 控制所有协程的生命周期
		ctx, cancel := context.WithCancel(context.Background())

		// 启动消息处理MC1
		go func() {
			defer func() {
				cancel() // 一旦任一协程退出，取消所有
			}()
			broadcast.HandleStcCmd(ctx, conn1) // 改造 HandleStcCmd 接收 ctx
		}()

		// 启动心跳MC1
		go func() {
			defer func() {
				cancel()
			}()
			Sendheartbeat(ctx, conn1) // 改造 Sendheartbeat 接收 ctx
		}()

		// 等待连接断开（监听连接状态）
		<-ctx.Done()

		// 连接已断开，清理
		cancel() // 确保所有 cancel 被调用
		conn1.Close()

		//set connection log
		if logTag == 1 {
			utils.Logger.Printf("Train Information: CabNumber %s, Message: Connection to Cab%s STC is down !", active.ActivedCab, cab)
			logTag = 0
		}

		//lfshook.NewLogger().Logger.Info("Reconnecting in 1 second...")
		time.Sleep(time.Second) // 重连前等待
	}
}

// 返回错误而不是终止程序
func CreateConnection(RemoteCab string) (net.Conn, error) {

	if RemoteCab == "1" { // connect to MC1
		//lfshook.NewLogger().Logger.Infof("========Connect Server MC1 IP:%s  :Port:%d", socket.RemoteAddr, socket.RemotePort)
		// 创建 Dialer
		if active.CabNum == "1" { //in cab1
			dialer := &net.Dialer{
				LocalAddr: &net.TCPAddr{IP: net.ParseIP("0.0.0.0"), Port: socket.LocalPort}, // 固定本地端口
				Control:   controlTCPConn,
				Timeout:   5 * time.Second,
			}

			DialAddr := fmt.Sprintf("%s:%d", socket.RemoteAddr, socket.RemotePort)
			conn, err := dialer.Dial("tcp", DialAddr)
			if err != nil {
				//lfshook.NewLogger().Logger.Infof("========Connect server err :%+v", err)
				return nil, err
			}
			lfshook.NewLogger().Logger.Infof("Connect success :%s:%d", socket.RemoteAddr, socket.RemotePort)
			return conn, nil
		} else { //in cab 8
			dialer := &net.Dialer{
				LocalAddr: &net.TCPAddr{IP: net.ParseIP("0.0.0.0"), Port: socket.LocalPort8}, // 固定本地端口
				Control:   controlTCPConn,
				Timeout:   5 * time.Second,
			}

			DialAddr := fmt.Sprintf("%s:%d", socket.RemoteAddr, socket.RemotePort)
			conn, err := dialer.Dial("tcp", DialAddr)
			if err != nil {
				//lfshook.NewLogger().Logger.Infof("========Connect server err :%+v", err)
				return nil, err
			}
			lfshook.NewLogger().Logger.Infof("Connect success :%s:%d", socket.RemoteAddr, socket.RemotePort)
			return conn, nil
		}
	} else { // connect to MC8
		//lfshook.NewLogger().Logger.Infof("========Connect server MC8 IP:%s  :Port:%d", socket.RemoteAddr8, socket.RemotePort)
		// 创建 Dialer
		if active.CabNum == "1" { //in cab1
			dialer := &net.Dialer{
				LocalAddr: &net.TCPAddr{IP: net.ParseIP("0.0.0.0"), Port: socket.LocalPort}, // 固定本地端口
				Control:   controlTCPConn,
				Timeout:   5 * time.Second,
			}

			DialAddr := fmt.Sprintf("%s:%d", socket.RemoteAddr8, socket.RemotePort)
			conn, err := dialer.Dial("tcp", DialAddr)
			if err != nil {
				//lfshook.NewLogger().Logger.Infof("========Connect server err :%+v", err)
				return nil, err
			}
			lfshook.NewLogger().Logger.Infof("Connect success MC8:%s:%d", socket.RemoteAddr8, socket.RemotePort)
			return conn, nil
		} else { //in cab 8
			dialer := &net.Dialer{
				LocalAddr: &net.TCPAddr{IP: net.ParseIP("0.0.0.0"), Port: socket.LocalPort8}, // 固定本地端口
				Control:   controlTCPConn,
				Timeout:   5 * time.Second,
			}

			DialAddr := fmt.Sprintf("%s:%d", socket.RemoteAddr8, socket.RemotePort)
			conn, err := dialer.Dial("tcp", DialAddr)
			if err != nil {
				//lfshook.NewLogger().Logger.Infof("========Connect server err :%+v", err)
				return nil, err
			}
			lfshook.NewLogger().Logger.Infof("Connect success MC8:%s:%d", socket.RemoteAddr8, socket.RemotePort)
			return conn, nil
		}
	}
}

/*
func StartConnectionToSipServer(conn net.Conn) {

	var connMux sync.Mutex // 保护 conn 的读写
	var conn1 net.Conn
	var err error

	for {
		// 尝试建立连接MC
		conn1, err = CreateConnectionSipServer()
		if err != nil || conn1 == nil {
			time.Sleep(2 * time.Second)
			lfshook.NewLogger().Logger.Infof("===========Reconnecting==Sip Server=======")
			continue
		}

		connMux.Lock()
		oldConn := conn

		socket.ConnToMaster = conn1

		connMux.Unlock()

		// 关闭旧连接（如果存在）
		if oldConn != nil {
			oldConn.Close()
			lfshook.NewLogger().Logger.Infof("Closed previous connection")
		}

		// 使用 context 控制所有协程的生命周期
		ctx, cancel := context.WithCancel(context.Background())

		// 启动心跳MC1
		go func() {
			defer func() {
				cancel()
			}()
			SendheartbeatToSipServer(ctx, conn1) // 改造 Sendheartbeat 接收 ctx
		}()

		// 等待连接断开（监听连接状态）
		<-ctx.Done()

		// 连接已断开，清理
		cancel() // 确保所有 cancel 被调用
		conn1.Close()

		lfshook.NewLogger().Logger.Info("Reconnecting in 1 second...")
		time.Sleep(time.Second) // 重连前等待

		//check connected Master tag; connection err change to Master role ,exit
		//if socket.ConnectedMaster {
		//	active.Master = true
		//	return
		//}
	}
}
*/
// 连接Master sipserver
func CreateConnectionSipServer() (net.Conn, error) {
	dialer := &net.Dialer{
		LocalAddr: &net.TCPAddr{IP: net.ParseIP("0.0.0.0"), Port: socket.LocalPort8}, // 固定本地端口
		Control:   controlTCPConn,
		Timeout:   5 * time.Second,
	}

	DialAddr := fmt.Sprintf("%s:%d", socket.RemoteAddr, socket.LocalPort) // Connect to Cab1 Sip server
	conn, err := dialer.Dial("tcp", DialAddr)
	if err != nil {
		//lfshook.NewLogger().Logger.Infof("========Connect server err :%+v", err)
		return nil, err
	}
	lfshook.NewLogger().Logger.Infof("Connect SIP Server success :%s:%d", socket.RemoteAddr, socket.LocalPort)
	return conn, nil
}

func controlTCPConn(network, address string, c syscall.RawConn) error {
	return c.Control(func(fd uintptr) {
		syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_REUSEADDR, 1)
		// 注意：SO_REUSEPORT 在某些系统可用（如 Linux），但非标准
	})
}

func Sendheartbeat(ctx context.Context, conn net.Conn) {
	var count uint8

	protocol := msgdata.NewProtocol()
	protocol.MessageID = 0x21
	protocol.DataLength = 0x04
	protocol.Data = make([]byte, 4)

	// 初始化协议...
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			//lfshook.NewLogger().Logger.Infof("Sendheartbeat===ctx==ret======")
			return

		case <-ticker.C:
			count++
			protocol.Data[0] = count
			// 编码并发送数据...
			encoded, err := protocol.Encode()
			if err != nil {
				//fmt.Printf("encode err : %v\n", err)
				return
			}
			if conn != nil {
				_, err = conn.Write(encoded)
				if err != nil {
					//fmt.Printf("Send hearbeat err: %v\n", err)
					conn.Close()
					return // 触发重连
				}
				//lfshook.NewLogger().Logger.Infof("Sendheartbeat===send ======%x", encoded)
			}
		}
	}
}

func SendheartbeatToSipServer(ctx context.Context, conn net.Conn) {
	var count uint8

	protocol := msgdata.NewProtocol()
	protocol.MessageID = 0x21
	protocol.DataLength = 0x04
	protocol.Data = make([]byte, 4)

	// 初始化协议...
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			//lfshook.NewLogger().Logger.Infof("Sendheartbeat===ctx==ret======")
			return

		case <-ticker.C:
			count++
			protocol.Data[0] = count
			// 编码并发送数据...
			encoded, err := protocol.Encode()
			if err != nil {
				//fmt.Printf("encode err : %v\n", err)
				return
			}
			if conn != nil {
				_, err = conn.Write(encoded)
				if err != nil {
					//fmt.Printf("Send hearbeat err: %v\n", err)
					conn.Close()
					return // 触发重连
				}

				//Set connected Master tag
				if !socket.ConnectedMaster {
					socket.ConnectedMaster = true
				}
				//lfshook.NewLogger().Logger.Infof("Sendheartbeat===send ======%x", encoded)
			}
		}
	}
}