Skip to content

Commit

Permalink
Rewrite fan-in/out messaging patterns
Browse files Browse the repository at this point in the history
  • Loading branch information
tmrts committed Apr 16, 2016
1 parent 8b0943a commit 4bd71e7
Show file tree
Hide file tree
Showing 4 changed files with 54 additions and 40 deletions.
4 changes: 2 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -77,8 +77,8 @@ __Messaging Patterns__:

| Pattern | Description |
|:-------:| ----------- |
| [Fan-In](fan/fan_in.go) | Funnels tasks to a work sink (e.g. server) |
| [Fan-Out](fan/fan_out.go) | Distributes tasks amongs workers |
| [Fan-In](messaging/fan_in.md) | Funnels tasks to a work sink (e.g. server) |
| [Fan-Out](messaging/fan_out.md) | Distributes tasks amongs workers (e.g. producer) |
| [Futures & Promises](futures_promises.go) | Acts as a place-holder of a result that is initally unknown for synchronization purposes |
| [Publish/Subscribe](messaging/publish_subscribe.md) | Passes information to a collection of recipients who subscribed to a topic |
| [Push & Pull](push_pull.go) | Distributes messages to multiple workers, arranged in a pipeline |
Expand Down
33 changes: 0 additions & 33 deletions fan/fan_in.go

This file was deleted.

40 changes: 40 additions & 0 deletions messaging/fan_in.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
Fan-In Messaging Patterns
===================================
Fan-In is a messaging pattern used to create a funnel for work amongst workers (clients: source, server: destination).

We can model fan-in using the Go channels.

```go
// Merge different channels in one channel
func Merge(cs ...<-chan int) <-chan int {
var wg sync.WaitGroup

out := make(chan int)

// Start an send goroutine for each input channel in cs. send
// copies values from c to out until c is closed, then calls wg.Done.
send := func(c <-chan int) {
for n := range c {
out <- n
}
wg.Done()
}

wg.Add(len(cs))
for _, c := range cs {
go send(c)
}

// Start a goroutine to close out once all the send goroutines are
// done. This must start after the wg.Add call.
go func() {
wg.Wait()
close(out)
}()
return out
}
```

The `Merge` function converts a list of channels to a single channel by starting a goroutine for each inbound channel that copies the values to the sole outbound channel.

Once all the output goroutines have been started, `Merge` a goroutine is started to close the main channel.
17 changes: 12 additions & 5 deletions fan/fan_out.go → messaging/fan_out.md
Original file line number Diff line number Diff line change
@@ -1,9 +1,12 @@
package fan
Fan-Out Messaging Pattern
=========================
Fan-Out is a messaging pattern used for distributing work amongst workers (producer: source, consumers: destination).

// Out implements fan.Out messaging pattern
// Split a channel into n channels that receive messages
// in a round-robin fashion.
func Out(ch <-chan int, n int) []<-chan int {
We can model fan-out using the Go channels.

```go
// Split a channel into n channels that receive messages in a round-robin fashion.
func Split(ch <-chan int, n int) []<-chan int {
cs := make([]chan int)
for i := 0; i < n; i++ {
cs = append(cs, make(chan int))
Expand Down Expand Up @@ -38,3 +41,7 @@ func Out(ch <-chan int, n int) []<-chan int {

return cs
}
```

The `Split` function converts a single channel into a list of channels by using
a goroutine to copy received values to channels in the list in a round-robin fashion.

0 comments on commit 4bd71e7

Please sign in to comment.