cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

read from sub.Done()

Browse Source
This commit is contained in:
Yusef Napora 2020-06-25 13:00:55 -04:00
parent 3cbf5efddd
commit 8b0f6c355c
1 changed files with 17 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
lotus-soup/drand.go
View File

@ -97,20 +97,21 @@ func prepareDrandNode(t *TestEnvironment) (*DrandInstance, error) {
var publicAddrs []string var publicAddrs []string
var leaderAddr string var leaderAddr string
ch := make(chan *NodeAddr) ch := make(chan *NodeAddr)
t.SyncClient.MustPublishSubscribe(ctx, addrTopic, &NodeAddr{ _, sub := t.SyncClient.MustPublishSubscribe(ctx, addrTopic, &NodeAddr{
PrivateAddr: n.PrivateAddr(), PrivateAddr: n.PrivateAddr(),
PublicAddr: n.PublicAddr(), PublicAddr: n.PublicAddr(),
IsLeader: isLeader, IsLeader: isLeader,
}, ch) }, ch)
for i := 0; i < nNodes; i++ { for i := 0; i < nNodes; i++ {
msg, ok := <-ch select {
if !ok { case msg := <-ch:
return nil, fmt.Errorf("failed to read drand node addr from sync service")
}
publicAddrs = append(publicAddrs, fmt.Sprintf("http://%s", msg.PublicAddr)) publicAddrs = append(publicAddrs, fmt.Sprintf("http://%s", msg.PublicAddr))
if msg.IsLeader { if msg.IsLeader {
leaderAddr = msg.PrivateAddr leaderAddr = msg.PrivateAddr
} }
case err := <-sub.Done():
return nil, fmt.Errorf("unable to read drand addrs from sync service: %w", err)
}
} }
if leaderAddr == "" { if leaderAddr == "" {
return nil, fmt.Errorf("got %d drand addrs, but no leader", len(publicAddrs)) return nil, fmt.Errorf("got %d drand addrs, but no leader", len(publicAddrs))
@ -196,10 +197,15 @@ func prepareDrandNode(t *TestEnvironment) (*DrandInstance, error) {
} }
infoCh := make(chan *peer.AddrInfo, nNodes) infoCh := make(chan *peer.AddrInfo, nNodes)
infoTopic := sync.NewTopic("drand-gossip-addrs", &peer.AddrInfo{}) infoTopic := sync.NewTopic("drand-gossip-addrs", &peer.AddrInfo{})
t.SyncClient.MustPublishSubscribe(ctx, infoTopic, relayInfo, infoCh)
_, sub := t.SyncClient.MustPublishSubscribe(ctx, infoTopic, relayInfo, infoCh)
for i := 0; i < nNodes; i++ { for i := 0; i < nNodes; i++ {
ai := <-infoCh select {
case ai := <-infoCh:
relayAddrs = append(relayAddrs, *ai) relayAddrs = append(relayAddrs, *ai)
case err := <-sub.Done():
return nil, fmt.Errorf("unable to get drand relay addr from sync service: %w", err)
}
} }
} }