feat(bitswap) make offline exchange query datastore
License: MIT Signed-off-by: Brian Tiger Chow <brian@perfmode.com>
Brian Tiger Chow committed
Nov 26, 2014 at 16:55 UTC
670d0244b44de0e89eb71e42ef350f9ae666ad97
8 files changed
+101
-29
blockservice/blocks_test.go
+2
-3
@@ -19,9 +19,8 @@ import (
19
)
20
21
func TestBlocks(t *testing.T) {
22
- d := ds.NewMapDatastore()
23
- tsds := dssync.MutexWrap(d)
24
- bs, err := New(blockstore.NewBlockstore(tsds), offline.Exchange())
22
+ bstore := blockstore.NewBlockstore(dssync.MutexWrap(ds.NewMapDatastore()))
23
+ bs, err := New(bstore, offline.Exchange(bstore))
24
if err != nil {
25
t.Error("failed to construct block service", err)
26
return
core/core.go
+1
-1
@@ -130,7 +130,7 @@ func NewIpfsNode(cfg *config.Config, online bool) (n *IpfsNode, err error) {
130
return nil, debugerror.Wrap(err)
131
}
132
133
- n.Exchange = offline.Exchange()
133
+ n.Exchange = offline.Exchange(blockstore.NewBlockstore(n.Datastore))
134
135
// setup online services
136
if online {
core/mock.go
+2
-1
@@ -45,7 +45,8 @@ func NewMockNode() (*IpfsNode, error) {
45
nd.Routing = dht
46
47
// Bitswap
48
- bserv, err := blockservice.New(blockstore.NewBlockstore(nd.Datastore), offline.Exchange())
48
+ bstore := blockstore.NewBlockstore(nd.Datastore)
49
+ bserv, err := blockservice.New(bstore, offline.Exchange(bstore))
50
if err != nil {
51
return nil, err
52
}
exchange/offline/offline.go
+38
-14
@@ -3,42 +3,66 @@
3
package offline
4
5
import (
6
- "errors"
7
-
6
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
9
-
7
blocks "github.com/jbenet/go-ipfs/blocks"
8
+ "github.com/jbenet/go-ipfs/blocks/blockstore"
9
exchange "github.com/jbenet/go-ipfs/exchange"
10
u "github.com/jbenet/go-ipfs/util"
11
)
12
15
-var OfflineMode = errors.New("Block unavailable. Operating in offline mode")
16
-
17
-func Exchange() exchange.Interface {
18
- return &offlineExchange{}
13
+func Exchange(bs blockstore.Blockstore) exchange.Interface {
14
+ return &offlineExchange{bs: bs}
15
}
16
17
// offlineExchange implements the Exchange interface but doesn't return blocks.
18
// For use in offline mode.
23
-type offlineExchange struct{}
19
+type offlineExchange struct {
20
+ bs blockstore.Blockstore
21
+}
22
23
// GetBlock returns nil to signal that a block could not be retrieved for the
24
// given key.
25
// NB: This function may return before the timeout expires.
28
-func (_ *offlineExchange) GetBlock(context.Context, u.Key) (*blocks.Block, error) {
29
- return nil, OfflineMode
26
+func (e *offlineExchange) GetBlock(_ context.Context, k u.Key) (*blocks.Block, error) {
27
+ return e.bs.Get(k)
28
}
29
30
// HasBlock always returns nil.
33
-func (_ *offlineExchange) HasBlock(context.Context, *blocks.Block) error {
34
- return nil
31
+func (e *offlineExchange) HasBlock(_ context.Context, b *blocks.Block) error {
32
+ return e.bs.Put(b)
33
}
34
35
// Close always returns nil.
36
func (_ *offlineExchange) Close() error {
37
+ // NB: exchange doesn't own the blockstore's underlying datastore, so it is
38
+ // not responsible for closing it.
39
return nil
40
}
41
42
-func (_ *offlineExchange) GetBlocks(context.Context, []u.Key) (<-chan *blocks.Block, error) {
43
- return nil, OfflineMode
42
+func (e *offlineExchange) GetBlocks(ctx context.Context, ks []u.Key) (<-chan *blocks.Block, error) {
43
+ out := make(chan *blocks.Block, 0)
44
+ go func() {
45
+ defer close(out)
46
+ var misses []u.Key
47
+ for _, k := range ks {
48
+ hit, err := e.bs.Get(k)
49
+ if err != nil {
50
+ misses = append(misses, k)
51
+ // a long line of misses should abort when context is cancelled.
52
+ select {
53
+ // TODO case send misses down channel
54
+ case <-ctx.Done():
55
+ return
56
+ default:
57
+ continue
58
+ }
59
+ }
60
+ select {
61
+ case out <- hit:
62
+ case <-ctx.Done():
63
+ return
64
+ }
65
+ }
66
+ }()
67
+ return out, nil
68
}
exchange/offline/offline_test.go
+54
-5
@@ -4,13 +4,16 @@ import (
4
"testing"
5
6
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
7
-
7
+ ds "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
8
+ ds_sync "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
9
blocks "github.com/jbenet/go-ipfs/blocks"
10
+ "github.com/jbenet/go-ipfs/blocks/blockstore"
11
+ "github.com/jbenet/go-ipfs/blocks/blocksutil"
12
u "github.com/jbenet/go-ipfs/util"
13
)
14
15
func TestBlockReturnsErr(t *testing.T) {
13
- off := Exchange()
16
+ off := Exchange(bstore())
17
_, err := off.GetBlock(context.Background(), u.Key("foo"))
18
if err != nil {
19
return // as desired
@@ -19,10 +22,56 @@ func TestBlockReturnsErr(t *testing.T) {
22
}
23
24
func TestHasBlockReturnsNil(t *testing.T) {
22
- off := Exchange()
25
+ store := bstore()
26
+ ex := Exchange(store)
27
block := blocks.NewBlock([]byte("data"))
24
- err := off.HasBlock(context.Background(), block)
28
+
29
+ err := ex.HasBlock(context.Background(), block)
30
if err != nil {
26
- t.Fatal("")
31
+ t.Fail()
32
+ }
33
+
34
+ if _, err := store.Get(block.Key()); err != nil {
35
+ t.Fatal(err)
36
+ }
37
+}
38
+
39
+func TestGetBlocks(t *testing.T) {
40
+ store := bstore()
41
+ ex := Exchange(store)
42
+ g := blocksutil.NewBlockGenerator()
43
+
44
+ expected := g.Blocks(2)
45
+
46
+ for _, b := range expected {
47
+ if err := ex.HasBlock(context.Background(), b); err != nil {
48
+ t.Fail()
49
+ }
50
}
51
+
52
+ request := func() []u.Key {
53
+ var ks []u.Key
54
+
55
+ for _, b := range expected {
56
+ ks = append(ks, b.Key())
57
+ }
58
+ return ks
59
+ }()
60
+
61
+ received, err := ex.GetBlocks(context.Background(), request)
62
+ if err != nil {
63
+ t.Fatal(err)
64
+ }
65
+
66
+ var count int
67
+ for _ = range received {
68
+ count++
69
+ }
70
+ if len(expected) != count {
71
+ t.Fail()
72
+ }
73
+}
74
+
75
+func bstore() blockstore.Blockstore {
76
+ return blockstore.NewBlockstore(ds_sync.MutexWrap(ds.NewMapDatastore()))
77
}
pin/pin_test.go
+1
-1
@@ -23,7 +23,7 @@ func randNode() (*mdag.Node, util.Key) {
23
func TestPinnerBasic(t *testing.T) {
24
dstore := ds.NewMapDatastore()
25
bstore := blockstore.NewBlockstore(dssync.MutexWrap(dstore))
26
- bserv, err := bs.New(bstore, offline.Exchange())
26
+ bserv, err := bs.New(bstore, offline.Exchange(bstore))
27
if err != nil {
28
t.Fatal(err)
29
}
unixfs/io/dagmodifier_test.go
+1
-1
@@ -24,7 +24,7 @@ func getMockDagServ(t *testing.T) mdag.DAGService {
24
dstore := ds.NewMapDatastore()
25
tsds := sync.MutexWrap(dstore)
26
bstore := blockstore.NewBlockstore(tsds)
27
- bserv, err := bs.New(bstore, offline.Exchange())
27
+ bserv, err := bs.New(bstore, offline.Exchange(bstore))
28
if err != nil {
29
t.Fatal(err)
30
}
util/testutil/gen.go
+2
-3
@@ -16,9 +16,8 @@ import (
16
)
17
18
func GetDAGServ(t testing.TB) dag.DAGService {
19
- dstore := ds.NewMapDatastore()
20
- tsds := dssync.MutexWrap(dstore)
21
- bserv, err := bsrv.New(blockstore.NewBlockstore(tsds), offline.Exchange())
19
+ bstore := blockstore.NewBlockstore(dssync.MutexWrap(ds.NewMapDatastore()))
20
+ bserv, err := bsrv.New(bstore, offline.Exchange(bstore))
21
if err != nil {
22
t.Fatal(err)
23
}