| 485 | } |
| 486 | |
| 487 | func merge(ctx context.Context, cs ...chan P2PMessage) chan P2PMessage { |
| 488 | var wg sync.WaitGroup |
| 489 | out := make(chan P2PMessage) |
| 490 | |
| 491 | // Start an output goroutine for each input channel in cs. output |
| 492 | // copies values from c to out until c is closed, then calls wg.Done. |
| 493 | output := func(c <-chan P2PMessage) { |
| 494 | for n := range c { |
| 495 | select { |
| 496 | case <-ctx.Done(): |
| 497 | return |
| 498 | case out <- n: |
| 499 | } |
| 500 | } |
| 501 | wg.Done() |
| 502 | } |
| 503 | wg.Add(len(cs)) |
| 504 | for _, c := range cs { |
| 505 | go output(c) |
| 506 | } |
| 507 | |
| 508 | // Start a goroutine to close out once all the output goroutines are |
| 509 | // done. This must start after the wg.Add call. |
| 510 | go func() { |
| 511 | wg.Wait() |
| 512 | close(out) |
| 513 | }() |
| 514 | return out |
| 515 | } |
| 516 | |
| 517 | // SubscribeMsg is a message subscription operation binding the P2PMessage |
| 518 | func (n *server) SubscribeMsg(chanBuffer int, peersFeed ...interface{}) (outch chan P2PMessage, err error) { |