Wait, isn't this just a bunch of functions? Why use channels?

You could definitely just pass a slice of data through a series of functions. But the pipeline pattern isn't about the transformation itself—it's about concurrency and streaming. If you have a 10GB log file, you can't load that into a slice. You want to start processing the first line while the disk is still reading the second.

In Go, a pipeline is a series of stages connected by channels. Each stage is a group of goroutines running the same function. This lets you utilize all your CPU cores; while stage three is aggregating data, stage one is still ingestion, and stage two is filtering. It's a conveyor belt for your data.

How do I actually structure a "stage" so it connects to the next one?

The pattern I always use is: a function that takes a receive-only channel as input and returns a receive-only channel as output. This makes the pipeline composable. You can plug in new stages or swap them out without breaking the rest of the chain.

Let's look at a real-world example: processing a stream of raw log strings. We want to convert them to structs and then filter out anything that isn't an "ERROR".

type LogEntry struct {
    Level   string
    Message string
}

// Stage 1: Generator - converts a slice of strings into a channel
func generateLogs(logs []string) <-chan string {
    out := make(chan string)
    go func() {
        for _, l := range logs {
            out <- l
        }
        close(out)
    }()
    return out
}

// Stage 2: Parser - converts raw strings to LogEntry structs
func parseLogs(in <-chan string) <-chan LogEntry {
    out := make(chan LogEntry)
    go func() {
        for s := range in {
            // Simple split logic for example purposes
            parts := strings.SplitN(s, ":", 2)
            out <- LogEntry{Level: parts[0], Message: parts[1]}
        }
        close(out)
    }()
    return out
}

// Stage 3: Filter - only lets ERROR logs through
func filterErrors(in <-chan LogEntry) <-chan LogEntry {
    out := make(chan LogEntry)
    go func() {
        for entry := range in {
            if entry.Level == "ERROR" {
                out <- entry
            }
        }
        close(out)
    }()
    return out
}

How do I prevent goroutine leaks when the pipeline finishes?

This is where most developers trip up. If you stop reading from the final channel (the "sink") but the earlier stages are still trying to send data, those goroutines will hang forever. They're blocked on a channel send that will never be received. That's a memory leak.

I handle this in two ways. First, always ensure your stages close(out) when the input channel is exhausted. That's why I used for s := range in in the examples above—the loop terminates automatically when the previous stage closes its channel.

Second, if you need to be able to cancel the pipeline early (say, you found the first error and don't care about the rest), you must pass a context.Context through every stage. Inside your select block, you check for ctx.Done(). If the context is cancelled, you exit the goroutine immediately.

Here is how you'd wire it all together in your main function:

func main() {
    rawLogs := []string{"INFO:System start", "ERROR:Disk full", "DEBUG:Checking port", "ERROR:Network timeout"}

    // Pipeline wiring
    source := generateLogs(rawLogs)
    parsed := parseLogs(source)
    errors := filterErrors(parsed)

    // The "Sink" - this consumes the final output
    for errLog := range errors {
        fmt.Printf("Alert: %s\n", errLog.Message)
    }
}

📋 Practical Task

Exercise: Build a High-Value Transaction Monitor

You need to build a data pipeline that processes a stream of financial transactions. Your goal is to identify "High Value" transactions (those over $10,000) and format them for a reporting system.

Implement the following pipeline stages:

1. generateTransactions: Takes a slice of floats (transaction amounts) and sends them into a channel.
2. filterHighValue: Receives amounts and only forwards those greater than 10,000.
3. formatReport: Receives the high-value amounts and converts them into a string format: "HIGH VALUE: $X.XX".

In your main function, provide a slice of at least 5 different amounts (some above and some below 10,000), wire the pipeline together, and print the final results. Ensure that all channels are closed correctly to avoid goroutine leaks.