blob: f72cc1be713b9c0f0ca5438f95a089369fbcd531 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
|
package workerpool
import (
"net/http"
log "github.com/sirupsen/logrus"
)
const (
// ActivityPubWorkerPoolSize defines the number of concurrent HTTP ActivityPub requests.
ActivityPubWorkerPoolSize = 10
)
// Job struct bundling the ActivityPub and the payload in one struct.
type Job struct {
request *http.Request
}
var queue chan Job
// InitOutboundWorkerPool starts n go routines that await ActivityPub jobs.
func InitOutboundWorkerPool() {
queue = make(chan Job)
// start workers
for i := 1; i <= ActivityPubWorkerPoolSize; i++ {
go worker(i, queue)
}
}
// AddToOutboundQueue will queue up an outbound http request.
func AddToOutboundQueue(req *http.Request) {
log.Tracef("Queued request for ActivityPub destination %s", req.RequestURI)
queue <- Job{req}
}
func worker(workerID int, queue <-chan Job) {
log.Debugf("Started ActivityPub worker %d", workerID)
for job := range queue {
if err := sendActivityPubMessageToInbox(job); err != nil {
log.Errorf("ActivityPub destination %s failed to send Error: %s", job.request.RequestURI, err)
}
log.Tracef("Done with ActivityPub destination %s using worker %d", job.request.RequestURI, workerID)
}
}
func sendActivityPubMessageToInbox(job Job) error {
client := &http.Client{}
resp, err := client.Do(job.request)
if err != nil {
return err
}
defer resp.Body.Close()
return nil
}
|
