()
| 289 | } |
| 290 | |
| 291 | func (a *Acquirer) subscribe() { |
| 292 | subscribed := make(chan struct{}) |
| 293 | go func() { |
| 294 | defer close(subscribed) |
| 295 | eb := backoff.NewExponentialBackOff() |
| 296 | eb.MaxElapsedTime = 0 // retry indefinitely |
| 297 | eb.MaxInterval = dbMaxBackoff |
| 298 | bkoff := backoff.WithContext(eb, a.ctx) |
| 299 | var cancel context.CancelFunc |
| 300 | err := backoff.Retry(func() error { |
| 301 | cancelFn, err := a.ps.SubscribeWithErr(provisionerjobs.EventJobPosted, a.jobPosted) |
| 302 | if err != nil { |
| 303 | a.logger.Warn(a.ctx, "failed to subscribe to job postings", slog.Error(err)) |
| 304 | return err |
| 305 | } |
| 306 | cancel = cancelFn |
| 307 | return nil |
| 308 | }, bkoff) |
| 309 | if err != nil { |
| 310 | if a.ctx.Err() == nil { |
| 311 | a.logger.Error(a.ctx, "code bug: retry failed before context canceled", slog.Error(err)) |
| 312 | } |
| 313 | return |
| 314 | } |
| 315 | defer cancel() |
| 316 | bkoff.Reset() |
| 317 | a.logger.Debug(a.ctx, "subscribed to job postings") |
| 318 | |
| 319 | // unblock the outer function from returning |
| 320 | subscribed <- struct{}{} |
| 321 | |
| 322 | // hold subscriptions open until context is canceled |
| 323 | <-a.ctx.Done() |
| 324 | }() |
| 325 | <-subscribed |
| 326 | } |
| 327 | |
| 328 | func (a *Acquirer) jobPosted(ctx context.Context, message []byte, err error) { |
| 329 | if xerrors.Is(err, pubsub.ErrDroppedMessages) { |
no test coverage detected