526互联

11、从0到1实现SECS协议之HSMS协议中的handler

发布时间 2023-08-27 14:25:26作者: 画个一样的我

11、从0到1实现SECS协议之HSMS协议中的handler

前面实现了 发送事件的处理机制，接下来我们即将实现ISecsHandler 接口，将前面实现的各种功能组合起来，从而提供一个完整可用的服务。

1、handler 的具体实现

package handler

import (
	"context"
	"errors"
	"fmt"
	"github.com/looplab/fsm"
	"go.uber.org/zap"
	"math/rand"
	"runtime/debug"
	"secs-gem/functions"
	"sync"
	"sync/atomic"
	"time"

	"secs-gem/common"
	secs_errors "secs-gem/errors"
	"secs-gem/hsms/connection"
	packets "secs-gem/hsms/packet"
	"secs-gem/hsms/statemachine"
	driver_log "secs-gem/log"
	"secs-gem/secsgem"
)

// HsmsHandler hsms 协议的处理者
type HsmsHandler struct {
	// 建立连接时应当做的事情
	secsgem.IConnection

	// 消息监听者(接收到 pkt 时如何处理这些 pkt )
	listener secsgem.MessageListener

	// device id
	sessionId uint16

	// hsms 协议中的 连接状态机
	connectionState *statemachine.ConnectionState
	connected       bool // 是否连接
	// hsms 协议中的唯一标识
	// 唯一地标识此消息事务, ReplyMessage 的 SystemBytes 应与其对应回复的Prymary Message的 System Bytes 相同
	systemCounter uint32
	systemQueues  sync.Map // 存储一个 systemCounter 使用到的 优先级队列

	// hsms 协议规定, 客户端需要定时发送 LinkTest.req , 这样才认为通信正常?
	linkTestTimer   *time.Timer
	linkTestTimeout time.Duration
	timeMx          sync.Mutex

	// 日志
	log *zap.SugaredLogger
	mx  sync.RWMutex

	// 桥接模式下的消息转发
	bridge secsgem.ISecsBridge
}

func NewHsmsHandler(cfg *secsgem.ConnectionCfg, listener secsgem.MessageListener) *HsmsHandler {
	// 构造函数

	hh := &HsmsHandler{
		sessionId:       uint16(cfg.SessionId),
		listener:        listener,
		linkTestTimeout: 30 * time.Second,
	}

	// 日志
	hh.log = driver_log.S().With("port", cfg.Port, "active", cfg.Active)

	// 连接
	hh.IConnection = connection.NewConnection(cfg, hh)

	// system bytes
	rand.Seed(time.Now().Unix())
	hh.systemCounter = rand.Uint32()

	//连接状态
	callbacks := fsm.Callbacks{
		"on_enter_CONNECTED":          hh.onStateConnect,
		"on_exit_CONNECTED":           hh.onStateDisconnect,
		"on_enter_CONNECTED_SELECTED": hh.onStateSelect,
	}
	hh.connectionState = statemachine.NewConnectionState("hsms", callbacks)

	return hh
}

func (hh *HsmsHandler) IsActive() bool {
	return hh.GetConfig().Active
}

func (hh *HsmsHandler) SetSessionID(sessionId uint16) {
	/*
			设置sessionID
		使用者: 1.自动获取DeviceID, 当接收到 S9F1 指令时, 说明 device id 设置错误, 这时自动修正 device id 为接收到的数据
	*/
	hh.sessionId = sessionId
}

func (hh *HsmsHandler) Enable() error {
	return hh.IConnection.Enable()
}

func (hh *HsmsHandler) Disable() error {
	return hh.IConnection.Disable()
}

// 连接状态
func (hh *HsmsHandler) onStateConnect(ctx context.Context, e *fsm.Event) {
	hh.log.Debug("on state connect")

	if hh.IsActive() { // 客户端才主动发送心跳 (host)
		//心跳检测(应该在init时启动timer)
		hh.startLinkTester()

		go hh.sendSelectReq()
	}
}

func (hh *HsmsHandler) onStateDisconnect(ctx context.Context, e *fsm.Event) {
	//重置
	hh.timeMx.Lock()
	if hh.linkTestTimer != nil {
		hh.linkTestTimer.Stop()
	}
	hh.timeMx.Unlock()
}

func (hh *HsmsHandler) onStateSelect(ctx context.Context, e *fsm.Event) {
}

func (hh *HsmsHandler) onLinkTestTimer() {
	//发送心跳
	if hh.connected {
		hh.sendLinkTestReq()
	}

	//重置
	hh.timeMx.Lock()
	timer := hh.linkTestTimer
	if timer != nil {
		timer.Reset(hh.linkTestTimeout)
	} else {
		//timer过期了
	}
	hh.timeMx.Unlock()
}

// startLinkTester 开启定时发送心跳包 LinkTest.req
func (hh *HsmsHandler) startLinkTester() {
	//连接心跳
	hh.timeMx.Lock()
	timer := hh.linkTestTimer
	if timer == nil {
		hh.linkTestTimer = time.AfterFunc(hh.linkTestTimeout, hh.onLinkTestTimer)
	} else {
		timer.Reset(hh.linkTestTimeout)
	}
	hh.timeMx.Unlock()
}

// sendSelectReq select.req
// 用于使用Select.req和Select.rsp消息在TCP/IP连接上建立HSMS通信
// 实体由 Not Selected 状态转换为 Selected 状态所使用的消息 (Active Entity 发送Select.req)；
func (hh *HsmsHandler) sendSelectReq() *packets.HsmsPacket {
	hh.log.Debug("select.req")

	//packet
	systemId := hh.GetNextSystemCounter()
	queueChan := hh.getQueueForSystem(systemId)
	outPacket := &packets.HsmsPacket{
		Header: packets.HsmsSelectReqHeader(systemId),
	}

	//发送
	hh.IConnection.SendPacket(outPacket, secsgem.PriorityNORMAL)

	//等待结果
	response, err := hh.waitResponse(&queueChan, hh.GetConfig().T6)
	if err != nil {
		hh.log.Debugf("[system=0x%x] %s", systemId, err)
	}

	//移除queue
	hh.removeQueue(systemId)
	return response
}

//  sendSelectRsp select.rsp
func (hh *HsmsHandler) sendSelectRsp(systemId uint32) {
	//packet
	outPacket := &packets.HsmsPacket{
		Header: packets.HsmsSelectRspHeader(systemId),
	}
	hh.IConnection.SendPacket(outPacket, secsgem.PriorityRESPONSE)
}

// sendLinkTestReq linktest.req
func (hh *HsmsHandler) sendLinkTestReq() *packets.HsmsPacket {
	hh.log.Debugf("linktest.req (connected=%v)", hh.connected)
	//packet
	systemId := hh.GetNextSystemCounter()
	queueChan := hh.getQueueForSystem(systemId)
	outPacket := &packets.HsmsPacket{
		Header: packets.HsmsLinkTestReqHeader(systemId),
	}
	//日志
	secsgem.PublishEvent(&secsgem.SecsMessageSend{
		PortID:   hh.GetConfig().ID,
		System:   systemId,
		Stream:   outPacket.Header.Stream,
		Function: outPacket.Header.Function,
		Packet:   outPacket,
	})

	//发送
	hh.IConnection.SendPacket(outPacket, secsgem.PriorityNORMAL)

	//等待结果
	response, err := hh.waitResponse(&queueChan, hh.GetConfig().T6)
	if err != nil {
		hh.log.Warnf("[system=0x%x] %s", systemId, err)
	}

	//移除queue
	hh.removeQueue(systemId)
	return response
}

// sendLinktestRsp linktest.rsp
func (hh *HsmsHandler) sendLinktestRsp(systemId uint32) {
	//packet
	outPacket := &packets.HsmsPacket{
		Header: packets.HsmsLinkTestRspHeader(systemId),
	}
	//日志
	secsgem.PublishEvent(&secsgem.SecsMessageSend{
		PortID:   hh.GetConfig().ID,
		System:   systemId,
		Stream:   outPacket.Header.Stream,
		Function: outPacket.Header.Function,
		Packet:   outPacket,
	})
	hh.IConnection.SendPacket(outPacket, secsgem.PriorityRESPONSE)
}

//deselect.req
func (hh *HsmsHandler) sendDeselectReq() *packets.HsmsPacket {
	hh.log.Debug("deselect.req")
	systemId := hh.GetNextSystemCounter()
	queueChan := hh.getQueueForSystem(systemId)
	outPacket := &packets.HsmsPacket{
		Header: packets.HsmsDeselectReqHeader(systemId),
	}

	//发送
	hh.IConnection.SendPacket(outPacket, secsgem.PriorityNORMAL)

	//等待结果
	response, err := hh.waitResponse(&queueChan, hh.GetConfig().T6)
	if err != nil {
		hh.log.Debug(err)
	}

	//移除queue
	hh.removeQueue(systemId)
	return response
}

//deselect.rsp
func (hh *HsmsHandler) sendDeselectRsp(systemId uint32) {
	outPacket := &packets.HsmsPacket{
		Header: packets.HsmsDeselectRspHeader(systemId),
	}
	//发送
	hh.IConnection.SendPacket(outPacket, secsgem.PriorityRESPONSE)
}

//separate.req
func (hh *HsmsHandler) sendSeparateReq() {
	systemId := hh.GetNextSystemCounter()
	outPacket := &packets.HsmsPacket{
		Header: packets.HsmsSeparateReqHeader(systemId),
	}
	//发送
	hh.IConnection.SendPacket(outPacket, secsgem.PriorityNORMAL)
}

// GetNextSystemCounter 获取一个未使用的 system id
func (hh *HsmsHandler) GetNextSystemCounter() uint32 {
	atomic.AddUint32(&hh.systemCounter, 1)
	return hh.systemCounter

}

// waitResponse 等待结果, 等待从 queue chan 中获取收到的 pkt
// queueChan: 对于每一个 system id 都有一个专属的 queue chan
// timeout: 最大等待时长
// 现在就需要知道哪个地方在往 queueChan 中发送 pkt
// 实际上往 queueChan 中发送 pkt 的地方是 OnConnectionPacketReceived 函数,
// 函数中会先根据 system id 尝试从 systemQueues(sync.Map) 获取 chan ,能得到说明是自己主动发送的消息的回复
func (hh *HsmsHandler) waitResponse(queueChan *chan packets.HsmsPacket, timeout time.Duration) (*packets.HsmsPacket, error) {
	timer := common.GlobalTimerPool.Get(timeout)
	defer common.GlobalTimerPool.Put(timer)

	select {
	case response, ok := <-*queueChan:
		if !ok {
			return nil, errors.New("wait channel closed")
		}
		return &response, nil
	case <-hh.IConnection.GetCtx().Done():
		return nil, fmt.Errorf("wait response context canceled")
	case <-timer.C:
		return nil, fmt.Errorf("wait response timeout(T3=%s) ", timeout)
	}
}

/*
	connection 事件接口
*/

func (hh *HsmsHandler) OnConnectionEstablished() {
	hh.connected = true
	hh.log.Info("Connection established")
	hh.connectionState.FSM.Event(context.Background(), "connect")
	secsgem.PublishEvent(&secsgem.ConnectionEstablished{PortID: hh.GetConfig().ID})
}

func (hh *HsmsHandler) OnConnectionPacketReceived(conn secsgem.IConnection, packet interface{}) {
	pkt, ok := packet.(*packets.HsmsPacket)
	if !ok {
		return
	}
	// 往 rxgo 中发送事件
	secsgem.PublishEvent(&secsgem.SecsMessageRecv{
		PortID:   hh.GetConfig().ID,
		System:   pkt.Header.System,
		Stream:   pkt.Header.Stream,
		Function: pkt.Header.Function,
		Packet:   pkt,
	})

	systemId := pkt.Header.System
	if pkt.Header.SType > 0 {
		// 此时是非 data message
		hh._handleHsmsRequests(pkt)
	} else { // data message
		if _queueChan, ok := hh.systemQueues.Load(systemId); ok {
			queueChan := _queueChan.(chan packets.HsmsPacket)
			// 这里就是 waitResponse 函数中等待数据发送的地方
			queueChan <- *pkt
		} else {
			//未进行标记的消息
			hh._OnSecsPacketReceived(pkt)
		}
	}
}
func (hh *HsmsHandler) _handleHsmsRequests(pkt *packets.HsmsPacket) {
	systemId := pkt.Header.System

	hh.log.Debugf("[system=0x%x] < %s", systemId, pkt)

	switch pkt.Header.SType {
	case 0x01:
		// 说明此 pkt 是 select.req, 需要进行回复
		hh.sendSelectRsp(pkt.Header.System)
		hh.connectionState.FSM.Event(context.Background(), "select")
	case 0x02:
		// 说明此 pkt 是 select.rsp, 即前面发送 select.req 消息的回复
		hh.connectionState.FSM.Event(context.Background(), "select")
		if _queueChan, ok := hh.systemQueues.Load(systemId); ok {
			queueChan := _queueChan.(chan packets.HsmsPacket)
			queueChan <- *pkt
		}
	case 0x03:
		// 说明此 pkt 是 deselect.req
		// 用于在断开TCP/IP连接之前，为一个实体提供一个优雅的HSMS通信终止
		// HSMS要求使用该Procedure时连接处于Selected状态。
		// 这里并不用严格按照 协议来实现
		hh.sendDeselectRsp(pkt.Header.System)
		hh.connectionState.FSM.Event(context.Background(), "deselect")
	case 0x04:
		hh.connectionState.FSM.Event(context.Background(), "deselect")
		if _queueChan, ok := hh.systemQueues.Load(systemId); ok {
			queueChan := _queueChan.(chan packets.HsmsPacket)
			queueChan <- *pkt
		}
	case 0x05:
		hh.sendLinktestRsp(pkt.Header.System)
	default:
		if _queueChan, ok := hh.systemQueues.Load(systemId); ok {
			queueChan := _queueChan.(chan packets.HsmsPacket)
			queueChan <- *pkt
		}
	}
}

func (hh *HsmsHandler) OnConnectionBeforeClosed() {
	hh.log.Info("Connection to be close")
	hh.sendSeparateReq()
}

func (hh *HsmsHandler) OnConnectionClosed() {
	hh.log.Info("Connection closed")
	hh.connectionState.FSM.Event(context.Background(), "disconnect")
	hh.connected = false
}

func (hh *HsmsHandler) _OnSecsPacketReceived(packet *packets.HsmsPacket) (err error) {
	defer func() {
		if err1 := recover(); err1 != nil {
			err = secs_errors.CovertError(err1)
			hh.log.Warn(string(debug.Stack()))
		}
	}()

	//自动修改device id
	if packet.Header.Stream == 9 && packet.Header.Function == 1 {
		hh.sessionId = packet.Header.SessionID
	}

	if hh.listener != nil {
		// 下面是决定收到的消息如何处理, 为什么这里并没有在此仓库中实现如何处理接收到的消息呢?
		// 往高一层级思考, 此仓库的作用是如何编解码、建立连接、接受pkt 等公共的功能, 但是收到的 pkt 应当如何处理应该由使用此仓库的者去决定
		// 这样此仓库的功能更加纯粹, 后面也有更多扩展性的可能
		if packet.Header.Function%2 == 1 {
			hh.listener.PrimaryIn(hh, packet)
		} else {
			hh.listener.SecondaryIn(hh, packet)
		}
	}
	return nil
}

// getQueueForSystem 创建 存储元素为 HsmsPacket 的 channel, 并将其存放在并发安全的 sync.Map 中
// 这里使用不当时, 会造成 queue 中的 数据丢失
func (hh *HsmsHandler) getQueueForSystem(systemId uint32) chan packets.HsmsPacket {
	//发送队列
	queue := make(chan packets.HsmsPacket, 10)
	hh.systemQueues.Store(systemId, queue)
	return queue
}

// removeQueue 删除 sync.Map 中的 queue
func (hh *HsmsHandler) removeQueue(systemId uint32) {
	if _, ok := hh.systemQueues.Load(systemId); ok {
		hh.systemQueues.Delete(systemId)
	}
}

// SendResponse 接收到消息时发送回复的响应
// 后期调用此方式的地方主要有两个地方, 一个是 flow 流程中的 PrimaryIn, 另
// 外一个是 StandardEquipment 中的 PrimaryIn
func (hh *HsmsHandler) SendResponse(packet interface{}, system uint32) error {
	// 消息响应
	var err error
	var pkt *secsgem.Packet

	if sf, ok := packet.(*functions.SecsStreamFunction); ok {
		if pkt, err = sf.CastPacket(); err != nil {
			return err
		}
	} else {
		if pkt, err = secsgem.CastPacket(packet); err != nil {
			return err
		}
	}

	// 发送
	header := &packets.HsmsHeader{
		SessionID:       hh.sessionId,
		RequireResponse: false,
		Stream:          pkt.Stream,
		Function:        pkt.Function,
		System:          system,
	}
	outPacket := &packets.HsmsPacket{
		Header: header,
		Data:   pkt.Data,
	}
	//往 rxgo 中发送数据
	secsgem.PublishEvent(&secsgem.SecsMessageSend{
		PortID:   hh.GetConfig().ID,
		System:   header.System,
		Stream:   header.Stream,
		Function: header.Function,
		Packet:   outPacket,
	})

	return hh.IConnection.SendPacket(outPacket, secsgem.PriorityRESPONSE)
}

// SendAndWaitForResponse 发送并等待回复的响应
func (hh *HsmsHandler) SendAndWaitForResponse(packet interface{}, priority int) (interface{}, error) {
	var err error
	var pkt *secsgem.Packet
	if sf, ok := packet.(*functions.SecsStreamFunction); ok {
		if pkt, err = sf.CastPacket(); err != nil {
			return nil, err
		}
	} else {
		if pkt, err = secsgem.CastPacket(packet); err != nil {
			return nil, err
		}
	}

	if _, ok := hh.systemQueues.Load(pkt.System); pkt.System != 0 && ok {
		// 指定 system id,system id已存在
		// 这个时候不能够再使用此 system id , 否则影响数据的正确性
		pkt.System = 0
	}

	if pkt.System == 0 {
		//自动id
		pkt.System = hh.GetNextSystemCounter()
	}
	systemId := pkt.System

	queueChan := hh.getQueueForSystem(pkt.System)
	defer hh.removeQueue(pkt.System)

	//发送
	header := &packets.HsmsHeader{
		SessionID:       hh.sessionId,
		RequireResponse: true,
		Stream:          pkt.Stream,
		Function:        pkt.Function,
		System:          pkt.System,
	}
	outPacket := &packets.HsmsPacket{
		Header: header,
		Data:   pkt.Data,
	}

	//日志
	secsgem.PublishEvent(&secsgem.SecsMessageSend{
		PortID:   hh.GetConfig().ID,
		System:   header.System,
		Stream:   header.Stream,
		Function: header.Function,
		Packet:   outPacket,
	})

	if err1 := hh.IConnection.SendPacket(outPacket, priority|secsgem.PriorityNORMAL); err1 != nil {
		hh.log.Debugf("[system=0x%x] %s", systemId, err1)
		return nil, err1
	}

	hh.log.Debugf("[system=0x%x] start t3 timer", systemId)
	response, err2 := hh.waitResponse(&queueChan, hh.GetConfig().T3)
	if err2 != nil {
		hh.log.Debugf("[system=0x%x] %s", systemId, err2)
		return nil, fmt.Errorf("system=0x%x %s", systemId, err2)
	}
	hh.log.Debugf("[system=0x%x] end t3 timer", systemId)

	return response, nil
}

// SendStreamFunction 发送消息, 只发送不等待
func (hh *HsmsHandler) SendStreamFunction(packet interface{}) error {
	//只发送不等待
	var err error
	var pkt *secsgem.Packet
	if sf, ok := packet.(*functions.SecsStreamFunction); ok {
		if pkt, err = sf.CastPacket(); err != nil {
			return err
		}
	} else {
		if pkt, err = secsgem.CastPacket(packet); err != nil {
			return err
		}
	}

	if pkt.System == 0 {
		pkt.System = hh.GetNextSystemCounter()
	}

	//发送
	header := &packets.HsmsHeader{
		SessionID:       hh.sessionId,
		RequireResponse: pkt.RequireResponse,
		Stream:          pkt.Stream,
		Function:        pkt.Function,
		System:          pkt.System,
	}
	outPacket := &packets.HsmsPacket{
		Header: header,
		Data:   pkt.Data,
	}
	//日志
	secsgem.PublishEvent(&secsgem.SecsMessageSend{
		PortID:   hh.GetConfig().ID,
		System:   header.System,
		Stream:   header.Stream,
		Function: header.Function,
		Packet:   packet,
	})

	return hh.IConnection.SendPacket(outPacket, secsgem.PriorityNORMAL)
}

func (hh *HsmsHandler) Connected() bool {
	return hh.connected
}

func (hh *HsmsHandler) SetBridge(bridge secsgem.ISecsBridge) {
	hh.bridge = bridge
}

func (hh *HsmsHandler) SupportBridgeRecv(fromHandler secsgem.ISecsHandler, packet interface{}, originalResponseUnused interface{}) (interface{}, error) {
	if hh.bridge != nil {
		return hh.bridge.SupportBridgeRecv(fromHandler, packet, originalResponseUnused)
	}
	return originalResponseUnused, nil
}

func (hh *HsmsHandler) GetView() interface{} {
	return hh.IConnection.GetView()
}

func (hh *HsmsHandler) SetConfig(cfg *secsgem.ConnectionCfg) {
	view := hh.IConnection.GetView().(*secsgem.ConnView)
	if !view.Enabled.Load() { // 未连接的才可以修改 cfg
		//连接
		hh.IConnection = connection.NewConnection(cfg, hh)
		hh.IConnection.SetConfig(cfg)
	}
}

2、handler 的单元测试

package handler

import (
	"fmt"
	"runtime"
	"secs-gem/secsgem"
	"testing"
)

func TestHsmsHandlerForActive(t *testing.T) {
	fmt.Println("-------------------- active 启动两次 --------------------")
	activeConf := secsgem.ConnectionCfg{
		Active:     true,
		DriverType: "hsms",
		Host:       "localhost",
		Port:       5000,
	}

	handler := NewHsmsHandler(&activeConf, nil)
	err := handler.Enable()
	if err != nil {
		t.Fatalf("enable active failed, err:%s", err)
	}
	goroutine1 := runtime.NumGoroutine()

	err2 := handler.Enable()
	if err2 != nil {
		t.Fatalf("enable active failed, err:%s", err)
	}

	goroutine2 := runtime.NumGoroutine()

	if goroutine1 != goroutine2 {
		t.Fatalf("active enable goroutine not equal, goroutine1=%d,goroutine2=%d", goroutine1, goroutine2)
	}

}

func TestHsmsHandlerForPassive(t *testing.T) {
	fmt.Println("-------------------- passive 启动两次 --------------------")
	passiveConf := secsgem.ConnectionCfg{
		Active:     false,
		DriverType: "hsms",
		Host:       "localhost",
		Port:       5001,
	}

	handler := NewHsmsHandler(&passiveConf, nil)
	err := handler.Enable()
	if err != nil {
		t.Fatalf("enable passive failed, err:%s", err)
	}
	goroutine1 := runtime.NumGoroutine()

	err2 := handler.Enable()
	if err2 != nil {
		t.Fatalf("enable passive failed, err:%s", err)
	}

	goroutine2 := runtime.NumGoroutine()

	if goroutine1 != goroutine2 {
		t.Fatalf("passive enable goroutine not equal, goroutine1=%d,goroutine2=%d", goroutine1, goroutine2)
	}

}