cic-chain-events/internal/events/jetstream.go
Mohamed Sohail 9d1b77e907
refactor: filters, pipeline, minor fixes, remove RPC support (#27)
* devnet: snapshot

* refactor: pass struct through pipeline fllters

* refactor: replace timer with ticker

* refactor: filters, jetstream emitter

* add register filter
* update gas filter

* refactor: remove RPC fetcher support
2023-02-24 13:28:30 +03:00

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
}