Minor fixes:
- Fixed function name typo - Fetcher/Run now returns a combination of errors if any workers error out. All other workers are ordered to shutdown if any one does error out - findATender now returns a pointer instead of an instance from New
This commit is contained in:
1 parent
5cb3601e1f
commit
005b335371
6 files changed
+31
-15
No files matched your search
@@ -47,7 +47,7 @@ opportunities.
|
|||||||
HiddenDefaultCmd: true,
|
HiddenDefaultCmd: true,
|
||||||
},
|
},
|
||||||
PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
|
PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
|
||||||
cli.SetupIntialLogging(cmd)
|
cli.SetupInitialLogging(cmd)
|
||||||
|
|
||||||
slog.Debug("root.PersistentPreRun")
|
slog.Debug("root.PersistentPreRun")
|
||||||
slog.Debug(
|
slog.Debug(
|
||||||
|
|||||||
@@ -52,9 +52,7 @@ func commandService(opts *cli.Options) *cobra.Command {
|
|||||||
stop()
|
stop()
|
||||||
}()
|
}()
|
||||||
|
|
||||||
fetcher.Run(ctx, conf, since)
|
return fetcher.Run(ctx, conf, since)
|
||||||
|
|
||||||
return nil
|
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
cmd.Flags().Time(
|
cmd.Flags().Time(
|
||||||
|
|||||||
@@ -7,7 +7,7 @@ import (
|
|||||||
"github.com/spf13/cobra"
|
"github.com/spf13/cobra"
|
||||||
)
|
)
|
||||||
|
|
||||||
func SetupIntialLogging(cmd *cobra.Command) {
|
func SetupInitialLogging(cmd *cobra.Command) {
|
||||||
// Start logging to terminal first - any commands that want to use file
|
// Start logging to terminal first - any commands that want to use file
|
||||||
// logging should activate it themselves
|
// logging should activate it themselves
|
||||||
logging.LogToTerminal()
|
logging.LogToTerminal()
|
||||||
|
|||||||
@@ -2,6 +2,7 @@ package fetcher
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log/slog"
|
"log/slog"
|
||||||
"sync"
|
"sync"
|
||||||
@@ -13,10 +14,21 @@ import (
|
|||||||
|
|
||||||
var ErrNoNewReleases = fmt.Errorf("%w: no new releases", util.ErrWorkerNoWork)
|
var ErrNoNewReleases = fmt.Errorf("%w: no new releases", util.ErrWorkerNoWork)
|
||||||
|
|
||||||
|
// Run starts a worker for each enabled source and blocks until every worker
|
||||||
|
// has stopped. Workers stop when ctx is cancelled, or when any worker stops
|
||||||
|
// with an error, which starts a shutdown of all of them. The errors are
|
||||||
|
// logged as they happen and returned joined; nil means a clean shutdown.
|
||||||
func Run(
|
func Run(
|
||||||
ctx context.Context, conf *config.JobSearchConfig, since time.Time,
|
ctx context.Context, conf *config.JobSearchConfig, since time.Time,
|
||||||
) {
|
) error {
|
||||||
var wg sync.WaitGroup
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
|
var (
|
||||||
|
wg sync.WaitGroup
|
||||||
|
mu sync.Mutex
|
||||||
|
errs []error
|
||||||
|
)
|
||||||
|
|
||||||
start := func(
|
start := func(
|
||||||
name string,
|
name string,
|
||||||
@@ -26,7 +38,11 @@ func Run(
|
|||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
if err := run(ctx, conf, since); err != nil {
|
if err := run(ctx, conf, since); err != nil {
|
||||||
slog.Error("source stopped", "source", name, "err", err)
|
slog.Error("source stopped, shutting down", "source", name, "err", err)
|
||||||
|
mu.Lock()
|
||||||
|
errs = append(errs, fmt.Errorf("%s: %w", name, err))
|
||||||
|
mu.Unlock()
|
||||||
|
cancel()
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
@@ -42,4 +58,5 @@ func Run(
|
|||||||
}
|
}
|
||||||
|
|
||||||
wg.Wait()
|
wg.Wait()
|
||||||
|
return errors.Join(errs...)
|
||||||
}
|
}
|
||||||
@@ -119,6 +119,8 @@ func recursiveRepairDates(v any) bool {
|
|||||||
return changed
|
return changed
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// isDateField reports whether a field holds a date, by its name (see
|
||||||
|
// repairDates).
|
||||||
func isDateField(name string) bool {
|
func isDateField(name string) bool {
|
||||||
return strings.HasPrefix(name, "date") ||
|
return strings.HasPrefix(name, "date") ||
|
||||||
strings.Contains(name, "Date") ||
|
strings.Contains(name, "Date") ||
|
||||||
@@ -322,10 +324,9 @@ func (ft *FindATender) fetchURI(
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
if ft.ctx == nil {
|
ctx := ft.ctx
|
||||||
copied := *ft
|
if ctx == nil {
|
||||||
copied.ctx = context.Background()
|
ctx = context.Background()
|
||||||
ft = &copied
|
|
||||||
}
|
}
|
||||||
|
|
||||||
if res.StatusCode() == 429 {
|
if res.StatusCode() == 429 {
|
||||||
@@ -344,7 +345,7 @@ func (ft *FindATender) fetchURI(
|
|||||||
slog.Warn(fmt.Sprintf(
|
slog.Warn(fmt.Sprintf(
|
||||||
"API rate limit hit, waiting as requested for %d seconds", retrySec,
|
"API rate limit hit, waiting as requested for %d seconds", retrySec,
|
||||||
))
|
))
|
||||||
if rateSleep(ft.ctx, time.Duration(retrySec)*time.Second) != nil {
|
if rateSleep(ctx, time.Duration(retrySec)*time.Second) != nil {
|
||||||
return nil, ErrFetchCancelled
|
return nil, ErrFetchCancelled
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -16,8 +16,8 @@ type FindATender struct {
|
|||||||
ctx context.Context
|
ctx context.Context
|
||||||
}
|
}
|
||||||
|
|
||||||
func New(conf *config.JobSearchConfig, apiRoot string) FindATender {
|
func New(conf *config.JobSearchConfig, apiRoot string) *FindATender {
|
||||||
ft := FindATender{
|
ft := &FindATender{
|
||||||
config: conf,
|
config: conf,
|
||||||
apiRoot: apiRoot,
|
apiRoot: apiRoot,
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in new issue
Block a user