beforeTaskStart updates Task status from queued to processing
(ctx context.Context, task Task, to task.Status)
| 344 | |
| 345 | // beforeTaskStart updates Task status from queued to processing |
| 346 | func (q *queue) transitStatus(ctx context.Context, task Task, to task.Status) (err error) { |
| 347 | old := task.Status() |
| 348 | transition, ok := stateTransitions[task.Status()][to] |
| 349 | if !ok { |
| 350 | err = fmt.Errorf("invalid state transition from %s to %s", old, to) |
| 351 | } else { |
| 352 | if innerErr := transition(ctx, task, to, q); innerErr != nil { |
| 353 | err = fmt.Errorf("failed to transit Task status from %s to %s: %w", old, to, innerErr) |
| 354 | } |
| 355 | } |
| 356 | |
| 357 | l := logging.FromContext(ctx) |
| 358 | if err != nil { |
| 359 | l.Error(err.Error()) |
| 360 | } |
| 361 | |
| 362 | l.Info("Task %d status changed from %q to %q.", task.ID(), old, to) |
| 363 | return |
| 364 | } |
| 365 | |
| 366 | // schedule to check worker number |
| 367 | func (q *queue) schedule() { |