feat(exch:bitswap) simply get method
Brian Tiger Chow committed
Sep 18, 2014 at 20:30 UTC
98a6e9fac27f22be200583e650c9fc27f0fd720f
1 file changed
+33
-46
exchange/bitswap/bitswap.go
+33
-46
@@ -65,63 +65,38 @@ type bitswap struct {
65
// deadline enforced by the context
66
//
67
// TODO ensure only one active request per key
68
-func (bs *bitswap) Block(ctx context.Context, k u.Key) (*blocks.Block, error) {
68
+func (bs *bitswap) Block(parent context.Context, k u.Key) (*blocks.Block, error) {
69
70
- const maxProviders = 20
71
- provs_ch := bs.routing.FindProvidersAsync(ctx, k, maxProviders)
70
+ ctx, cancelFunc := context.WithCancel(parent)
71
+ promise := bs.notifications.Subscribe(ctx, k)
72
73
- blockChannel := make(chan blocks.Block)
74
-
75
- // TODO: when the data is received, shut down this for loop ASAP
73
go func() {
77
- for p := range provs_ch {
78
- go func(pr *peer.Peer) {
79
- blk, err := bs.getBlock(ctx, k, pr)
74
+ const maxProviders = 20
75
+ peersToQuery := bs.routing.FindProvidersAsync(ctx, k, maxProviders)
76
+ message := bsmsg.New()
77
+ message.AppendWanted(k)
78
+ for i := range peersToQuery {
79
+ go func(p *peer.Peer) {
80
+ response, err := bs.sender.SendRequest(ctx, p, message)
81
if err != nil {
82
return
83
}
83
- select {
84
- case blockChannel <- *blk:
85
- default:
86
- }
87
- }(p)
84
+ // FIXME ensure accounting is handled correctly when
85
+ // communication fails. May require slightly different API to
86
+ // get better guarantees. May need shared sequence numbers.
87
+ bs.strategy.MessageSent(p, message)
88
+
89
+ bs.ReceiveMessage(ctx, p, response)
90
+ }(i)
91
}
92
}()
93
94
select {
92
- case block := <-blockChannel:
93
- close(blockChannel)
95
+ case block := <-promise:
96
+ cancelFunc()
97
return &block, nil
95
- case <-ctx.Done():
96
- return nil, ctx.Err()
97
- }
98
-}
99
-
100
-func (bs *bitswap) getBlock(ctx context.Context, k u.Key, p *peer.Peer) (*blocks.Block, error) {
101
-
102
- blockChannel := bs.notifications.Subscribe(ctx, k)
103
-
104
- message := bsmsg.New()
105
- message.AppendWanted(k)
106
-
107
- bs.send(ctx, p, message)
108
-
109
- block, ok := <-blockChannel
110
- if !ok {
111
- return nil, u.ErrTimeout
112
- }
113
- return &block, nil
114
-}
115
-
116
-func (bs *bitswap) sendToPeersThatWant(ctx context.Context, block blocks.Block) {
117
- for _, p := range bs.strategy.Peers() {
118
- if bs.strategy.BlockIsWantedByPeer(block.Key(), p) {
119
- if bs.strategy.ShouldSendBlockToPeer(block.Key(), p) {
120
- message := bsmsg.New()
121
- message.AppendBlock(block)
122
- go bs.send(ctx, p, message)
123
- }
124
- }
98
+ case <-parent.Done():
99
+ return nil, parent.Err()
100
}
101
}
102
@@ -173,3 +148,15 @@ func (bs *bitswap) send(ctx context.Context, p *peer.Peer, m bsmsg.BitSwapMessag
148
func numBytes(b blocks.Block) int {
149
return len(b.Data)
150
}
151
+
152
+func (bs *bitswap) sendToPeersThatWant(ctx context.Context, block blocks.Block) {
153
+ for _, p := range bs.strategy.Peers() {
154
+ if bs.strategy.BlockIsWantedByPeer(block.Key(), p) {
155
+ if bs.strategy.ShouldSendBlockToPeer(block.Key(), p) {
156
+ message := bsmsg.New()
157
+ message.AppendBlock(block)
158
+ go bs.send(ctx, p, message)
159
+ }
160
+ }
161
+ }
162
+}