urdt/term
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use durable consumer

Browse Source
This commit is contained in:
lash 2024-10-24 16:30:24 +01:00
parent e29cb4b5a2
commit 51389476ca
Signed by: lash
GPG Key ID: 21D2E7BB88C2A746
1 changed files with 8 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
event/nats/nats.go
View File

@ -42,8 +42,10 @@ func(n *NatsSubscription) Connect(ctx context.Context, connStr string) error {
if err != nil { if err != nil {
return err return err
} }
n.cs, err = n.js.OrderedConsumer(ctx, "TRACKER", jetstream.OrderedConsumerConfig{ n.cs, err = n.js.CreateConsumer(ctx, "TRACKER", jetstream.ConsumerConfig{
//FilterSubjects: []string{"TRACKER.*"}, Name: "omnom",
Durable: "omnom",
FilterSubjects: []string{"TRACKER.*"},
}) })
if err != nil { if err != nil {
return err return err
@ -87,14 +89,13 @@ func(n *NatsSubscription) handleEvent(m jetstream.Msg) {
err = n.Route(&ev) err = n.Route(&ev)
if err != nil { if err != nil {
logg.Error("handler route fail", "err", err) logg.Error("handler route fail", "err", err)
fail(m) //fail(m)
return
} }
} }
err = m.Term() err = m.Ack()
if err != nil { if err != nil {
logg.Error("term fail", "err", err) logg.Error("ack fail", "err", err)
panic("term fail") panic("ack fail")
} }
logg.Debug("handle msg complete") logg.Debug("handle msg complete")
} }