package main
 
import (
	"context"
	"fmt"
	"math/rand"
	"time"
)
 
type Result struct {
	Source string
	Value  string
	Err    error
}
 
func fetch(ctx context.Context, source
	// Simulate actual network latency with random sleep
	d := min + time.Duration(rand.Int63n(int64(max
	timer := time.NewTimer(d)
	defer timer.Stop()
 
	select {
	case <-ctx.Done():
		return Result{Source: source, Err: ctx.Err()}
	case <-timer.C:
		// Assume success
		return Result{Source: source, Value: "data from " +
	}
}
 
func firstN(ctx context.Context, sources
	if n <= 0 {
		return nil, fmt.Errorf("n must be > 0")
	}
	if n > len(sources) {
		n = len(sources)
	}
 
	// Create a cancelable context to stop slower workers later
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()
 
	// Use a buffered channel to prevent blocking if we return early (preventing goroutine leaks)
	resultsCh := make(chan Result, len(sources))
 
	// Fan-out: Start workers
	for _, s := range sources {
		s := s
		go func() {
			r := fetch(ctx, s, 150*time.Millisecond, 900*
			resultsCh <- r
		}()
	}
 
	// Fan-in: Collect only N results
	out := make([]Result, 0, n)
	var firstErr error
 
	for len(out) < n {
		r := <-resultsCh
		if r.Err != nil {
			// Ignore errors caused by our own cancellation, record actual errors
			if r.Err != context.Canceled && r.Err != context.DeadlineExceeded
				firstErr = r.Err
			}
			continue
		}
		out = append(out, r)
	}
 
	cancel() // We have N results, cancel the remaining workers immediately
 
	return out, firstErr
}
 
func main() {
	// Note: In Go 1.20+, global seeding is deprecated/automatic,
	// but keeping it for compatibility with older examples.
	rand.New(rand.NewSource(time.Now().UnixNano()))
 
	// Timeout is a safety net. Never trust the network.
	ctx, cancel := context.WithTimeout(context.Background(), 2*time
	defer cancel()
 
	sources := []string{"A-Search", "B-Catalog", "C-Pricing
	res, err := firstN(ctx, sources, 2)
 
	fmt.Println("Got results:")
	for _, r := range res {
		fmt.Printf("- %s: %s\n", r.Source,
	}
	fmt.Println("Error:", err)
}

package main
 
import (
	"context"
	"fmt"
	"sync"
	"time"
)
 
type Job struct {
	ID int
}
 
type Output struct {
	ID     int
	Status string
}
 
func worker(ctx context.Context, wid
	defer wg.Done()
 
	for {
		select {
		case <-ctx.Done():
			return
		case j, ok := <-jobs:
			if !ok {
				return
			}
			// Simulate actual work (CPU/IO intensive)
			time.Sleep(80 * time.Millisecond)
			results <- Output{ID: j.ID, Status: fmt
		}
	}
}
 
func main() {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
 
	const (
		workerCount = 4
		jobCount    = 20
	)
 
	// Buffer serves as a backpressure lever.
	// If the buffer is full, the producer will block.
	jobs := make(chan Job, 8)
	results := make(chan Output, 8)
 
	var wg sync.WaitGroup
 
	// 1. Start Workers
	wg.Add(workerCount)
	for i := 1; i <= workerCount; i++ {
		go worker(ctx, i, jobs, results, &wg)
	}
 
	// 2. Producer: Enqueue jobs
	go func() {
		for i := 1; i <= jobCount; i++ {
			jobs <- Job{ID: i}
		}
		close(jobs) // Signal that no more jobs are coming
	}()
 
	// 3. Closer: Wait for workers to finish, then close results
	go func() {
		wg.Wait()
		close(results)
	}()
 
	// 4. Collector: Consume results
	for r := range results {
		fmt.Printf("job %02d -> %s\n", r.ID,
	}
 
	fmt.Println("all done")
}

package main
 
import (
	"fmt"
	"sync"
	"time"
)
 
// Stage 1: Ingest (Generator)
func ingest(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}
 
// Stage 2: Parse (Parallel Processing)
// This stage scales horizontally (fan-out)
func parseParallel(in <-chan int, workers int)
	out := make(chan int)
	var wg sync.WaitGroup
 
	wg.Add(workers)
	for i := 0; i < workers; i++ {
		go func() {
			defer wg.Done()
			for n := range in {
				// Simulate expensive parsing
				time.Sleep(10 * time.Millisecond)
				out <- n * n // Square the number
			}
		}()
	}
 
	// Closer goroutine: wait for all parsers to finish, then close out
	go func() {
		wg.Wait()
		close(out)
	}()
 
	return out
}
 
// Stage 3: Enrich (Linear Processing)
func enrich(in <-chan int) <-chan string {
	out := make(chan string)
	go func() {
		for n := range in {
			out <- fmt.Sprintf("Value:%d", n)
		}
		close(out)
	}()
	return out
}
 
func main() {
	// 1. Ingest data
	rawCh := ingest(1, 2, 3, 4, 5, 6,
 
	// 2. Parse (parallelize this bottleneck with 3 workers)
	parsedCh := parseParallel(rawCh, 3)
 
	// 3. Enrich (add metadata)
	enrichedCh := enrich(parsedCh)
 
	// 4. Sink (write/print output)
	for data := range enrichedCh {
		fmt.Println(data)
	}
}