Navy.gif/owncast
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

⚡ execute webhooks in parallel (and 🐛 update webhooks.last_used correctly) (#1505)

Browse Source 
* ⚡ execute webhooks in parallel

* 🐛 update the webhooks.last_used column correctly
This commit is contained in:
Jannik 2021-11-02 03:13:20 +01:00 committed by GitHub
parent 3fdae7afb9
commit 2278fec70a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 18 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
core/data/webhooks.go
View File

@ -102,15 +102,15 @@ func GetWebhooksForEvent(event models.EventType) []models.Webhook {
webhooks := make([]models.Webhook, 0)
var query = `SELECT * FROM (
WITH RECURSIVE split(url, event, rest) AS (
SELECT url, '', events || ',' FROM webhooks
WITH RECURSIVE split(id, url, event, rest) AS (
SELECT id, url, '', events || ',' FROM webhooks
UNION ALL
SELECT url,
SELECT id, url,
substr(rest, 0, instr(rest, ',')),
substr(rest, instr(rest, ',')+1)
FROM split
WHERE rest <> '')
SELECT url, event
SELECT id, url, event
FROM split
WHERE event <> ''
) AS webhook WHERE event IS "` + event + `"`
@ -123,15 +123,17 @@ func GetWebhooksForEvent(event models.EventType) []models.Webhook {
defer rows.Close()
for rows.Next() {
var id int
var url string
if err := rows.Scan(&url, &event); err != nil {
if err := rows.Scan(&id, &url, &event); err != nil {
log.Debugln(err)
log.Error("There is a problem with the database.")
break
}
singleWebhook := models.Webhook{
ID: id,
URL: url,
}
@ -194,7 +196,7 @@ func GetWebhooks() ([]models.Webhook, error) { //nolint
}
// SetWebhookAsUsed will update the last used time for a webhook.
func SetWebhookAsUsed(id string) error {
func SetWebhookAsUsed(webhook models.Webhook) error {
tx, err := _db.Begin()
if err != nil {
return err
@ -206,7 +208,7 @@ func SetWebhookAsUsed(id string) error {
}
defer stmt.Close()
if _, err := stmt.Exec(id); err != nil {
if _, err := stmt.Exec(webhook.ID); err != nil {
return err
}

16
core/webhooks/webhooks.go
View File

@ -36,20 +36,22 @@ func SendEventToWebhooks(payload WebhookEvent) {
webhooks := data.GetWebhooksForEvent(payload.Type)
for _, webhook := range webhooks {
log.Debugf("Event %s sent to Webhook %s", payload.Type, webhook.URL)
if err := sendWebhook(webhook.URL, payload); err != nil {
log.Errorf("Event: %s failed to send to webhook: %s Error: %s", payload.Type, webhook.URL, err)
}
go func(webhook models.Webhook, payload WebhookEvent) {
log.Debugf("Event %s sent to Webhook %s", payload.Type, webhook.URL)
if err := sendWebhook(webhook, payload); err != nil {
log.Errorf("Event: %s failed to send to webhook: %s Error: %s", payload.Type, webhook.URL, err)
}
}(webhook, payload)
}
}
func sendWebhook(url string, payload WebhookEvent) error {
func sendWebhook(webhook models.Webhook, payload WebhookEvent) error {
jsonText, err := json.Marshal(payload)
if err != nil {
return err
}
req, err := http.NewRequest("POST", url, bytes.NewReader(jsonText))
req, err := http.NewRequest("POST", webhook.URL, bytes.NewReader(jsonText))
if err != nil {
return err
}
@ -65,7 +67,7 @@ func sendWebhook(url string, payload WebhookEvent) error {
defer resp.Body.Close()
if err := data.SetWebhookAsUsed(url); err != nil {
if err := data.SetWebhookAsUsed(webhook); err != nil {
log.Warnln(err)
}