TOPO
A library to create in process topologies of goroutines connected by channels.
Topo does boilerplate work as outlined in http://blog.golang.org/pipelines.
You receive correctly connected input and output channels, leaving the
message processing for you while handling the plumbing. Topo strives to be
simple, all interaction are via proper Go channels, no wrapping interfaces.
Example Code
package main
import (
"fmt"
"sync"
"github.com/mdmarek/topo"
"github.com/mdmarek/topo/topoutil"
)
const nworkers = 2
func worker(in <-chan topo.Mesg, out chan<- topo.Mesg) {
... do something ...
}
func main() {
wg := new(sync.WaitGroup)
wg.Add(nworkers)
// Create a new topo and source of streaming data from meetup.com.
t, err := topo.New()
if err != nil {
fmt.Printf("Failed to create topo: %v\n", err)
return
}
source, err := topoutil.NewMeetupSource(t)
if err != nil {
fmt.Printf("Failed to open source: %v\n", err)
return
}
// Shuffles messages read from the source
// to each worker.
outputs := t.Shuffle(nworkers, worker, source)
// Each output channel is read by one Sink, which
// prints to stdout the messages it receives.
for i := 0; i < nworkers; i++ {
go topoutil.Sink(i, wg, outputs[i])
}
// Wait for the sinks to finish, if ever.
wg.Wait()
}
Messages
Topo creates channels of type chan Mesg, and a Mesg is defined as the
interface:
type Mesg interface {
Key() uint64
Body() interface{}
}
Compositions
Topo works through three simple compositions of channels to form pipelines:
Merge, Shuffle, and Partition.
Merge takes n input channels and merges them into one output channel.
Shuffle takes n input channels and connects them to m functions writing their output
to m output channels. Messages from the n input channels are sent to the first
available function.
Partition takes n input channels and connects them to m functions writing their output
to m output channels. Messages from the n input channels are routed by taking the
message's key value modulo m.
Sources
When writing a source of data for the topology it should use the topology's exit channel
in its select statement, otherwise a deadlock panic may occur. The basic structure is
as follows:
func NewMySource(... params ..., t topo.Topo) (<-chan topo.Mesg, error) {
...
out := make(chan topo.Mesg)
go func(exit <-chan bool) {
defer close(out)
for ... {
select {
case out <- produce():
case <-exit:
return
}
}
}(t.ExitChan())
...
return out, nil
}
Keep in mind to pass the exit channel as a parameter to any started goroutiness rather
than as a closure.