wip

client/rpc/pin.go

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

client/rpc/unixfs.go

@@ -144,16 +144,12 @@ type lsOutput struct {
 	Objects []lsObject
 }
 
-func (api *UnixfsAPI) Ls(ctx context.Context, p path.Path, opts ...caopts.UnixfsLsOption) (<-chan iface.DirEntry, <-chan error) {
-	out := make(chan iface.DirEntry)
-	errOut := make(chan error, 1)
+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 {
-		errOut <- err
-		close(out)
-		close(errOut)
-		return out, errOut
+		return err
 	}
 
 	resp, err := api.core().Request("ls", p.String()).
@@ -162,79 +158,66 @@ func (api *UnixfsAPI) Ls(ctx context.Context, p path.Path, opts ...caopts.Unixfs
 		Option("stream", true).
 		Send(ctx)
 	if err != nil {
-		errOut <- err
-		close(out)
-		close(errOut)
-		return out, errOut
+		return err
 	}
 	if resp.Error != nil {
-		errOut <- resp.Error
-		close(out)
-		close(errOut)
-		return out, errOut
+		return err
 	}
+	defer resp.Close()
 
 	dec := json.NewDecoder(resp.Output)
 
-	go func() {
-		defer resp.Close()
-		defer close(out)
-		defer close(errOut)
-
-		for {
-			var link lsOutput
-			if err := dec.Decode(&link); err != nil {
-				if err != io.EOF {
-					errOut <- err
-				}
-				return
+	for {
+		var link lsOutput
+		if err = dec.Decode(&link); err != nil {
+			if err != io.EOF {
+				return err
 			}
+			return nil
+		}
 
-			if len(link.Objects) != 1 {
-				errOut <- errors.New("unexpected Objects len")
-				return
-			}
+		if len(link.Objects) != 1 {
+			return errors.New("unexpected Objects len")
+		}
 
-			if len(link.Objects[0].Links) != 1 {
-				errOut <- errors.New("unexpected Links len")
-				return
-			}
+		if len(link.Objects[0].Links) != 1 {
+			return errors.New("unexpected Links len")
+		}
 
-			l0 := link.Objects[0].Links[0]
+		l0 := link.Objects[0].Links[0]
 
-			c, err := cid.Decode(l0.Hash)
-			if err != nil {
-				errOut <- err
-				return
-			}
+		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
-			}
+		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():
-			}
+		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()
 		}
-	}()
+	}
 
-	return out, errOut
+	return nil
 }
 
 func (api *UnixfsAPI) core() *HttpApi {

core/commands/ls.go

@@ -133,14 +133,21 @@ The JSON output contains type information.
 			}
 		}
 
+		lsCtx, cancel := context.WithCancel(req.Context)
+		defer cancel()
+
 		for i, fpath := range paths {
 			pth, err := cmdutils.PathOrCidPath(fpath)
 			if err != nil {
 				return err
 			}
 
-			results, errCh := api.Unixfs().Ls(req.Context, pth,
-				options.Unixfs.ResolveChildren(resolveSize || resolveType))
+			results := make(chan iface.DirEntry)
+			var lsErr error
+			go func() {
+				lsErr = api.Unixfs().Ls(lsCtx, pth, results,
+					options.Unixfs.ResolveChildren(resolveSize || resolveType))
+			}()
 
 			processLink, dirDone = processDir()
 			for link := range results {
@@ -164,12 +171,12 @@ The JSON output contains type information.
 					Mode:    link.Mode,
 					ModTime: link.ModTime,
 				}
-				if err := processLink(paths[i], lsLink); err != nil {
+				if err = processLink(paths[i], lsLink); err != nil {
 					return err
 				}
 			}
-			if err = <-errCh; err != nil {
-				return err
+			if lsErr != nil {
+				return lsErr
 			}
 			dirDone(i)
 		}

core/commands/pin/pin.go

@@ -557,7 +557,15 @@ func pinLsAll(req *cmds.Request, typeStr string, detailed bool, name string, api
 		panic("unhandled pin type")
 	}
 
-	pins, errCh := api.Pin().Ls(req.Context, opt, options.Pin.Ls.Detailed(detailed), options.Pin.Ls.Name(name))
+	var lsErr error
+	pins := make(chan iface.Pin)
+	lsCtx, cancel := context.WithCancel(req.Context)
+	defer cancel()
+
+	go func() {
+		lsErr = api.Pin().Ls(lsCtx, opt, pins, options.Pin.Ls.Detailed(detailed), options.Pin.Ls.Name(name))
+	}()
+
 	for p := range pins {
 		err = emit(PinLsOutputWrapper{
 			PinLsObject: PinLsObject{
@@ -570,7 +578,7 @@ func pinLsAll(req *cmds.Request, typeStr string, detailed bool, name string, api
 			return err
 		}
 	}
-	return <-errCh
+	return lsErr
 }
 
 const (

core/commands/pin/remotepin.go

@@ -313,7 +313,7 @@ Pass '--status=queued,pinning,pinned,failed' to list pins in all states.
 }
 
 // Executes GET /pins/?query-with-filters
-func lsRemote(ctx context.Context, req *cmds.Request, c *pinclient.Client) (<-chan pinclient.PinStatusGetter, <-chan error) {
+func lsRemote(ctx context.Context, req *cmds.Request, c *pinclient.Client, out chan<- pinclient.PinStatusGetter) error {
 	opts := []pinclient.LsOption{}
 	if name, nameFound := req.Options[pinNameOptionName]; nameFound {
 		nameStr := name.(string)
@@ -326,12 +326,8 @@ func lsRemote(ctx context.Context, req *cmds.Request, c *pinclient.Client) (<-ch
 		for _, rawCID := range cidsRawArr {
 			parsedCID, err := cid.Decode(rawCID)
 			if err != nil {
-				psCh := make(chan pinclient.PinStatusGetter)
-				errCh := make(chan error, 1)
-				errCh <- fmt.Errorf("CID %q cannot be parsed: %v", rawCID, err)
-				close(psCh)
-				close(errCh)
-				return psCh, errCh
+				close(out)
+				return fmt.Errorf("CID %q cannot be parsed: %v", rawCID, err)
 			}
 			parsedCIDs = append(parsedCIDs, parsedCID)
 		}
@@ -343,19 +339,15 @@ func lsRemote(ctx context.Context, req *cmds.Request, c *pinclient.Client) (<-ch
 		for _, rawStatus := range statusRawArr {
 			s := pinclient.Status(rawStatus)
 			if s.String() == string(pinclient.StatusUnknown) {
-				psCh := make(chan pinclient.PinStatusGetter)
-				errCh := make(chan error, 1)
-				errCh <- fmt.Errorf("status %q is not valid", rawStatus)
-				close(psCh)
-				close(errCh)
-				return psCh, errCh
+				close(out)
+				return fmt.Errorf("status %q is not valid", rawStatus)
 			}
 			parsedStatuses = append(parsedStatuses, s)
 		}
 		opts = append(opts, pinclient.PinOpts.FilterStatus(parsedStatuses...))
 	}
 
-	return c.Ls(ctx, opts...)
+	return c.Ls(ctx, out, opts...)
 }
 
 var rmRemotePinCmd = &cmds.Command{