dataflow/util/util.go

// util
package util

import (
	"encoding/json"
	"time"

	"github.com/nats-io/nats.go"
)

type MsgInfo struct {
	Id       string                 //消息唯一id
	CurrSetp string                 //当前步骤
	Data     map[string]interface{} //数据内容
	Stime    int64
	Etime    int64
}

func SendRequest(nc *nats.Conn, subject, step string, requestData *MsgInfo, timeout time.Duration) (*MsgInfo, error) {
	requestData.CurrSetp = step
	stime := time.Now().UnixMilli()
	// 发送请求并等待响应
	bs, err := json.Marshal(requestData)
	if err != nil {
		return nil, err
	}
	rep, err := nc.Request(subject+"."+step, bs, timeout)
	if err != nil {
		return nil, err
	}
	// 返回响应数据
	msgInfo := &MsgInfo{}
	msgInfo.Etime = time.Now().UnixMilli()
	msgInfo.Stime = stime
	err = json.Unmarshal(rep.Data, msgInfo)
	if err != nil {
		return nil, err
	}
	return msgInfo, nil
}