go-nats/publisher.go at main · darwinOrg/go-nats · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
package dgnats

import (
	"strconv"
	"time"

	"github.com/darwinOrg/go-common/constants"
	dgctx "github.com/darwinOrg/go-common/context"
	dglogger "github.com/darwinOrg/go-logger"
	"github.com/nats-io/nats.go"
)

func Publish(ctx *dgctx.DgContext, subject *NatsSubject, obj any) error {
	bytes, err := ToBytes(ctx, obj)
	if err != nil {
		return err
	}
	dglogger.Infof(ctx, "publish subject[%s] json message: %s", subject.Name, string(bytes))

	return PublishRaw(ctx, subject, bytes)
}

func PublishDelay(ctx *dgctx.DgContext, subject *NatsSubject, obj any, duration time.Duration) error {
	bytes, err := ToBytes(ctx, obj)
	if err != nil {
		return err
	}
	dglogger.Infof(ctx, "publish subject[%s] json delay message: %s", subject.Name, string(bytes))

	header := map[string][]string{
		constants.TraceId: {ctx.TraceId},
		headerPubAt:       {strconv.FormatInt(time.Now().UnixNano(), 10)},
		headerDelay:       {strconv.FormatInt(int64(duration), 10)},
	}

	msg := &nats.Msg{
		Subject: subject.Name,
		Reply:   subject.Name,
		Header:  header,
		Data:    bytes,
	}

	return publishMsg(ctx, subject, msg)
}

func PublishRaw(ctx *dgctx.DgContext, subject *NatsSubject, data []byte) error {
	return publishRawWithHeader(ctx, subject, map[string][]string{}, data)
}

func PublishRawWithTag(ctx *dgctx.DgContext, subject *NatsSubject, tag string, data []byte) error {
	if tag == "" {
		return PublishRaw(ctx, subject, data)
	}

	return publishRawWithHeader(ctx, subject, map[string][]string{headerTag: {tag}}, data)
}

func publishRawWithHeader(ctx *dgctx.DgContext, subject *NatsSubject, header map[string][]string, data []byte) error {
	header[constants.TraceId] = []string{ctx.TraceId}

	msg := &nats.Msg{
		Subject: subject.Name,
		Header:  header,
		Data:    data,
	}

	return publishMsg(ctx, subject, msg)
}

func publishMsg(ctx *dgctx.DgContext, subject *NatsSubject, msg *nats.Msg) error {
	err := InitStream(ctx, subject)
	if err != nil {
		return err
	}

	js, err := GetJs()
	if err != nil {
		return err
	}
	_, err = js.PublishMsg(msg, buildPubOpts(subject)...)

	return err
}

func buildPubOpts(subject *NatsSubject) []nats.PubOpt {
	return []nats.PubOpt{
		nats.MsgId(nats.NewInbox()),
		nats.ExpectStream(subject.Category),
	}
}