bwilliams
18db462d89
fetcher.Run(ctx, conf, since) becomes New(&conf.Sources, opts...) and Run(ctx, since). Each worker now takes only its own source's config, and Run starts each one through a closure that binds it. Cancelling ctx stops the workers through the Fetcher's background context, and Run always waits for every worker before returning their joined errors. The findatender client no longer holds the configuration: New takes only the API root, ReloadConfig is gone, and request tracing follows logging.IsDebugEnabled(). The FindAOpportunity worker is now FindATender. job-search serve runs the fetcher alongside the HTTP server. A source that fails stops the server, and the server stopping stops the sources; with no sources enabled the server keeps running. fetcher service and fetcher fts-once use the new APIs.
61 lines
1.4 KiB
Go
61 lines
1.4 KiB
Go
package fetcher
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
"time"
|
|
|
|
"git.avsdev.uk/Go/job-search/internal/config"
|
|
"git.avsdev.uk/Go/job-search/internal/util"
|
|
)
|
|
|
|
// sleepDuration is how long the sleeper's simulated work takes.
|
|
const sleepDuration = 60 * time.Second
|
|
|
|
// runSleeper runs the sleeper test source until ctx is cancelled: every
|
|
// configured interval it simulates work by sleeping for sleepDuration.
|
|
func runSleeper(
|
|
ctx context.Context,
|
|
logger *slog.Logger,
|
|
conf *config.JSCSourceSleeper,
|
|
since time.Time,
|
|
) error {
|
|
blockShutdown := conf.BlockShutdown
|
|
worker := util.Worker{
|
|
Logger: tagged(logger),
|
|
Name: "Sleeper",
|
|
Interval: conf.Interval,
|
|
Task: func(taskCtx context.Context, since time.Time) error {
|
|
if blockShutdown {
|
|
return sleep(taskCtx, true, sleepDuration)
|
|
}
|
|
return sleep(ctx, false, sleepDuration)
|
|
},
|
|
}
|
|
return worker.Run(ctx, since)
|
|
}
|
|
|
|
// sleep simulates work taking duration, returning early with ctx.Err() if ctx
|
|
// is cancelled. With blockShutdown, the work itself ignores ctx (as a
|
|
// non-cooperative call would) and only the wait for it stops.
|
|
func sleep(
|
|
ctx context.Context, blockShutdown bool, duration time.Duration,
|
|
) error {
|
|
return util.AwaitTaskE(ctx, func() error {
|
|
if blockShutdown {
|
|
time.Sleep(duration)
|
|
return nil
|
|
}
|
|
|
|
timer := time.NewTimer(duration)
|
|
defer timer.Stop()
|
|
|
|
select {
|
|
case <-timer.C:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
})
|
|
}
|