Chapter 7: Channel Patterns

// Illustrative snippet — not a complete program
func generate(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case <-ctx.Done():
                return
            case out <- n:
            }
        }
    }()
    return out
}

// Illustrative snippet — not a complete program
func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case <-ctx.Done():
                return
            case out <- n * n:
            }
        }
    }()
    return out
}

// Illustrative snippet — not a complete program
func consume(in <-chan int) {
    for n := range in {
        fmt.Println(n)
    }
}

// Illustrative snippet — not a complete program
func expensiveConsumer(ctx context.Context, in <-chan Item) {
    for item := range in {
        if ctx.Err() != nil {
            return  // Don't start expensive work if cancelled
        }
        expensiveWork(item)
    }
}

// Pseudocode — replace T and U with concrete types
func stageName(ctx context.Context, in <-chan T) <-chan U {
    out := make(chan U)
    go func() {
        defer close(out)          // 1. Always close output
        for value := range in {   // 2. Process until input closes
            result := transform(value)
            select {
            case <-ctx.Done():    // 3. Respect cancellation
                return
            case out <- result:   // 4. Send result
            }
        }
    }()
    return out
}

// Illustrative snippet — not a complete program
// ✗ DANGEROUS: Blocks forever if receiver stops
for value := range in {
    out <- transform(value)  // Can't cancel mid-send
}

// ✓ SAFE: Can exit immediately when cancelled
for value := range in {
    select {
    case <-ctx.Done():
        return
    case out <- transform(value):
    }
}

// Illustrative snippet — not a complete program
for {
    select {
    case <-ctx.Done():
        return
    case n, ok := <-in:
        if !ok {
            return
        }
        // process n
    }
}

// Illustrative snippet — not a complete program
func main() {
    ctx := context.Background()

    // Build pipeline: generate → square
    numbers := generate(ctx, 1, 2, 3, 4, 5)
    squares := square(ctx, numbers)

    // Consume results
    for n := range squares {
        fmt.Println(n)  // 1, 4, 9, 16, 25
    }
}

// Illustrative snippet — not a complete program
func filter(
    ctx context.Context,
    in <-chan int,
    predicate func(int) bool,
) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if predicate(n) {
                select {
                case <-ctx.Done():
                    return
                case out <- n:
                }
            }
        }
    }()
    return out
}

func main() {
    ctx := context.Background()

    // Four-stage pipeline:
    // generate → square → filter evens → consume
    numbers := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    squared := square(ctx, numbers)
    evens := filter(ctx, squared, func(n int) bool {
        return n%2 == 0
    })

    for n := range evens {
        fmt.Println(n)  // 4, 16, 36, 64, 100
    }
}

// Illustrative snippet — not a complete program
func brokenSquare(
    ctx context.Context,
    in <-chan int,
) <-chan int {
    out := make(chan int)
    go func() {
        // Missing: defer close(out)
        for n := range in {
            select {
            case <-ctx.Done():
                return
            case out <- n * n:
            }
        }
    }()
    return out
}

// Consumer blocks forever after receiving all values
for n := range brokenSquare(ctx, generate(ctx, 1, 2, 3)) {
    // Receives 1, 4, 9, then blocks forever
}

// Illustrative snippet — not a complete program
func main() {
    ctx := context.Background()

    numbers := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    squares := square(ctx, numbers)

    // Only consume first 3 values
    count := 0
    for n := range squares {
        fmt.Println(n)
        count++
        if count >= 3 {
            break  // Stop consuming
        }
    }

    // Problem: generate and square goroutines
    // still running! They block forever.
}

// Illustrative snippet — not a complete program
func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()  // Ensure cleanup on any exit

    numbers := generate(ctx, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    squares := square(ctx, numbers)

    count := 0
    for n := range squares {
        fmt.Println(n)
        count++
        if count >= 3 {
            cancel()  // Signal all stages; cancel is idempotent
            break
        }
    }

    // All goroutines exit cleanly via ctx.Done()
}

package main

import (
    "context"
    "fmt"
    "time"
)

// generate emits integers 1..max (range-based variant)
func generate(ctx context.Context, max int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 1; i <= max; i++ {
            time.Sleep(10 * time.Millisecond) // Simulate I/O
            select {
            case <-ctx.Done():
                return
            case out <- i:
            }
        }
    }()
    return out
}

func square(ctx context.Context, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            select {
            case <-ctx.Done():
                return
            case out <- n * n:
            }
        }
    }()
    return out
}

func filter(
    ctx context.Context,
    in <-chan int,
    pred func(int) bool,
) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if pred(n) {
                select {
                case <-ctx.Done():
                    return
                case out <- n:
                }
            }
        }
    }()
    return out
}

func main() {
    ctx, cancel := context.WithTimeout(
        context.Background(),
        100*time.Millisecond,
    )
    defer cancel()

    // Build pipeline:
    // generate → square → filter (evens only)
    numbers := generate(ctx, 100)
    squared := square(ctx, numbers)
    evens := filter(ctx, squared, func(n int) bool {
        return n%2 == 0
    })

    // Consume results
    for n := range evens {
        fmt.Println(n)
    }

    fmt.Println("Done")
}

// Illustrative snippet — not a complete program
for value := range in {
    // Takes 10s — can't cancel mid-work
    result := expensiveTransform(value)
    select {
    case <-ctx.Done():
        return
    case out <- result:
    }
}

// Illustrative snippet — not a complete program
for value := range in {
    // Check before expensive work
    select {
    case <-ctx.Done():
        return
    default:
    }

    result := expensiveTransform(value)

    select {
    case <-ctx.Done():
        return
    case out <- result:
    }
}

// Illustrative snippet — not a complete program
type Result struct {
    Value int
    Err   error
}

func process(
    ctx context.Context,
    in <-chan string,
) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for s := range in {
            n, err := strconv.Atoi(s)
            select {
            case <-ctx.Done():
                return
            case out <- Result{Value: n, Err: err}:
            }
        }
    }()
    return out
}

// Consumer: ALWAYS check Err before using Value
// Value is the zero value of its type when Err != nil
for r := range process(ctx, strings) {
    if r.Err != nil {
        log.Printf("error: %v", r.Err)
        continue
    }
    use(r.Value)
}

// Illustrative snippet — not a complete program
func squareBuffered(
    ctx context.Context,
    in <-chan int,
) <-chan int {
    out := make(chan int, 10)  // Buffer smooths throughput
    go func() {
        defer close(out)
        for n := range in {
            select {
            case <-ctx.Done():
                return
            case out <- n * n:
            }
        }
    }()
    return out
}

// Illustrative snippet — not a complete program
// ✗ OVERKILL: Pipeline for 5 numbers
// Creates 2 goroutines, 2 channels
for n := range square(ctx, generate(ctx, 1, 2, 3, 4, 5)) {
    process(n)
}

// ✓ SIMPLER: Direct loop is clearer and faster
for _, n := range []int{1, 2, 3, 4, 5} {
    process(n * n)
}

// Rule of thumb: Pipelines pay off when:
// - Items involve non-trivial computation or I/O, OR
// - Stages involve I/O or blocking, OR
// - Data is streaming (won't fit in memory)

// Illustrative snippet — not a complete program
func main() {
    ctx := context.Background()

    urls := generate(ctx, getURLs()...)  // Fast: generates instantly
    responses := fetch(ctx, urls)       // SLOW: network I/O
    results := parse(ctx, responses)     // Fast: CPU parsing

    for r := range results {
        process(r)
    }
}

// Illustrative snippet — not a complete program
func fanOut(
    ctx context.Context,
    in <-chan int,
    workers int,
    work func(int) int,
) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    // Start N workers, all reading from
    // same input, writing to same output
    for i := 0; i < workers; i++ {
        wg.Go(func() {
            for n := range in {
                result := work(n)
                select {
                case <-ctx.Done():
                    return
                case out <- result:
                }
            }
        })
    }

    // Close output when all workers finish
    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

// Illustrative snippet — not a complete program
result := work(n)  // Runs to completion even if ctx cancelled
select {
case <-ctx.Done():
    return
case out <- result:
}

// Illustrative snippet — not a complete program
func fanIn(
    ctx context.Context,
    channels ...<-chan int,
) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    // Start a forwarder for each input channel
    for _, ch := range channels {
        wg.Go(func() {
            for n := range ch {
                select {
                case <-ctx.Done():
                    return
                case out <- n:
                }
            }
        })
    }

    // Close when all inputs are exhausted
    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

// Illustrative snippet — not a complete program
// drain consumes all values and returns the count
func drain[T any](ch <-chan T) int {
    count := 0
    for range ch {
        count++
    }
    return count
}

// Pseudocode — define generate, process for your workload
for workers := 1; workers <= 64; workers *= 2 {
    tasks := generate(ctx, 10000)  // Fresh channel each iteration
    start := time.Now()
    results := fanOut(ctx, tasks, workers, process)
    count := drain(results)
    throughput := float64(count) / time.Since(start).Seconds()
    fmt.Printf("%2d workers: %.0f items/sec\n",
        workers, throughput)
}

// Illustrative snippet — not a complete program
// ✗ DANGEROUS: Unbounded — one goroutine per task
for task := range tasks {
    go func() {
        process(task)  // 1M tasks = 1M goroutines
    }()
}

// ✓ SAFE: Fixed worker count
results := fanOut(ctx, tasks, 10, process)

package main

import (
    "context"
    "fmt"
    "io"
    "net/http"
    "net/http/httptest"
    "sync"
    "time"
)

type FetchResult struct {
    URL  string
    Size int
    Err  error
}

// generate emits each URL (string variant of the pipeline generator)
func generate(
    ctx context.Context,
    urls ...string,
) <-chan string {
    out := make(chan string)
    go func() {
        defer close(out)
        for _, url := range urls {
            select {
            case <-ctx.Done():
                return
            case out <- url:
            }
        }
    }()
    return out
}

func fetch(
    ctx context.Context,
    url string,
) FetchResult {
    req, err := http.NewRequestWithContext(
        ctx, "GET", url, nil,
    )
    if err != nil {
        return FetchResult{URL: url, Err: err}
    }

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return FetchResult{URL: url, Err: err}
    }
    defer resp.Body.Close()

    body, err := io.ReadAll(resp.Body)
    if err != nil {
        return FetchResult{URL: url, Err: err}
    }

    return FetchResult{URL: url, Size: len(body)}
}

func fanOutFetch(
    ctx context.Context,
    urls <-chan string,
    workers int,
) <-chan FetchResult {
    out := make(chan FetchResult)
    var wg sync.WaitGroup

    for i := 0; i < workers; i++ {
        wg.Go(func() {
            for url := range urls {
                result := fetch(ctx, url)
                select {
                case <-ctx.Done():
                    return
                case out <- result:
                }
            }
        })
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

// A local server, so the byte counts and the timing below are the
// same on every machine. Each page takes 200ms to produce.
func newSlowServer() *httptest.Server {
    return httptest.NewServer(http.HandlerFunc(
        func(w http.ResponseWriter, r *http.Request) {
            time.Sleep(200 * time.Millisecond)
            size := map[string]int{
                "/a": 1024, "/b": 2048,
                "/c": 4096, "/d": 8192,
            }[r.URL.Path]
            w.Write(make([]byte, size))
        },
    ))
}

func main() {
    ctx, cancel := context.WithTimeout(
        context.Background(),
        10*time.Second,
    )
    defer cancel()

    srv := newSlowServer()
    defer srv.Close()

    urls := []string{
        srv.URL + "/a", srv.URL + "/b",
        srv.URL + "/c", srv.URL + "/d",
    }

    start := time.Now()

    // Pipeline: generate → fan-out fetch → consume
    urlChan := generate(ctx, urls...)
    results := fanOutFetch(ctx, urlChan, 3)

    for r := range results {
        if r.Err != nil {
            fmt.Printf("Error fetching %s: %v\n",
                r.URL, r.Err)
        } else {
            fmt.Printf("Fetched %s: %d bytes\n",
                r.URL[len(srv.URL):], r.Size)
        }
    }

    // 4 pages at 200ms each: ~800ms one at a time, ~400ms
    // with 3 workers (three in the first wave, one in the second).
    fmt.Printf("Total time: %v\n",
        time.Since(start).Round(10*time.Millisecond))
}

// Illustrative snippet — not a complete program
type IndexedResult struct {
    Index int
    Value int
}

// orderedFanOut processes values in parallel
// and returns results in input order.
// On cancellation, unprocessed indices contain zero values.
func orderedFanOut(
    ctx context.Context,
    values []int,
    workers int,
    work func(int) int,
) ([]int, error) {

    type indexedTask struct {
        index int
        value int
    }

    // Send indexed tasks
    tasks := make(chan indexedTask)
    go func() {
        defer close(tasks)
        for i, v := range values {
            select {
            case <-ctx.Done():
                return
            case tasks <- indexedTask{i, v}:
            }
        }
    }()

    // Fan-out processing
    results := make(chan IndexedResult)
    var wg sync.WaitGroup

    for i := 0; i < workers; i++ {
        wg.Go(func() {
            for t := range tasks {
                r := IndexedResult{
                    Index: t.index,
                    Value: work(t.value),
                }
                select {
                case <-ctx.Done():
                    return
                case results <- r:
                }
            }
        })
    }

    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect and reorder
    ordered := make([]int, len(values))
    received := 0
    for r := range results {
        ordered[r.Index] = r.Value
        received++
    }

    // Check for partial completion
    if received < len(values) {
        return ordered, fmt.Errorf(
            "cancelled: received %d of %d",
            received, len(values),
        )
    }

    return ordered, nil
}

// Illustrative snippet — not a complete program
func processPipeline(
    ctx context.Context,
    urls []string,
) <-chan Analyzed {
    // Stage 1: Generate URLs (fast — single)
    urlsCh := generate(ctx, urls...)

    // Stage 2: Fetch (slow — fan out 10 workers)
    fetched := fanOutFetch(ctx, urlsCh, 10)

    // Stage 3: Parse (fast — single stage)
    parsed := parse(ctx, fetched)

    // Stage 4: Analyze (slow — fan out 5 workers)
    analyzed := fanOutAnalyze(ctx, parsed, 5)

    return analyzed
}

// Illustrative snippet — not a complete program
// ✗ OVERKILL: Fan-out for fast, simple operations
results := fanOut(ctx, numbers, 10, func(n int) int {
    return n * 2
})

// ✓ SIMPLER: Sequential for lightweight work
for n := range numbers {
    process(n * 2)
}

// Illustrative snippet — not a complete program
func handleUpload(w http.ResponseWriter, r *http.Request) {
    // ✗ PROBLEM: Unbounded concurrency
    go processImage(r.Body)  // One goroutine per request
    w.WriteHeader(http.StatusAccepted)
}

// Illustrative snippet — not a complete program
type Job struct {
    ID   int
    Data string
}

type Pool struct {
    jobs    chan Job
    workers int
    wg      sync.WaitGroup
}

func NewPool(workers, queueSize int) *Pool {
    p := &Pool{
        jobs:    make(chan Job, queueSize),
        workers: workers,
    }

    // Start workers immediately
    for i := 0; i < workers; i++ {
        p.wg.Go(func() {
            p.worker(i)
        })
    }

    return p
}

func (p *Pool) worker(id int) {
    for job := range p.jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job.ID)
        time.Sleep(100 * time.Millisecond)  // Simulate work
    }
}

func (p *Pool) Submit(job Job) {
    p.jobs <- job  // Blocks when queue is full
}

func (p *Pool) Stop() {
    close(p.jobs)  // Signal workers to stop
    p.wg.Wait()    // Wait for workers to finish
}

// Illustrative snippet — not a complete program
var (
    ErrPoolStopped  = errors.New("pool stopped")
    ErrQueueFull    = errors.New("queue full")
    ErrSubmitTimeout = errors.New("submit timed out")
)

type Pool struct {
    jobs    chan Job
    workers int
    wg      sync.WaitGroup
    ctx     context.Context
    cancel  context.CancelFunc
}

func NewPool(ctx context.Context, workers, queueSize int) *Pool {
    ctx, cancel := context.WithCancel(ctx)

    p := &Pool{
        jobs:    make(chan Job, queueSize),
        workers: workers,
        ctx:     ctx,
        cancel:  cancel,
    }

    for i := 0; i < workers; i++ {
        p.wg.Go(func() {
            p.worker(i)
        })
    }

    return p
}

func (p *Pool) worker(id int) {

    for {
        select {
        case <-p.ctx.Done():
            return  // Immediate exit on cancellation
        case job, ok := <-p.jobs:
            if !ok {
                return  // Jobs channel closed
            }
            p.process(job)
        }
    }
}

func (p *Pool) Submit(job Job) error {
    select {
    case <-p.ctx.Done():
        return ErrPoolStopped
    case p.jobs <- job:
        return nil
    }
}

// Illustrative snippet — not a complete program
jobs: make(chan Job, queueSize)  // Buffered queue

// Illustrative snippet — not a complete program
func (p *Pool) TrySubmit(job Job) error {
    select {
    case <-p.ctx.Done():
        return ErrPoolStopped
    case p.jobs <- job:
        return nil
    default:
        return ErrQueueFull
    }
}

func (p *Pool) SubmitTimeout(job Job, timeout time.Duration) error {
    timer := time.NewTimer(timeout)
    defer timer.Stop()  // Release now, not at next GC

    select {
    case <-p.ctx.Done():
        return ErrPoolStopped
    case <-timer.C:
        return ErrSubmitTimeout
    case p.jobs <- job:
        return nil
    }
}

// Illustrative snippet — not a complete program
func handleUpload(w http.ResponseWriter, r *http.Request) {
    body, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "Bad request", http.StatusBadRequest)
        return
    }

    job := Job{ID: generateID(), Data: string(body)}

    if err := pool.TrySubmit(job); err != nil {
        http.Error(w, "Server busy", http.StatusServiceUnavailable)
        return
    }

    w.WriteHeader(http.StatusAccepted)
}

// Illustrative snippet — not a complete program
type Result struct {
    JobID int
    Value string
    Err   error
}

type Pool struct {
    jobs    chan Job
    results chan Result
    wg      sync.WaitGroup
    ctx     context.Context
    cancel  context.CancelFunc
}

func NewPool(ctx context.Context, workers, queueSize int) *Pool {
    ctx, cancel := context.WithCancel(ctx)

    p := &Pool{
        jobs:    make(chan Job, queueSize),
        results: make(chan Result, queueSize),  // Buffer for results
        ctx:     ctx,
        cancel:  cancel,
    }

    for i := 0; i < workers; i++ {
        p.wg.Go(func() {
            p.worker(i)
        })
    }

    return p
}

func (p *Pool) worker(id int) {

    for {
        select {
        case <-p.ctx.Done():
            return
        case job, ok := <-p.jobs:
            if !ok {
                return
            }
            result := p.safeProcess(job)
            select {
            case <-p.ctx.Done():
                return
            case p.results <- result:
            }
        }
    }
}

func (p *Pool) Results() <-chan Result {
    return p.results
}

// Illustrative snippet — not a complete program
// Start consumer FIRST — drains results so workers don't block
go func() {
    for result := range pool.Results() {
        handleResult(result)
    }
}()

// THEN submit jobs — safe because consumer is already draining
for i := 0; i < 100; i++ {
    pool.Submit(Job{ID: i})
}

// Illustrative snippet — not a complete program
// ✗ DANGEROUS: Race between Submit and Stop
func (p *Pool) Submit(job Job) error {
    select {
    case <-p.ctx.Done():
        return ErrPoolStopped
    case p.jobs <- job:  // Can panic if Stop() closes p.jobs
        return nil
    }
}

func (p *Pool) Stop() {
    close(p.jobs)  // Race: Submit might be about to send
    p.wg.Wait()
}

// Illustrative snippet — not a complete program
type Pool struct {
    jobs    chan Job
    results chan Result
    wg      sync.WaitGroup
    ctx     context.Context
    cancel  context.CancelFunc

    // Shutdown coordination
    shutdownOnce sync.Once
    shutdownDone chan struct{} // closed when shutdown finishes
    shutdownErr  error         // written once, before the close
    stopped      atomic.Bool
}

func (p *Pool) Submit(job Job) (err error) {
    if p.stopped.Load() {
        return ErrPoolStopped
    }
    defer func() {
        // Catches send-on-closed-channel panic from the race window
        // between stopped.Load() and the channel send in select.
        if recover() != nil {
            err = ErrPoolStopped
        }
    }()

    select {
    case <-p.ctx.Done():
        return ErrPoolStopped
    case p.jobs <- job:
        return nil
    }
}

func (p *Pool) TrySubmit(job Job) (err error) {
    if p.stopped.Load() {
        return ErrPoolStopped
    }
    defer func() {
        if recover() != nil {
            err = ErrPoolStopped
        }
    }()

    select {
    case <-p.ctx.Done():
        return ErrPoolStopped
    case p.jobs <- job:
        return nil
    default:
        return ErrQueueFull
    }
}

// Shutdown gracefully stops the pool with optional timeout.
// Pass 0 for no timeout (wait indefinitely for drain).
// Safe for concurrent calls and idempotent: every caller blocks
// until the shutdown finishes and every caller gets the same error.
func (p *Pool) Shutdown(timeout time.Duration) error {
    p.stopped.Store(true)  // Reject new submissions

    p.shutdownOnce.Do(func() {
        defer close(p.shutdownDone)
        close(p.jobs)  // Signal workers: no more jobs

        drained := make(chan struct{})
        go func() {
            p.wg.Wait()
            close(drained)
        }()

        if timeout == 0 {
            // Graceful: wait indefinitely for drain
            <-drained
            close(p.results)
            return
        }

        // Graceful with timeout: try to drain, then force
        select {
        case <-drained:
            // Workers finished within timeout
            close(p.results)
            return
        case <-time.After(timeout):
        }

        // Timeout: cancel, then give workers a grace period to
        // notice. A worker that ignores ctx must not be able to
        // block Shutdown forever — that is the case the timeout
        // exists for, so it has to be bounded here too.
        p.cancel()
        select {
        case <-drained:
            close(p.results)
            p.shutdownErr = errors.New(
                "shutdown timeout: jobs abandoned",
            )
        case <-time.After(timeout):
            // results stays open: a stuck worker may still send.
            p.shutdownErr = errors.New(
                "shutdown timeout: workers did not stop",
            )
        }
    })

    <-p.shutdownDone  // late callers wait for the real outcome
    return p.shutdownErr
}

// Illustrative snippet — not a complete program
for result := range pool.Results() {
    if result.Err != nil {
        log.Printf("Job %d failed: %v", result.JobID, result.Err)
        continue
    }
    use(result.Value)
}

// Illustrative snippet — not a complete program
func (p *Pool) safeProcess(job Job) (result Result) {
    defer func() {
        if r := recover(); r != nil {
            // Log with stack trace for debugging
            log.Printf("Worker panic on job %d: %v\n%s",
                job.ID, r, debug.Stack())

            result = Result{
                JobID: job.ID,
                Err:   fmt.Errorf("panic: %v", r),
            }
        }
    }()
    return p.process(job)
}

// Illustrative snippet — not a complete program
func (p *Pool) worker(id int) {
    for {
        select {
        case <-p.ctx.Done():
            return
        case job, ok := <-p.jobs:
            if !ok {
                return
            }
            result := p.process(job)
            select {
            case <-p.ctx.Done():
                return
            case p.results <- result:
            }
            if result.Err != nil {
                p.cancel()  // Cancel AFTER sending the error result
            }
        }
    }
}

// Illustrative snippet — not a complete program
func (p *Pool) process(ctx context.Context, job Job) Result {
    for i := 0; i < 100; i++ {
        select {
        case <-ctx.Done():
            return Result{JobID: job.ID, Err: ctx.Err()}
        default:
        }
        // Do chunk of work...
    }
    return Result{JobID: job.ID, Value: "done"}
}

package main

import (
    "context"
    "errors"
    "fmt"
    "log"
    "runtime/debug"
    "sync"
    "sync/atomic"
    "time"
)

var (
    ErrPoolStopped   = errors.New("pool stopped")
    ErrQueueFull     = errors.New("queue full")
    ErrSubmitTimeout = errors.New("submit timed out")
)

type Job struct {
    ID      int
    Payload string
}

type Result struct {
    JobID int
    Value string
    Err   error
}

type Pool struct {
    jobs    chan Job
    results chan Result
    wg      sync.WaitGroup
    ctx     context.Context
    cancel  context.CancelFunc

    shutdownOnce sync.Once
    shutdownDone chan struct{} // closed when shutdown finishes
    shutdownErr  error         // written once, before the close
    stopped      atomic.Bool
}

func NewPool(ctx context.Context, workers, queueSize int) *Pool {
    ctx, cancel := context.WithCancel(ctx)

    p := &Pool{
        jobs:         make(chan Job, queueSize),
        results:      make(chan Result, queueSize),
        ctx:          ctx,
        cancel:       cancel,
        shutdownDone: make(chan struct{}),
    }

    for i := 0; i < workers; i++ {
        p.wg.Go(func() {
            p.worker(i)
        })
    }

    return p
}

func (p *Pool) worker(id int) {

    for {
        select {
        case <-p.ctx.Done():
            log.Printf("Worker %d: stopping (cancelled)", id)
            return
        case job, ok := <-p.jobs:
            if !ok {
                log.Printf("Worker %d: stopping (jobs closed)", id)
                return
            }
            result := p.safeProcess(job)
            select {
            case <-p.ctx.Done():
                return
            case p.results <- result:
            }
        }
    }
}

func (p *Pool) safeProcess(job Job) (result Result) {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("PANIC processing job %d: %v\n%s",
                job.ID, r, debug.Stack())
            result = Result{
                JobID: job.ID,
                Err:   fmt.Errorf("panic: %v", r),
            }
        }
    }()
    return p.process(job)
}

func (p *Pool) process(job Job) Result {
    time.Sleep(100 * time.Millisecond) // Simulate work

    return Result{
        JobID: job.ID,
        Value: fmt.Sprintf("processed: %s", job.Payload),
    }
}

func (p *Pool) Submit(job Job) (err error) {
    if p.stopped.Load() {
        return ErrPoolStopped
    }
    defer func() {
        if recover() != nil {
            err = ErrPoolStopped
        }
    }()

    select {
    case <-p.ctx.Done():
        return ErrPoolStopped
    case p.jobs <- job:
        return nil
    }
}

func (p *Pool) TrySubmit(job Job) (err error) {
    if p.stopped.Load() {
        return ErrPoolStopped
    }
    defer func() {
        if recover() != nil {
            err = ErrPoolStopped
        }
    }()

    select {
    case <-p.ctx.Done():
        return ErrPoolStopped
    case p.jobs <- job:
        return nil
    default:
        return ErrQueueFull
    }
}

func (p *Pool) Results() <-chan Result {
    return p.results
}

// Shutdown is the implementation from "Production-Safe
// Implementation" above: every caller waits for the real outcome and
// every caller gets the same error.
func (p *Pool) Shutdown(timeout time.Duration) error {
    p.stopped.Store(true)

    p.shutdownOnce.Do(func() {
        defer close(p.shutdownDone)
        close(p.jobs)

        drained := make(chan struct{})
        go func() {
            p.wg.Wait()
            close(drained)
        }()

        if timeout == 0 {
            <-drained
            close(p.results)
            return
        }

        select {
        case <-drained:
            close(p.results)
            return
        case <-time.After(timeout):
        }

        // A worker that ignores ctx must not be able to block
        // Shutdown forever — that is the case the timeout exists
        // for, so the wait after cancel is bounded too.
        p.cancel()
        select {
        case <-drained:
            close(p.results)
            p.shutdownErr = errors.New(
                "shutdown timeout: jobs abandoned",
            )
        case <-time.After(timeout):
            p.shutdownErr = errors.New(
                "shutdown timeout: workers did not stop",
            )
        }
    })

    <-p.shutdownDone
    return p.shutdownErr
}

func main() {
    ctx := context.Background()
    pool := NewPool(ctx, 3, 10)

    // Start result consumer FIRST
    var consumerWg sync.WaitGroup
    consumerWg.Go(func() {
        for result := range pool.Results() {
            if result.Err != nil {
                log.Printf("job %d err: %v",
                    result.JobID, result.Err)
            } else {
                fmt.Printf("Job %d: %s\n",
                    result.JobID, result.Value)
            }
        }
    })

    // Submit jobs
    for i := 0; i < 10; i++ {
        job := Job{ID: i, Payload: fmt.Sprintf("task-%d", i)}
        if err := pool.Submit(job); err != nil {
            log.Printf("Submit failed: %v", err)
        }
    }

    // Graceful shutdown with timeout
    if err := pool.Shutdown(5 * time.Second); err != nil {
        log.Printf("Shutdown: %v", err)
    }

    consumerWg.Wait()
    fmt.Println("All jobs complete")
}

// Illustrative snippet — not a complete program
func fetchWithTimeout(
    ctx context.Context,
    url string,
    timeout time.Duration,
) ([]byte, error) {
    result := make(chan []byte, 1)
    errCh := make(chan error, 1)

    go func() {
        data, err := fetch(ctx, url)
        if err != nil {
            errCh <- err
            return
        }
        result <- data
    }()

    timer := time.NewTimer(timeout)
    defer timer.Stop()

    select {
    case data := <-result:
        return data, nil
    case err := <-errCh:
        return nil, err
    case <-timer.C:
        return nil, fmt.Errorf("timeout after %v", timeout)
    }
}

// Illustrative snippet — not a complete program
func or(channels ...<-chan struct{}) <-chan struct{} {
    switch len(channels) {
    case 0:
        return nil  // nil channel blocks forever in select (no-op)
    case 1:
        return channels[0]
    }

    orDone := make(chan struct{})
    go func() {
        defer close(orDone)

        switch len(channels) {
        case 2:
            select {
            case <-channels[0]:
            case <-channels[1]:
            }
        default:
            // Clip the capacity so append allocates instead of
            // writing orDone into the caller's backing array.
            rest := channels[3:len(channels):len(channels)]
            select {
            case <-channels[0]:
            case <-channels[1]:
            case <-channels[2]:
            case <-or(append(rest, orDone)...):
            }
        }
    }()

    return orDone
}

// Illustrative snippet — not a complete program
func doWork(
    ctx context.Context,
    userCancel, systemShutdown <-chan struct{},
) error {
    // Wait for ANY cancellation source
    select {
    case <-or(ctx.Done(), userCancel, systemShutdown):
        return errors.New("cancelled")
    case result := <-doExpensiveWork(ctx):
        return processResult(result)
    }
}

// Illustrative snippet — not a complete program
// fetchFirstResponse returns the first response (success OR error).
// For first successful response, use fetchFirstSuccess.
func fetchFirstResponse(
    ctx context.Context, urls []string,
) ([]byte, error) {
    if len(urls) == 0 {
        return nil, errors.New("no URLs provided")
    }

    ctx, cancel := context.WithCancel(ctx)
    defer cancel()  // Cancel remaining requests when first completes

    type result struct {
        data []byte
        err  error
    }

    results := make(chan result, len(urls)) // Buffered: no leaks

    // One slot per goroutine, so this send can never block and
    // never needs a select on ctx.Done to stay leak-free.
    for _, url := range urls {
        go func() {
            data, err := fetch(ctx, url)
            results <- result{data, err}
        }()
    }

    r := <-results
    return r.data, r.err
}

// Illustrative snippet — not a complete program
func fetchFirstSuccess(
    ctx context.Context,
    urls []string,
) ([]byte, error) {
    if len(urls) == 0 {
        return nil, errors.New("no URLs provided")
    }

    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    type result struct {
        data []byte
        err  error
    }

    results := make(chan result, len(urls))

    for _, url := range urls {
        go func() {
            data, err := fetch(ctx, url)
            results <- result{data, err}
        }()
    }

    var firstErr error
    for i := 0; i < len(urls); i++ {
        r := <-results
        if r.err == nil {
            return r.data, nil  // First success wins
        }
        if firstErr == nil {
            firstErr = r.err
        }
    }

    // Returns first error only; use errors.Join to collect all
    return nil, firstErr
}

// Illustrative snippet — not a complete program
func hedgedFetch(
    ctx context.Context,
    url string,
    hedgeDelay time.Duration,
) ([]byte, error) {
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    type result struct {
        data []byte
        err  error
    }

    // Two slots, two senders: neither send can ever block.
    results := make(chan result, 2)

    // Primary request
    go func() {
        data, err := fetch(ctx, url)
        results <- result{data, err}
    }()

    // Hedged request after delay. In production, send this to a
    // different replica or endpoint than the primary.
    go func() {
        select {
        case <-ctx.Done():
            results <- result{nil, ctx.Err()}
        case <-time.After(hedgeDelay):
            data, err := fetch(ctx, url)
            results <- result{data, err}
        }
    }()

    // Take the first SUCCESS, not the first response. A primary that
    // fails fast is exactly the case hedging exists to survive, so
    // returning its error would defeat the pattern. See "First-Response
    // vs First-Success" above.
    var firstErr error
    for i := 0; i < 2; i++ {
        r := <-results
        if r.err == nil {
            return r.data, nil
        }
        if firstErr == nil {
            firstErr = r.err
        }
    }
    return nil, firstErr
}

// Illustrative snippet — not a complete program
var (
    totalRequests = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "requests_total",
    })
    hedgesSent = prometheus.NewCounter(prometheus.CounterOpts{
        Name: "hedged_requests_total",
    })
)

// In hedgedFetch, before sending hedge request:
hedgesSent.Inc()

// Illustrative snippet — not a complete program
// ✗ VERBOSE: Check done in every receive
for {
    select {
    case <-ctx.Done():
        return
    case v, ok := <-in:
        if !ok {
            return
        }
        process(v)
    }
}

// Illustrative snippet — not a complete program
// orDone wraps a channel to respect context cancellation.
func orDone[T any](ctx context.Context, in <-chan T) <-chan T {
    out := make(chan T)
    go func() {
        defer close(out)
        for {
            select {
            case <-ctx.Done():
                return
            case v, ok := <-in:
                if !ok {
                    return
                }
                select {
                case <-ctx.Done():
                    return
                case out <- v:
                }
            }
        }
    }()
    return out
}

// Illustrative snippet — not a complete program
// ✓ CLEAN: Range over wrapped channel
for v := range orDone(ctx, in) {
    process(v)
}

// Illustrative snippet — not a complete program
// ✗ BROKEN: Only one select
case v, ok := <-in:
    if !ok { return }
    out <- v  // ← BLOCKS if no receiver!

// Illustrative snippet — not a complete program
// ✓ CORRECT: Second select checks cancellation on send
select {
case <-ctx.Done():
    return  // Exit instead of blocking on send
case out <- v:
}

// Illustrative snippet — not a complete program
// Third-party library returns channel, doesn't accept context
dataChan := legacyLib.Subscribe(topic)

// Wrap it to make it cancellable
for event := range orDone(ctx, dataChan) {
    handleEvent(event)
}

// Illustrative snippet — not a complete program
// ✗ DANGEROUS: Unbuffered channel
results := make(chan []byte)  // Unbuffered!

for _, url := range urls {
    go func() {
        results <- fetch(url)  // Losers block here
    }()
}

return <-results  // Returns first, others leak

// Illustrative snippet — not a complete program
results := make(chan Result, len(urls)) // Buffered

for _, url := range urls {
    go func() {
        results <- fetch(ctx, url) // Buffer: no block
    }()
}

return <-results  // Losers send to buffer and exit

// Illustrative snippet — not a complete program
ctx, cancel := context.WithCancel(ctx)
defer cancel()  // Cancels losing goroutines

for _, url := range urls {
    go func() {
        result := fetch(ctx, url)  // Respects context
        select {
        case <-ctx.Done():
            return  // Exit without sending
        case results <- result:
        }
    }()
}

package main

import (
    "context"
    "errors"
    "fmt"
    "io"
    "net/http"
    "net/http/httptest"
    "time"
)

type Response struct {
    Data   []byte
    Source string
}

func fetch(ctx context.Context, url string) ([]byte, error) {
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
    if err != nil {
        return nil, err
    }

    // Production: use http.Client{Timeout: 10*time.Second}
    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()

    // A 500 from one replica has to count as a failure, or the
    // failover never happens: the caller would accept the error
    // page as data from the first source that answered.
    if resp.StatusCode >= 400 {
        return nil, fmt.Errorf("%s: %s", url, resp.Status)
    }

    return io.ReadAll(resp.Body)
}

func fetchWithRedundancy(
    ctx context.Context,
    urls []string,
    timeout time.Duration,
) (Response, error) {
    if len(urls) == 0 {
        return Response{}, errors.New("no URLs provided")
    }

    ctx, cancel := context.WithTimeout(ctx, timeout)
    defer cancel()

    type result struct {
        data   []byte
        source string
        err    error
    }

    results := make(chan result, len(urls))

    // One slot per source, so no send can block and none leaks.
    for _, url := range urls {
        go func() {
            data, err := fetch(ctx, url)
            results <- result{data, url, err}
        }()
    }

    var lastErr error
    for i := 0; i < len(urls); i++ {
        select {
        case <-ctx.Done():
            if lastErr != nil {
                return Response{}, lastErr
            }
            return Response{}, ctx.Err()
        case r := <-results:
            if r.err == nil {
                return Response{Data: r.data, Source: r.source}, nil
            }
            lastErr = r.err
        }
    }

    return Response{}, fmt.Errorf("all sources failed: %w", lastErr)
}

// Three stand-in replicas so this file runs as written. The primary
// fails fast, replica 1 is slow, replica 2 answers first.
func replicas() (*httptest.Server, []string) {
    mux := http.NewServeMux()
    mux.HandleFunc("/primary", func(
        w http.ResponseWriter, r *http.Request,
    ) {
        http.Error(w, "boom", http.StatusInternalServerError)
    })
    mux.HandleFunc("/replica1", func(
        w http.ResponseWriter, r *http.Request,
    ) {
        time.Sleep(300 * time.Millisecond)
        w.Write([]byte("payload from replica 1"))
    })
    mux.HandleFunc("/replica2", func(
        w http.ResponseWriter, r *http.Request,
    ) {
        time.Sleep(50 * time.Millisecond)
        w.Write([]byte("payload from replica 2"))
    })

    srv := httptest.NewServer(mux)
    return srv, []string{
        srv.URL + "/primary",
        srv.URL + "/replica1",
        srv.URL + "/replica2",
    }
}

func main() {
    ctx := context.Background()

    srv, urls := replicas()
    defer srv.Close()

    start := time.Now()
    resp, err := fetchWithRedundancy(ctx, urls, 5*time.Second)

    if err != nil {
        fmt.Printf("Error: %v\n", err)
        return
    }

    fmt.Printf("Success from %s in %v (%d bytes)\n",
        resp.Source[len(srv.URL):],
        time.Since(start).Round(10*time.Millisecond),
        len(resp.Data))
}

// Illustrative snippet — not a complete program
// ✗ OVERKILL: Or-channel for simple timeout
done := or(operation(), timeoutCh)

// ✓ SIMPLER: Context handles timeout + cancellation
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
result := operation(ctx)

// Illustrative snippet — not a complete program
func tee[T any](
    ctx context.Context,
    in <-chan T,
) (<-chan T, <-chan T) {
    out1 := make(chan T)
    out2 := make(chan T)

    go func() {
        defer close(out1)
        defer close(out2)

        for val := range in {
            // Local copies for select cases
            o1, o2 := out1, out2

            // Send to both outputs
            for i := 0; i < 2; i++ {
                select {
                case <-ctx.Done():
                    return
                case o1 <- val:
                    o1 = nil  // Disable this case after sending
                case o2 <- val:
                    o2 = nil
                }
            }
        }
    }()

    return out1, out2
}

// Illustrative snippet — not a complete program
func main() {
    ctx := context.Background()

    events := generateEvents(ctx)
    forLogging, forProcessing := tee(ctx, events)

    var wg sync.WaitGroup
    // Consumer 1: Log everything
    wg.Go(func() {
        for e := range forLogging {
            log.Printf("Event: %v", e)
        }
    })

    // Consumer 2: Process everything
    wg.Go(func() {
        for e := range forProcessing {
            process(e)
        }
    })

    wg.Wait()
}

// Illustrative snippet — not a complete program
func bufferedTee[T any](
    ctx context.Context,
    in <-chan T,
    bufSize int,
) (<-chan T, <-chan T) {
    out1 := make(chan T, bufSize)  // Buffered
    out2 := make(chan T, bufSize)

    go func() {
        defer close(out1)
        defer close(out2)

        for val := range in {
            o1, o2 := out1, out2
            for i := 0; i < 2; i++ {
                select {
                case <-ctx.Done():
                    return
                case o1 <- val:
                    o1 = nil
                case o2 <- val:
                    o2 = nil
                }
            }
        }
    }()

    return out1, out2
}

// Illustrative snippet — not a complete program
func lossyTee[T any](
    ctx context.Context,
    in <-chan T,
    bufSize int,
) (<-chan T, <-chan T) {
    out1 := make(chan T, bufSize)
    out2 := make(chan T, bufSize)

    go func() {
        defer close(out1)
        defer close(out2)

        for val := range in {
            // Exit early if context cancelled
            select {
            case <-ctx.Done():
                return
            default:
            }

            // Non-blocking sends: each consumer
            // receives independently. If buffer
            // full, val dropped for that consumer.
            select {
            case out1 <- val:
            default:
                // Buffer full—drop for consumer 1
            }

            select {
            case out2 <- val:
            default:
                // Buffer full—drop for consumer 2
            }
        }
    }()

    return out1, out2
}

// Illustrative snippet — not a complete program
func broadcast[T any](
    ctx context.Context,
    in <-chan T,
    n int,
) []<-chan T {
    outputs := make([]chan T, n)
    for i := range outputs {
        outputs[i] = make(chan T)
    }

    go func() {
        defer func() {
            for _, ch := range outputs {
                close(ch)
            }
        }()

        for val := range in {
            for _, ch := range outputs {
                select {
                case <-ctx.Done():
                    return
                case ch <- val:
                }
            }
        }
    }()

    // Convert to receive-only
    result := make([]<-chan T, n)
    for i, ch := range outputs {
        result[i] = ch
    }
    return result
}

// Illustrative snippet — not a complete program
events := generateEvents(ctx)
consumers := broadcast(ctx, events, 3)

// In production, use a WaitGroup to wait for all consumers
go processEvents(consumers[0])
go logEvents(consumers[1])
go updateMetrics(consumers[2])

// Illustrative snippet — not a complete program
type Hub[T any] struct {
    mu          sync.RWMutex
    subscribers map[int]chan T
    nextID      int
    bufSize     int
    closed      bool
}

func NewHub[T any](bufSize int) *Hub[T] {
    return &Hub[T]{
        subscribers: make(map[int]chan T),
        bufSize:     bufSize,
    }
}

// Subscribe returns a subscription ID and channel.
// If the hub is closed, returns (-1, closed channel) so the caller's
// range loop exits immediately.
func (h *Hub[T]) Subscribe() (int, <-chan T) {
    h.mu.Lock()
    defer h.mu.Unlock()

    if h.closed {
        ch := make(chan T)
        close(ch)
        return -1, ch
    }

    id := h.nextID
    h.nextID++

    ch := make(chan T, h.bufSize)
    h.subscribers[id] = ch

    return id, ch
}

func (h *Hub[T]) Unsubscribe(id int) {
    h.mu.Lock()
    defer h.mu.Unlock()

    if ch, ok := h.subscribers[id]; ok {
        close(ch)
        delete(h.subscribers, id)
    }
}

func (h *Hub[T]) Publish(val T) {
    h.mu.RLock()
    defer h.mu.RUnlock()

    if h.closed {
        return  // Silently ignore publishes after close
    }

    for _, ch := range h.subscribers {
        select {
        case ch <- val:
        default:
            // Subscriber too slow, drop
        }
    }
}

// Close shuts down the hub. Safe to call multiple times.
func (h *Hub[T]) Close() {
    h.mu.Lock()
    defer h.mu.Unlock()

    if h.closed {
        return  // Idempotent
    }
    h.closed = true

    for _, ch := range h.subscribers {
        close(ch)
    }
    h.subscribers = nil
}

// Illustrative snippet — not a complete program
hub := NewHub[Event](100)
defer hub.Close()

// Subscribers can join anytime
id1, events1 := hub.Subscribe()
id2, events2 := hub.Subscribe()

go func() {
    for event := range events1 {
        processEvent(event)
    }
}()

go func() {
    for event := range events2 {
        logEvent(event)
    }
}()

// Publisher sends to all current subscribers
for event := range eventSource {
    hub.Publish(event)
}

// Subscribers can leave
hub.Unsubscribe(id1)

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"
)

type Event struct {
    ID   int
    Type string
    Data string
}

func generateEvents(ctx context.Context) <-chan Event {
    out := make(chan Event)
    go func() {
        defer close(out)
        events := []Event{
            {1, "click", "button-submit"},
            {2, "view", "page-home"},
            {3, "purchase", "item-123"},
            {4, "click", "button-cancel"},
            {5, "view", "page-checkout"},
        }
        for _, e := range events {
            select {
            case <-ctx.Done():
                return
            case out <- e:
                time.Sleep(100 * time.Millisecond)
            }
        }
    }()
    return out
}

func tee[T any](
    ctx context.Context,
    in <-chan T,
) (<-chan T, <-chan T) {
    out1, out2 := make(chan T), make(chan T)
    go func() {
        defer close(out1)
        defer close(out2)
        for val := range in {
            o1, o2 := out1, out2
            for i := 0; i < 2; i++ {
                select {
                case <-ctx.Done():
                    return
                case o1 <- val:
                    o1 = nil
                case o2 <- val:
                    o2 = nil
                }
            }
        }
    }()
    return out1, out2
}

func main() {
    ctx, cancel := context.WithTimeout(
        context.Background(), 2*time.Second,
    )
    defer cancel()

    events := generateEvents(ctx)
    forLogging, forProcessing := tee(ctx, events)

    var wg sync.WaitGroup
    // Consumer 1: Log all events
    wg.Go(func() {
        for e := range forLogging {
            log.Printf(
                "[AUDIT] Event %d: %s - %s",
                e.ID, e.Type, e.Data,
            )
        }
    })

    // Consumer 2: Count by type
    wg.Go(func() {
        counts := make(map[string]int)
        for e := range forProcessing {
            counts[e.Type]++
        }
        fmt.Printf("Event counts: %v\n", counts)
    })

    wg.Wait()
}

// Illustrative snippet — not a complete program
// Composed: Pipeline with fan-out, tee for monitoring
func processWithMonitoring(
    ctx context.Context,
    items <-chan Item,
) <-chan Result {
    // Stage 1: Fan-out the slow processing (10 workers)
    processed := fanOutProcess(ctx, items, 10)

    // Stage 2: Buffered tee for monitoring + storage
    forStorage, forMetrics := bufferedTee(ctx, processed, 64)

    // Observer: metrics collection (buffered, won't block pipeline)
    go func() {
        for p := range forMetrics {
            recordMetric(p)
        }
    }()

    // Primary path continues
    return forStorage
}

// Illustrative snippet — not a complete program
// Worker pool feeding into pipeline
func serviceWithPool(ctx context.Context) {
    pool := NewPool(ctx, 10, 100)

    // Results flow into pipeline for post-processing
    validated := validate(ctx, pool.Results())
    stored := store(ctx, validated)

    // Consume final results
    for result := range stored {
        respond(result)
    }
}

// Or-done wrapping third-party channel in pipeline
func processExternalEvents(ctx context.Context) {
    // Third-party library doesn't respect context
    externalEvents := thirdPartyLib.Subscribe()

    // Wrap with or-done, then pipeline
    events := orDone(ctx, externalEvents)
    enriched := enrich(ctx, events)

    for e := range enriched {
        handle(e)
    }
}

// Hedged requests with fan-in for redundancy
func resilientFetch(
    ctx context.Context, urls []string,
) <-chan Response {
    channels := make([]<-chan Response, len(urls))

    for i, url := range urls {
        channels[i] = hedgedFetch(ctx, url, 50*time.Millisecond)
    }

    // Fan-in all hedged results
    return fanIn(ctx, channels...)
}