pcm-coordinator/api/internal/mqs/kq/ScheduleCloud.go

package kq

import (
	"context"
	"encoding/json"
	"github.com/zeromicro/go-zero/core/logx"
	"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/svc"
	"gitlink.org.cn/jcce-pcm/pcm-coordinator/api/internal/types"
	"gitlink.org.cn/jcce-pcm/utils/tool"
)

/*
*
Listening to the payment flow status change notification message queue
*/
type ScheduleCloudMq struct {
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewScheduleCloudMq(ctx context.Context, svcCtx *svc.ServiceContext) *ScheduleCloudMq {
	return &ScheduleCloudMq{
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *ScheduleCloudMq) Consume(_, val string) error {
	var task *types.TaskInfo
	json.Unmarshal([]byte(val), &task)
	// 构建提交作业到云算的结构体
	bytes, err := json.Marshal(task.Metadata)
	if err != nil {
		return err
	}
	cloud := tool.UnMarshalK8sStruct(string(bytes), task.TaskId)
	cloud.YamlString = string(bytes)
	// 存储数据
	tx := l.svcCtx.DbEngin.Create(&cloud)
	if tx.Error != nil {
		logx.Error(tx.Error)
		return tx.Error
	}
	return nil
}