44 "context"
55 "sync"
66
7+ nc "github.com/nats-io/nats.go"
78 "go.infratographer.com/permissions-api/internal/query"
89 "go.infratographer.com/permissions-api/internal/types"
910 "go.infratographer.com/x/events"
@@ -26,6 +27,7 @@ type Subscriber struct {
2627 changeChannels []<- chan * message.Message
2728 logger * zap.SugaredLogger
2829 subscriber * events.Subscriber
30+ subOpts []nc.SubOpt
2931 qe query.Engine
3032}
3133
@@ -39,25 +41,33 @@ func WithLogger(l *zap.SugaredLogger) SubscriberOption {
3941 }
4042}
4143
42- // NewSubscriber creates a new Subscriber
43- func NewSubscriber (ctx context.Context , cfg events.SubscriberConfig , engine query.Engine , opts ... SubscriberOption ) (* Subscriber , error ) {
44- sub , err := events .NewSubscriber (cfg )
45- if err != nil {
46- return nil , err
44+ // WithNatsSubOpts sets the logger for the Subscriber
45+ func WithNatsSubOpts (options ... nc.SubOpt ) SubscriberOption {
46+ return func (s * Subscriber ) {
47+ s .subOpts = append (s .subOpts , options ... )
4748 }
49+ }
4850
51+ // NewSubscriber creates a new Subscriber
52+ func NewSubscriber (ctx context.Context , cfg events.SubscriberConfig , engine query.Engine , opts ... SubscriberOption ) (* Subscriber , error ) {
4953 s := & Subscriber {
50- ctx : ctx ,
51- logger : zap .NewNop ().Sugar (),
52- subscriber : sub ,
53- qe : engine ,
54+ ctx : ctx ,
55+ logger : zap .NewNop ().Sugar (),
56+ qe : engine ,
5457 }
5558
5659 for _ , opt := range opts {
5760 opt (s )
5861 }
5962
60- s .logger .Debugw ("subscriber configuration" , cfg )
63+ sub , err := events .NewSubscriber (cfg , s .subOpts ... )
64+ if err != nil {
65+ return nil , err
66+ }
67+
68+ s .subscriber = sub
69+
70+ s .logger .Debugw ("subscriber configuration" , "config" , cfg )
6171
6272 return s , nil
6373}
0 commit comments