Go 9 min read

Building ch.Read: Context-Aware Channel Reading in Go

Published on May 12, 2026 By Berkan Çetinkaya

Series Building ch: Small Channel Helpers From First Principles Part 4 of 6 See list Previous Next

In the previous posts, we started building small channel helpers instead of hiding Go channels behind a larger framework.

ch.Send made channel sends context-aware.

ch.Receive made channel receives context-aware.

Now we can build a slightly higher-level helper on top of them:

ch.Read

The goal is simple:

Keep reading from a channel until either the input closes or the context is cancelled.

At first, this may look too small to deserve its own helper.

But the problem appears quickly once we start from a real loop.

Start with a Normal Channel Reader

Imagine we have a worker that processes image jobs.

type ImageJob struct {
	ID   string
	Path string
}

The worker receives jobs from a channel:

func processImages(jobs <-chan ImageJob) {
	for job := range jobs {
		processImage(job)
	}
}

This is clean Go.

It is simple.

It is readable.

And in many cases, it is enough.

The loop keeps reading from jobs until the channel is closed.

jobs channel
     |
     v
processImages

But there is a hidden assumption here.

The loop only stops when jobs is closed.

The Hidden Problem

Now imagine the application is shutting down.

The parent context is cancelled.

Maybe the HTTP server is stopping.

Maybe the worker pool is being replaced.

Maybe a request was cancelled.

But the jobs channel is still open.

This loop does not know anything about the context:

for job := range jobs {
	processImage(job)
}

If no new job arrives and the channel is not closed, the goroutine waits.

Context cancellation does not interrupt a plain channel range.

That means this worker can get stuck waiting for input that may never come.

The problem is not the processing logic.

The problem is channel lifecycle.

The worker needs to stop when either:

jobs is closed
ctx is cancelled

A plain for range only handles the first case.

Manual Context-Aware Reading

We can write the loop manually with select:

func processImages(ctx context.Context, jobs <-chan ImageJob) {
	for {
		select {
		case <-ctx.Done():
			return

		case job, ok := <-jobs:
			if !ok {
				return
			}

			processImage(job)
		}
	}
}

This is more correct.

Now the worker exits when the context is cancelled.

It also exits when the input channel is closed.

The behavior is right, but the loop is noisier.

The business logic is only this:

processImage(job)

Everything else is channel coordination.

As soon as this pattern appears in multiple places, the repetition becomes obvious.

We Already Have `ch.Receive`

In the previous article, we built ch.Receive.

It turns this select:

select {
case <-ctx.Done():
	return

case job, ok := <-jobs:
	if !ok {
		return
	}

	processImage(job)
}

into a reusable receive helper:

job, ok, err := ch.Receive(ctx, jobs)

So the worker can become:

func processImages(ctx context.Context, jobs <-chan ImageJob) {
	for {
		job, ok, err := ch.Receive(ctx, jobs)
		if err != nil || !ok {
			return
		}

		processImage(job)
	}
}

This is better.

The context-aware receive logic is no longer repeated.

But we still have the same loop shape everywhere:

for {
	value, ok, err := ch.Receive(ctx, input)
	if err != nil || !ok {
		return
	}

	// use value
}

At this point, another helper starts to appear.

Sometimes we do not want to consume the value immediately.

Sometimes we want a context-aware view of the input channel that still behaves like a normal channel.

That is what ch.Read gives us.

What `ch.Read` Should Do

ch.Read should take one input channel and return one output channel.

input channel
      |
      v
   ch.Read
      |
      v
output channel

It should keep forwarding values from input to output.

It should stop when the input channel closes.

It should also stop when the context is cancelled.

So this:

for job := range ch.Read(ctx, jobs) {
	processImage(job)
}

means:

Read jobs until jobs is closed or ctx is cancelled.

The caller still gets a normal receive-only channel.

The caller can still use for range.

But the loop is now context-aware.

Extracting `ch.Read`

Here is the helper:

package ch

import "context"

func Read[T any](
	ctx context.Context,
	in <-chan T,
) <-chan T {
	out := make(chan T)

	go func() {
		defer close(out)

		for {
			value, ok, err := Receive(ctx, in)
			if err != nil || !ok {
				return
			}

			if err := Send(ctx, out, value); err != nil {
				return
			}
		}
	}()

	return out
}

This is small, but every line matters.

Let’s break it down.

The Function Signature

func Read[T any](
	ctx context.Context,
	in <-chan T,
) <-chan T

Read is generic.

It works with any channel element type.

ch.Read[ImageJob](ctx, jobs)
ch.Read[OrderCreated](ctx, orders)
ch.Read[string](ctx, logs)

The input is receive-only:

in <-chan T

That means Read can only receive from it.

It does not own the input channel.

It does not close the input channel.

The output is also receive-only from the caller’s point of view:

<-chan T

The caller can read from it, but cannot send into it.

That matters because Read owns the output channel.

Read is responsible for sending values into it and closing it.

Creating the Output Channel

Inside the function, we create a new channel:

out := make(chan T)

This is the channel returned to the caller.

The caller ranges over it:

for job := range ch.Read(ctx, jobs) {
	processImage(job)
}

By default, this output channel is unbuffered.

That keeps the helper simple and preserves backpressure.

If the caller stops reading from the output, Read should not silently keep buffering values forever.

Backpressure is still part of the design.

Starting the Forwarding Goroutine

Read starts a goroutine:

go func() {
	// ...
}()

That goroutine is responsible for moving values from in to out.

Without this goroutine, Read could not return a channel immediately.

The caller needs a channel to range over.

The forwarding work happens concurrently.

This is the same general shape used by many channel combinators:

create output channel
start goroutine
forward values
close output channel
return output channel

Closing the Output Channel

The goroutine starts with:

defer close(out)

This is important.

The caller may write:

for job := range ch.Read(ctx, jobs) {
	processImage(job)
}

That loop only exits when the returned output channel is closed.

So Read must close out when forwarding stops.

Forwarding can stop because:

input channel closed
context cancelled while receiving
context cancelled while sending

In all cases, closing out tells the caller:

There will be no more values.

This also follows the channel ownership rule.

Read creates out.

Read sends into out.

So Read closes out.

Receiving from the Input

The main loop starts like this:

for {
	value, ok, err := Receive(ctx, in)
	if err != nil || !ok {
		return
	}

	// ...
}

This is where Read uses the lower-level helper.

Receive(ctx, in) waits for one of two things:

a value from in
ctx.Done()

If the context is cancelled, Receive returns an error.

If the input channel is closed, ok is false.

In both cases, Read stops.

That gives Read the behavior we wanted:

read from input or stop when context is done

This is also why Read is closely related to the classic orDone pattern.

But the name Read keeps the API aligned with the ch package:

ch.Send
ch.Receive
ch.Read
ch.Merge
ch.Broadcast

Sending to the Output

After receiving a value, Read forwards it:

if err := Send(ctx, out, value); err != nil {
	return
}

This may look unnecessary at first.

Why not write this?

out <- value

Because sending can block too.

If the caller stops reading from the returned channel, this send can wait forever:

out <- value

So Read uses Send(ctx, out, value) instead.

That way, if the context is cancelled while Read is trying to forward a value, the goroutine can exit.

This is the part people often miss.

Context-aware receiving is not enough.

A forwarding helper also needs context-aware sending.

Otherwise, it can still leak a goroutine while trying to send downstream.

Using `ch.Read`

Now the worker can be written like this:

func processImages(ctx context.Context, jobs <-chan ImageJob) {
	for job := range ch.Read(ctx, jobs) {
		processImage(job)
	}
}

The loop is readable again.

The processing logic is visible.

The channel lifecycle logic is hidden inside a small helper.

This is the main value of Read.

It does not process jobs.

It does not transform values.

It does not start a worker pool.

It only gives us a safer way to range over a channel with context cancellation.

Why Not Just Use `ch.Receive` Everywhere?

We can.

This is valid:

func processImages(ctx context.Context, jobs <-chan ImageJob) {
	for {
		job, ok, err := ch.Receive(ctx, jobs)
		if err != nil || !ok {
			return
		}

		processImage(job)
	}
}

Sometimes this is the better choice.

Use ch.Receive when you want to handle one value at a time and make decisions around each receive.

Use ch.Read when you want to keep the for range style:

for value := range ch.Read(ctx, input) {
	// use value
}

Receive is the lower-level helper.

Read is built on top of it.

That does not mean Read is always the right choice inside every higher-level helper.

Sometimes a helper such as Merge is already running its own forwarding goroutine and can stay leaner by using Receive directly instead of creating one more channel through Read.

So Read is best understood as a useful tool, not a mandatory layer.

What `ch.Read` Is Not

ch.Read is not Each.

This matters.

An Each function would probably look like this:

stream.Each(ctx, jobs, func(job ImageJob) error {
	return processImage(job)
})

That is a stream operator.

It applies a function to every item.

ch.Read does not do that.

It only returns a channel.

for job := range ch.Read(ctx, jobs) {
	processImage(job)
}

The caller still owns the processing logic.

So the distinction is:

ch.Read:
  channel lifecycle helper

stream.Each:
  item processing operator

That keeps the ch package small.

It improves channel coordination without turning the package into a stream framework.

How `ch.Read` Supports Other Primitives

Read becomes useful when building larger channel helpers.

For example, ch.Merge can read from each input safely:

for value := range ch.Read(ctx, input) {
	if err := ch.Send(ctx, out, value); err != nil {
		return
	}
}

ch.Broadcast can read from one input safely:

for value := range ch.Read(ctx, in) {
	for _, out := range outs {
		if err := ch.Send(ctx, out, value); err != nil {
			return
		}
	}
}

That is the real reason small helpers matter.

They compose.

Send handles cancellable sending.

Receive handles cancellable receiving.

Read turns cancellable receiving and sending into a reusable channel forwarding helper.

Then Merge and Broadcast can build on top of it.

Final Version

Here is the full implementation again:

package ch

import "context"

func Read[T any](
	ctx context.Context,
	in <-chan T,
) <-chan T {
	out := make(chan T)

	go func() {
		defer close(out)

		for {
			value, ok, err := Receive(ctx, in)
			if err != nil || !ok {
				return
			}

			if err := Send(ctx, out, value); err != nil {
				return
			}
		}
	}()

	return out
}

The shape is simple:

input channel
      |
      v
    ch.Read
      |
      v
context-aware output channel

ch.Read keeps native channel usage intact.

The caller still writes:

for value := range ch.Read(ctx, input) {
	// use value
}

But now the loop can stop when the input closes or when the context is cancelled.

That small difference matters.

It prevents channel lifecycle logic from spreading across the codebase, and it gives higher-level helpers like Merge and Broadcast a safer foundation.