grassrootseconomics/eth-indexer
7
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: jetstream switch to new API, discard buffer on close

Browse Source
This commit is contained in:
Mohamed Sohail 2024-10-31 09:52:00 +03:00
parent d8bb140f94
commit 44a9b5cd29
Signed by: kamikazechaser
GPG Key ID: 7DD45520C01CD85D
3 changed files with 33 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
internal/sub/jetstream.go
View File

@ -20,11 +20,11 @@ type (
} }
JetStreamSub struct { JetStreamSub struct {
jsConsumer jetstream.Consumer jsIter jetstream.MessagesContext
logg *slog.Logger logg *slog.Logger
natsConn *nats.Conn natsConn *nats.Conn
router *router.Router router *router.Router
durableID string durableID string
} }
) )
@ -33,7 +33,7 @@ const (
pullSubject = "TRACKER.*" pullSubject = "TRACKER.*"
) )
func NewJetStreamSub(o JetStreamOpts) (Sub, error) { func NewJetStreamSub(o JetStreamOpts) (*JetStreamSub, error) {
natsConn, err := nats.Connect(o.Endpoint) natsConn, err := nats.Connect(o.Endpoint)
if err != nil { if err != nil {
return nil, err return nil, err
@ -53,52 +53,52 @@ func NewJetStreamSub(o JetStreamOpts) (Sub, error) {
} }
consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{ consumer, err := stream.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{
Durable: o.JetStreamID, Durable: o.JetStreamID,
AckPolicy: jetstream.AckExplicitPolicy, AckPolicy: jetstream.AckExplicitPolicy,
FilterSubject: pullSubject,
}) })
if err != nil { if err != nil {
return nil, err return nil, err
} }
o.Logg.Info("successfully connected to NATS server") o.Logg.Info("successfully connected to NATS server")
iter, err := consumer.Messages(
jetstream.WithMessagesErrOnMissingHeartbeat(false),
jetstream.PullMaxMessages(100),
)
if err != nil {
return nil, err
}
return &JetStreamSub{ return &JetStreamSub{
jsConsumer: consumer, jsIter: iter,
router: o.Router, router: o.Router,
natsConn: natsConn, natsConn: natsConn,
logg: o.Logg, logg: o.Logg,
durableID: o.JetStreamID, durableID: o.JetStreamID,
}, nil }, nil
} }
func (s *JetStreamSub) Close() { func (s *JetStreamSub) Close() {
if s.natsConn != nil { s.jsIter.Stop()
s.natsConn.Close()
}
} }
func (s *JetStreamSub) Process() error { func (s *JetStreamSub) Process() {
iter, err := s.jsConsumer.Messages(jetstream.WithMessagesErrOnMissingHeartbeat(false))
if err != nil {
return err
}
defer iter.Stop()
for { for {
msg, err := iter.Next() msg, err := s.jsIter.Next()
if err != nil { if err != nil {
if errors.Is(err, nats.ErrTimeout) { if errors.Is(err, jetstream.ErrMsgIteratorClosed) {
s.logg.Error("jetstream: iter fetch timeout") s.logg.Debug("jetstream: iterator closed")
continue return
} else if errors.Is(err, nats.ErrConnectionClosed) {
return nil
} else { } else {
return err s.logg.Debug("jetstream: unknown iterator error", "error", err)
continue
} }
} }
s.logg.Info("processing nats message", "subject", msg.Subject()) s.logg.Debug("processing nats message", "subject", msg.Subject())
if err := s.router.Handle(context.Background(), msg); err != nil { if err := s.router.Handle(context.Background(), msg); err != nil {
s.logg.Error("router: error processing nats message", "error", err) s.logg.Error("jetstream: router: error processing nats message", "error", err)
} }
} }
} }

6
internal/sub/sub.go
View File

@ -1,6 +0,0 @@
package sub
type Sub interface {
Process() error
Close()
}

3
internal/telegram/telegram.go
View File

@ -19,8 +19,7 @@ type (
) )
const ( const (
NOTIFY_LOW_BALANCE_ON_GAS_FAUCET = ` NOTIFY_LOW_BALANCE_ON_GAS_FAUCET = "Gas faucet balance is low. Top-up is required soon!"
Gas faucet balance is low. Top is required soon!`
) )
func New(o TelegramOpts) *Telegram { func New(o TelegramOpts) *Telegram {