Go concurrency patterns with channels

Need common Go concurrency patterns: fan-out/fan-in, pipeline, worker pool, and timeout.

Essential Go concurrency patterns:

// 1. Worker Pool
func workerPool(jobs <-chan int, results chan<- int, numWorkers int) {
	var wg sync.WaitGroup
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for job := range jobs {
				results <- process(job)
			}
		}()
	}
	wg.Wait()
	close(results)
}

// 2. Fan-out / Fan-in
func fanOut(input <-chan int, n int) []<-chan int {
	channels := make([]<-chan int, n)
	for i := 0; i < n; i++ {
		channels[i] = worker(input)
	}
	return channels
}

func fanIn(channels ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)
	for _, ch := range channels {
		wg.Add(1)
		go func(c <-chan int) {
			defer wg.Done()
			for v := range c {
				out <- v
			}
		}(ch)
	}
	go func() { wg.Wait(); close(out) }()
	return out
}

// 3. Timeout with select
func fetchWithTimeout(url string, timeout time.Duration) (string, error) {
	ch := make(chan string, 1)
	errCh := make(chan error, 1)
	go func() {
		resp, err := http.Get(url)
		if err != nil {
			errCh <- err
			return
		}
		body, _ := io.ReadAll(resp.Body)
		ch <- string(body)
	}()

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

// 4. Pipeline
func generate(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

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

// Usage: for v := range square(square(generate(1,2,3))) { ... }

// 5. Done channel for cancellation
func doWork(done <-chan struct{}) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for {
			select {
			case <-done:
				return
			case out <- compute():
			}
		}
	}()
	return out
}

Go's concurrency model (goroutines + channels) enables clean concurrent code. These patterns handle the common needs: bounded concurrency, parallel processing, and cancellation.

Go concurrent programming