feat(commands): --stream option for ls

Convert LS Command to use current cmds lib
Update LS Command to support streaming
Rebase fixes

License: MIT
Signed-off-by: hannahhoward <hannah@hannahhoward.net>

core/commands/ls.go

 package commands
 
 import (
-	"bytes"
 	"fmt"
 	"io"
 	"text/tabwriter"
 
-	cmds "github.com/ipfs/go-ipfs/commands"
+	cmdenv "github.com/ipfs/go-ipfs/core/commands/cmdenv"
 	e "github.com/ipfs/go-ipfs/core/commands/e"
 	iface "github.com/ipfs/go-ipfs/core/coreapi/interface"
 
@@ -16,29 +15,40 @@ import (
 	unixfspb "gx/ipfs/QmUnHNqhSB1JgzVCxL1Kz3yb4bdyB4q1Z9AD5AUBVmt3fZ/go-unixfs/pb"
 	blockservice "gx/ipfs/QmVDTbzzTwnuBwNbJdhW3u7LoBQp46bezm9yp4z1RoEepM/go-blockservice"
 	offline "gx/ipfs/QmYZwey1thDTynSrvd6qQkX24UpTka6TFhQ2v569UpoqxD/go-ipfs-exchange-offline"
+	cmds "gx/ipfs/Qma6uuSyjkecGhMFFLfzyJDPyoDtNJSHJNweDccZhaWkgU/go-ipfs-cmds"
 	merkledag "gx/ipfs/QmcGt25mrjuB2kKW2zhPbXVZNHc4yoTDQ65NA8m6auP2f1/go-merkledag"
 	ipld "gx/ipfs/QmcKKBwfz6FyQdHR2jsXrrF6XeSBXYL86anmWNewpFpoF5/go-ipld-format"
 	"gx/ipfs/Qmde5VP1qUkyQXKCfmEUA7bP64V2HAptbJ7phuPp7jXWwg/go-ipfs-cmdkit"
 )
 
+// LsLink contains printable data for a single ipld link in ls output
 type LsLink struct {
 	Name, Hash string
 	Size       uint64
 	Type       unixfspb.Data_DataType
 }
 
+// LsObject is an element of LsOutput
+// It can represent a whole directory, a directory header, one or more links,
+// Or a the end of a directory
 type LsObject struct {
 	Hash      string
 	Links     []LsLink
+	HasHeader bool
+	HasLinks  bool
+	HasFooter bool
 }
 
+// LsOutput is a set of printable data for directories
 type LsOutput struct {
+	MultipleFolders bool
 	Objects         []LsObject
 }
 
 const (
 	lsHeadersOptionNameTime = "headers"
 	lsResolveTypeOptionName = "resolve-type"
+	lsStreamOptionName      = "stream"
 )
 
 var LsCmd = &cmds.Command{
@@ -60,32 +70,20 @@ The JSON output contains type information.
 	Options: []cmdkit.Option{
 		cmdkit.BoolOption(lsHeadersOptionNameTime, "v", "Print table headers (Hash, Size, Name)."),
 		cmdkit.BoolOption(lsResolveTypeOptionName, "Resolve linked objects to find out their types.").WithDefault(true),
+		cmdkit.BoolOption(lsStreamOptionName, "s", "Stream directory entries as they are found."),
 	},
-	Run: func(req cmds.Request, res cmds.Response) {
-		nd, err := req.InvocContext().GetNode()
+	Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
+		nd, err := cmdenv.GetNode(env)
 		if err != nil {
-			res.SetError(err, cmdkit.ErrNormal)
-			return
+			return err
 		}
 
-		api, err := req.InvocContext().GetApi()
+		api, err := cmdenv.GetApi(env)
 		if err != nil {
-			res.SetError(err, cmdkit.ErrNormal)
-			return
-		}
-
-		// get options early -> exit early in case of error
-		if _, _, err := req.Option(lsHeadersOptionNameTime).Bool(); err != nil {
-			res.SetError(err, cmdkit.ErrNormal)
-			return
-		}
-
-		resolve, _, err := req.Option(lsResolveTypeOptionName).Bool()
-		if err != nil {
-			res.SetError(err, cmdkit.ErrNormal)
-			return
+			return err
 		}
 
+		resolve, _ := req.Options[lsResolveTypeOptionName].(bool)
 		dserv := nd.DAG
 		if !resolve {
 			offlineexch := offline.Exchange(nd.Blockstore)
@@ -93,125 +91,204 @@ The JSON output contains type information.
 			dserv = merkledag.NewDAGService(bserv)
 		}
 
-		paths := req.Arguments()
+		err = req.ParseBodyArgs()
+		if err != nil {
+			return err
+		}
+
+		paths := req.Arguments
 
 		var dagnodes []ipld.Node
 		for _, fpath := range paths {
 			p, err := iface.ParsePath(fpath)
 			if err != nil {
-				res.SetError(err, cmdkit.ErrNormal)
-				return
+				return err
 			}
 
-			dagnode, err := api.ResolveNode(req.Context(), p)
+			dagnode, err := api.ResolveNode(req.Context, p)
 			if err != nil {
-				res.SetError(err, cmdkit.ErrNormal)
-				return
+				return err
 			}
 			dagnodes = append(dagnodes, dagnode)
 		}
-
-		output := make([]LsObject, len(req.Arguments()))
-		ng := merkledag.NewSession(req.Context(), nd.DAG)
+		ng := merkledag.NewSession(req.Context, nd.DAG)
 		ro := merkledag.NewReadOnlyDagService(ng)
 
+		stream, _ := req.Options[lsStreamOptionName].(bool)
+		multipleFolders := len(req.Arguments) > 1
+		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 {
-				res.SetError(fmt.Errorf("the data in %s (at %q) is not a UnixFS directory: %s", dagnode.Cid(), paths[i], err), cmdkit.ErrNormal)
-				return
+					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())
+					links, err = dir.Links(req.Context)
 					if err != nil {
-					res.SetError(err, cmdkit.ErrNormal)
-					return
+						return err
 					}
 				}
+				outputLinks := make([]LsLink, len(links))
+				for j, link := range links {
+					lsLink, err := makeLsLink(req, dserv, resolve, link)
+					if err != nil {
+						return err
+					}
+					outputLinks[j] = *lsLink
+				}
+				output[i] = newFullDirectoryLsObject(paths[i], outputLinks)
+			}
 
-			output[i] = LsObject{
-				Hash:  paths[i],
-				Links: make([]LsLink, len(links)),
+			return cmds.EmitOnce(res, &LsOutput{multipleFolders, output})
 		}
 
-			for j, link := range links {
-				t := unixfspb.Data_DataType(-1)
+		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)
+			}
 
-				switch link.Cid.Type() {
-				case cid.Raw:
-					// No need to check with raw leaves
-					t = unixfs.TFile
-				case cid.DagProtobuf:
-					linkNode, err := link.GetNode(req.Context(), dserv)
-					if err == ipld.ErrNotFound && !resolve {
-						// not an error
-						linkNode = nil
-					} else if err != nil {
-						res.SetError(err, cmdkit.ErrNormal)
-						return
+			var linkResults <-chan unixfs.LinkResult
+			if dir == nil {
+				linkResults = makeDagNodeLinkResults(req, dagnode)
+			} else {
+				linkResults = dir.EnumLinksAsync(req.Context)
 			}
 
-					if pn, ok := linkNode.(*merkledag.ProtoNode); ok {
-						d, err := unixfs.FSNodeFromBytes(pn.Data())
-						if err != nil {
-							res.SetError(err, cmdkit.ErrNormal)
-							return
+			output := make([]LsObject, 1)
+			outputLinks := make([]LsLink, 1)
+
+			output[0] = newDirectoryHeaderLsObject(paths[i])
+			if err = res.Emit(&LsOutput{multipleFolders, output}); err != nil {
+				return nil
 			}
-						t = d.Type()
+			for linkResult := range linkResults {
+				if linkResult.Err != nil {
+					return linkResult.Err
 				}
+				link := linkResult.Link
+				lsLink, err := makeLsLink(req, dserv, resolve, link)
+				if err != nil {
+					return err
 				}
-				output[i].Links[j] = LsLink{
-					Name: link.Name,
-					Hash: link.Cid.String(),
-					Size: link.Size,
-					Type: t,
+				outputLinks[0] = *lsLink
+				output[0] = newDirectoryLinksLsObject(outputLinks)
+				if err = res.Emit(&LsOutput{multipleFolders, output}); err != nil {
+					return err
 				}
 			}
+			output[0] = newDirectoryFooterLsObject()
+			if err = res.Emit(&LsOutput{multipleFolders, output}); err != nil {
+				return err
 			}
-
-		res.SetOutput(&LsOutput{output})
-	},
-	Marshalers: cmds.MarshalerMap{
-		cmds.Text: func(res cmds.Response) (io.Reader, error) {
-
-			v, err := unwrapOutput(res.Output())
-			if err != nil {
-				return nil, err
 		}
-
-			headers, _, _ := res.Request().Option(lsHeadersOptionNameTime).Bool()
+		return nil
+	},
+	Encoders: cmds.EncoderMap{
+		cmds.Text: cmds.MakeEncoder(func(req *cmds.Request, w io.Writer, v interface{}) error {
+			headers, _ := req.Options[lsHeadersOptionNameTime].(bool)
 			output, ok := v.(*LsOutput)
 			if !ok {
-				return nil, e.TypeErr(output, v)
+				return e.TypeErr(output, v)
 			}
 
-			buf := new(bytes.Buffer)
-			w := tabwriter.NewWriter(buf, 1, 2, 1, ' ', 0)
+			tw := tabwriter.NewWriter(w, 1, 2, 1, ' ', 0)
 			for _, object := range output.Objects {
-				if len(output.Objects) > 1 {
-					fmt.Fprintf(w, "%s:\n", object.Hash)
+				if object.HasHeader {
+					if output.MultipleFolders {
+						fmt.Fprintf(tw, "%s:\n", object.Hash)
 					}
 					if headers {
-					fmt.Fprintln(w, "Hash\tSize\tName")
+						fmt.Fprintln(tw, "Hash\tSize\tName")
 					}
+				}
+				if object.HasLinks {
 					for _, link := range object.Links {
 						if link.Type == unixfs.TDirectory {
 							link.Name += "/"
 						}
-					fmt.Fprintf(w, "%s\t%v\t%s\n", link.Hash, link.Size, link.Name)
+
+						fmt.Fprintf(tw, "%s\t%v\t%s\n", link.Hash, link.Size, link.Name)
 					}
-				if len(output.Objects) > 1 {
-					fmt.Fprintln(w)
 				}
+				if object.HasFooter {
+					if output.MultipleFolders {
+						fmt.Fprintln(tw)
 					}
-			w.Flush()
-
-			return buf, nil
-		},
+				}
+			}
+			tw.Flush()
+			return nil
+		}),
 	},
 	Type: LsOutput{},
 }
+
+func makeDagNodeLinkResults(req *cmds.Request, dagnode ipld.Node) <-chan unixfs.LinkResult {
+	linkResults := make(chan unixfs.LinkResult)
+	go func() {
+		defer close(linkResults)
+		for _, l := range dagnode.Links() {
+			select {
+			case linkResults <- unixfs.LinkResult{
+				Link: l,
+				Err:  nil,
+			}:
+			case <-req.Context.Done():
+				return
+			}
+		}
+	}()
+	return linkResults
+}
+
+func newFullDirectoryLsObject(hash string, links []LsLink) LsObject {
+	return LsObject{hash, links, true, true, true}
+}
+func newDirectoryHeaderLsObject(hash string) LsObject {
+	return LsObject{hash, nil, true, false, false}
+}
+func newDirectoryLinksLsObject(links []LsLink) LsObject {
+	return LsObject{"", links, false, true, false}
+}
+func newDirectoryFooterLsObject() LsObject {
+	return LsObject{"", nil, false, false, true}
+}
+
+func makeLsLink(req *cmds.Request, dserv ipld.DAGService, resolve bool, link *ipld.Link) (*LsLink, error) {
+	t := unixfspb.Data_DataType(-1)
+
+	switch link.Cid.Type() {
+	case cid.Raw:
+		// No need to check with raw leaves
+		t = unixfs.TFile
+	case cid.DagProtobuf:
+		linkNode, err := link.GetNode(req.Context, dserv)
+		if err == ipld.ErrNotFound && !resolve {
+			// 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
+			}
+			t = d.Type()
+		}
+	}
+	return &LsLink{
+		Name: link.Name,
+		Hash: link.Cid.String(),
+		Size: link.Size,
+		Type: t,
+	}, nil
+}

core/commands/root.go

@@ -3,7 +3,6 @@ package commands
 import (
 	"errors"
 
-	lgc "github.com/ipfs/go-ipfs/commands/legacy"
 	dag "github.com/ipfs/go-ipfs/core/commands/dag"
 	name "github.com/ipfs/go-ipfs/core/commands/name"
 	ocmd "github.com/ipfs/go-ipfs/core/commands/object"
@@ -127,7 +126,7 @@ var rootSubcommands = map[string]*cmds.Command{
 	"id":        IDCmd,
 	"key":       KeyCmd,
 	"log":       LogCmd,
-	"ls":        lgc.NewCommand(LsCmd),
+	"ls":        LsCmd,
 	"mount":     MountCmd,
 	"name":      name.NameCmd,
 	"object":    ocmd.ObjectCmd,
@@ -165,7 +164,7 @@ var rootROSubcommands = map[string]*cmds.Command{
 	},
 	"get": GetCmd,
 	"dns": DNSCmd,
-	"ls":  lgc.NewCommand(LsCmd),
+	"ls":  LsCmd,
 	"name": {
 		Subcommands: map[string]*cmds.Command{
 			"resolve": name.IpnsCmd,