Skip to content

@leehambley /widget-pipeline.go secret
Created

Embed URL

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
package main
import (
"fmt"
"os"
)
type Pipeline struct {
Emitter chan *Widget
Drain chan *Widget
processes []func(chan *Widget, chan *Widget)
}
func NewPipeline() (*Pipeline, chan *Widget, chan *Widget) {
e, d := make(chan *Widget), make(chan *Widget)
return &Pipeline{Emitter: e, Drain: d}, e, d
}
func (pipe *Pipeline) Add(process func(chan *Widget, chan *Widget)) {
pipe.processes = append(pipe.processes, process)
}
func (pipe *Pipeline) Execute() {
var in, out chan *Widget
for i, process := range pipe.processes {
switch {
case i == 0:
in, out = pipe.Emitter, make(chan *Widget)
case i == len(pipe.processes):
in, out = make(chan *Widget), pipe.Drain
default:
in, out = out, make(chan *Widget)
}
fmt.Fprintln(os.Stderr, "Pipeline reading from", in, "writing to", out)
go process(in, out)
}
}
type Widget struct {
Whiz bool
Pop bool
Bang bool
}
var whizWidgets = func(chi chan *Widget, cho chan *Widget) {
for widget := range chi {
widget.Whiz = true
cho <- widget
}
close(cho)
}
var popWidgets = func(chi chan *Widget, cho chan *Widget) {
for widget := range chi {
widget.Pop = true
cho <- widget
}
close(cho)
}
var bangWidgets = func(chi chan *Widget, cho chan *Widget) {
for widget := range chi {
widget.Bang = true
cho <- widget
}
close(cho)
}
func main() {
p, e, d := NewPipeline()
fmt.Fprintln(os.Stderr, "Input Emitted On", e)
fmt.Fprintln(os.Stderr, "Output Will Drain From", d)
p.Add(whizWidgets)
p.Add(popWidgets)
p.Add(bangWidgets)
go emit(e)
p.Execute()
drain(d)
}
func emit(cho chan *Widget) {
for i := 0; i < 1000; i++ {
cho <- &Widget{}
}
close(cho)
}
func drain(chi chan *Widget) {
for widget := range chi {
fmt.Println(widget)
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Something went wrong with that request. Please try again.