Channels

package main

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

// 1) Unbuffered — synchronous handshake
func main() {
    ch := make(chan int)
    go func() { ch <- 42 }()
    fmt.Println(<-ch)      // 42
}

// 2) Buffered — async up to N
ch := make(chan int, 3)
ch <- 1; ch <- 2; ch <- 3       // doesn't block
// ch <- 4                       // blocks — buffer full
fmt.Println(<-ch)               // 1

// 3) Closing — signals 'no more values'
func producer(ch chan<- int) {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch)
}

func main() {
    ch := make(chan int)
    go producer(ch)
    for v := range ch {         // loops until ch is closed
        fmt.Println(v)
    }
}

// 4) Two-value receive — detect close
v, ok := <-ch
if !ok {
    fmt.Println("channel closed")
}

// 5) Direction-typed channels — readable API
func send(ch chan<- int, v int) { ch <- v }     // send-only
func recv(ch <-chan int) int    { return <-ch } // receive-only

// 6) select — wait on multiple channels
select {
case v := <-a:
    fmt.Println("from a:", v)
case v := <-b:
    fmt.Println("from b:", v)
case out <- 42:
    fmt.Println("sent 42")
case <-time.After(2 * time.Second):
    fmt.Println("timeout")
default:
    fmt.Println("non-blocking — nothing ready")
}

// 7) Worker pool — distribute work across N goroutines
func workerPool(jobs <-chan Job, results chan<- Result, n int) {
    var wg sync.WaitGroup
    for i := 0; i < n; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for j := range jobs {
                results <- process(j)
            }
        }(i)
    }
    wg.Wait()
    close(results)
}

func main() {
    jobs := make(chan Job, 100)
    results := make(chan Result, 100)

    go workerPool(jobs, results, 4)

    // Send jobs
    go func() {
        for _, j := range allJobs {
            jobs <- j
        }
        close(jobs)
    }()

    // Receive results
    for r := range results {
        fmt.Println(r)
    }
}

// 8) Fan-out (one producer, many consumers) + fan-in (merge results)
func fanIn(chs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    wg.Add(len(chs))
    for _, c := range chs {
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(c)
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

// 9) Cancellation via context
func worker(ctx context.Context, jobs <-chan Job) {
    for {
        select {
        case <-ctx.Done():
            return                                // cancel signal
        case j, ok := <-jobs:
            if !ok { return }
            process(j)
        }
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    go worker(ctx, jobs)
    // ...
}

// 10) done channel — manual signal
done := make(chan struct{})
go func() {
    // ... work ...
    close(done)
}()
<-done                        // blocks until close

// 11) Pipeline — stage1 → stage2 → stage3
func gen(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums { out <- n }
    }()
    return out
}

func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for v := range in { out <- v * v }
    }()
    return out
}

func main() {
    for v := range square(gen(1, 2, 3, 4, 5)) {
        fmt.Println(v)    // 1, 4, 9, 16, 25
    }
}

// 12) Channel direction in API — clarity
// Accept the narrowest channel type your function needs:
//   func read(ch <-chan T)
//   func write(ch chan<- T)
//   func bidir(ch chan T)
// Compiler enforces; safer + more readable.

// 13) Rate limiting with a ticker
ticker := time.NewTicker(100 * time.Millisecond)
defer ticker.Stop()
for range ticker.C {
    process()
}

// 14) Common bugs
//   • Sending to a CLOSED channel → panic
//   • Closing twice → panic
//   • Receiving from a closed channel returns the zero value forever
//   • Nil channels block forever (in select, useful for disabling cases)
//   • Goroutine leaks: goroutine waits on a channel that no one writes to → orphan
//   • Buffered channel with no consumer → eventually blocks

// 15) When to use channels vs other sync
// Channels    : data flow, hand-off between goroutines, cancellation, pipelines
// sync.Mutex  : protect shared state (counters, caches, maps)
// sync.WaitGroup : wait for N goroutines to finish
// sync.Once    : single initialisation
// atomic      : single-value lock-free updates (counters, flags)
// errgroup    : goroutine-aware error propagation + cancellation

// 16) errgroup — common modern pattern
import "golang.org/x/sync/errgroup"

func main() {
    g, ctx := errgroup.WithContext(context.Background())
    for _, url := range urls {
        url := url
        g.Go(func() error {
            return fetch(ctx, url)
        })
    }
    if err := g.Wait(); err != nil {
        log.Fatal(err)
    }
}
// Errgroup cancels the ctx on first error → all other goroutines exit cleanly.

// 17) Tips
//   • 'Don't communicate by sharing memory; share memory by communicating'
//   • Close on the SENDER side, not the receiver
//   • Use context for cancellation; channels for data
//   • Avoid buffered channels as a quick fix — usually means you need backpressure
//   • For high-throughput pipelines, consider chunking values (batch of 100 per send)

ch := make(chan int, 2)
ch <- 1
ch <- 2
close(ch)
for v := range ch { fmt.Println(v) }