Composable Concurrency in Go: Building Pipelines Without the Boilerplate

If you've ever written concurrent Go code, you know the drill. You spin up goroutines, create channels, manage error propagation, handle cancellation, and somewhere in the middle of all that plumbing, the actual logic of your program gets buried. Rill is a toolkit that aims to change that by letting you build concurrent programs from simple, reusable parts—while keeping Go's natural channel-based model intact.

What It Does

Rill is a lightweight Go library that brings composable concurrency to your programs. Most functions in the library take Go channels as inputs and return new, transformed channels as outputs. This means you can chain them together to build pipelines from simpler parts—similar to how Unix pipes work. The result is concurrent code that reads as a clear sequence of operations rather than a tangle of goroutine management.

The library provides built-in functions for common tasks like parallel job execution, real-time event processing, batching, ordered fan-in, map-reduce, stream splitting, and merging. It handles error propagation automatically through the pipeline, so you can deal with errors in one place at the end. And because it operates on standard Go channels, you retain full control over concurrency levels at each step. It has zero dependencies and a small, type-safe API, making it straightforward to drop into existing projects.

Why It's Cool

• It keeps Go's mental model intact. Rill doesn't try to replace channels or introduce a new abstraction layer on top of them. It builds on what you already know. If you understand how channels and backpressure work in Go, you already understand the foundation of Rill. The library just removes the repetitive parts.

• Composability is the real win here. Because functions take channels in and return channels out, you can chain them in any cycle-free topology—not just linear pipelines. This makes it easy to build reusable components and assemble them in different ways for different problems. You're not locked into a single pattern.

• Error handling is centralized. Anyone who's written concurrent Go code knows the pain of collecting errors from multiple goroutines. Rill propagates errors through the pipeline automatically, so you can handle them in a single place. For more complex scenarios, you can intercept and handle errors at any point in the pipeline.

• Stream processing comes naturally. Since everything is built on channels, Rill handles potentially infinite streams without issue. Items are processed as they arrive, which makes it suitable for real-time processing or working with datasets that don't fit in memory. You're not forced to load everything upfront.

• Resource usage stays flat. The library is designed so that the number of memory allocations and goroutines doesn't grow with input size. That's a meaningful detail—it means you can use Rill for long-running processes without worrying about gradual resource creep.

• Custom extensions are straightforward. Because Rill works with standard Go channels, writing your own functions that integrate with the library is just a matter of following the same input/output pattern. There's no special interface to implement or framework to learn.

How to Try It

Getting started with Rill is a single command:

go get -u github.com/destel/rill

Here's a quick example that shows the core idea. The following code fetches users from an API, activates them, and saves the changes back—with configurable concurrency at each step:

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    // Convert a slice of user IDs into a channel
    ids := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

    // Read users from the API.
    // Concurrency = 3
    users := rill.Map(ids, 3, func(id int) (*mockapi.User, error) {
        return mockapi.GetUser(ctx, id)
    })

    // Activate users.
    // Concurrency = 2
    err := rill.ForEach(users, 2, func(u *mockapi.User) error {
        if u.IsActive {
            fmt.Printf("User %d is already active
", u.ID)
            return nil
        }

        u.IsActive = true
        err := mockapi.SaveUser(ctx, u)
        if err != nil {
            return err
        }

        fmt.Printf("User saved: %+v
", u)
        return nil
    })

    // Handle errors
    fmt.Println("Error:", err)
}

Notice how each step specifies its own concurrency level—three for fetching, two for saving. The error from ForEach returns on the first failure, and the deferred cancel() stops all remaining fetches. That's a lot of behavior from a few lines of code.

You can find the full documentation and more examples at github.com/destel/rill.

Final Thoughts

Rill is for Go developers who write concurrent code regularly and are tired of the boilerplate that comes with it. It's not trying to reinvent concurrency in Go—it's trying to make the existing model more ergonomic and composable. If you work with pipelines, stream processing, or parallel job execution, it's worth a look. The library is small, has no dependencies, and integrates with standard Go channels, so there's not much risk in trying it out. For anyone building concurrent systems in Go, Rill offers a practical way to keep your code clean without giving up control.

Follow @githubprojects for more developer tools and open source projects.