refactor(cmds): do not return errors embedded in result type (#10527)

incl. https://github.com/ipfs/boxo/pull/738

client/rpc/pin.go

@@ -62,10 +62,12 @@ type pinLsObject struct {
 	Type string
 }
 
-func (api *PinAPI) Ls(ctx context.Context, opts ...caopts.PinLsOption) (<-chan iface.Pin, error) {
+func (api *PinAPI) Ls(ctx context.Context, pins chan<- iface.Pin, opts ...caopts.PinLsOption) error {
+	defer close(pins)
+
 	options, err := caopts.PinLsOptions(opts...)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	res, err := api.core().Request("pin/ls").
@@ -73,48 +75,32 @@ func (api *PinAPI) Ls(ctx context.Context, opts ...caopts.PinLsOption) (<-chan i
 		Option("stream", true).
 		Send(ctx)
 	if err != nil {
-		return nil, err
+		return err
 	}
+	defer res.Output.Close()
 
-	pins := make(chan iface.Pin)
-	go func(ch chan<- iface.Pin) {
-		defer res.Output.Close()
-		defer close(ch)
-
-		dec := json.NewDecoder(res.Output)
-		var out pinLsObject
-		for {
-			switch err := dec.Decode(&out); err {
-			case nil:
-			case io.EOF:
-				return
-			default:
-				select {
-				case ch <- pin{err: err}:
-					return
-				case <-ctx.Done():
-					return
-				}
-			}
-
-			c, err := cid.Parse(out.Cid)
-			if err != nil {
-				select {
-				case ch <- pin{err: err}:
-					return
-				case <-ctx.Done():
-					return
-				}
-			}
-
-			select {
-			case ch <- pin{typ: out.Type, name: out.Name, path: path.FromCid(c)}:
-			case <-ctx.Done():
-				return
+	dec := json.NewDecoder(res.Output)
+	var out pinLsObject
+	for {
+		err := dec.Decode(&out)
+		if err != nil {
+			if err != io.EOF {
+				return err
 			}
+			return nil
 		}
-	}(pins)
-	return pins, nil
+
+		c, err := cid.Parse(out.Cid)
+		if err != nil {
+			return err
+		}
+
+		select {
+		case pins <- pin{typ: out.Type, name: out.Name, path: path.FromCid(c)}:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
 }
 
 // IsPinned returns whether or not the given cid is pinned

client/rpc/unixfs.go

@@ -144,10 +144,12 @@ type lsOutput struct {
 	Objects []lsObject
 }
 
-func (api *UnixfsAPI) Ls(ctx context.Context, p path.Path, opts ...caopts.UnixfsLsOption) (<-chan iface.DirEntry, error) {
+func (api *UnixfsAPI) Ls(ctx context.Context, p path.Path, out chan<- iface.DirEntry, opts ...caopts.UnixfsLsOption) error {
+	defer close(out)
+
 	options, err := caopts.UnixfsLsOptions(opts...)
 	if err != nil {
-		return nil, err
+		return err
 	}
 
 	resp, err := api.core().Request("ls", p.String()).
@@ -156,86 +158,64 @@ func (api *UnixfsAPI) Ls(ctx context.Context, p path.Path, opts ...caopts.Unixfs
 		Option("stream", true).
 		Send(ctx)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	if resp.Error != nil {
-		return nil, resp.Error
+		return err
 	}
+	defer resp.Close()
 
 	dec := json.NewDecoder(resp.Output)
-	out := make(chan iface.DirEntry)
 
-	go func() {
-		defer resp.Close()
-		defer close(out)
-
-		for {
-			var link lsOutput
-			if err := dec.Decode(&link); err != nil {
-				if err == io.EOF {
-					return
-				}
-				select {
-				case out <- iface.DirEntry{Err: err}:
-				case <-ctx.Done():
-				}
-				return
-			}
-
-			if len(link.Objects) != 1 {
-				select {
-				case out <- iface.DirEntry{Err: errors.New("unexpected Objects len")}:
-				case <-ctx.Done():
-				}
-				return
-			}
-
-			if len(link.Objects[0].Links) != 1 {
-				select {
-				case out <- iface.DirEntry{Err: errors.New("unexpected Links len")}:
-				case <-ctx.Done():
-				}
-				return
-			}
-
-			l0 := link.Objects[0].Links[0]
-
-			c, err := cid.Decode(l0.Hash)
-			if err != nil {
-				select {
-				case out <- iface.DirEntry{Err: err}:
-				case <-ctx.Done():
-				}
-				return
-			}
-
-			var ftype iface.FileType
-			switch l0.Type {
-			case unixfs.TRaw, unixfs.TFile:
-				ftype = iface.TFile
-			case unixfs.THAMTShard, unixfs.TDirectory, unixfs.TMetadata:
-				ftype = iface.TDirectory
-			case unixfs.TSymlink:
-				ftype = iface.TSymlink
-			}
-
-			select {
-			case out <- iface.DirEntry{
-				Name:   l0.Name,
-				Cid:    c,
-				Size:   l0.Size,
-				Type:   ftype,
-				Target: l0.Target,
-
-				Mode:    l0.Mode,
-				ModTime: l0.ModTime,
-			}:
-			case <-ctx.Done():
+	for {
+		var link lsOutput
+		if err = dec.Decode(&link); err != nil {
+			if err != io.EOF {
+				return err
 			}
+			return nil
 		}
-	}()
 
-	return out, nil
+		if len(link.Objects) != 1 {
+			return errors.New("unexpected Objects len")
+		}
+
+		if len(link.Objects[0].Links) != 1 {
+			return errors.New("unexpected Links len")
+		}
+
+		l0 := link.Objects[0].Links[0]
+
+		c, err := cid.Decode(l0.Hash)
+		if err != nil {
+			return err
+		}
+
+		var ftype iface.FileType
+		switch l0.Type {
+		case unixfs.TRaw, unixfs.TFile:
+			ftype = iface.TFile
+		case unixfs.THAMTShard, unixfs.TDirectory, unixfs.TMetadata:
+			ftype = iface.TDirectory
+		case unixfs.TSymlink:
+			ftype = iface.TSymlink
+		}
+
+		select {
+		case out <- iface.DirEntry{
+			Name:   l0.Name,
+			Cid:    c,
+			Size:   l0.Size,
+			Type:   ftype,
+			Target: l0.Target,
+
+			Mode:    l0.Mode,
+			ModTime: l0.ModTime,
+		}:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
 }
 
 func (api *UnixfsAPI) core() *HttpApi {