Method
Enqueue
(ctx context.Context, fn func(context.Context))
Source from the content-addressed store, hash-verified
| 18 | } |
| 19 | |
| 20 | func (q *Queue) Enqueue(ctx context.Context, fn func(context.Context)) { |
| 21 | q.mu.RLock() |
| 22 | if q.closed { |
| 23 | q.mu.RUnlock() |
| 24 | return |
| 25 | } |
| 26 | q.mu.RUnlock() |
| 27 | |
| 28 | // Don't start new tasks if context is already cancelled |
| 29 | if ctx.Err() != nil { |
| 30 | return |
| 31 | } |
| 32 | |
| 33 | q.wg.Go(func() { |
| 34 | // Check context again before executing |
| 35 | if ctx.Err() != nil { |
| 36 | return |
| 37 | } |
| 38 | fn(ctx) |
| 39 | }) |
| 40 | } |
| 41 | |
| 42 | // Wait waits for all active tasks to complete. |
| 43 | // It does not prevent new tasks from being enqueued while waiting. |