When we use channels to serialize tasks, we create an asynchronous flow. It is easy to mess things up in asynchronous programming, especially if we have additional requirements like timeout and cancelation.
A real-world example of chained channels is reading data line by line from a file reader and then passing it to a text parser, which also works asynchronously.
I found a common pattern that may cover most such scenarios with minimum and understandable codes.
Asynchronous provider followed by a synchronous consumer.
Concept:
1go provider()
2consumer()
Example:
1var (
2 provider = make(chan int)
3 consumer = make(chan int)
4 ctx, cancel = context.WithTimeout(context.Background, time.Second)
5)
6defer cancel()
7
8// provider
9go func() {
10 defer close(consumer)
11
12 for {
13 select {
14 case <-ctx.Done():
15 return
16 case v, ok := <-provider:
17 if !ok {
18 return
19 }
20 consumer <- v
21 }
22 }
23}()
24
25consume:
26for {
27 select {
28 case <-ctx.Done():
29 return
30 case v, ok := <-consumer:
31 if !ok {
32 break consume
33 }
34 }
35}
36
37// follow-up logic
The whole process has a blocking nature due to the last consumer for-loop. This is desired because we want to ensure the program completes the task before quitting. It is also simple to handle errors, like how I return/break when the channel is not OK. For example, in the consumer loop:
1var err error
2
3consume:
4for {
5 select {
6 case <-ctx.Done():
7 return
8 case v, ok := <-consumer:
9 if !ok {
10 break consume
11 }
12
13 if err = process(v); err != nil {
14 break consume
15 }
16 }
17}
18
19// follow-up logic
20if err != nil {
21 log.Error(err)
22}
At last, may we all handle channels well.