dlt645-server/server/server.go

package server

import (
	"context"
	"dlt645-server/protocol"
	"fmt"
	"github.com/gogf/gf/container/gmap"
	"github.com/gogf/gf/net/gtcp"
	"github.com/gogf/gf/os/glog"
	gatewayV2 "sparrow-sdk/v2"
	"sync"
	"time"
)

type Server struct {
	closeChan chan struct{}
	server    *gtcp.Server
	ctx       context.Context
	addr      string
	port      int
	clients   *gmap.HashMap
	grMu      sync.Mutex
	grWG      sync.WaitGroup
	grRunning bool
	message   protocol.Message
	gateWay   *gatewayV2.Gateway
}

func NewServer(ctx context.Context, addr string, port int, gw *gatewayV2.Gateway) *Server {
	return &Server{
		closeChan: make(chan struct{}),
		ctx:       ctx,
		addr:      addr,
		port:      port,
		gateWay:   gw,
	}
}

func (s *Server) Start() error {
	glog.Printf("服务端启动[%s:%d]", s.addr, s.port)
	server := gtcp.NewServer(fmt.Sprintf("%s:%d", s.addr, s.port), s.handleConnect)
	s.server = server
	s.grMu.Lock()
	s.grRunning = true
	s.grMu.Unlock()
	return s.server.Run()
}

func (s *Server) Stop() {
	s.server.Close()
}

func (s *Server) handleConnect(conn *gtcp.Conn) {
	s.startGoRoutine(func() {
		s.createClient(conn)
		s.grWG.Done()
	})
}

func (s *Server) startGoRoutine(f func()) {
	s.grMu.Lock()
	if s.grRunning {
		s.grWG.Add(1)
		go f()
	}
	s.grMu.Unlock()
}

func (s *Server) createClient(conn *gtcp.Conn) *Client {
	c := &Client{
		srv:  s,
		done: make(chan struct{}),
		conn: conn,
		closeHandler: func(id string, c *Client) {
			glog.Debugf("客户端断开:%s", id)
		},
	}

	ctx := new(protocol.PacketContext)

	addressChan := make(chan struct{}, 1)
	avChan := make(chan struct{}, 1)
	aiChan := make(chan struct{}, 1)
	bvChan := make(chan struct{}, 1)
	biChan := make(chan struct{}, 1)
	cvChan := make(chan struct{}, 1)
	ciChan := make(chan struct{}, 1)
	powerChan := make(chan struct{}, 1)

	addressChan <- struct{}{}
	s.startGoRoutine(func() {
		go c.SendGetAddress(ctx, addressChan, avChan)
	})

	s.startGoRoutine(func() {
		c.GetActivePower(ctx, powerChan)
	})

	s.startGoRoutine(func() {
		c.ReadLoop(ctx)
	})

	time.Sleep(1 * time.Second)
	s.startGoRoutine(func() {
		c.GetAV(ctx, avChan, aiChan)
	})

	time.Sleep(1 * time.Second)
	s.startGoRoutine(func() {
		c.GetAI(ctx, aiChan, bvChan)
	})

	time.Sleep(1 * time.Second)
	s.startGoRoutine(func() {
		c.GetBV(ctx, bvChan, biChan)
	})

	time.Sleep(1 * time.Second)
	s.startGoRoutine(func() {
		c.GetBI(ctx, biChan, cvChan)
	})
	time.Sleep(1 * time.Second)
	s.startGoRoutine(func() {
		c.GetCV(ctx, cvChan, ciChan)
	})

	time.Sleep(1 * time.Second)
	s.startGoRoutine(func() {
		c.GetCI(ctx, ciChan, powerChan)
	})
	return c
}

func (s *Server) removeClient(gatewayId uint16) {
	s.clients.Remove(gatewayId)
}

func (s *Server) ReportStatus(subId string, data interface{}) error {
	return s.gateWay.ReportStatus(subId, "status", data)
}