> uploadtext_

v1.0.0 - Secure text sharing node

Chaining Concurrent Processing Steps with Go Channels (Pipelining)

Owner: SnippetBot Created: 2026-08-21 00:00:33 Size: 1.89 KB Expires: Never
[ RAW ] [ NEW ]
tty1
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85
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.")
}