Merge pull request #5667 from ipfs/fix/simplify-dag-add

simplify dag put and correctly take pin lock

core/commands/dag/dag.go

@@ -6,7 +6,6 @@ import (
 	"math"
 
 	"github.com/ipfs/go-ipfs/core/commands/cmdenv"
-	"github.com/ipfs/go-ipfs/core/commands/e"
 	"github.com/ipfs/go-ipfs/core/coredag"
 	"github.com/ipfs/go-ipfs/pin"
 
@@ -14,7 +13,6 @@ import (
 	mh "gx/ipfs/QmPnFwZ2JXKnXgMw8CdBPxn7FWh6LLdjUjxV1fKHuJnkr8/go-multihash"
 	path "gx/ipfs/QmRKuTyCzg7HFBcV1YUhzStroGtJSb8iWgyxfsDCwFhWTS/go-path"
 	cmds "gx/ipfs/QmSXUokcP4TJpFfqozT69AVAYRtzXVMUjzQVkYX41R9Svs/go-ipfs-cmds"
-	files "gx/ipfs/QmZMWMvWMVKCbHetJ4RgndbuEF1io2UpUxwQwtNjtYPzSC/go-ipfs-files"
 	ipld "gx/ipfs/QmdDXJs4axxefSPgK6Y1QhpJWKuDPnGJiqgq4uncb4rFHL/go-ipld-format"
 	cmdkit "gx/ipfs/Qmde5VP1qUkyQXKCfmEUA7bP64V2HAptbJ7phuPp7jXWwg/go-ipfs-cmdkit"
 )
@@ -87,14 +85,15 @@ into an object of the specified format.
 			}
 		}
 
-		outChan := make(chan interface{}, 8)
-
-		addAllAndPin := func(f files.File) error {
 		cids := cid.NewSet()
 		b := ipld.NewBatch(req.Context, nd.DAG)
 
+		if dopin {
+			defer nd.Blockstore.PinLock().Unlock()
+		}
+
 		for {
-				file, err := f.NextFile()
+			file, err := req.Files.NextFile()
 			if err == io.EOF {
 				// Finished the list of files.
 				break
@@ -119,11 +118,8 @@ into an object of the specified format.
 
 			cid := nds[0].Cid()
 			cids.Add(cid)
-
+			if err := res.Emit(&OutputObject{Cid: cid}); err != nil {
-				select {
+				return err
-				case outChan <- &OutputObject{Cid: cid}:
-				case <-req.Context.Done():
-					return nil
 			}
 		}
 
@@ -132,8 +128,6 @@ into an object of the specified format.
 		}
 
 		if dopin {
-				defer nd.Blockstore.PinLock().Unlock()
-
 			cids.ForEach(func(c cid.Cid) error {
 				nd.Pinning.PinWithMode(c, pin.Recursive)
 				return nil
@@ -144,24 +138,7 @@ into an object of the specified format.
 				return err
 			}
 		}
-
 		return nil
-		}
-
-		errC := make(chan error)
-		go func() {
-			var err error
-			defer func() { errC <- err }()
-			defer close(outChan)
-			err = addAllAndPin(req.Files)
-		}()
-
-		err = res.Emit(outChan)
-		if err != nil {
-			return err
-		}
-
-		return <-errC
 	},
 	Type: OutputObject{},
 	Encoders: cmds.EncoderMap{
@@ -260,17 +237,3 @@ var DagResolveCmd = &cmds.Command{
 	},
 	Type: ResolveOutput{},
 }
-
-// copy+pasted from ../commands.go
-func unwrapOutput(i interface{}) (interface{}, error) {
-	var (
-		ch <-chan interface{}
-		ok bool
-	)
-
-	if ch, ok = i.(<-chan interface{}); !ok {
-		return nil, e.TypeErr(ch, i)
-	}
-
-	return <-ch, nil
-}