Create WebhookBroker struct (#80)

* Create WebhookBroker struct

* stop using global config in webhook.go

* ticker resource leak

* linter

Author: zzzming

src/broker/webhook.go

@@ -21,72 +21,77 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+type WebhookBroker struct {
+	// key is the hash of topic full name and pulsar url, and subscription name
+	webhooks  map[string]chan *SubCloseSignal
+	dbHandler db.Db
+	l         *log.Entry
+	sync.RWMutex
+}
+
 // SubCloseSignal is a signal object to pass for channel
 type SubCloseSignal struct{}
 
-// key is the hash of topic full name and pulsar url, and subscription name
-var webhooks = make(map[string]chan *SubCloseSignal)
-var whLock = sync.RWMutex{}
+func NewWebhookBroker(config *util.Configuration) *WebhookBroker {
+	return &WebhookBroker{
+		dbHandler: db.NewDbWithPanic(config.PbDbType),
+		webhooks:  make(map[string]chan *SubCloseSignal),
+		l:         log.WithFields(log.Fields{"app": "webhookbroker"}),
+	}
+}
 
 // ReadWebhook reads a thread safe map
-func ReadWebhook(key string) (chan *SubCloseSignal, bool) {
-	whLock.RLock()
-	defer whLock.RUnlock()
-	c, ok := webhooks[key]
+func (wb *WebhookBroker) ReadWebhook(key string) (chan *SubCloseSignal, bool) {
+	wb.RLock()
+	defer wb.RUnlock()
+	c, ok := wb.webhooks[key]
 	return c, ok
 }
 
 // WriteWebhook writes a key/value to a thread safe map
-func WriteWebhook(key string, c chan *SubCloseSignal) {
-	whLock.Lock()
-	defer whLock.Unlock()
-	webhooks[key] = c
+func (wb *WebhookBroker) WriteWebhook(key string, c chan *SubCloseSignal) {
+	wb.Lock()
+	defer wb.Unlock()
+	wb.webhooks[key] = c
 }
 
 // DeleteWebhook deletes a key from a thread safe map
-func DeleteWebhook(key string) bool {
-	whLock.Lock()
-	defer whLock.Unlock()
-	if c, ok := webhooks[key]; ok {
+func (wb *WebhookBroker) DeleteWebhook(key string) bool {
+	wb.Lock()
+	defer wb.Unlock()
+	if c, ok := wb.webhooks[key]; ok {
 		c <- &SubCloseSignal{}
-		delete(webhooks, key)
+		delete(wb.webhooks, key)
 		//channel is deleted where it's been created with `defer`
 		return ok
 	}
 	return false
 }
 
-var singleDb db.Db
-
 // Init initializes webhook configuration database
-func Init() {
-	NewDbHandler()
-	durationStr := util.AssignString(util.GetConfig().PbDbInterval, "180s")
+func Init(config *util.Configuration) {
+	svr := NewWebhookBroker(config)
+	durationStr := util.AssignString(config.PbDbInterval, "180s")
 	duration, err := time.ParseDuration(durationStr)
 	if err != nil {
-		log.Errorf("specified duration %s error %v", durationStr, err)
+		svr.l.Errorf("specified duration %s error %v", durationStr, err)
 		duration, _ = time.ParseDuration("180s")
 	}
-	log.Warnf("beam database pull every %.0f seconds", duration.Seconds())
+	svr.l.Infof("beam database pull every %.0f seconds", duration.Seconds())
 
 	go func() {
-		run()
+		svr.run()
 		ticker := time.NewTicker(duration)
+		defer ticker.Stop()
 		for {
 			select {
 			case <-ticker.C:
-				run()
+				svr.run()
 			}
 		}
 	}()
 }
 
-// NewDbHandler gets a local copy of Db handler
-func NewDbHandler() {
-	log.Infof("webhook database init...")
-	singleDb = db.NewDbWithPanic(util.GetConfig().PbDbType)
-}
-
 // pushWebhook sends data to a webhook interface
 func pushWebhook(url string, data []byte, headers []string) (int, *http.Response) {
 
@@ -163,7 +168,7 @@ func pushAndAck(c pulsar.Consumer, msg pulsar.Message, url string, data []byte,
 
 // ConsumeLoop consumes data from Pulsar topic
 // Do not use context since go vet will puke that requires cancel invoked in the same function
-func ConsumeLoop(url, token, topic, subscriptionKey string, whCfg model.WebhookConfig) error {
+func (wb *WebhookBroker) ConsumeLoop(url, token, topic, subscriptionKey string, whCfg model.WebhookConfig) error {
 	headers := whCfg.Headers
 	_, err := model.GetSubscriptionType(whCfg.SubscriptionType)
 	if err != nil {
@@ -175,11 +180,11 @@ func ConsumeLoop(url, token, topic, subscriptionKey string, whCfg model.WebhookC
 	}
 	c, err := pulsardriver.GetPulsarConsumer(url, token, topic, whCfg.Subscription, whCfg.InitialPosition, whCfg.SubscriptionType, subscriptionKey)
 	if err != nil {
-		return fmt.Errorf("Failed to create Pulsar subscription %v", err)
+		return fmt.Errorf("failed to create Pulsar subscription %v", err)
 	}
 
 	terminate := make(chan *SubCloseSignal, 2)
-	WriteWebhook(subscriptionKey, terminate)
+	wb.WriteWebhook(subscriptionKey, terminate)
 	defer close(terminate)
 	ctx := context.Background()
 
@@ -189,18 +194,18 @@ func ConsumeLoop(url, token, topic, subscriptionKey string, whCfg model.WebhookC
 	retryMax := 3
 	for {
 		if retry > retryMax {
-			cancelConsumer(subscriptionKey)
+			wb.cancelConsumer(subscriptionKey)
 			return fmt.Errorf("consumer retried %d times, max reached", retryMax)
 		}
 		msg, err := c.Receive(ctx)
 		if err != nil {
-			log.Infof("error from consumer loop receive: %v\n", err)
+			wb.l.Infof("error from consumer loop receive: %v\n", err)
 			retry++
 			ticker := time.NewTicker(time.Duration(2*retry) * time.Second)
 			defer ticker.Stop()
 			select {
 			case <-terminate:
-				log.Infof("subscription %s received signal to exit consumer loop", subscriptionKey)
+				wb.l.Infof("subscription %s received signal to exit consumer loop", subscriptionKey)
 				return nil
 			case <-ticker.C:
 				//reconnect after error
@@ -211,8 +216,8 @@ func ConsumeLoop(url, token, topic, subscriptionKey string, whCfg model.WebhookC
 			}
 		} else if msg != nil {
 			retry = 0
-			if log.GetLevel() == log.DebugLevel {
-				log.Debugf("PulsarMessageId:%#v", msg.ID())
+			if wb.l.Level == log.DebugLevel {
+				wb.l.Debugf("PulsarMessageId:%v", msg.ID())
 			}
 			headers = append(headers, fmt.Sprintf("PulsarMessageId:%#v", msg.ID()))
 			headers = append(headers, "PulsarPublishedTime:"+msg.PublishTime().String())
@@ -229,61 +234,58 @@ func ConsumeLoop(url, token, topic, subscriptionKey string, whCfg model.WebhookC
 			if json.Valid(data) {
 				headers = append(headers, "content-type:application/json")
 			}
-			if log.GetLevel() == log.DebugLevel {
-				log.Debug(string(data))
-			}
 			pushAndAck(c, msg, whCfg.URL, data, headers)
 		}
 	}
 
 }
 
-func run() {
+func (wb *WebhookBroker) run() {
 	// key is hash of topic name and pulsar url, and subscription name
 	subscriptionSet := make(map[string]bool)
 
-	for _, cfg := range LoadConfig() {
+	for _, cfg := range wb.LoadConfig() {
 		for _, whCfg := range cfg.Webhooks {
 			topic := cfg.TopicFullName
 			token := cfg.Token
 			url := cfg.PulsarURL
 			subscriptionKey := cfg.Key + whCfg.URL
 			status := whCfg.WebhookStatus
-			_, ok := ReadWebhook(subscriptionKey)
+			_, ok := wb.ReadWebhook(subscriptionKey)
 			if status == model.Activated {
 				subscriptionSet[subscriptionKey] = true
 				if !ok {
-					log.Infof("start activated webhook for topic subscription %v", subscriptionKey)
-					go ConsumeLoop(url, token, topic, subscriptionKey, whCfg)
+					wb.l.Infof("start activated webhook for topic subscription %v", subscriptionKey)
+					go wb.ConsumeLoop(url, token, topic, subscriptionKey, whCfg)
 				}
 			}
 		}
 	}
 
 	// cancel any webhook which is no longer required to be activated by the database
-	for k := range webhooks {
-		if subscriptionSet[k] != true {
-			log.Infof("cancel webhook consumer subscription key %s", k)
-			cancelConsumer(k)
+	for k := range wb.webhooks {
+		if !subscriptionSet[k] {
+			wb.l.Infof("cancel webhook consumer subscription key %s", k)
+			wb.cancelConsumer(k)
 		}
 	}
-	log.Infof("load webhooks size %d", len(webhooks))
+	wb.l.Infof("load webhooks size %d", len(wb.webhooks))
 }
 
 // LoadConfig loads the entire topic documents from the database
-func LoadConfig() []*model.TopicConfig {
-	cfgs, err := singleDb.Load()
+func (wb *WebhookBroker) LoadConfig() []*model.TopicConfig {
+	cfgs, err := wb.dbHandler.Load()
 	if err != nil {
-		log.Errorf("failed to load topics from database error %v", err.Error())
+		wb.l.Errorf("failed to load topics from database error %v", err.Error())
 	}
 
 	return cfgs
 }
 
-func cancelConsumer(key string) error {
-	ok := DeleteWebhook(key)
+func (wb *WebhookBroker) cancelConsumer(key string) error {
+	ok := wb.DeleteWebhook(key)
 	if ok {
-		log.Infof("cancel consumer %v", key)
+		wb.l.Infof("cancel consumer %v", key)
 		pulsardriver.CancelPulsarConsumer(key)
 		return nil
 	}

src/main.go

@@ -28,7 +28,7 @@ func main() {
 	}
 
 	exit := make(chan bool) // future use to exit the main program if in broker only mode
-	util.Init()
+	config := util.Init()
 
 	flag.Parse()
 	log.Warnf("start server mode %s", mode)
@@ -37,7 +37,7 @@ func main() {
 	}
 
 	if util.IsBrokerRequired(&mode) {
-		broker.Init()
+		broker.Init(config)
 	}
 	if util.IsHTTPRouterRequired(&mode) {
 		route.Init()

src/route/router_test.go

@@ -0,0 +1,19 @@
+package route
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestEffectiveRoutes(t *testing.T) {
+	receiverRoutesLen := len(ReceiverRoutes)
+	restRoutesLen := len(RestRoutes)
+	prometheusLen := len(PrometheusRoute)
+	// mode := "hybrid"
+	// assert.Equal(t, len(GetEffectiveRoutes(&mode)), (receiverRoutesLen + restRoutesLen + prometheusLen))
+	mode := "rest"
+	assert.Equal(t, len(GetEffectiveRoutes(&mode)), (receiverRoutesLen + restRoutesLen + prometheusLen))
+	mode = "receiver"
+	assert.Equal(t, len(GetEffectiveRoutes(&mode)), (receiverRoutesLen + restRoutesLen + prometheusLen))
+}