opensource/kubo
1
0
Fork 0
mirror of https://github.com/ipfs/kubo.git synced 2025-06-30 01:52:26 +08:00
Code Issues Projects Releases Wiki Activity

address comments from CR

Browse Source
This commit is contained in:
Jeromy
2015-05-06 00:50:44 -07:00
parent 0324b4b283
commit 9049dae674
1 changed files with 17 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
exchange/bitswap/bitswap.go
View File

@ -349,7 +349,8 @@ func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
keys = append(keys, block.Key()) keys = append(keys, block.Key())
} }
return bs.cancelBlocks(ctx, keys) bs.cancelBlocks(ctx, keys)
return nil
} }
// Connected/Disconnected warns bitswap about peer connections // Connected/Disconnected warns bitswap about peer connections
@ -369,9 +370,9 @@ func (bs *Bitswap) PeerDisconnected(p peer.ID) {
bs.engine.PeerDisconnected(p) bs.engine.PeerDisconnected(p)
} }
func (bs *Bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) error { func (bs *Bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) {
if len(bkeys) < 1 { if len(bkeys) < 1 {
return nil return
} }
message := bsmsg.New() message := bsmsg.New()
message.SetFull(false) message.SetFull(false)
@ -379,14 +380,21 @@ func (bs *Bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) error {
log.Debug("cancel block: %s", k) log.Debug("cancel block: %s", k)
message.Cancel(k) message.Cancel(k)
} }
wg := sync.WaitGroup{}
for _, p := range bs.engine.Peers() { for _, p := range bs.engine.Peers() {
err := bs.send(ctx, p, message) wg.Add(1)
if err != nil { go func(p peer.ID) {
log.Debugf("Error sending message: %s", err) defer wg.Done()
return err err := bs.send(ctx, p, message)
} if err != nil {
log.Warningf("Error sending message: %s", err)
return
}
}(p)
} }
return nil wg.Wait()
return
} }
func (bs *Bitswap) wantNewBlocks(ctx context.Context, bkeys []u.Key) { func (bs *Bitswap) wantNewBlocks(ctx context.Context, bkeys []u.Key) {