Listener interface draft solution v3

client.go

@@ -1,3 +1,5 @@
+// SPDX-FileCopyrightText: 2024 Comcast Cable Communications Management, LLC
+// SPDX-License-Identifier: Apache-2.0
 package main
 
 import "net/http"
@@ -7,3 +9,14 @@ import "net/http"
 type Client interface {
 	Do(*http.Request) (*http.Response, error)
 }
+
+func nopClient(next Client) Client {
+	return next
+}
+
+// DoerFunc implements Client
+type doerFunc func(*http.Request) (*http.Response, error)
+
+func (d doerFunc) Do(req *http.Request) (*http.Response, error) {
+	return d(req)
+}

config.go

@@ -44,15 +44,15 @@ type Service struct {
 }
 type Consul struct {
 	Client            ConsulClient
-	Registrations     []Registration
+	Registrations     []ConsulRegistration
 	DisableGenerateId bool
 }
 type ConsulClient struct {
 	Address  string
 	Scheme   string
 	WaitTime string
 }
-type Registration struct {
+type ConsulRegistration struct {
 	Id      string
 	Name    string
 	Tags    []string

httpClient.go

@@ -16,17 +16,6 @@ var (
 	errNilHistogram = errors.New("histogram cannot be nil")
 )
 
-func nopClient(next Client) Client {
-	return next
-}
-
-// DoerFunc implements HTTPClient
-type doerFunc func(*http.Request) (*http.Response, error)
-
-func (d doerFunc) Do(req *http.Request) (*http.Response, error) {
-	return d(req)
-}
-
 type metricWrapper struct {
 	now          func() time.Time
 	queryLatency prometheus.ObserverVec

listenerStub.go

@@ -3,6 +3,7 @@
 package main
 
 import (
+	"encoding/json"
 	"time"
 
 	webhook "github.com/xmidt-org/webhook-schema"
@@ -11,8 +12,8 @@ import (
 //This is a stub for the webhook and kafka listeners. This will be removed once the webhook-schema configuration is approved
 
 type ListenerStub struct {
-	PartnerIds []string
-	Webhook    webhook.Registration
+	PartnerIds   []string
+	Registration Registration
 }
 
 // Webhook is a substructure with data related to event delivery.
@@ -139,3 +140,86 @@ type RegistrationV2 struct {
 	// TODO: list of supported formats
 	Expires time.Time `json:"expires"`
 }
+
+// Deprecated: This structure should only be used for backwards compatibility
+// matching. Use RegistrationV2 instead.
+// RegistrationV1 is a special struct for unmarshaling a webhook as part of a webhook registration request.
+type RegistrationV1 struct {
+	// Address is the subscription request origin HTTP Address.
+	Address string `json:"registered_from_address"`
+
+	// Config contains data to inform how events are delivered.
+	Config DeliveryConfig `json:"config"`
+
+	// FailureURL is the URL used to notify subscribers when they've been cut off due to event overflow.
+	// Optional, set to "" to disable notifications.
+	FailureURL string `json:"failure_url"`
+
+	// Events is the list of regular expressions to match an event type against.
+	Events []string `json:"events"`
+
+	// Matcher type contains values to match against the metadata.
+	Matcher MetadataMatcherConfig `json:"matcher,omitempty"`
+
+	// Duration describes how long the subscription lasts once added.
+	Duration webhook.CustomDuration `json:"duration"`
+
+	// Until describes the time this subscription expires.
+	Until time.Time `json:"until"`
+}
+
+// MetadataMatcherConfig is Webhook substructure with config to match event metadata.
+type MetadataMatcherConfig struct {
+	// DeviceID is the list of regular expressions to match device id type against.
+	DeviceID []string `json:"device_id"`
+}
+
+// Deprecated: This substructure should only be used for backwards compatibility
+// matching. Use Webhook instead.
+// DeliveryConfig is a Webhook substructure with data related to event delivery.
+type DeliveryConfig struct {
+	// URL is the HTTP URL to deliver messages to.
+	ReceiverURL string `json:"url"`
+
+	// ContentType is content type value to set WRP messages to (unless already specified in the WRP).
+	ContentType string `json:"content_type"`
+
+	// Secret is the string value for the SHA1 HMAC.
+	// (Optional, set to "" to disable behavior).
+	Secret string `json:"secret,omitempty"`
+
+	// AlternativeURLs is a list of explicit URLs that should be round robin through on failure cases to the main URL.
+	AlternativeURLs []string `json:"alt_urls,omitempty"`
+}
+
+type Registration interface {
+	GetId() string
+	GetAddress() string
+	Marshal() ([]byte, error)
+}
+
+func MarshalRegistration[R Registration](reg R) ([]byte, error) {
+	return reg.Marshal()
+}
+
+func (v1 *RegistrationV1) GetId() string {
+	return v1.Config.ReceiverURL
+}
+
+func (v1 *RegistrationV1) Marshal() ([]byte, error) {
+	return json.Marshal(v1)
+
+}
+
+func (v1 *RegistrationV1) GetAddress() string {
+	return v1.Address
+}
+
+// TODO: is this what we want to return for the ids Map for V2?
+func (v2 *RegistrationV2) GetId() string {
+	return v2.CanonicalName
+}
+
+func (v2 *RegistrationV2) Marshal() ([]byte, error) {
+	return nil, nil
+}

metrics.go

@@ -47,7 +47,9 @@ const (
 	CodeLabel   = "code"
 )
 
-type SenderMetricsIn struct {
+// MetricsIn will be populated automatically by the ProvideMetrics function
+// and then used to populate the Metrics struct
+type MetricsIn struct {
 	fx.In
 	QueryLatency                    prometheus.ObserverVec `name:"query_duration_histogram_seconds"`
 	DeliveryCounter                 *prometheus.CounterVec `name:"delivery_count"`
@@ -64,6 +66,23 @@ type SenderMetricsIn struct {
 	ConsumerRenewalTimeGauge        *prometheus.GaugeVec   `name:"consumer_renewal_time"`
 }
 
+// Metrics will be used to set up the metrics for each sink
+type Metrics struct {
+	DeliveryCounter                 *prometheus.CounterVec
+	DeliveryRetryCounter            *prometheus.CounterVec
+	DeliveryRetryMaxGauge           *prometheus.GaugeVec
+	CutOffCounter                   *prometheus.CounterVec
+	SlowConsumerDroppedMsgCounter   *prometheus.CounterVec
+	DropsDueToPanic                 *prometheus.CounterVec
+	ConsumerDeliverUntilGauge       *prometheus.GaugeVec
+	ConsumerDropUntilGauge          *prometheus.GaugeVec
+	ConsumerDeliveryWorkersGauge    *prometheus.GaugeVec
+	ConsumerMaxDeliveryWorkersGauge *prometheus.GaugeVec
+	OutgoingQueueDepth              *prometheus.GaugeVec
+	ConsumerRenewalTimeGauge        *prometheus.GaugeVec
+	QueryLatency                    prometheus.ObserverVec
+}
+
 // TODO: do these need to be annonated/broken into groups based on where the metrics are being used/called
 func ProvideMetrics() fx.Option {
 	return fx.Options(