forked from TrueCloudLab/frostfs-s3-gw
[#195] Add logger to notification controller
Signed-off-by: Denis Kirillov <denis@nspcc.ru>
This commit is contained in:
parent
9d19acadcd
commit
46e4b28489
2 changed files with 7 additions and 5 deletions
|
@ -8,6 +8,7 @@ import (
|
||||||
|
|
||||||
"github.com/nats-io/nats.go"
|
"github.com/nats-io/nats.go"
|
||||||
"github.com/nspcc-dev/neofs-s3-gw/api/layer"
|
"github.com/nspcc-dev/neofs-s3-gw/api/layer"
|
||||||
|
"go.uber.org/zap"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
|
@ -23,6 +24,7 @@ type Options struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
type Controller struct {
|
type Controller struct {
|
||||||
|
logger *zap.Logger
|
||||||
taskQueueConnection *nats.Conn
|
taskQueueConnection *nats.Conn
|
||||||
jsClient nats.JetStreamContext
|
jsClient nats.JetStreamContext
|
||||||
handlers map[string]Stream
|
handlers map[string]Stream
|
||||||
|
@ -34,7 +36,7 @@ type Stream struct {
|
||||||
ch chan *nats.Msg
|
ch chan *nats.Msg
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewController(p *Options) (*Controller, error) {
|
func NewController(p *Options, l *zap.Logger) (*Controller, error) {
|
||||||
if p == nil {
|
if p == nil {
|
||||||
return nil, nil
|
return nil, nil
|
||||||
}
|
}
|
||||||
|
@ -61,6 +63,7 @@ func NewController(p *Options) (*Controller, error) {
|
||||||
}
|
}
|
||||||
|
|
||||||
return &Controller{
|
return &Controller{
|
||||||
|
logger: l,
|
||||||
taskQueueConnection: nc,
|
taskQueueConnection: nc,
|
||||||
jsClient: js,
|
jsClient: js,
|
||||||
handlers: make(map[string]Stream),
|
handlers: make(map[string]Stream),
|
||||||
|
@ -103,11 +106,10 @@ func (c *Controller) Listen(ctx context.Context) {
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case msg := <-stream.ch:
|
case msg := <-stream.ch:
|
||||||
fmt.Printf("got message: %s\n", msg.Data)
|
|
||||||
if err := stream.h.HandleMessage(ctx, msg); err != nil {
|
if err := stream.h.HandleMessage(ctx, msg); err != nil {
|
||||||
fmt.Printf("could not handle message: %s", err)
|
c.logger.Error("could not handle message", zap.Error(err))
|
||||||
} else if err = msg.Ack(); err != nil {
|
} else if err = msg.Ack(); err != nil {
|
||||||
fmt.Printf("could not ACK message: %s", err)
|
c.logger.Error("could not ACK message", zap.Error(err))
|
||||||
}
|
}
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return
|
return
|
||||||
|
|
|
@ -144,7 +144,7 @@ func newApp(ctx context.Context, l *zap.Logger, v *viper.Viper) *App {
|
||||||
}
|
}
|
||||||
|
|
||||||
nopts := getNotificationsOptions(v, l)
|
nopts := getNotificationsOptions(v, l)
|
||||||
nc, err := notifications.NewController(nopts)
|
nc, err := notifications.NewController(nopts, l)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
l.Fatal("failed to enable notifications", zap.Error(err))
|
l.Fatal("failed to enable notifications", zap.Error(err))
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue