revamp BatchFetch a bit

io/dagreader.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"io"
 
+	"code.google.com/p/go.net/context"
+
 	proto "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/goprotobuf/proto"
 	mdag "github.com/jbenet/go-ipfs/merkledag"
 	ft "github.com/jbenet/go-ipfs/unixfs"
@@ -17,8 +19,8 @@ var ErrIsDir = errors.New("this dag node is a directory")
 type DagReader struct {
 	serv      mdag.DAGService
 	node      *mdag.Node
-	position int
 	buf       io.Reader
+	fetchChan <-chan *mdag.Node
 }
 
 // NewDagReader creates a new reader object that reads the data represented by the given
@@ -39,6 +41,7 @@ func NewDagReader(n *mdag.Node, serv mdag.DAGService) (io.Reader, error) {
 			node:      n,
 			serv:      serv,
 			buf:       bytes.NewBuffer(pb.GetData()),
+			fetchChan: serv.BatchFetch(context.TODO(), n),
 		}, nil
 	case ftpb.Data_Raw:
 		// Raw block will just be a single level, return a byte buffer
@@ -51,19 +54,20 @@ func NewDagReader(n *mdag.Node, serv mdag.DAGService) (io.Reader, error) {
 // precalcNextBuf follows the next link in line and loads it from the DAGService,
 // setting the next buffer to read from
 func (dr *DagReader) precalcNextBuf() error {
-	if dr.position >= len(dr.node.Links) {
+	var nxt *mdag.Node
+	var ok bool
+	select {
+	case nxt, ok = <-dr.fetchChan:
+		if !ok {
 			return io.EOF
 		}
-	nxt, err := dr.node.Links[dr.position].GetNode(dr.serv)
-	if err != nil {
-		return err
 	}
+
 	pb := new(ftpb.Data)
-	err = proto.Unmarshal(nxt.Data, pb)
+	err := proto.Unmarshal(nxt.Data, pb)
 	if err != nil {
 		return err
 	}
-	dr.position++
 
 	switch pb.GetType() {
 	case ftpb.Data_Directory: