package main

import (
	"fmt"
	"strconv"
	"sync"
	"time"
)

// generateNumbers sends integers from start to end into an output channel.
func generateNumbers(start, end int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for i := start; i <= end; i++ {
			out <- i
			time.Sleep(50 * time.Millisecond) // Simulate data generation time
		}
	}()
	return out
}

// multiplyByTwo receives numbers from an input channel, multiplies them by two,
// and sends the result to an output channel.
func multiplyByTwo(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			fmt.Printf("[Multiply Stage] Received %d, multiplying...
", n)
			time.Sleep(100 * time.Millisecond) // Simulate processing time
			out <- n * 2
		}
	}()
	return out
}

// convertToString receives integers, converts them to strings with a prefix,
// and sends the result to an output channel.
func convertToString(in <-chan int) <-chan string {
	out := make(chan string)
	go func() {
		defer close(out)
		for n := range in {
			fmt.Printf("[Convert Stage] Received %d, converting...
", n)
			time.Sleep(70 * time.Millisecond) // Simulate processing time
			out <- "Result-" + strconv.Itoa(n)
		}
	}()
	return out
}

func main() {
	fmt.Println("Starting data processing pipeline...")

	// Stage 1: Generate numbers
	numbers := generateNumbers(1, 5)

	// Stage 2: Multiply by two
	multiplied := multiplyByTwo(numbers)

	// Stage 3: Convert to string
	finalResults := convertToString(multiplied)

	// Consumer: Print final results
	var wg sync.WaitGroup
	wg.Add(1)
	go func() {
		defer wg.Done()
		fmt.Println("
--- Final Results ---")
		for res := range finalResults {
			fmt.Printf("[*] Final Output: %s
", res)
		}
		fmt.Println("--- End of Results ---")
	}()

	// Wait for the consumer to finish (which means all pipeline stages are done)
	wg.Wait()
	fmt.Println("Pipeline execution complete.")
}