unistack-org/micro-broker-nats
3
0
Fork 1
Code Issues 2 Pull Requests Actions Packages Projects Releases 8 Wiki Activity

update for latest micro logger changes

Browse Source 
Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>
This commit is contained in:
Василий Толстов
2024-10-12 13:13:52 +03:00
parent 39fda18176
commit 8802dbf1eb
3 changed files with 1261 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
nats.go
View File

@@ -1,5 +1,5 @@
// Package nats provides a NATS broker
package nats // import "go.unistack.org/micro-broker-nats/v3"
package nats
import (
"context"
@@ -42,6 +42,7 @@ type publication struct {
topic string
err error
msg *broker.Message
ctx context.Context
}
func (p *publication) Topic() string {
@@ -65,6 +66,10 @@ func (p *publication) Error() error {
return p.err
}
func (p *publication) Context() context.Context {
return p.ctx
}
type subscriber struct {
s *nats.Subscription
opts broker.SubscribeOptions
@@ -135,7 +140,7 @@ func (n *natsBroker) Connect(ctx context.Context) error {
c, err := opts.Connect()
if err != nil {
if n.opts.Logger.V(logger.WarnLevel) {
n.opts.Logger.Warnf(n.opts.Context, "error connecting to broker: %v", err)
n.opts.Logger.Error(n.opts.Context, "error connecting to broker", err)
}
return err
@@ -306,6 +311,7 @@ func (n *natsBroker) Subscribe(ctx context.Context, topic string, handler broker
pub.msg.Body = nil
pub.topic = msg.Subject
pub.err = nil
pub.ctx = ctx
if options.BodyOnly {
pub.msg.Body = msg.Data
@@ -324,7 +330,7 @@ func (n *natsBroker) Subscribe(ctx context.Context, topic string, handler broker
eh(pub)
} else {
if n.opts.Logger.V(logger.ErrorLevel) {
n.opts.Logger.Error(n.opts.Context, err)
n.opts.Logger.Error(n.opts.Context, "handler error", err)
}
}
return
@@ -336,7 +342,7 @@ func (n *natsBroker) Subscribe(ctx context.Context, topic string, handler broker
eh(pub)
} else {
if n.opts.Logger.V(logger.ErrorLevel) {
n.opts.Logger.Error(n.opts.Context, err)
n.opts.Logger.Error(n.opts.Context, "handler error", err)
}
}
}
@@ -420,7 +426,7 @@ func NewBroker(opts ...broker.Option) *natsBroker {
options := broker.NewOptions(opts...)
if options.Codec.String() != "noop" {
options.Logger.Infof(options.Context, "broker codec not noop, disable plain nats headers usage")
options.Logger.Info(options.Context, "broker codec not noop, disable plain nats headers usage")
}
n := &natsBroker{