Matcher v2 init

internal/sink/matcher.go

@@ -43,6 +43,10 @@ type MatcherV1 struct {
 	CommonWebhook
 }
 
+type MatcherV2 struct {
+	matcher map[string]*regexp.Regexp
+	CommonWebhook
+}
 type CommonWebhook struct {
 	mutex  sync.RWMutex
 	logger *zap.Logger
@@ -58,6 +62,13 @@ func NewMatcher(l ancla.Register, logger *zap.Logger) (Matcher, error) {
 			return nil, err
 		}
 		return m, nil
+	case *ancla.RegistryV2:
+		m := &MatcherV2{}
+		m.logger = logger
+		if err := m.update(*v); err != nil {
+			return nil, err
+		}
+		return m, nil
 	default:
 		return nil, fmt.Errorf("invalid listener")
 	}
@@ -194,3 +205,76 @@ func (m1 *MatcherV1) getUrls() (urls *ring.Ring) {
 	m1.urls = m1.urls.Next()
 	return
 }
+
+// Update applies user configurable values for the outbound sender when a
+// webhook is registered
+func (m2 *MatcherV2) update(l ancla.RegistryV2) error {
+
+	//TODO: don't believe the logger for webhook is being set anywhere just yet
+	m2.logger = m2.logger.With(zap.String("webhook.address", l.Registration.Address))
+
+	if l.Registration.FailureURL != "" {
+		_, err := url.ParseRequestURI(l.Registration.FailureURL)
+		if err != nil {
+			return err
+		}
+	}
+
+	//TODO: should we be checking that the l.Registration.Matcher.Field is a field in the wrp.Message?
+	matcher := make(map[string]*regexp.Regexp)
+	for _, item := range l.Registration.Matcher {
+		if item.Regex == ".*" {
+			// Match everything - skip the filtering
+			matcher[item.Field] = &regexp.Regexp{}
+			break
+		}
+
+		var re *regexp.Regexp
+		re, err := regexp.Compile(item.Regex)
+		if err != nil {
+			return fmt.Errorf("invalid matcher item: '%s'", item.Regex)
+		}
+		matcher[item.Field] = re
+	}
+
+	// write/update sink sender
+	m2.mutex.Lock()
+	defer m2.mutex.Unlock()
+
+	// if matcher list is empty set it nil for Queue() logic
+	if len(matcher) == 0 {
+		m2.matcher = nil
+	} else {
+		m2.matcher = matcher
+	}
+
+	return nil
+
+}
+
+func (m2 *MatcherV2) IsMatch(msg *wrp.Message) bool {
+	m2.mutex.RLock()
+	matcher := m2.matcher
+	m2.mutex.RUnlock()
+
+	var (
+		matchDevice = false
+	)
+	for field, deviceRegex := range matcher {
+		if deviceRegex.MatchString(field) || deviceRegex.MatchString(strings.TrimPrefix(field, "event:")) {
+			matchDevice = true
+			break
+		}
+	}
+
+	if !matchDevice {
+		m2.logger.Debug("device regex doesn't match", zap.String("event.source", msg.Source))
+		return false
+	}
+	return true
+}
+
+// TODO: this is a big reason why I want to refactor the Matcher logic
+func (m2 *MatcherV2) getUrls() (urls *ring.Ring) {
+	return
+}

internal/sink/sink.go

@@ -52,10 +52,9 @@ type Kafka struct {
 }
 
 func NewSink(c Config, logger *zap.Logger, listener ancla.Register) Sink {
-	var sink Sink
 	switch l := listener.(type) {
 	case *ancla.RegistryV1:
-		sink = &WebhookV1{
+		sink := &WebhookV1{
 			id:               l.GetId(),
 			deliveryInterval: c.DeliveryInterval,
 			deliveryRetries:  c.DeliveryRetries,
@@ -68,31 +67,32 @@ func NewSink(c Config, logger *zap.Logger, listener ancla.Register) Sink {
 			kafka := &Kafka{
 				id:         l.Registration.CanonicalName,
 				brokerAddr: k.BootstrapServers,
-				topic:      "test",
+				topic:      "quickstart-events",
+				logger:     logger,
 			}
 
+			config := sarama.NewConfig()
 			//TODO: this is basic set up for now - will need to add more options to config
 			//once we know what we are allowing users to send
-			kafka.config.Producer.Return.Successes = true			
-			kafka.config.Producer.RequiredAcks = sarama.WaitForAll
-			kafka.config.Producer.Retry.Max = c.DeliveryRetries //should we be using retryhint for this?
 
+			config.Producer.Return.Successes = true
+			config.Producer.RequiredAcks = sarama.WaitForAll
+			config.Producer.Retry.Max = c.DeliveryRetries //should we be using retryhint for this?
+			kafka.config = config
 			// Create a new Kafka producer
 			producer, err := sarama.NewSyncProducer(kafka.brokerAddr, config)
 			if err != nil {
 				kafka.logger.Error("Could not create Kafka producer", zap.Error(err))
 				return nil
 			}
-			defer producer.Close()
 
 			kafka.producer = producer
 			sink = append(sink, kafka)
 		}
-
+		return sink
 	default:
 		return nil
 	}
-	return sink
 }
 
 func (v1 *WebhookV1) Update(l ancla.Register) (err error) {
@@ -346,6 +346,7 @@ func (k *Kafka) send(secret string, acceptType string, msg *wrp.Message) error {
 
 	// Send the message to Kafka
 	partition, offset, err := k.producer.SendMessage(kafkaMsg)
+	defer k.producer.Close()
 	if err != nil {
 		k.logger.Error("Failed to send message to Kafka", zap.Error(err))
 		return err