use @maybebtc's ForwardBlocks function
Jeromy committed
Nov 21, 2014 at 23:33 UTC
50b00eb90fc3f1d493f4c1c05592c09ae7958b8a
3 files changed
+7
-26
blockservice/blockservice.go
+1
-2
@@ -98,7 +98,7 @@ func (s *BlockService) GetBlock(ctx context.Context, k u.Key) (*blocks.Block, er
98
// GetBlocks gets a list of blocks asynchronously and returns through
99
// the returned channel.
100
// NB: No guarantees are made about order.
101
-func (s *BlockService) GetBlocks(parent context.Context, ks []u.Key) <-chan *blocks.Block {
101
+func (s *BlockService) GetBlocks(ctx context.Context, ks []u.Key) <-chan *blocks.Block {
102
out := make(chan *blocks.Block, 32)
103
go func() {
104
var toFetch []u.Key
@@ -112,7 +112,6 @@ func (s *BlockService) GetBlocks(parent context.Context, ks []u.Key) <-chan *blo
112
out <- block
113
}
114
115
- ctx, cancel := context.WithCancel(parent)
115
nblocks, err := s.Remote.GetBlocks(ctx, toFetch)
116
if err != nil {
117
log.Errorf("Error with GetBlocks: %s", err)
exchange/bitswap/bitswap.go
+2
-24
@@ -16,6 +16,7 @@ import (
16
strategy "github.com/jbenet/go-ipfs/exchange/bitswap/strategy"
17
peer "github.com/jbenet/go-ipfs/peer"
18
u "github.com/jbenet/go-ipfs/util"
19
+ async "github.com/jbenet/go-ipfs/util/async"
20
"github.com/jbenet/go-ipfs/util/eventlog"
21
)
22
@@ -128,35 +129,12 @@ func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.
129
promise := bs.notifications.Subscribe(ctx, keys...)
130
select {
131
case bs.batchRequests <- keys:
131
- return pipeBlocks(ctx, promise, len(keys)), nil
132
+ return async.ForwardN(ctx, promise, len(keys)), nil
133
case <-ctx.Done():
134
return nil, ctx.Err()
135
}
136
}
137
137
-func pipeBlocks(ctx context.Context, in <-chan *blocks.Block, count int) <-chan *blocks.Block {
138
- out := make(chan *blocks.Block, 1)
139
- go func() {
140
- defer close(out)
141
- for i := 0; i < count; i++ {
142
- select {
143
- case blk, ok := <-in:
144
- if !ok {
145
- return
146
- }
147
- select {
148
- case out <- blk:
149
- case <-ctx.Done():
150
- return
151
- }
152
- case <-ctx.Done():
153
- return
154
- }
155
- }
156
- }()
157
- return out
158
-}
159
-
138
func (bs *bitswap) sendWantListTo(ctx context.Context, peers <-chan peer.Peer) error {
139
if peers == nil {
140
panic("Cant send wantlist to nil peerchan")
util/async/forward.go
+4
@@ -3,8 +3,11 @@ package async
3
import (
4
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5
"github.com/jbenet/go-ipfs/blocks"
6
+ u "github.com/jbenet/go-ipfs/util"
7
)
8
9
+var log = u.Logger("async")
10
+
11
// ForwardN forwards up to |num| blocks to the returned channel.
12
func ForwardN(ctx context.Context, in <-chan *blocks.Block, num int) <-chan *blocks.Block {
13
out := make(chan *blocks.Block)
@@ -14,6 +17,7 @@ func ForwardN(ctx context.Context, in <-chan *blocks.Block, num int) <-chan *blo
17
select {
18
case block, ok := <-in:
19
if !ok {
20
+ log.Error("Forwarder exiting early!")
21
return // otherwise nil value is forwarded to output
22
}
23
select {