mirror of https://gitlab.com/ngerakines/tavern.git
79 lines
1.6 KiB
Go
79 lines
1.6 KiB
Go
package job
|
|
|
|
import (
|
|
"context"
|
|
"net/http"
|
|
"time"
|
|
|
|
"go.uber.org/zap"
|
|
|
|
"github.com/ngerakines/tavern/common"
|
|
"github.com/ngerakines/tavern/config"
|
|
"github.com/ngerakines/tavern/fed"
|
|
"github.com/ngerakines/tavern/storage"
|
|
)
|
|
|
|
type webfinger struct {
|
|
ctx context.Context
|
|
cancel context.CancelFunc
|
|
logger *zap.Logger
|
|
queue common.StringQueue
|
|
storage storage.Storage
|
|
httpClient common.HTTPClient
|
|
fedConfig config.FedConfig
|
|
}
|
|
|
|
func NewWebFingerWorker(logger *zap.Logger, queue common.StringQueue, storage storage.Storage, fedConfig config.FedConfig, httpClient *http.Client) Job {
|
|
return &webfinger{
|
|
logger: logger,
|
|
queue: queue,
|
|
storage: storage,
|
|
httpClient: httpClient,
|
|
fedConfig: fedConfig,
|
|
}
|
|
}
|
|
|
|
func (job *webfinger) Run(parent context.Context) error {
|
|
job.ctx, job.cancel = context.WithCancel(parent)
|
|
defer job.cancel()
|
|
|
|
for {
|
|
select {
|
|
case <-time.After(time.Second):
|
|
err := job.work()
|
|
if err != nil {
|
|
job.logger.Error("error processing work", zap.Error(err))
|
|
return err
|
|
}
|
|
case <-job.ctx.Done():
|
|
return ignoreCanceled(job.ctx.Err())
|
|
}
|
|
}
|
|
}
|
|
|
|
func (job *webfinger) Shutdown(parent context.Context) error {
|
|
job.cancel()
|
|
select {
|
|
case <-parent.Done():
|
|
return parent.Err()
|
|
case <-job.ctx.Done():
|
|
return job.ctx.Err()
|
|
}
|
|
}
|
|
|
|
func (job *webfinger) work() error {
|
|
work, err := job.queue.Take()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if len(work) == 0 {
|
|
return nil
|
|
}
|
|
|
|
_, err = fed.GetOrFetchActor(context.Background(), job.storage, job.logger, job.httpClient, work)
|
|
if err != nil {
|
|
job.logger.Warn("webfinger error", zap.Error(err), zap.String("location", work))
|
|
}
|
|
return nil
|
|
}
|