mirror of
https://github.com/tmrts/go-patterns.git
synced 2024-11-21 20:46:08 +03:00
Rewrite fan-in/out messaging patterns
This commit is contained in:
parent
8b0943a7b2
commit
4bd71e70e3
@ -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 |
|
||||
|
@ -1,33 +0,0 @@
|
||||
package fan
|
||||
|
||||
import "sync"
|
||||
|
||||
// In implements fan.In messaging pattern
|
||||
// Merges different channels in one channel
|
||||
func In(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
|
||||
}
|
40
messaging/fan_in.md
Normal file
40
messaging/fan_in.md
Normal file
@ -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.
|
@ -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))
|
||||
@ -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.
|
Loading…
Reference in New Issue
Block a user