SoXX
802764092e
As mentioned in Issue #11, the folder structure got an overall as some file names Co-authored-by: Fenpaws <soxx@fenpa.ws> Reviewed-on: anthrove/e621-to-graph#14 Reviewed-by: Lennard Brinkhaus <lennard.brinkhaus@noreply.localhost> Co-authored-by: SoXX <fenpaws@noreply.localhost> Co-committed-by: SoXX <fenpaws@noreply.localhost>
63 lines
1.3 KiB
Go
63 lines
1.3 KiB
Go
package scheduler
|
|
|
|
import (
|
|
"context"
|
|
"git.dragse.it/anthrove/e621-to-graph/pkg/e621"
|
|
"git.dragse.it/anthrove/e621-to-graph/pkg/util/queue"
|
|
log "github.com/sirupsen/logrus"
|
|
"golang.org/x/time/rate"
|
|
)
|
|
|
|
type Scheduler struct {
|
|
queue queue.Queue
|
|
limiter *rate.Limiter
|
|
}
|
|
|
|
func NewScheduler() *Scheduler {
|
|
scheduler := &Scheduler{
|
|
queue: queue.NewQueue(),
|
|
limiter: nil,
|
|
}
|
|
go scheduler.StartExecutionHandler()
|
|
return scheduler
|
|
}
|
|
|
|
func Schedule[T e621.DataType](s *Scheduler, t e621.Task[T], username string, apiKey string) func() (T, error) {
|
|
channel := make(chan e621.DataResponse[T])
|
|
schedulerTask := NewSchedulerTaskImpl[T](t, channel, username, apiKey)
|
|
log.Debug("Psuh task")
|
|
err := s.queue.Push(schedulerTask)
|
|
log.Debug("Element pushed")
|
|
|
|
if err != nil {
|
|
return func() (T, error) {
|
|
var nil T
|
|
return nil, err
|
|
}
|
|
}
|
|
return func() (T, error) {
|
|
data := <-channel
|
|
return data.Data, data.Error
|
|
}
|
|
}
|
|
|
|
func (s *Scheduler) SetLimiter(limiter *rate.Limiter) {
|
|
s.limiter = limiter
|
|
}
|
|
|
|
func (s *Scheduler) StartExecutionHandler() {
|
|
for {
|
|
if s.limiter != nil {
|
|
s.limiter.Wait(context.Background())
|
|
}
|
|
s.queue.WaitForElement()
|
|
log.Debug("element found")
|
|
task, err := s.queue.Pop()
|
|
if err != nil {
|
|
continue
|
|
}
|
|
GetAPIRequest(task)
|
|
}
|
|
|
|
}
|