Port ipfs ls to CoreAPI

core/commands/ls.go

@@ -8,18 +8,12 @@ import (
 
 	cmdenv "github.com/ipfs/go-ipfs/core/commands/cmdenv"
 	iface "github.com/ipfs/go-ipfs/core/coreapi/interface"
+	options "github.com/ipfs/go-ipfs/core/coreapi/interface/options"
 
 	unixfs "gx/ipfs/QmQ1JnYpnzkaurjW1yxkQxC2w3K1PorNE1nv1vaP5Le7sq/go-unixfs"
-	uio "gx/ipfs/QmQ1JnYpnzkaurjW1yxkQxC2w3K1PorNE1nv1vaP5Le7sq/go-unixfs/io"
 	unixfspb "gx/ipfs/QmQ1JnYpnzkaurjW1yxkQxC2w3K1PorNE1nv1vaP5Le7sq/go-unixfs/pb"
 	cmds "gx/ipfs/QmR77mMvvh8mJBBWQmBfQBu8oD38NUN4KE9SL2gDgAQNc6/go-ipfs-cmds"
-	cid "gx/ipfs/QmR8BauakNcBa3RbE4nbQu76PDiJgoQgz8AJdhJuiU4TAw/go-cid"
-	ipld "gx/ipfs/QmRL22E4paat7ky7vx9MLpR97JHHbFPrg3ytFQw6qp1y1s/go-ipld-format"
-	blockservice "gx/ipfs/QmVKQHuzni68SWByzJgBUCwHvvr4TWiXfutNWWwpZpp4rE/go-blockservice"
-	offline "gx/ipfs/QmYZwey1thDTynSrvd6qQkX24UpTka6TFhQ2v569UpoqxD/go-ipfs-exchange-offline"
-	merkledag "gx/ipfs/Qmb2UEG2TAeVrEJSjqsZF7Y2he7wRDkrdt6c3bECxwZf4k/go-merkledag"
-	cidenc "gx/ipfs/QmdPQx9fvN5ExVwMhRmh7YpCQJzJrFhd1AjVBwJmRMFJeX/go-cidutil/cidenc"
-	"gx/ipfs/Qmde5VP1qUkyQXKCfmEUA7bP64V2HAptbJ7phuPp7jXWwg/go-ipfs-cmdkit"
+	cmdkit "gx/ipfs/Qmde5VP1qUkyQXKCfmEUA7bP64V2HAptbJ7phuPp7jXWwg/go-ipfs-cmdkit"
 )
 
 // LsLink contains printable data for a single ipld link in ls output
@@ -72,24 +66,14 @@ The JSON output contains type information.
 		cmdkit.BoolOption(lsStreamOptionName, "s", "Enable exprimental streaming of directory entries as they are traversed."),
 	},
 	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
-		nd, err := cmdenv.GetNode(env)
-		if err != nil {
-			return err
-		}
-
 		api, err := cmdenv.GetApi(env, req)
 		if err != nil {
 			return err
 		}
 
 		resolveType, _ := req.Options[lsResolveTypeOptionName].(bool)
 		resolveSize, _ := req.Options[lsSizeOptionName].(bool)
-		dserv := nd.DAG
-		if !resolveType && !resolveSize {
-			offlineexch := offline.Exchange(nd.Blockstore)
-			bserv := blockservice.New(nd.Blockstore, offlineexch)
-			dserv = merkledag.NewDAGService(bserv)
-		}
+		stream, _ := req.Options[lsStreamOptionName].(bool)
 
 		err = req.ParseBodyArgs()
 		if err != nil {
@@ -102,90 +86,80 @@ The JSON output contains type information.
 			return err
 		}
 
-		var dagnodes []ipld.Node
-		for _, fpath := range paths {
-			p, err := iface.ParsePath(fpath)
-			if err != nil {
-				return err
-			}
-			dagnode, err := api.ResolveNode(req.Context, p)
-			if err != nil {
-				return err
-			}
-			dagnodes = append(dagnodes, dagnode)
-		}
-		ng := merkledag.NewSession(req.Context, nd.DAG)
-		ro := merkledag.NewReadOnlyDagService(ng)
+		var processLink func(path string, link LsLink) error
+		var dirDone func(i int)
 
-		stream, _ := req.Options[lsStreamOptionName].(bool)
+		processDir := func() (func(path string, link LsLink) error, func(i int)) {
+			return func(path string, link LsLink) error {
+				output := []LsObject{{
+					Hash:  path,
+					Links: []LsLink{link},
+				}}
+				return res.Emit(&LsOutput{output})
+			}, func(i int) {}
+		}
+		done := func() error { return nil }
 
 		if !stream {
 			output := make([]LsObject, len(req.Arguments))
 
-			for i, dagnode := range dagnodes {
-				dir, err := uio.NewDirectoryFromNode(ro, dagnode)
-				if err != nil && err != uio.ErrNotADir {
-					return fmt.Errorf("the data in %s (at %q) is not a UnixFS directory: %s", dagnode.Cid(), paths[i], err)
-				}
-
-				var links []*ipld.Link
-				if dir == nil {
-					links = dagnode.Links()
-				} else {
-					links, err = dir.Links(req.Context)
-					if err != nil {
-						return err
-					}
-				}
-				outputLinks := make([]LsLink, len(links))
-				for j, link := range links {
-					lsLink, err := makeLsLink(req, dserv, resolveType, resolveSize, link, enc)
-					if err != nil {
-						return err
+			processDir = func() (func(path string, link LsLink) error, func(i int)) {
+				// for each dir
+				outputLinks := make([]LsLink, 0)
+				return func(path string, link LsLink) error {
+						// for each link
+						outputLinks = append(outputLinks, link)
+						return nil
+					}, func(i int) {
+						// after each dir
+						output[i] = LsObject{
+							Hash:  paths[i],
+							Links: outputLinks,
+						}
 					}
-					outputLinks[j] = *lsLink
-				}
-				output[i] = LsObject{
-					Hash:  paths[i],
-					Links: outputLinks,
-				}
 			}
 
-			return cmds.EmitOnce(res, &LsOutput{output})
+			done = func() error {
+				return cmds.EmitOnce(res, &LsOutput{output})
+			}
 		}
 
-		for i, dagnode := range dagnodes {
-			dir, err := uio.NewDirectoryFromNode(ro, dagnode)
-			if err != nil && err != uio.ErrNotADir {
-				return fmt.Errorf("the data in %s (at %q) is not a UnixFS directory: %s", dagnode.Cid(), paths[i], err)
+		for i, fpath := range paths {
+			p, err := iface.ParsePath(fpath)
+			if err != nil {
+				return err
 			}
 
-			var linkResults <-chan unixfs.LinkResult
-			if dir == nil {
-				linkResults = makeDagNodeLinkResults(req, dagnode)
-			} else {
-				linkResults = dir.EnumLinksAsync(req.Context)
+			results, err := api.Unixfs().Ls(req.Context, p,
+				options.Unixfs.Async(stream),
+				options.Unixfs.ResolveType(resolveType),
+				options.Unixfs.ResolveSize(resolveSize))
+			if err != nil {
+				return err
 			}
 
-			for linkResult := range linkResults {
-
-				if linkResult.Err != nil {
-					return linkResult.Err
+			processLink, dirDone = processDir()
+			for link := range results {
+				if link.Err != nil {
+					return link.Err
 				}
-				link := linkResult.Link
-				lsLink, err := makeLsLink(req, dserv, resolveType, resolveSize, link, enc)
-				if err != nil {
-					return err
+				lsLink := LsLink{
+					Name: link.Link.Name,
+					Hash: enc.Encode(link.Link.Cid),
+
+					Size: link.Size,
+					Type: link.Type,
 				}
-				output := []LsObject{{
-					Hash:  paths[i],
-					Links: []LsLink{*lsLink},
-				}}
-				if err = res.Emit(&LsOutput{output}); err != nil {
+				if err := processLink(paths[i], lsLink); err != nil {
 					return err
 				}
 			}
+			dirDone(i)
 		}
+		if err := done(); err != nil {
+			return err
+		}
+
 		return nil
 	},
 	PostRun: cmds.PostRunMap{
@@ -219,58 +193,6 @@ The JSON output contains type information.
 	Type: LsOutput{},
 }
 
-func makeDagNodeLinkResults(req *cmds.Request, dagnode ipld.Node) <-chan unixfs.LinkResult {
-	links := dagnode.Links()
-	linkResults := make(chan unixfs.LinkResult, len(links))
-	defer close(linkResults)
-	for _, l := range links {
-		linkResults <- unixfs.LinkResult{
-			Link: l,
-			Err:  nil,
-		}
-	}
-	return linkResults
-}
-
-func makeLsLink(req *cmds.Request, dserv ipld.DAGService, resolveType bool, resolveSize bool, link *ipld.Link, enc cidenc.Encoder) (*LsLink, error) {
-	t := unixfspb.Data_DataType(-1)
-	var size uint64
-
-	switch link.Cid.Type() {
-	case cid.Raw:
-		// No need to check with raw leaves
-		t = unixfs.TFile
-		size = link.Size
-	case cid.DagProtobuf:
-		linkNode, err := link.GetNode(req.Context, dserv)
-		if err == ipld.ErrNotFound && !resolveType && !resolveSize {
-			// not an error
-			linkNode = nil
-		} else if err != nil {
-			return nil, err
-		}
-
-		if pn, ok := linkNode.(*merkledag.ProtoNode); ok {
-			d, err := unixfs.FSNodeFromBytes(pn.Data())
-			if err != nil {
-				return nil, err
-			}
-			if resolveType {
-				t = d.Type()
-			}
-			if d.Type() == unixfs.TFile && resolveSize {
-				size = d.FileSize()
-			}
-		}
-	}
-	return &LsLink{
-		Name: link.Name,
-		Hash: enc.Encode(link.Cid),
-		Size: size,
-		Type: t,
-	}, nil
-}
-
 func tabularOutput(req *cmds.Request, w io.Writer, out *LsOutput, lastObjectHash string, ignoreBreaks bool) string {
 	headers, _ := req.Options[lsHeadersOptionNameTime].(bool)
 	stream, _ := req.Options[lsStreamOptionName].(bool)

core/coreapi/interface/options/unixfs.go

@@ -42,7 +42,15 @@ type UnixfsAddSettings struct {
 	Progress bool
 }
 
+type UnixfsLsSettings struct {
+	Async bool
+
+	ResolveType bool
+	ResolveSize bool
+}
+
 type UnixfsAddOption func(*UnixfsAddSettings) error
+type UnixfsLsOption func(*UnixfsLsSettings) error
 
 func UnixfsAddOptions(opts ...UnixfsAddOption) (*UnixfsAddSettings, cid.Prefix, error) {
 	options := &UnixfsAddSettings{
@@ -122,6 +130,24 @@ func UnixfsAddOptions(opts ...UnixfsAddOption) (*UnixfsAddSettings, cid.Prefix,
 	return options, prefix, nil
 }
 
+func UnixfsLsOptions(opts ...UnixfsLsOption) (*UnixfsLsSettings, error) {
+	options := &UnixfsLsSettings{
+		Async: true,
+
+		ResolveSize: true,
+		ResolveType: true,
+	}
+
+	for _, opt := range opts {
+		err := opt(options)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return options, nil
+}
+
 type unixfsOpts struct{}
 
 var Unixfs unixfsOpts
@@ -290,3 +316,27 @@ func (unixfsOpts) Nocopy(enable bool) UnixfsAddOption {
 		return nil
 	}
 }
+
+// Async tells ls to return results as soon as they are available, which can be
+// useful for listing HAMT directories. When this option is set to true returned
+// results won't be returned in order
+func (unixfsOpts) Async(async bool) UnixfsLsOption {
+	return func(settings *UnixfsLsSettings) error {
+		settings.Async = async
+		return nil
+	}
+}
+
+func (unixfsOpts) ResolveSize(resolve bool) UnixfsLsOption {
+	return func(settings *UnixfsLsSettings) error {
+		settings.ResolveSize = resolve
+		return nil
+	}
+}
+
+func (unixfsOpts) ResolveType(resolve bool) UnixfsLsOption {
+	return func(settings *UnixfsLsSettings) error {
+		settings.ResolveType = resolve
+		return nil
+	}
+}

core/coreapi/interface/tests/unixfs.go

@@ -749,22 +749,42 @@ func (tp *provider) TestLs(t *testing.T) {
 		t.Error(err)
 	}
 
-	links, err := api.Unixfs().Ls(ctx, p)
+	links, err := api.Unixfs().Ls(ctx, p, options.Unixfs.Async(false))
 	if err != nil {
 		t.Error(err)
 	}
 
-	if len(links) != 1 {
-		t.Fatalf("expected 1 link, got %d", len(links))
+	link := (<-links).Link
+	if link.Size != 23 {
+		t.Fatalf("expected size = 23, got %d", link.Size)
 	}
-	if links[0].Size != 23 {
-		t.Fatalf("expected size = 23, got %d", links[0].Size)
+	if link.Name != "name-of-file" {
+		t.Fatalf("expected name = name-of-file, got %s", link.Name)
 	}
-	if links[0].Name != "name-of-file" {
-		t.Fatalf("expected name = name-of-file, got %s", links[0].Name)
+	if link.Cid.String() != "QmX3qQVKxDGz3URVC3861Z3CKtQKGBn6ffXRBBWGMFz9Lr" {
+		t.Fatalf("expected cid = QmX3qQVKxDGz3URVC3861Z3CKtQKGBn6ffXRBBWGMFz9Lr, got %s", link.Cid)
 	}
-	if links[0].Cid.String() != "QmX3qQVKxDGz3URVC3861Z3CKtQKGBn6ffXRBBWGMFz9Lr" {
-		t.Fatalf("expected cid = QmX3qQVKxDGz3URVC3861Z3CKtQKGBn6ffXRBBWGMFz9Lr, got %s", links[0].Cid)
+	if _, ok := <-links; ok {
+		t.Errorf("didn't expect a second link")
+	}
+
+	links, err = api.Unixfs().Ls(ctx, p, options.Unixfs.Async(true))
+	if err != nil {
+		t.Error(err)
+	}
+
+	link = (<-links).Link
+	if link.Size != 23 {
+		t.Fatalf("expected size = 23, got %d", link.Size)
+	}
+	if link.Name != "name-of-file" {
+		t.Fatalf("expected name = name-of-file, got %s", link.Name)
+	}
+	if link.Cid.String() != "QmX3qQVKxDGz3URVC3861Z3CKtQKGBn6ffXRBBWGMFz9Lr" {
+		t.Fatalf("expected cid = QmX3qQVKxDGz3URVC3861Z3CKtQKGBn6ffXRBBWGMFz9Lr, got %s", link.Cid)
+	}
+	if _, ok := <-links; ok {
+		t.Errorf("didn't expect a second link")
 	}
 }