example.go

package main

import (
	"fmt"

	"gopkg.in/tomb.v2"
)

type Pipeline struct {
	tomb.Tomb
	next     int
	numbers  chan int
	filtered chan int
}

func NewPipeline() *Pipeline {
	p := &Pipeline{}
	p.numbers = make(chan int)
	p.filtered = make(chan int)

	p.Go(p.generate)
	p.Go(p.filter)
	p.Go(p.print)

	return p
}

func (p *Pipeline) generate() error {
	for {
		select {
		case p.numbers <- p.next:
		case <-p.Dying():
			return nil
		}
		p.next += 1
	}
	return nil
}

func (p *Pipeline) filter() error {
	v := 0
	for {
		select {
		case v = <-p.numbers:
		case <-p.Dying():
			return nil
		}

		if v%2 == 1 {
			continue
		}

		select {
		case p.filtered <- v:
		case <-p.Dying():
			return nil
		}
	}
	return nil
}

func (p *Pipeline) print() error {
	for {
		select {
		case v := <-p.filtered:
			if v > 10 {
				return p.Killf("the number %v is too big", v)
			}
			fmt.Println(v)
		case <-p.Dying():
			return nil
		}
	}
	return nil
}

func main() {
	p := NewPipeline()
	if err := p.Wait(); err != nil {
		fmt.Println(err)
	}
}