property-device/impl/v1/gate_command/command_use.go

package gate_command

import (
	"context"
	"encoding/json"
	"fmt"
	"git.getensh.com/common/gopkgs/cache"
	"git.getensh.com/common/gopkgs/database"
	"git.getensh.com/common/gopkgs/logger"
	"go.uber.org/zap"
	"google.golang.org/grpc/status"
	"gorm.io/gorm"
	"property-device/errors"
	dbmodel "property-device/model"
	pb_v1 "property-device/pb/v1"
	"property-device/utils/gate_utils"
	"time"
)

func checkGateCommandUseParam(req *pb_v1.GateCommandUseRequest) error {
	switch {
	case req.Sn == "" || req.Protocol == 0:
		return status.Error(10003, "sn 和协议不能为空")
	}
	return nil
}
func GateWhiteLock(sn string, protocol int32) error {
	if cache.RedisLock("gate_white_lock_" + fmt.Sprintf("%v----%v", sn, protocol)) {
		return nil
	}
	return status.Error(10003, "系统繁忙，请稍后")
}
func GateWhiteUnlock(sn string, protocol int32) {
	cache.RedisUnlock("gate_white_lock_" + fmt.Sprintf("%v----%v", sn, protocol))
}

func GateCommandUse(ctx context.Context, req *pb_v1.GateCommandUseRequest) (reply *pb_v1.GateCommandUseReply, err error) {

	reply = &pb_v1.GateCommandUseReply{}

	// 捕获各个task中的异常并返回给调用者
	defer func() {
		if r := recover(); r != nil {
			err = fmt.Errorf("%+v", r)
			e := &status.Status{}
			if er := json.Unmarshal([]byte(err.Error()), e); er != nil {
				logger.Error("err",
					zap.String("system_err", err.Error()),
					zap.Stack("stacktrace"))
			}
		}
	}()
	err = checkGateCommandUseParam(req)
	if err != nil {
		return nil, err
	}

	if !gate_utils.HasCommand(req.Sn, req.Protocol) {
		return reply, nil
	}

	if err = GateWhiteLock(req.Sn, req.Protocol); err != nil {
		return nil, err
	}
	defer GateWhiteUnlock(req.Sn, req.Protocol)

	p := &dbmodel.TGateCommand{}

	gateInfo, err := getGateInfo(0, req.Sn, req.Protocol)
	if err != nil {
		return nil, err
	}
	where := [][2]interface{}{}
	where = dbmodel.WhereAdd(where, "device_id", gateInfo.ID)
	where = dbmodel.WhereAdd(where, "status", gate_utils.CommandStatusWait)
	err = p.Find(database.DB(), where)
	if err != nil && err != gorm.ErrRecordNotFound {
		return nil, errors.DataBaseError
	}
	if p.ID == 0 {
		_ = gate_utils.CommandCacheDecrease(req.Sn, req.Protocol)
		return reply, nil
	}

	values := map[string]interface{}{
		"status":     gate_utils.CommandStatusRuning,
		"updated_at": time.Now(),
	}

	db := database.DB().Begin()
	filter := [][2]interface{}{}
	filter = dbmodel.WhereAdd(filter, "id", p.ID)
	err = p.Update(db, filter, values)
	if err != nil {
		db.Rollback()
		return nil, errors.DataBaseError
	}

	count, err := p.Count(db, where, nil)
	if err != nil {
		db.Rollback()
		return nil, errors.DataBaseError
	}
	if count == 0 {
		if err = gate_utils.CommandCacheDecrease(req.Sn, req.Protocol); err != nil {
			db.Rollback()
			return nil, err
		}
	}

	reply.Id = p.ID
	reply.CmdCode = p.Code
	reply.CmdParams = p.Param
	db.Commit()
	return reply, nil
}