79 lines
1.5 KiB
Go
79 lines
1.5 KiB
Go
package events
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/goccy/go-json"
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
const (
|
|
StreamName string = "CHAIN"
|
|
StreamSubjects string = "CHAIN.*"
|
|
)
|
|
|
|
type JetStreamOpts struct {
|
|
ServerUrl string
|
|
PersistDuration time.Duration
|
|
DedupDuration time.Duration
|
|
}
|
|
|
|
type JetStream struct {
|
|
jsCtx nats.JetStreamContext
|
|
nc *nats.Conn
|
|
}
|
|
|
|
func NewJetStreamEventEmitter(o JetStreamOpts) (EventEmitter, error) {
|
|
natsConn, err := nats.Connect(o.ServerUrl)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
js, err := natsConn.JetStream()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
// Bootstrap stream if it doesn't exist.
|
|
stream, _ := js.StreamInfo(StreamName)
|
|
if stream == nil {
|
|
_, err = js.AddStream(&nats.StreamConfig{
|
|
Name: StreamName,
|
|
MaxAge: o.PersistDuration,
|
|
Storage: nats.FileStorage,
|
|
Subjects: []string{StreamSubjects},
|
|
Duplicates: o.DedupDuration,
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
return &JetStream{
|
|
jsCtx: js,
|
|
nc: natsConn,
|
|
}, nil
|
|
}
|
|
|
|
// Close gracefully shutdowns the JetStream connection.
|
|
func (js *JetStream) Close() {
|
|
if js.nc != nil {
|
|
js.nc.Close()
|
|
}
|
|
}
|
|
|
|
// Publish publishes the JSON data to the NATS stream.
|
|
func (js *JetStream) Publish(subject string, dedupId string, eventPayload interface{}) error {
|
|
jsonData, err := json.Marshal(eventPayload)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
_, err = js.jsCtx.Publish(subject, jsonData, nats.MsgId(dedupId))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return nil
|
|
}
|