Matcher and Webhook Updates

caduceus.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"os"
 	"runtime/debug"
-	"time"
 
 	"github.com/alecthomas/kong"
 	"github.com/goschtalt/goschtalt"

internal/sink/matcher.go

@@ -47,9 +47,24 @@ type CommonWebhook struct {
 	logger *zap.Logger
 }
 
+func NewMatcher(l Listener, logger *zap.Logger) (matcher Matcher, version int, err error) {
+	switch v := l.(type) {
+	case *ListenerV1:
+		m := &MatcherV1{}
+		m.logger = logger
+		if err := m.update(*v); err != nil {
+			return nil, 0, err
+		}
+		matcher = m
+		return matcher, 1, nil
+	default:
+		return nil, 0, fmt.Errorf("invalid listner")
+	}
+}
+
 // Update applies user configurable values for the outbound sender when a
 // webhook is registered
-func (m1 *MatcherV1) Update(l ListenerV1) error {
+func (m1 *MatcherV1) update(l ListenerV1) error {
 
 	//TODO: don't believe the logger for webhook is being set anywhere just yet
 	m1.logger = m1.logger.With(zap.String("webhook.address", l.Registration.Address))

internal/sink/sink.go

@@ -39,15 +39,20 @@ type WebhookV1 struct {
 	// clientMiddleware func(http.Client) http.Client
 }
 
-func NewWebhookV1(s *sender) {
-	v1 := &WebhookV1{
-		id:               s.id,
-		deliveryInterval: s.deliveryInterval,
-		deliveryRetries:  s.deliveryRetries,
-		logger:           s.logger,
+func NewSink(c Config, logger *zap.Logger, id string, version int) Sink {
+	var sink Sink
+	if version == 1 {
+		v1 := &WebhookV1{
+			id:               id,
+			deliveryInterval: c.DeliveryInterval,
+			deliveryRetries:  c.DeliveryRetries,
+			logger:           logger,
+		}
+		sink = v1
 	}
-	s.sink = v1
+	return sink
 }
+
 func (v1 *WebhookV1) Update(l Listener) (err error) {
 	v1.id = l.GetId()
 	return nil

internal/sink/sinkSender.go

@@ -57,6 +57,7 @@ type sender struct {
 	disablePartnerIDs bool
 	customPIDs        []string
 	mutex             sync.RWMutex
+	config            Config
 	deliverUntil      time.Time
 	dropUntil         time.Time
 	deliveryInterval  time.Duration
@@ -120,6 +121,7 @@ func NewSender(w *wrapper, l Listener) (s *sender, err error) {
 		queueSize:    w.config.QueueSizePerSender,
 		deliverUntil: l.GetUntil(),
 		logger:       w.logger,
+		config:       w.config, //TODO: need to figure out which config options are used for just sender, just sink, and both
 		// dropUntil:        where is this being set in old caduceus?,
 		cutOffPeriod:     w.config.CutOffPeriod,
 		deliveryRetries:  w.config.DeliveryRetries,
@@ -157,19 +159,9 @@ func NewSender(w *wrapper, l Listener) (s *sender, err error) {
 }
 
 func (s *sender) Update(l Listener) (err error) {
-	switch v := l.(type) {
-	case *ListenerV1:
-		m := &MatcherV1{}
-		m.logger = s.logger
-		if err = m.Update(*v); err != nil {
-			return
-		}
-		s.matcher = m
-		NewWebhookV1(s)
-
-	default:
-		err = fmt.Errorf("invalid listner")
-	}
+	var version int
+	s.matcher, version, err = NewMatcher(l, s.logger)
+	s.sink = NewSink(s.config, s.logger, s.id, version)
 
 	s.renewalTimeGauge.Set(float64(time.Now().Unix()))