Gabe KangasandGitHub
a82efb0e9a
* fix(ap): add safe constructors, getters, and validators to ActivityPub actors to address #4701 * chore: replace nil check with the new Validate() method * chore(test): added test to verify the values extracted from actors * fix: return the specific error type + test for it * fix(ap): add recovery for each AP inbox worker for a worst case scenario * fix(ap): add additional safe accessor methods to other AP entities other than actors * chore(tests): add tests for the new safe accessors * fix(ap): handle empty public keys in AP actors
52 lines
1.1 KiB
Go
52 lines
1.1 KiB
Go
package inbox
|
|
|
|
import (
|
|
"runtime"
|
|
|
|
"github.com/owncast/owncast/activitypub/apmodels"
|
|
log "github.com/sirupsen/logrus"
|
|
)
|
|
|
|
// workerPoolSize defines the number of concurrent ActivityPub handlers.
|
|
var workerPoolSize = runtime.GOMAXPROCS(0)
|
|
|
|
// Job struct bundling the ActivityPub and the payload in one struct.
|
|
type Job struct {
|
|
request apmodels.InboxRequest
|
|
}
|
|
|
|
var queue chan Job
|
|
|
|
// InitInboxWorkerPool starts n go routines that await ActivityPub jobs.
|
|
func InitInboxWorkerPool() {
|
|
queue = make(chan Job)
|
|
|
|
// start workers
|
|
for i := 1; i <= workerPoolSize; i++ {
|
|
go worker(i, queue)
|
|
}
|
|
}
|
|
|
|
// AddToQueue will queue up an outbound http request.
|
|
func AddToQueue(req apmodels.InboxRequest) {
|
|
log.Tracef("Queued request for ActivityPub inbox handler")
|
|
queue <- Job{req}
|
|
}
|
|
|
|
func worker(workerID int, queue <-chan Job) {
|
|
log.Debugf("Started ActivityPub worker %d", workerID)
|
|
|
|
for job := range queue {
|
|
func() {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Errorf("Recovered from panic in ActivityPub worker %d: %v", workerID, r)
|
|
}
|
|
}()
|
|
handle(job.request)
|
|
}()
|
|
|
|
log.Tracef("Done with ActivityPub inbox handler using worker %d", workerID)
|
|
}
|
|
}
|