lbryio/chainquery

package notifications

import (
    "net/http"
    "net/url"

    "github.com/lbryio/chainquery/metrics"

    "github.com/lbryio/lbry.go/v2/extras/errors"

    "github.com/sirupsen/logrus"
    "golang.org/x/net/context"
    "golang.org/x/sync/semaphore"
)

type subscriber struct {
    URL    string
    Type   string
    Params url.Values
}

var subscriptions map[string][]subscriber

// AddSubscriber adds a subscriber to the subscribers list for a type
func AddSubscriber(address, subType string, params map[string]interface{}) {
    if subscriptions == nil {
        subscriptions = make(map[string][]subscriber)
    }
    urlParams := url.Values{}
    for param, v := range params {
        value, ok := v.(string)
        if ok {
            urlParams.Set(param, value)
        }
    }
    subscriptions[subType] = append(subscriptions[subType], subscriber{URL: address, Type: subType, Params: urlParams})
}

// ClearSubscribers clears the list of subscribers
func ClearSubscribers() {
    subscriptions = make(map[string][]subscriber)
}

var notificationSem = semaphore.NewWeighted(20)

// Notify notifies the list of subscribers for a type
func Notify(t string, values url.Values) {
    err := notificationSem.Acquire(context.Background(), 1)
    if err != nil {
        logrus.Error(errors.Prefix("Notify", errors.Err(err)))
        return
    }
    defer notificationSem.Release(1)
    subs, ok := subscriptions[t]
    if ok {
        for _, s := range subs {
            for param, value := range s.Params {
                values.Set(param, value[0])
            }
            s.notify(values)
            metrics.Notifications.WithLabelValues(t).Inc()
        }
    }
}

func (s subscriber) notify(values url.Values) {
    res, err := http.PostForm(s.URL, values)
    if err != nil {
        logrus.Error(errors.Prefix("Notify", errors.Err(err)))
        return
    }
    defer func() {
        err = res.Body.Close()
        if err != nil {
            logrus.Error(errors.Prefix("Notify", errors.Err(err)))
        }
    }()
    //b, err := io.ReadAll(res.Body)
    //if err != nil {
    //    logrus.Error(errors.Prefix("Notify", errors.Err(err)))
    //}
    //if res.StatusCode != http.StatusOK {
    //    logrus.Errorln(string(b))
    //}
}