Use a bitswap session for 'Cat'
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
Feb 1, 2018 at 22:24 UTC
1de040d0b7925b4222971fbecec30ed8185a2ea2
12 files changed
+212
-14
core/coreapi/coreapi.go
+14
-4
@@ -5,10 +5,12 @@ import (
5
6
core "github.com/ipfs/go-ipfs/core"
7
coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
8
+ namesys "github.com/ipfs/go-ipfs/namesys"
9
ipfspath "github.com/ipfs/go-ipfs/path"
10
uio "github.com/ipfs/go-ipfs/unixfs/io"
11
12
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
13
+ ipld "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
14
)
15
16
type CoreAPI struct {
@@ -49,12 +51,16 @@ func (api *CoreAPI) Object() coreiface.ObjectAPI {
51
// ResolveNode resolves the path `p` using Unixfx resolver, gets and returns the
52
// resolved Node.
53
func (api *CoreAPI) ResolveNode(ctx context.Context, p coreiface.Path) (coreiface.Node, error) {
52
- p, err := api.ResolvePath(ctx, p)
54
+ return resolveNode(ctx, api.node.DAG, api.node.Namesys, p)
55
+}
56
+
57
+func resolveNode(ctx context.Context, ng ipld.NodeGetter, nsys namesys.NameSystem, p coreiface.Path) (coreiface.Node, error) {
58
+ p, err := resolvePath(ctx, ng, nsys, p)
59
if err != nil {
60
return nil, err
61
}
62
57
- node, err := api.node.DAG.Get(ctx, p.Cid())
63
+ node, err := ng.Get(ctx, p.Cid())
64
if err != nil {
65
return nil, err
66
}
@@ -65,17 +71,21 @@ func (api *CoreAPI) ResolveNode(ctx context.Context, p coreiface.Path) (coreifac
71
// resolved path.
72
// TODO: store all of ipfspath.Resolver.ResolvePathComponents() in Path
73
func (api *CoreAPI) ResolvePath(ctx context.Context, p coreiface.Path) (coreiface.Path, error) {
74
+ return resolvePath(ctx, api.node.DAG, api.node.Namesys, p)
75
+}
76
+
77
+func resolvePath(ctx context.Context, ng ipld.NodeGetter, nsys namesys.NameSystem, p coreiface.Path) (coreiface.Path, error) {
78
if p.Resolved() {
79
return p, nil
80
}
81
82
r := &ipfspath.Resolver{
73
- DAG: api.node.DAG,
83
+ DAG: ng,
84
ResolveOnce: uio.ResolveUnixfsOnce,
85
}
86
87
p2 := ipfspath.FromString(p.String())
78
- node, err := core.Resolve(ctx, api.node.Namesys, r, p2)
88
+ node, err := core.Resolve(ctx, nsys, r, p2)
89
if err == core.ErrNoNamesys {
90
return nil, coreiface.ErrOffline
91
} else if err != nil {
core/coreapi/unixfs.go
+5
-2
@@ -6,6 +6,7 @@ import (
6
7
coreiface "github.com/ipfs/go-ipfs/core/coreapi/interface"
8
coreunix "github.com/ipfs/go-ipfs/core/coreunix"
9
+ dag "github.com/ipfs/go-ipfs/merkledag"
10
uio "github.com/ipfs/go-ipfs/unixfs/io"
11
12
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
@@ -30,12 +31,14 @@ func (api *UnixfsAPI) Add(ctx context.Context, r io.Reader) (coreiface.Path, err
31
32
// Cat returns the data contained by an IPFS or IPNS object(s) at path `p`.
33
func (api *UnixfsAPI) Cat(ctx context.Context, p coreiface.Path) (coreiface.Reader, error) {
33
- dagnode, err := api.core().ResolveNode(ctx, p)
34
+ ses := dag.NewSession(ctx, api.node.DAG)
35
+
36
+ dagnode, err := resolveNode(ctx, ses, api.node.Namesys, p)
37
if err != nil {
38
return nil, err
39
}
40
38
- r, err := uio.NewDagReader(ctx, dagnode, api.node.DAG)
41
+ r, err := uio.NewDagReader(ctx, dagnode, ses)
42
if err == uio.ErrIsDir {
43
return nil, coreiface.ErrIsDir
44
} else if err != nil {
merkledag/errservice.go
new
+41
@@ -0,0 +1,41 @@
1
+package merkledag
2
+
3
+import (
4
+ "context"
5
+
6
+ cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
7
+ ipld "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
8
+)
9
+
10
+// ErrorService implements ipld.DAGService, returning 'Err' for every call.
11
+type ErrorService struct {
12
+ Err error
13
+}
14
+
15
+var _ ipld.DAGService = (*ErrorService)(nil)
16
+
17
+func (cs *ErrorService) Add(ctx context.Context, nd ipld.Node) error {
18
+ return cs.Err
19
+}
20
+
21
+func (cs *ErrorService) AddMany(ctx context.Context, nds []ipld.Node) error {
22
+ return cs.Err
23
+}
24
+
25
+func (cs *ErrorService) Get(ctx context.Context, c *cid.Cid) (ipld.Node, error) {
26
+ return nil, cs.Err
27
+}
28
+
29
+func (cs *ErrorService) GetMany(ctx context.Context, cids []*cid.Cid) <-chan *ipld.NodeOption {
30
+ ch := make(chan *ipld.NodeOption)
31
+ close(ch)
32
+ return ch
33
+}
34
+
35
+func (cs *ErrorService) Remove(ctx context.Context, c *cid.Cid) error {
36
+ return cs.Err
37
+}
38
+
39
+func (cs *ErrorService) RemoveMany(ctx context.Context, cids []*cid.Cid) error {
40
+ return cs.Err
41
+}
merkledag/merkledag.go
+4
@@ -146,6 +146,10 @@ func (sg *sesGetter) GetMany(ctx context.Context, keys []*cid.Cid) <-chan *ipld.
146
return getNodesFromBG(ctx, sg.bs, keys)
147
}
148
149
+func (ds *dagService) Session(ctx context.Context) ipld.NodeGetter {
150
+ return &sesGetter{bserv.NewSession(ctx, ds.Blocks)}
151
+}
152
+
153
// FetchGraph fetches all nodes that are children of the given node
154
func FetchGraph(ctx context.Context, root *cid.Cid, serv ipld.DAGService) error {
155
var ng ipld.NodeGetter = serv
merkledag/readonly.go
new
+16
@@ -0,0 +1,16 @@
1
+package merkledag
2
+
3
+import (
4
+ "fmt"
5
+
6
+ ipld "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
7
+)
8
+
9
+var ErrReadOnly = fmt.Errorf("cannot write to readonly DAGService")
10
+
11
+func NewReadOnlyDagService(ng ipld.NodeGetter) ipld.DAGService {
12
+ return &ComboService{
13
+ Read: ng,
14
+ Write: &ErrorService{ErrReadOnly},
15
+ }
16
+}
merkledag/readonly_test.go
new
+64
@@ -0,0 +1,64 @@
1
+package merkledag_test
2
+
3
+import (
4
+ "context"
5
+ "testing"
6
+
7
+ . "github.com/ipfs/go-ipfs/merkledag"
8
+ dstest "github.com/ipfs/go-ipfs/merkledag/test"
9
+
10
+ cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
11
+ ipld "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
12
+)
13
+
14
+func TestReadonlyProperties(t *testing.T) {
15
+ ds := dstest.Mock()
16
+ ro := NewReadOnlyDagService(ds)
17
+
18
+ ctx := context.Background()
19
+ nds := []ipld.Node{
20
+ NewRawNode([]byte("foo1")),
21
+ NewRawNode([]byte("foo2")),
22
+ NewRawNode([]byte("foo3")),
23
+ NewRawNode([]byte("foo4")),
24
+ }
25
+ cids := []*cid.Cid{
26
+ nds[0].Cid(),
27
+ nds[1].Cid(),
28
+ nds[2].Cid(),
29
+ nds[3].Cid(),
30
+ }
31
+
32
+ // add to the actual underlying datastore
33
+ if err := ds.Add(ctx, nds[2]); err != nil {
34
+ t.Fatal(err)
35
+ }
36
+ if err := ds.Add(ctx, nds[3]); err != nil {
37
+ t.Fatal(err)
38
+ }
39
+
40
+ if err := ro.Add(ctx, nds[0]); err != ErrReadOnly {
41
+ t.Fatal("expected ErrReadOnly")
42
+ }
43
+ if err := ro.Add(ctx, nds[2]); err != ErrReadOnly {
44
+ t.Fatal("expected ErrReadOnly")
45
+ }
46
+
47
+ if err := ro.AddMany(ctx, nds[0:1]); err != ErrReadOnly {
48
+ t.Fatal("expected ErrReadOnly")
49
+ }
50
+
51
+ if err := ro.Remove(ctx, cids[3]); err != ErrReadOnly {
52
+ t.Fatal("expected ErrReadOnly")
53
+ }
54
+ if err := ro.RemoveMany(ctx, cids[1:2]); err != ErrReadOnly {
55
+ t.Fatal("expected ErrReadOnly")
56
+ }
57
+
58
+ if _, err := ro.Get(ctx, cids[0]); err != ipld.ErrNotFound {
59
+ t.Fatal("expected ErrNotFound")
60
+ }
61
+ if _, err := ro.Get(ctx, cids[3]); err != nil {
62
+ t.Fatal(err)
63
+ }
64
+}
merkledag/rwservice.go
new
+41
@@ -0,0 +1,41 @@
1
+package merkledag
2
+
3
+import (
4
+ "context"
5
+
6
+ cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
7
+ ipld "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
8
+)
9
+
10
+// ComboService implements ipld.DAGService, using 'Read' for all fetch methods,
11
+// and 'Write' for all methods that add new objects.
12
+type ComboService struct {
13
+ Read ipld.NodeGetter
14
+ Write ipld.DAGService
15
+}
16
+
17
+var _ ipld.DAGService = (*ComboService)(nil)
18
+
19
+func (cs *ComboService) Add(ctx context.Context, nd ipld.Node) error {
20
+ return cs.Write.Add(ctx, nd)
21
+}
22
+
23
+func (cs *ComboService) AddMany(ctx context.Context, nds []ipld.Node) error {
24
+ return cs.Write.AddMany(ctx, nds)
25
+}
26
+
27
+func (cs *ComboService) Get(ctx context.Context, c *cid.Cid) (ipld.Node, error) {
28
+ return cs.Read.Get(ctx, c)
29
+}
30
+
31
+func (cs *ComboService) GetMany(ctx context.Context, cids []*cid.Cid) <-chan *ipld.NodeOption {
32
+ return cs.Read.GetMany(ctx, cids)
33
+}
34
+
35
+func (cs *ComboService) Remove(ctx context.Context, c *cid.Cid) error {
36
+ return cs.Write.Remove(ctx, c)
37
+}
38
+
39
+func (cs *ComboService) RemoveMany(ctx context.Context, cids []*cid.Cid) error {
40
+ return cs.Write.RemoveMany(ctx, cids)
41
+}
merkledag/session.go
new
+18
@@ -0,0 +1,18 @@
1
+package merkledag
2
+
3
+import (
4
+ "context"
5
+
6
+ ipld "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
7
+)
8
+
9
+type SessionMaker interface {
10
+ Session(context.Context) ipld.NodeGetter
11
+}
12
+
13
+func NewSession(ctx context.Context, g ipld.NodeGetter) ipld.NodeGetter {
14
+ if sm, ok := g.(SessionMaker); ok {
15
+ return sm.Session(ctx)
16
+ }
17
+ return g
18
+}
path/resolver.go
+3
-3
@@ -35,9 +35,9 @@ func (e ErrNoLink) Error() string {
35
// TODO: now that this is more modular, try to unify this code with the
36
// the resolvers in namesys
37
type Resolver struct {
38
- DAG ipld.DAGService
38
+ DAG ipld.NodeGetter
39
40
- ResolveOnce func(ctx context.Context, ds ipld.DAGService, nd ipld.Node, names []string) (*ipld.Link, []string, error)
40
+ ResolveOnce func(ctx context.Context, ds ipld.NodeGetter, nd ipld.Node, names []string) (*ipld.Link, []string, error)
41
}
42
43
// NewBasicResolver constructs a new basic resolver.
@@ -124,7 +124,7 @@ func (s *Resolver) ResolvePath(ctx context.Context, fpath Path) (ipld.Node, erro
124
125
// ResolveSingle simply resolves one hop of a path through a graph with no
126
// extra context (does not opaquely resolve through sharded nodes)
127
-func ResolveSingle(ctx context.Context, ds ipld.DAGService, nd ipld.Node, names []string) (*ipld.Link, []string, error) {
127
+func ResolveSingle(ctx context.Context, ds ipld.NodeGetter, nd ipld.Node, names []string) (*ipld.Link, []string, error) {
128
return nd.ResolveLink(names)
129
}
130
unixfs/io/dagreader.go
+1
-1
@@ -34,7 +34,7 @@ type ReadSeekCloser interface {
34
35
// NewDagReader creates a new reader object that reads the data represented by
36
// the given node, using the passed in DAGService for data retreival
37
-func NewDagReader(ctx context.Context, n ipld.Node, serv ipld.DAGService) (DagReader, error) {
37
+func NewDagReader(ctx context.Context, n ipld.Node, serv ipld.NodeGetter) (DagReader, error) {
38
switch n := n.(type) {
39
case *mdag.RawNode:
40
return NewBufDagReader(n.RawData()), nil
unixfs/io/pbdagreader.go
+2
-2
@@ -17,7 +17,7 @@ import (
17
18
// DagReader provides a way to easily read the data contained in a dag.
19
type pbDagReader struct {
20
- serv ipld.DAGService
20
+ serv ipld.NodeGetter
21
22
// the node being read
23
node *mdag.ProtoNode
@@ -51,7 +51,7 @@ type pbDagReader struct {
51
var _ DagReader = (*pbDagReader)(nil)
52
53
// NewPBFileReader constructs a new PBFileReader.
54
-func NewPBFileReader(ctx context.Context, n *mdag.ProtoNode, pb *ftpb.Data, serv ipld.DAGService) *pbDagReader {
54
+func NewPBFileReader(ctx context.Context, n *mdag.ProtoNode, pb *ftpb.Data, serv ipld.NodeGetter) *pbDagReader {
55
fctx, cancel := context.WithCancel(ctx)
56
curLinks := getLinkCids(n)
57
return &pbDagReader{
unixfs/io/resolve.go
+3
-2
@@ -12,7 +12,7 @@ import (
12
13
// ResolveUnixfsOnce resolves a single hop of a path through a graph in a
14
// unixfs context. This includes handling traversing sharded directories.
15
-func ResolveUnixfsOnce(ctx context.Context, ds ipld.DAGService, nd ipld.Node, names []string) (*ipld.Link, []string, error) {
15
+func ResolveUnixfsOnce(ctx context.Context, ds ipld.NodeGetter, nd ipld.Node, names []string) (*ipld.Link, []string, error) {
16
switch nd := nd.(type) {
17
case *dag.ProtoNode:
18
upb, err := ft.FromBytes(nd.Data())
@@ -28,7 +28,8 @@ func ResolveUnixfsOnce(ctx context.Context, ds ipld.DAGService, nd ipld.Node, na
28
29
switch upb.GetType() {
30
case ft.THAMTShard:
31
- s, err := hamt.NewHamtFromDag(ds, nd)
31
+ rods := dag.NewReadOnlyDagService(ds)
32
+ s, err := hamt.NewHamtFromDag(rods, nd)
33
if err != nil {
34
return nil, nil, err
35
}